Merge pull request #2648 from flow-php/pgsql-query-traversing-bug

fix(flow-php/postgresql): traverse every AST node through protobuf descriptors

Author: norberttech

documentation/components/libs/postgresql.md

@@ -87,6 +87,10 @@ foreach (sql_query_functions($query)->all() as $func) {
 }
 ```
 
+Extractors report every node of their kind anywhere in the statement — DDL targets (`CREATE TABLE x AS …`,
+`CREATE VIEW v …`, `SELECT … INTO t`), `FOR UPDATE OF t`, `excluded.*`, window `ORDER BY` — so filter the result
+when you need only some of them.
+
 ### Parsing Utilities
 
 ```php
@@ -621,17 +625,24 @@ interface NodeModifier
 
 The `ModificationContext` provides:
 
-- `$context->depth` - current traversal depth
-- `$context->ancestors` - array of parent nodes
-- `$context->getParent()` - immediate parent node
+- `$context->depth()` - traversal depth: 1 for a top-level statement, +1 per message edge
+- `$context->ancestors()` - parent messages from the root statement down, without `Node` wrappers
+- `$context->parent()` - immediate parent message
 - `$context->isTopLevel()` - whether this is the top-level statement
 
 Return values:
 
 - `null` - continue traversal
-- `Traverser::DONT_TRAVERSE_CHILDREN` - skip children
-- `Traverser::STOP_TRAVERSAL` - stop entire traversal
-- `object` - replace current node with returned object
+- `NodeModifier::DONT_TRAVERSE_CHILDREN` - skip children
+- `NodeModifier::STOP_TRAVERSAL` - stop entire traversal
+- `NodeModifier::REMOVE_NODE` - remove the node from its list (throws for a single-node slot)
+- `object` - replace the current node
+
+Replacement rules:
+
+- a slot holding a `Node` takes a `Node`; any other slot takes the same class; anything else throws
+- a replacement is not traversed (unlike PHP-Parser) - run a second traversal if needed
+- to clear a single slot, mutate the parent
 
 ### Using Modifiers Directly
 
@@ -668,6 +679,22 @@ echo $query->deparse();
 // SELECT * FROM users WHERE created_at > $1 OR (created_at = $1 AND id > $2) ORDER BY created_at, id LIMIT 10
 ```
 
+`ExplainModifier` wraps every statement before any of them is visited, so run it in its own `traverse()` after the
+other modifiers:
+
+```php
+<?php
+
+use Flow\PostgreSql\AST\Transformers\{ExplainConfig, ExplainModifier, PaginationConfig, PaginationModifier};
+
+use function Flow\PostgreSql\DSL\sql_parse;
+
+$query = sql_parse('SELECT * FROM users ORDER BY id');
+$query->traverse(new PaginationModifier(new PaginationConfig(limit: 10)));
+$query->traverse(new ExplainModifier(ExplainConfig::forEstimate()));
+echo $query->deparse(); // EXPLAIN (COSTS 1, FORMAT "json") SELECT * FROM users ORDER BY id LIMIT 10
+```
+
 Pass `param()` instead of a number to leave the value to the query's parameters - every page then sends the same SQL:
 
 ```php

documentation/components/libs/postgresql/client-explain.md

@@ -29,6 +29,11 @@ echo "Planning time: {$plan->planningTime()}ms\n";
 echo "Total cost: {$plan->rootNode()->cost()->totalCost()}\n";
 ```
 
+With `analyze` (the default), the statement runs inside a transaction - a savepoint when one is already open - that
+is always rolled back, so `INSERT`, `UPDATE`, `DELETE`, `MERGE` and `CREATE TABLE … AS` leave no rows behind (sequences
+still advance). SELECT, INSERT, UPDATE, DELETE, MERGE, CREATE TABLE AS, EXECUTE and DECLARE CURSOR can be explained;
+any other statement, including an `EXPLAIN …` query, throws `InvalidStatementException`.
+
 You can also pass raw SQL strings with parameters:
 
 ```php

documentation/components/libs/postgresql/client-telemetry.md

@@ -111,7 +111,7 @@ All query methods are traced with individual spans:
 - `fetchScalar()`, `fetchScalarInt()`, `fetchScalarString()`, etc.
 - `fetchInto()`, `fetchOneInto()`, `fetchAllInto()` - Object mapping queries
 - `execute()` - INSERT, UPDATE, DELETE operations
-- `explain()` - Query plan analysis
+- `explain()` - Query plan analysis; the BEGIN/SAVEPOINT/ROLLBACK it runs internally for ANALYZE get no span
 
 Span names follow the pattern: `{OPERATION} {table}` (e.g., `SELECT users`)
 

documentation/components/libs/postgresql/select-query-builder.md

