Skip to content

fix: failed reads keep the database error, reads that write are refused #2634

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
norberttech merged 1 commit into from
Sep 15, 2026
Merged

fix: failed reads keep the database error, reads that write are refused #2634

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions documentation/components/adapters/doctrine.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,9 @@ to_dbal_transaction($connection, to_dbal_table_insert($connection, 'users'))

This simple but powerful extractor let you extract data from a single or multiple parametrized queries.

`from_dbal_*()` extractors derive their schema from the query unless `->withSchema()` declares it - how that works
and fails: [Doctrine DBAL sources](../core/schema.md#doctrine-dbal-sources-describe-themselves-and-keep-withschema).

### Single Query
```php
use function Flow\ETL\DSL\{data_frame, to_output};
Expand Down
7 changes: 6 additions & 1 deletion documentation/components/adapters/postgresql.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,10 @@ All extractors support:
- Maximum row limits
- Custom schema definitions

Every extractor reads exactly one read-only `SELECT` or `VALUES` statement - anything else throws
`InvalidArgumentException` before a query runs. Schema derivation and its errors:
[PostgreSQL sources](../core/schema.md#postgresql-sources-describe-themselves-and-keep-withschema).

### Loader

A flexible loader supporting:
Expand All @@ -60,7 +64,8 @@ This is the **only way** to achieve true low-memory streaming with PHP's ext-pgs
has no unbuffered query mode.

> **Note:** This extractor automatically manages transactions. Cursors require a transaction context,
> which is auto-started if not already in one.
> which is auto-started if not already in one. A failed read rolls back the transaction the extractor started;
> inside a transaction you opened, the rollback is yours.

### Basic Usage

Expand Down
44 changes: 23 additions & 21 deletions documentation/components/core/schema.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,8 @@ carries its own schema.
**`DataFrame::schema()` describes the plan, not the data.** It walks the pipeline's steps once, threading each
step's output schema into the next, starting from the source's own `schema()` - so it may do the I/O that source
needs to describe itself (a CSV sniff, a Parquet footer), but it never reads a row. A column a step cannot name
before rows flow is a build error, raised there rather than mid-run: `select('nope')` and `ref('a')->greaterThan(ref('b'))`
before rows flow is a build error, raised there rather than mid-run: `select('nope')` and
`ref('a')->greaterThan(ref('b'))`
over incomparable types both refuse at the first trigger. `printSchema()` formats that same answer and runs
nothing. A plan containing `joinEach()` cannot be described this way and refuses.

Expand Down Expand Up @@ -119,10 +120,10 @@ below).
A column holds exactly one type. When two rows disagree, the schema merge widens them to the narrowest type that can
hold both, and the result is fixed:

| both sides are | result |
|---|---|
| containers (`json`, `list`, `map`, `structure`) | `json` |
| anything else | `string` |
| both sides are | result |
|-------------------------------------------------|----------|
| containers (`json`, `list`, `map`, `structure`) | `json` |
| anything else | `string` |

```php
data_frame()
Expand Down Expand Up @@ -245,15 +246,17 @@ How the derivation behaves:
expression column can always produce `null`.
- The schema is derived **once per read**, before the first row, and every batch conforms to it.

A query that cannot be described **does not read at all**; it throws `SchemaNotDerivableException` with the reason.
That happens for exactly two things:
A query that cannot be read or described **does not read at all**:

1. a query shape that cannot be wrapped in a zero-row `SELECT` - multi-statement, a data-modifying CTE, or
`INSERT ... RETURNING`;
2. a column whose PostgreSQL type Flow has no type for - after type mapping that is only `record` and the geometric
types (`point`, `line`, `lseg`, `box`, `path`, `polygon`, `circle`).
| What is wrong | What you get |
|------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------------|
| the SQL does not parse | `ParserException`, before any query runs |
| not exactly one read-only `SELECT` or `VALUES` (`INSERT ... RETURNING`, two statements, a data-modifying `WITH`, `SELECT ... INTO`) | `InvalidArgumentException`, before any query runs |
| a table, column, function, type or privilege the query names is missing | PostgreSQL's own `QueryException`, positioned in your SQL - what the read throws |
| any other refusal of the zero-row probe | `SchemaNotDerivableException`, `getPrevious()` is PostgreSQL's error |
| a column type Flow has no type for (`record`, `point`, `line`, `lseg`, `box`, `path`, `polygon`, `circle`) | `SchemaNotDerivableException` |

In both cases `->withSchema(...)` is the escape hatch, and it skips the probe.
`->withSchema(...)` skips the probe; it helps the last row, and the one before it when the query runs as written.

### Doctrine DBAL sources describe themselves, and keep `withSchema()`

Expand Down Expand Up @@ -304,17 +307,16 @@ What each driver answers:
makes the driver rename the second to `id:1`, so the schema carries a column the rows do not; alias the columns
apart, or declare the schema, when a SQLite query selects a name twice.

A query that cannot be described **does not read at all**; it throws `SchemaNotDerivableException` with the reason.
That happens for exactly three things:
A query that cannot be described **does not read at all**:

1. a query shape that cannot be wrapped in a zero-row `SELECT` - multi-statement, a data-modifying CTE, or
`INSERT ... RETURNING`;
2. a driver this adapter has no result-type probe for - `pdo_pgsql` and `pdo_mysql` hand out a plain `PDO`, which
has no arm yet, so they are refused by name; use the native `pgsql://` or `mysqli://` DSN, or `withSchema()`;
3. a column whose driver type Flow has no type for - on PostgreSQL that includes arrays, `money`, `inet`, ranges and
`interval` on this route (the PostgreSQL adapter above maps more of them); on MySQL `BIT` and `GEOMETRY`.
| What is wrong | What you get |
|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------|
| a table or column the query names is missing | DBAL's `TableNotFoundException` / `InvalidFieldNameException` (SQLite: `DriverException`) - what the read throws |
| any other refusal (a syntax error, a multi-statement query on SQLite or PostgreSQL) | `SchemaNotDerivableException`, `getPrevious()` is the DBAL exception |
| a driver without a result-type probe - `pdo_pgsql` and `pdo_mysql` hand out a plain `PDO` | `SchemaNotDerivableException` - use the native `pgsql://` or `mysqli://` DSN |
| a column whose driver type Flow has no type for - on PostgreSQL arrays, `money`, `inet`, ranges and `interval` on this route (the PostgreSQL adapter above maps more of them); on MySQL `BIT` and `GEOMETRY` | `SchemaNotDerivableException` |

In all three cases `->withSchema(...)` is the escape hatch, and it skips the probe.
`->withSchema(...)` skips the probe; it helps the last two rows, and the second one when the query runs as written.

## Inferring Types from an Array

Expand Down
18 changes: 18 additions & 0 deletions documentation/components/libs/postgresql.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -404,6 +404,10 @@ The library ships two default mappers, both available via DSL functions, plus an

Modify existing SQL queries programmatically - useful for adding pagination to queries.

The pagination helpers and modifiers take exactly one read-only `SELECT` or `VALUES` statement. Anything else -
`INSERT ... RETURNING`, two statements, a data-modifying `WITH`, `SELECT ... INTO` - throws
`InvalidStatementException`.

### Offset Pagination

Add LIMIT/OFFSET pagination to any SELECT query:
Expand Down Expand Up @@ -664,6 +668,20 @@ echo $query->deparse();
// SELECT * FROM users WHERE created_at > $1 OR (created_at = $1 AND id > $2) ORDER BY created_at, 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
<?php

use Flow\PostgreSql\AST\Transformers\{PaginationConfig, PaginationModifier};

use function Flow\PostgreSql\DSL\{param, sql_parse};

$query = sql_parse('SELECT * FROM users WHERE active = $1 ORDER BY id');
$query->traverse(new PaginationModifier(new PaginationConfig(limit: param(2), offset: param(3))));
echo $query->deparse(); // SELECT * FROM users WHERE active = $1 ORDER BY id LIMIT $2 OFFSET $3
```

### Raw AST Access

For full control, access the protobuf AST directly:
Expand Down
61 changes: 57 additions & 4 deletions documentation/upgrading.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1258,8 +1258,10 @@ Keep every column a string: `from_csv($path)->inferSchema(infer_schema()->allStr
| `pdo_pgsql`, `pdo_mysql`, any other | every column | driver values | throws `SchemaNotDerivableException` - use the `pgsql` / `mysqli` driver or `->withSchema(...)` |

Applies to `from_dbal_query()`, `from_dbal_queries()`, `from_dbal_limit_offset()`, `from_dbal_limit_offset_qb()` and
`from_dbal_key_set_qb()` without `->withSchema()`. A query the driver cannot describe (multi-statement, data-modifying
CTE, `INSERT ... RETURNING`) also throws. Declare `->withSchema(...)` to pick the types.
`from_dbal_key_set_qb()` without `->withSchema()`. A missing table or column throws what the read throws
(`TableNotFoundException`, `InvalidFieldNameException`, on SQLite `DriverException`); any other query the driver
refuses to describe throws
`SchemaNotDerivableException` with the DBAL exception as `getPrevious()`. Declare `->withSchema(...)` to pick the types.

### 83) `flow-php/etl-adapter-doctrine`, `-postgresql` - `withPageSize()` / `withFetchSize()` become `withBatchSize()`

Expand Down Expand Up @@ -1373,8 +1375,9 @@ Applies to `from_json()` and `from_json_lines()` unless the row names one.
| `record`, `point`, `line`, `lseg`, `box`, `path`, `polygon`, `circle` | `string` | throws `SchemaNotDerivableException` |

Applies to `from_pgsql_cursor()`, `from_pgsql_limit_offset()` and `from_pgsql_key_set()` without `->withSchema()`. A
query that cannot run as a subquery (multi-statement, data-modifying CTE, `INSERT ... RETURNING`) also throws
`SchemaNotDerivableException`. Declare `->withSchema(...)` to pick the types.
missing table, column, function, type or privilege throws PostgreSQL's `QueryException`, as the read does; a query
that writes (a data-modifying `WITH`, `SELECT ... INTO`) throws `InvalidArgumentException` before any query runs.
Declare `->withSchema(...)` to pick the types.

### 93) `flow-php/etl-adapter-postgresql` - `pgsql_table_to_flow_schema()` maps arrays, text-like types, `oid`, `timetz`

Expand Down Expand Up @@ -1532,6 +1535,56 @@ Recurse with `data/**/*.parquet`, not `data/**.parquet`. `webmozart/glob` is no
| `partitionBy(partition_by('date'))` - file body carries an all-null `date` column | file body without `date` |
| `from_parquet()` types `date` from that body column, e.g. `datetime` | `string` - declare it: `from_parquet($path)->partitionTypes(partition_types(date: type_datetime()))` |

### 109) `flow-php/etl-adapter-postgresql` - a failed `from_pgsql_cursor()` read throws its own error and rolls back

| Before | After |
|------------------------------------------------------------------------------------------------------------|---------------------------------------------------------|
| `QueryException` `[25P02] Invalid transaction state. SQL: CLOSE flow_cursor_...`, real error in `getPrevious()` | the failing statement's own `QueryException` |
| the client left inside an aborted transaction - every later query fails with `25P02` | the extractor's own transaction rolled back |
| a failure while reading rows (e.g. a row that does not match the schema) committed the transaction | rolled back |

### 110) `flow-php/etl-adapter-postgresql` - `from_pgsql_*()` read exactly one read-only `SELECT` or `VALUES` statement

| Query | Before | After |
|---------------------------------------------------------------------------|-----------------------------------------------|-----------------------------------------|
| `INSERT ... RETURNING` through `from_pgsql_cursor()` | the PHP process crashes (segfault) | `InvalidArgumentException`, nothing runs |
| `INSERT ... RETURNING` through `from_pgsql_key_set()` | the `INSERT` runs, then `QueryException` `08P01` | `InvalidArgumentException`, nothing runs |
| two statements through `from_pgsql_cursor()` | the second statement silently dropped | `InvalidArgumentException` |
| two statements through `from_pgsql_limit_offset()` / `from_pgsql_key_set()` | `QueryException` `42601` | `InvalidArgumentException` |
| a data-modifying `WITH` through `from_pgsql_key_set()` | the write runs once per page - an `INSERT` twice, an `UPDATE` never ends | `InvalidArgumentException`, nothing runs |
| `SELECT ... INTO` through `from_pgsql_key_set()` | the table is created | `InvalidArgumentException`, nothing runs |
| either through `from_pgsql_cursor()` / `from_pgsql_limit_offset()` | `QueryException` after a round trip | `InvalidArgumentException`, nothing runs |

### 111) `flow-php/postgresql` - `declare_cursor()` over SQL takes exactly one `SELECT` or `VALUES`

| Before | After |
|----------------------------------------------------------------------------------------|-----------------------------|
| `declare_cursor('c', 'INSERT INTO t VALUES (1) RETURNING id')->toSql()` - segfault | `InvalidArgumentException` |
| `declare_cursor('c', 'SELECT 1; SELECT 2')` - the second statement silently dropped | `InvalidArgumentException` |

### 112) `flow-php/postgresql` - `SelectStatement::hasIntoClause()` sees `SELECT ... INTO` in a `UNION` / `INTERSECT` / `EXCEPT`

| `sql_parse($sql)->statements()->first()->hasIntoClause()`, `$sql` | Before | After |
|-------------------------------------------------------------------|---------|--------|
| `SELECT id INTO t FROM x UNION SELECT 1` | `false` | `true` |

### 113) `flow-php/postgresql` - `sql_to_*_query()` and the pagination modifiers take exactly one read-only `SELECT` or `VALUES`

| Before | After |
|-------------------------------------------------------------------------------------|-----------------------------|
| `sql_to_paginated_query('UPDATE t SET a = 1 RETURNING id', 10)` - returned unchanged | `InvalidStatementException` |
| `sql_to_limited_query('SELECT 1; SELECT 2', 10)` - every statement paginated | `InvalidStatementException` |
| `sql_to_keyset_query()` over a data-modifying `WITH` - the write paginated | `InvalidStatementException` |
| `sql_to_count_query('SELECT id INTO t FROM x')` - counted | `InvalidStatementException` |

Applies to `PaginationModifier`, `CountModifier` and `KeysetPaginationModifier` passed to `ParsedQuery::traverse()`.

### 114) `flow-php/etl-adapter-postgresql` - `from_pgsql_limit_offset()` requires the query's own `ORDER BY`

| Query | Before | After |
|--------------------------------------------------|-----------------------------|----------------------------|
| `SELECT * FROM (SELECT id FROM t ORDER BY id) s` | pages in no defined order | `InvalidArgumentException` |

---

## Upgrading from 0.42.x to 0.43.x
Expand Down
25 changes: 20 additions & 5 deletions src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/DbalKeySetExtractor.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
use Flow\ETL\Adapter\Doctrine\Pagination\KeySet;
use Flow\ETL\Exception\InvalidArgumentException;
use Flow\ETL\Exception\RuntimeException;
use Flow\ETL\Exception\SchemaNotDerivableException;
use Flow\ETL\Extractor;
use Flow\ETL\Extractor\BatchableExtractor;
use Flow\ETL\Extractor\Batches;
Expand Down Expand Up @@ -50,6 +51,8 @@ final class DbalKeySetExtractor implements BatchableExtractor, Extractor, LimitP

private ?Schema $derived = null;

private ?SchemaNotDerivableException $refusal = null;

public function __construct(
private readonly Connection $connection,
private readonly QueryBuilder $queryBuilder,
Expand Down Expand Up @@ -191,14 +194,26 @@ public function extract(FlowContext $context): Generator

public function schema(): Schema
{
// The base builder, never the page SQL: the key_<sha1> alias lives on a per-page clone.
return (
$this->schema ?? ($this->derived ??= (new DbalResultSchema())->of(
if ($this->schema !== null) {
return $this->schema;
}

if ($this->refusal !== null) {
throw $this->refusal;
}

try {
// The base builder, never the page SQL: the key_<sha1> alias lives on a per-page clone.
return $this->derived ??= (new DbalResultSchema())->of(
$this->connection,
$this->queryBuilder->getSQL(),
self::class,
))
);
);
} catch (SchemaNotDerivableException $refusal) {
$this->refusal = $refusal;

throw $refusal;
}
}

public function withKeyAliasSuffix(string $keyAliasSuffix): self
Expand Down
Loading
Loading