Skip to content

Commit 2479a0b

Browse files
trowskitrowski
committed
Query defined types for casting with pgsql
1 parent 02d367f commit 2479a0b

10 files changed

+188
-262
lines changed

‎src/Internal/ArrayParser.php

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,18 +4,21 @@
44

55
use Amp\Postgres\ParseException;
66

7+
/**
8+
* @internal
9+
*/
710
final class ArrayParser
811
{
912
/**
1013
* @param string $data String representation of PostgreSQL array.
11-
* @param callable|null $cast Callback to cast parsed values.
14+
* @param callable $cast Callback to cast parsed values.
1215
* @param string $delimiter Delimiter used to separate values.
1316
*
1417
* @return array Parsed column data.
1518
*
1619
* @throws ParseException
1720
*/
18-
public function parse(string $data, callable $cast = null, string $delimiter = ','): array
21+
public function parse(string $data, callable $cast, string $delimiter = ','): array
1922
{
2023
$data = \trim($data);
2124

@@ -33,14 +36,14 @@ public function parse(string $data, callable $cast = null, string $delimiter = '
3336
* Recursive generator parser yielding array values.
3437
*
3538
* @param string $data Remaining buffer data.
36-
* @param callable|null $cast Callback to cast parsed values.
39+
* @param callable $cast Callback to cast parsed values.
3740
* @param string $delimiter Delimiter used to separate values.
3841
*
3942
* @return \Generator
4043
*
4144
* @throws ParseException
4245
*/
43-
private function parser(string $data, callable $cast = null, string $delimiter = ','): \Generator
46+
private function parser(string $data, callable $cast, string $delimiter = ','): \Generator
4447
{
4548
if ($data === '') {
4649
throw new ParseException("Unexpected end of data");
@@ -104,7 +107,7 @@ private function parser(string $data, callable $cast = null, string $delimiter =
104107
}
105108
}
106109

107-
yield $cast ? $cast($yield) : $yield;
110+
yield $cast($yield);
108111
} while ($end !== '}');
109112

110113
return $data;

‎src/PgSqlConnection.php

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -40,8 +40,9 @@ public static function connect(ConnectionConfig $connectionConfig, ?Cancellation
4040
}
4141

4242
$deferred = new Deferred;
43+
$id = \sha1($connectionConfig->getHost() . $connectionConfig->getPort() . $connectionConfig->getUser());
4344

44-
$callback = function ($watcher, $resource) use ($connection, $deferred): void {
45+
$callback = function ($watcher, $resource) use ($connection, $deferred, $id): void {
4546
switch (\pg_connect_poll($connection)) {
4647
case \PGSQL_POLLING_READING: // Connection not ready, poll again.
4748
case \PGSQL_POLLING_WRITING: // Still writing...
@@ -52,7 +53,7 @@ public static function connect(ConnectionConfig $connectionConfig, ?Cancellation
5253
return;
5354

5455
case \PGSQL_POLLING_OK:
55-
$deferred->resolve(new self($connection, $resource));
56+
$deferred->resolve(new self($connection, $resource, $id));
5657
return;
5758
}
5859
};
@@ -81,9 +82,10 @@ public static function connect(ConnectionConfig $connectionConfig, ?Cancellation
8182
/**
8283
* @param resource $handle PostgreSQL connection handle.
8384
* @param resource $socket PostgreSQL connection stream socket.
85+
* @param string $id Connection identifier for determining which cached type table to use.
8486
*/
85-
public function __construct($handle, $socket)
87+
public function __construct($handle, $socket, string $id = '')
8688
{
87-
parent::__construct(new PgSqlHandle($handle, $socket));
89+
parent::__construct(new PgSqlHandle($handle, $socket, $id));
8890
}
8991
}

‎src/PgSqlHandle.php

Lines changed: 48 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -30,10 +30,16 @@ final class PgSqlHandle implements Handle
3030
\PGSQL_DIAG_SOURCE_FUNCTION => "source_function",
3131
];
3232

33+
/** @var array<string, Promise<array<int, array{string, string}>>> */
34+
private static $typeCache;
35+
3336
/** @var resource PostgreSQL connection handle. */
3437
private $handle;
3538

36-
/** @var \Amp\Deferred|null */
39+
/** @var Promise<array<int, array{string, string}>> */
40+
private $types;
41+
42+
/** @var Deferred|null */
3743
private $deferred;
3844

3945
/** @var string */
@@ -42,7 +48,7 @@ final class PgSqlHandle implements Handle
4248
/** @var string */
4349
private $await;
4450

45-
/** @var \Amp\Emitter[] */
51+
/** @var Emitter[] */
4652
private $listeners = [];
4753

4854
/** @var Struct[] */
@@ -52,12 +58,11 @@ final class PgSqlHandle implements Handle
5258
private $lastUsedAt;
5359

5460
/**
55-
* Connection constructor.
56-
*
5761
* @param resource $handle PostgreSQL connection handle.
5862
* @param resource $socket PostgreSQL connection stream socket.
63+
* @param string $id Connection identifier for determining which cached type table to use.
5964
*/
60-
public function __construct($handle, $socket)
65+
public function __construct($handle, $socket, string $id = '')
6166
{
6267
$this->handle = $handle;
6368

@@ -147,9 +152,10 @@ public function __construct($handle, $socket)
147152
}
148153
});
149154

150-
//Loop::disable($this->poll);
151155
Loop::unreference($this->poll);
152156
Loop::disable($this->await);
157+
158+
$this->types = $this->fetchTypes($id);
153159
}
154160

155161
/**
@@ -160,6 +166,29 @@ public function __destruct()
160166
$this->close();
161167
}
162168

169+
private function fetchTypes(string $id): Promise
170+
{
171+
if (isset(self::$typeCache)) {
172+
return self::$typeCache[$id];
173+
}
174+
175+
return self::$typeCache[$id] = call(function (): \Generator {
176+
$result = yield from $this->send(
177+
"pg_send_query",
178+
"SELECT t.oid, t.typcategory, t.typdelim, t.typelem
179+
FROM pg_catalog.pg_type t JOIN pg_catalog.pg_namespace n ON t.typnamespace=n.oid
180+
WHERE t.typisdefined AND n.nspname IN ('pg_catalog', 'public')"
181+
);
182+
183+
$types = [];
184+
while ($row = \pg_fetch_array($result, null, \PGSQL_NUM)) {
185+
[$oid, $type, $delimiter, $element] = $row;
186+
$types[(int) $oid] = [$type, $delimiter, (int) $element];
187+
}
188+
return $types;
189+
});
190+
}
191+
163192
/**
164193
* {@inheritdoc}
165194
*/
@@ -257,7 +286,7 @@ private function send(callable $function, ...$args): \Generator
257286
* @throws FailureException
258287
* @throws QueryError
259288
*/
260-
private function createResult($result, string $sql)
289+
private function createResult($result, string $sql, array $types)
261290
{
262291
switch (\pg_result_status($result, \PGSQL_STATUS_LONG)) {
263292
case \PGSQL_EMPTY_QUERY:
@@ -267,7 +296,7 @@ private function createResult($result, string $sql)
267296
return new PgSqlCommandResult($result);
268297

269298
case \PGSQL_TUPLES_OK:
270-
return new PgSqlResultSet($result);
299+
return new PgSqlResultSet($result, $types);
271300

272301
case \PGSQL_NONFATAL_ERROR:
273302
case \PGSQL_FATAL_ERROR:
@@ -301,7 +330,11 @@ public function statementExecute(string $name, array $params): Promise
301330
{
302331
return call(function () use ($name, $params) {
303332
\assert(isset($this->statements[$name]), "Named statement not found when executing");
304-
return $this->createResult(yield from $this->send("pg_send_execute", $name, $params), $this->statements[$name]->sql);
333+
return $this->createResult(
334+
yield from $this->send("pg_send_execute", $name, $params),
335+
$this->statements[$name]->sql,
336+
yield $this->types
337+
);
305338
});
306339
}
307340

@@ -341,7 +374,7 @@ public function query(string $sql): Promise
341374
}
342375

343376
return call(function () use ($sql) {
344-
return $this->createResult(yield from $this->send("pg_send_query", $sql), $sql);
377+
return $this->createResult(yield from $this->send("pg_send_query", $sql), $sql, yield $this->types);
345378
});
346379
}
347380

@@ -358,7 +391,11 @@ public function execute(string $sql, array $params = []): Promise
358391
$params = Internal\replaceNamedParams($params, $names);
359392

360393
return call(function () use ($sql, $params) {
361-
return $this->createResult(yield from $this->send("pg_send_query_params", $sql, $params), $sql);
394+
return $this->createResult(
395+
yield from $this->send("pg_send_query_params", $sql, $params),
396+
$sql,
397+
yield $this->types
398+
);
362399
});
363400
}
364401

