|
19 | 19 | use Cycle\Database\Driver\Postgres\Query\PostgresInsertQuery; |
20 | 20 | use Cycle\Database\Driver\Postgres\Query\PostgresSelectQuery; |
21 | 21 | use Cycle\Database\Driver\Postgres\Query\PostgresUpdateQuery; |
| 22 | +use Cycle\Database\Driver\CursorableInterface; |
| 23 | +use Cycle\Database\Driver\CursorOptions; |
22 | 24 | use Cycle\Database\Exception\DriverException; |
23 | 25 | use Cycle\Database\Exception\StatementException; |
24 | 26 | use Cycle\Database\Query\QueryBuilder; |
| 27 | +use Cycle\Database\StatementInterface; |
25 | 28 |
|
26 | 29 | /** |
27 | 30 | * Talks to postgres databases. |
28 | 31 | */ |
29 | | -class PostgresDriver extends Driver |
| 32 | +class PostgresDriver extends Driver implements CursorableInterface |
30 | 33 | { |
31 | 34 | /** |
32 | 35 | * Cached list of primary keys associated with their table names. Used by InsertBuilder to |
@@ -196,6 +199,74 @@ public function beginTransaction(?string $isolationLevel = null): bool |
196 | 199 | return true; |
197 | 200 | } |
198 | 201 |
|
| 202 | + /** |
| 203 | + * Open a Postgres server-side cursor for the given SELECT and yield rows |
| 204 | + * lazily. Provides snapshot consistency within the enclosing transaction. |
| 205 | + * |
| 206 | + * Requires an active transaction (cursor lifetime is bound to it unless |
| 207 | + * {@see PostgresCursorOptions::$withHold} is set, in which case the cursor |
| 208 | + * survives `COMMIT` and the result is materialized on the server). The |
| 209 | + * cursor is declared with NO SCROLL — only forward fetches are performed. |
| 210 | + * The cursor is closed when the generator is fully consumed or |
| 211 | + * garbage-collected. |
| 212 | + * |
| 213 | + * @return \Generator<int, array<array-key, mixed>> |
| 214 | + * |
| 215 | + * @throws DriverException |
| 216 | + */ |
| 217 | + public function cursor( |
| 218 | + string $statement, |
| 219 | + iterable $parameters = [], |
| 220 | + CursorOptions $options = new CursorOptions(), |
| 221 | + int $mode = StatementInterface::FETCH_ASSOC, |
| 222 | + ): \Generator { |
| 223 | + $opts = PostgresCursorOptions::from($options); |
| 224 | + |
| 225 | + if ($opts->chunkSize < 1) { |
| 226 | + throw new DriverException('Chunk size must be a positive integer.'); |
| 227 | + } |
| 228 | + |
| 229 | + if ($this->getTransactionLevel() === 0) { |
| 230 | + throw new DriverException( |
| 231 | + 'Postgres server-side cursor requires an active transaction. ' |
| 232 | + . 'Wrap the cursor iteration in Database::transaction() or call beginTransaction() before cursor().', |
| 233 | + ); |
| 234 | + } |
| 235 | + |
| 236 | + $cursorName = '"c_' . \bin2hex(\random_bytes(8)) . '"'; |
| 237 | + $holdClause = $opts->withHold ? ' WITH HOLD' : ''; |
| 238 | + $declareSql = "DECLARE {$cursorName} NO SCROLL CURSOR{$holdClause} FOR {$statement}"; |
| 239 | + $fetchSql = "FETCH FORWARD {$opts->chunkSize} FROM {$cursorName}"; |
| 240 | + $closeSql = "CLOSE {$cursorName}"; |
| 241 | + |
| 242 | + try { |
| 243 | + $this->statement($declareSql, $parameters); |
| 244 | + |
| 245 | + do { |
| 246 | + $chunkStatement = $this->statement($fetchSql); |
| 247 | + $rows = $chunkStatement->fetchAll($mode); |
| 248 | + $chunkStatement->close(); |
| 249 | + |
| 250 | + foreach ($rows as $row) { |
| 251 | + yield $row; |
| 252 | + } |
| 253 | + } while (\count($rows) === $opts->chunkSize); |
| 254 | + } finally { |
| 255 | + try { |
| 256 | + $this->statement($closeSql); |
| 257 | + } catch (\Throwable) { |
| 258 | + // Cursor may already be gone (e.g. transaction was rolled back) — swallow. |
| 259 | + } |
| 260 | + |
| 261 | + // Avoid polluting the prepared-statement cache with single-use SQL strings. |
| 262 | + unset( |
| 263 | + $this->queryCache[$declareSql], |
| 264 | + $this->queryCache[$fetchSql], |
| 265 | + $this->queryCache[$closeSql], |
| 266 | + ); |
| 267 | + } |
| 268 | + } |
| 269 | + |
199 | 270 | /** |
200 | 271 | * Parse the table name and extract the schema and table. |
201 | 272 | * |
|
0 commit comments