@@ -177,7 +177,7 @@ echo $query->toSql();
 <?php
 
 use function Flow\PostgreSql\DSL\{
-    select, star, table, col, asc, desc, order_by
+    select, star, table, col, asc, desc, order_by, collate
 };
 
 use Flow\PostgreSql\QueryBuilder\Clause\{SortDirection, NullsPosition};
@@ -211,6 +211,14 @@ $query = select(star())
 
 echo $query->toSql();
 // SELECT * FROM products ORDER BY price ASC NULLS FIRST, name DESC NULLS LAST
+
+// COLLATE - the collation is written as in SQL: 'C', '"de_DE"', 'pg_catalog."C"'
+$query = select(star())
+    ->from(table('users'))
+    ->orderBy(asc(collate(col('name'), 'C')));
+
+echo $query->toSql();
+// SELECT * FROM users ORDER BY name COLLATE "C" ASC
 ```
 
 ## LIMIT and OFFSET

documentation/upgrading.md

@@ -278,6 +278,77 @@ final class MyExtractor implements Extractor
 |--------------------------------------------------|------------------------------------------------------------------------------------------------------|
 | `new Report(?Schema $schema, Statistics $stats)` | `new Report(?Schema $schema, Statistics $stats, ?array $sources)` - `null` unless analyzed with them |
 
+### 27) `flow-php/postgresql` - `Traverser` visits every node
+
+| Before                                                                                        | After                                                     |
+|-----------------------------------------------------------------------------------------------|-----------------------------------------------------------|
+| `sql_query_tables('CREATE TABLE x AS SELECT * FROM t')` - `[]`                                | `[t, x]`                                                  |
+| `sql_query_tables('SELECT * FROM t JOIN u ON true FOR UPDATE OF t')` - `[t, u]`               | `[t, u, t]` - every reference, filter duplicates yourself |
+| `sql_query_tables('CREATE VIEW v AS SELECT a FROM src')` - `[]`                               | `[v, src]`                                                |
+| `sql_query_tables('SELECT a INTO new_t FROM src')` - `[src]`                                  | `[new_t, src]`                                            |
+| `sql_query_columns('… ON CONFLICT (name) DO UPDATE SET name = excluded.name')` - `[]`         | `[excluded.name]`                                         |
+| `OrderBy` of `SELECT a, row_number() OVER (ORDER BY b) FROM t ORDER BY a` - 1 clause          | 2 clauses - window `ORDER BY` included                    |
+| `sql_query_tables('SELECT (SELECT x FROM a) FROM b')` - `[b, a]`                              | `[a, b]` - descriptor (PostgreSQL walker) order           |
+| `sql_query_depth()`: `EXPLAIN SELECT 1` 0, `CREATE VIEW v AS SELECT 1` 0, window subquery 1   | 1, 1, 2                                                   |
+| `sql_to_keyset_query()` cursor on `… WHERE $1 IN (SELECT …)` - `$1`, clashing with the user's | `$2`                                                      |
+| `TypeCastStripper` left casts under a `SubLink` test, a window and `COLLATE`                  | stripped                                                  |
+
+### 28) `flow-php/postgresql` - traversal contract: messages, depth, replacement, `REMOVE_NODE`
+
+| Before                                                            | After                                                                                    |
+|-------------------------------------------------------------------|------------------------------------------------------------------------------------------|
+| `ModificationContext::ancestors()` / `parent()` - `Node` wrappers | the real messages (`SelectStmt`, `RangeSubselect`, …), no `Node` wrappers                |
+| depth of a CTE body - 3                                           | 4 - `WithClause`, `WindowDef`, `IntoClause`, `OnConflictClause` are levels too           |
+| a replacement returned below the top-level statement - ignored    | written into its slot                                                                    |
+| a replacement of the wrong class - ignored                        | `ParserException`                                                                        |
+| `NodeVisitor::REMOVE_NODE` - declared, never honoured             | removed; `NodeModifier::REMOVE_NODE` removes a node from a list, throws on a single slot |
+
+### 29) `flow-php/postgresql` - `EXPLAIN` wraps every explainable statement, ANALYZE is rolled back
+
+| Before                                                                                                | After                                                                                                            |
+|-------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------|
+| `ExplainModifier::nodeClasses()` - `[SelectStmt::class]`                                              | `[ParseResult::class]`                                                                                           |
+| INSERT/UPDATE/DELETE/MERGE/CTAS/EXECUTE/DECLARE - returned unwrapped                                  | wrapped in `EXPLAIN`                                                                                             |
+| `sql_to_explain('CREATE TABLE x (a int)')`, `sql_to_explain('EXPLAIN SELECT 1')` - returned unwrapped | `InvalidStatementException`                                                                                      |
+| `$client->explain('INSERT …')` - the `INSERT` ran and committed                                       | EXPLAIN; with ANALYZE inside a transaction (savepoint when one is open) that is always rolled back               |
+| `traverse(new PaginationModifier(…), new ExplainModifier(…))` - both applied                          | the pagination is dropped (or the traversal throws) - call `traverse()` again with `ExplainModifier` alone, last |
+
+### 30) `flow-php/postgresql` - keyset pagination wraps `UNION` / `INTERSECT` / `EXCEPT`
+
+| Before                                                                                                                      | After                                                              |
+|-----------------------------------------------------------------------------------------------------------------------------|--------------------------------------------------------------------|
+| `sql_to_keyset_query('SELECT id FROM t UNION SELECT id FROM u ORDER BY id', …)` - cursor dropped, page 2 fails with `08P01` | `SELECT * FROM (…) _keyset_subq WHERE id > $1 ORDER BY id LIMIT …` |
+| qualified keyset column (`t.id`) on a set operation - `42P01` at run time                                                   | `PaginationException`                                              |
+
+### 31) `flow-php/postgresql` - schema keeps the declared expression text, compares normalised keys
+
+| Before                                                           | After                                                                              |
+|------------------------------------------------------------------|------------------------------------------------------------------------------------|
+| `Column::$generationExpression` - normalised (`lower(i)`)        | declared/catalog text (`lower(i::text)`); compare with `generationExpressionKey()` |
+| `CheckConstraint::$expression` - normalised                      | declared/catalog text; `expressionKey()`                                           |
+| `Index::$predicate` - normalised                                 | declared/catalog text; `predicateKey()`                                            |
+| `Trigger::$whenCondition` - compared verbatim                    | declared/catalog text; compared by `whenConditionKey()`                            |
+| `ColumnDefault` EXPRESSION `literal` - normalised                | declared/catalog text; `equals()` normalises                                       |
+| `CheckDefinitionParser::parse()` - normalised expression         | the catalog text, still validated                                                  |
+| `ExpressionParser::normalizeNode(Node)`                          | removed - `deparseNode(Node)` keeps the text, `normalize(string)` strips casts     |
+| DDL emitted the normalised text - `lower(i)` failed with `42883` | DDL emits the declared expression                                                  |
+
+### 32) `flow-php/postgresql` - schema DDL emits index `WHERE` and trigger `WHEN`
+
+| Before                                                         | After                                                                         |
+|----------------------------------------------------------------|-------------------------------------------------------------------------------|
+| `CREATE UNIQUE INDEX t_email_live ON s.t (email)`              | `CREATE UNIQUE INDEX t_email_live ON s.t (email) WHERE deleted_at IS NULL`    |
+| `CREATE TRIGGER t_trg … FOR EACH ROW EXECUTE FUNCTION f()`     | `CREATE TRIGGER t_trg … FOR EACH ROW WHEN (new.i > 0) EXECUTE FUNCTION s.f()` |
+| an unqualified trigger function resolved through `search_path` | resolved to the table's schema                                                |
+| introspected `Trigger::$functionName` - `name`                 | `schema.name`; new `Trigger::withFunctionSchema()`                            |
+| a declared `'s.f'` always drifted against the catalog          | no drift                                                                      |
+
+### 33) `flow-php/postgresql` - a failed `SAVEPOINT` leaves the outer transaction open
+
+| Before                                                                                      | After                                                                    |
+|---------------------------------------------------------------------------------------------|--------------------------------------------------------------------------|
+| nesting level reset to 0, the caller's `rollBack()` throws, the connection stays in `25P02` | nesting level kept; the caller's `rollBack()` ends the outer transaction |
+
 ---
 
 ## Upgrading from 0.43.x to 0.44.x

src/adapter/etl-adapter-postgresql/tests/Flow/ETL/Adapter/PostgreSql/Tests/Integration/PostgreSqlKeySetExtractorIntegrationTest.php

@@ -17,6 +17,7 @@
 use function Flow\ETL\DSL\from_array;
 use function Flow\ETL\DSL\int_schema;
 use function Flow\ETL\DSL\schema;
+use function Flow\PostgreSql\DSL\asc;
 use function Flow\PostgreSql\DSL\col;
 use function Flow\PostgreSql\DSL\column;
 use function Flow\PostgreSql\DSL\column_type_integer;
@@ -134,6 +135,49 @@ public function test_extracts_all_rows_with_keyset_pagination(): void
         static::assertSame(range(1, 25), array_column($rows, 'id'));
     }
 
+    public function test_extracts_all_rows_of_a_union(): void
+    {
+        $odd = $this->tableName . '_odd';
+        $even = $this->tableName . '_even';
+
+        $this->client->execute(create()->table($odd)->column(column('id', column_type_integer())->primaryKey()));
+        $this->client->execute(create()->table($even)->column(column('id', column_type_integer())->primaryKey()));
+        $this->client->execute(
+            insert()
+                ->into($odd)
+                ->columns('id')
+                ->values(literal(1))
+                ->values(literal(3))
+                ->values(literal(5))
+                ->values(literal(7))
+                ->values(literal(9)),
+        );
+        $this->client->execute(
+            insert()
+                ->into($even)
+                ->columns('id')
+                ->values(literal(2))
+                ->values(literal(4))
+                ->values(literal(6))
+                ->values(literal(8))
+                ->values(literal(10)),
+        );
+
+        $rows = df()
+            ->read(from_pgsql_key_set(
+                $this->client,
+                select(col('id'))
+                    ->from(table($odd))
+                    ->union(select(col('id'))->from(table($even)))
+                    ->orderBy(asc(col('id'))),
+                pgsql_pagination_key_set(pgsql_pagination_key_asc('id')),
+            )->withBatchSize(3))
+            ->fetch()
+            ->toArray();
+
+        static::assertSame(range(1, 10), array_column($rows, 'id'));
+    }
+
     public function test_extracts_limited_rows_with_maximum(): void
     {
         $rows = df()

src/lib/postgresql/src/Flow/PostgreSql/AST/ModificationContext.php

@@ -5,6 +5,7 @@
 namespace Flow\PostgreSql\AST;
 
 use Flow\PostgreSql\Protobuf\AST\ParseResult;
+use Google\Protobuf\Internal\Message;
 
 use function count;
 
@@ -17,8 +18,8 @@
 final readonly class ModificationContext
 {
     /**
-     * @param array<object> $ancestors Stack of parent nodes (from root to immediate parent)
-     * @param int $depth Current depth in the AST (1-based, root statements are at depth 1)
+     * @param list<Message> $ancestors Messages from the root statement down to the parent, excluding Node wrappers
+     * @param int $depth Current depth in the AST: root statements are at 1, +1 per message edge
      * @param ParseResult $parseResult The full parsed AST for context-aware operations
      */
     public function __construct(
@@ -28,7 +29,7 @@ public function __construct(
     ) {}
 
     /**
-     * @return array<object>
+     * @return list<Message>
      */
     public function ancestors(): array
     {
@@ -45,7 +46,7 @@ public function isTopLevel(): bool
         return $this->depth === 1;
     }
 
-    public function parent(): ?object
+    public function parent(): ?Message
     {
         return $this->ancestors[count($this->ancestors) - 1] ?? null;
     }

src/lib/postgresql/src/Flow/PostgreSql/AST/NodeModifier.php

@@ -24,6 +24,11 @@ interface NodeModifier
      */
     public const STOP_TRAVERSAL = 2;
 
+    /**
+     * Remove the current node from its list slot.
+     */
+    public const REMOVE_NODE = 3;
+
     /**
      * Returns the fully qualified class names of the node types this modifier handles.
      *
@@ -41,14 +46,15 @@ public static function nodeClasses(): array;
      * - Mutate the node in-place and return null to continue traversal
      * - Return DONT_TRAVERSE_CHILDREN to skip child nodes
      * - Return STOP_TRAVERSAL to stop the entire traversal
-     * - Return a new node to replace the current node (used for wrapping operations)
+     * - Return REMOVE_NODE to remove the node from a list slot; on a single-node slot the traverser throws
+     * - Return a new node to replace the current node; it is written into the slot and not descended into
      *
      * @param object $node The node instance to modify (one of the types listed in nodeClasses())
      * @param ModificationContext $context Context providing parent information
      *
      * @return null|int|object
      *                         - null: Continue traversal (node unchanged or modified in-place)
-     *                         - int (DONT_TRAVERSE_CHILDREN, STOP_TRAVERSAL): Control flow
+     *                         - int (DONT_TRAVERSE_CHILDREN, STOP_TRAVERSAL, REMOVE_NODE): Control flow
      *                         - object: Replace current node with returned node
      */
     public function modify(object $node, ModificationContext $context): int|object|null;

src/lib/postgresql/src/Flow/PostgreSql/AST/NodeVisitor.php

@@ -19,11 +19,6 @@ interface NodeVisitor
      */
     public const DONT_TRAVERSE_CHILDREN = 1;
 
-    /**
-     * Remove the node from its parent array.
-     */
-    public const REMOVE_NODE = 3;
-
     /**
      * Stop the entire traversal.
      */
@@ -58,7 +53,6 @@ public function enter(object $node): ?int;
      *
      * @return null|int Return value determines traversal behavior:
      *                  - null: Continue traversal
-     *                  - REMOVE_NODE: Remove node from parent
      *                  - STOP_TRAVERSAL: Stop entire traversal
      */
     public function leave(object $node): ?int;