‎src/PgSqlResultSet.php

Lines changed: 33 additions & 102 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,9 @@ final class PgSqlResultSet implements ResultSet
1111
/** @var resource PostgreSQL result resource. */
1212
private $handle;
1313

14+
/** @var array<int, array{string, string}> */
15+
private $types;
16+
1417
/** @var int */
1518
private $position = 0;
1619

@@ -28,10 +31,12 @@ final class PgSqlResultSet implements ResultSet
2831

2932
/**
3033
* @param resource $handle PostgreSQL result resource.
34+
* @param array<int, array{string, string}> $types
3135
*/
32-
public function __construct($handle)
36+
public function __construct($handle, array $types = [])
3337
{
3438
$this->handle = $handle;
39+
$this->types = $types;
3540

3641
$numFields = \pg_num_fields($this->handle);
3742
for ($i = 0; $i < $numFields; ++$i) {
@@ -91,123 +96,49 @@ public function getCurrent(): array
9196
continue;
9297
}
9398

94-
$result[$column] = $this->cast($column, $result[$column]);
99+
$result[$column] = $this->cast($this->fieldTypes[$column], $result[$column]);
95100
}
96101

97102
return $this->currentRow = \array_combine($this->fieldNames, $result);
98103
}
99104

100105
/**
101-
* @see https://github.com/postgres/postgres/blob/REL_10_STABLE/src/include/catalog/pg_type.h for OID types.
106+
* @see https://github.com/postgres/postgres/blob/REL_14_STABLE/src/include/catalog/pg_type.dat for OID types.
107+
* @see https://www.postgresql.org/docs/14/catalog-pg-type.html for pg_type catalog docs.
102108
*
103-
* @param int $column
109+
* @param int $oid
104110
* @param string $value
105111
*
106112
* @return array|bool|float|int Cast value.
107113
*
108114
* @throws ParseException
109115
*/
110-
private function cast(int $column, string $value)
116+
private function cast(int $oid, string $value)
111117
{
112-
switch ($this->fieldTypes[$column]) {
113-
case 16: // bool
118+
[$type, $delimiter, $element] = $this->types[$oid] ?? ['S', ',', 0];
119+
120+
switch ($type) {
121+
case 'A': // Arrays
122+
return $this->parser->parse($value, function (string $data) use ($element) {
123+
return $this->cast($element, $data);
124+
}, $delimiter);
125+
126+
case 'B': // Binary
114127
return $value === 't';
115128

116-
case 20: // int8
117-
case 21: // int2
118-
case 23: // int4
119-
case 26: // oid
120-
case 27: // tid
121-
case 28: // xid
122-
return (int) $value;
123-
124-
case 700: // real
125-
case 701: // double-precision
126-
return (float) $value;
127-
128-
case 1000: // boolean[]
129-
return $this->parser->parse($value, function (string $value): bool {
130-
return $value === 't';
131-
});
132-
133-
case 1005: // int2[]
134-
case 1007: // int4[]
135-
case 1010: // tid[]
136-
case 1011: // xid[]
137-
case 1016: // int8[]
138-
case 1028: // oid[]
139-
return $this->parser->parse($value, function (string $value): int {
140-
return (int) $value;
141-
});
142-
143-
case 1021: // real[]
144-
case 1022: // double-precision[]
145-
return $this->parser->parse($value, function (string $value): float {
146-
return (float) $value;
147-
});
148-
149-
case 1020: // box[] (semi-colon delimited)
150-
return $this->parser->parse($value, null, ';');
151-
152-
case 199: // json[]
153-
case 629: // line[]
154-
case 651: // cidr[]
155-
case 719: // circle[]
156-
case 775: // macaddr8[]
157-
case 791: // money[]
158-
case 1001: // bytea[]
159-
case 1002: // char[]
160-
case 1003: // name[]
161-
case 1006: // int2vector[]
162-
case 1008: // regproc[]
163-
case 1009: // text[]
164-
case 1013: // oidvector[]
165-
case 1014: // bpchar[]
166-
case 1015: // varchar[]
167-
case 1019: // path[]
168-
case 1023: // abstime[]
169-
case 1024: // realtime[]
170-
case 1025: // tinterval[]
171-
case 1027: // polygon[]
172-
case 1034: // aclitem[]
173-
case 1040: // macaddr[]
174-
case 1041: // inet[]
175-
case 1115: // timestamp[]
176-
case 1182: // date[]
177-
case 1183: // time[]
178-
case 1185: // timestampz[]
179-
case 1187: // interval[]
180-
case 1231: // numeric[]
181-
case 1263: // cstring[]
182-
case 1270: // timetz[]
183-
case 1561: // bit[]
184-
case 1563: // varbit[]
185-
case 2201: // refcursor[]
186-
case 2207: // regprocedure[]
187-
case 2208: // regoper[]
188-
case 2209: // regoperator[]
189-
case 2210: // regclass[]
190-
case 2211: // regtype[]
191-
case 2949: // txid_snapshot[]
192-
case 2951: // uuid[]
193-
case 3221: // pg_lsn[]
194-
case 3643: // tsvector[]
195-
case 3644: // gtsvector[]
196-
case 3645: // tsquery[]
197-
case 3735: // regconfig[]
198-
case 3770: // regdictionary[]
199-
case 3807: // jsonb[]
200-
case 3905: // int4range[]
201-
case 3907: // numrange[]
202-
case 3909: // tsrange[]
203-
case 3911: // tstzrange[]
204-
case 3913: // daterange[]
205-
case 3927: // int8range[]
206-
case 4090: // regnamespace[]
207-
case 4097: // regrole[]
208-
return $this->parser->parse($value);
209-
210-
default:
129+
case 'N': // Numeric
130+
switch ($oid) {
131+
case 700: // float4
132+
case 701: // float8
133+
case 790: // money
134+
case 1700: // numeric
135+
return (float) $value;
136+
137+
default: // Cast all other numeric types to an integer.
138+
return (int) $value;
139+
}
140+
141+
default: // Return all other types as strings.
211142
return $value;
212143
}
213144
}

0 commit comments

Comments
 (0)