feat(flow-php/etl): plan the query before running it

benchmarks/src/Partitioning/PartitionedReadScenario.php

@@ -27,12 +27,12 @@ public function run(): int
         $frame = data_frame(BenchmarkConfig::builder())->read(from_csv($tree->glob()));
 
         if ($this->pruned) {
-            $frame = $frame->filterPartitions(ref($this->cardinality->column())->equals(lit($tree->firstValue())));
+            $frame = $frame->filter(ref($this->cardinality->column())->equals(lit($tree->firstValue())));
         }
 
         $rows = 0;
 
-        $frame->run(static function (Rows $batch) use (&$rows): void {
+        $frame->forEach(static function (Rows $batch) use (&$rows): void {
             $rows += $batch->count();
         });
 

benchmarks/src/Partitioning/PartitionedTree.php

@@ -23,7 +23,7 @@
  * every phpbench iteration rebuild the whole tree before it could measure a read.
  *
  * The read path must be a glob. from_csv() over a bare partitioned directory silently reads 0 rows,
- * and a following filterPartitions() then throws "Column ... does not exist." from Row.php.
+ * and a following partition filter() then fails at bind: "Schema definition for entry ... not found."
  */
 final class PartitionedTree
 {

benchmarks/src/Pipeline/PlanDepthScenario.php

@@ -12,10 +12,10 @@
 use function Flow\ETL\DSL\ref;
 
 /**
- * schema() answers from the plan without executing it, so this prices PlanBinder::bind() alone rather
- * than $depth withEntry calls per row.
+ * schema() answers from the plan without executing it, so this prices Planner::plan() alone (optimize, translate,
+ * bind, split) rather than $depth withEntry calls per row.
  *
- * One call binds PLANS identical plans, because a single bind is far below timer resolution.
+ * One call plans PLANS identical plans, because a single plan is far below timer resolution.
  */
 final readonly class PlanDepthScenario
 {

benchmarks/src/Service/Doctrine/DoctrineWrappedWriteScenario.php

@@ -8,7 +8,6 @@
 
 use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
 use function Flow\ETL\DSL\data_frame;
-use function Flow\ETL\DSL\write_with_retries;
 use function Flow\Floe\DSL\from_floe;
 
 final readonly class DoctrineWrappedWriteScenario
@@ -39,7 +38,7 @@ public function run(): void
         data_frame()
             ->read(from_floe(Datasets::orders($this->rows)->floe()))
             ->batchSize(1000)
-            ->write(write_with_retries(to_dbal_table_insert($connection, $this->table())))
+            ->write(to_dbal_table_insert($connection, $this->table()))
             ->run();
 
         $connection->close();

benchmarks/src/Transformation/NestedTransformationScenario.php

@@ -7,6 +7,7 @@
 use Flow\Benchmarks\BenchmarkConfig;
 use Flow\Benchmarks\Datasets\Datasets;
 use Flow\ETL\Loader;
+use Flow\ETL\Sink;
 
 use function Flow\ETL\DSL\data_frame;
 use function Flow\Floe\DSL\from_floe;
@@ -15,15 +16,15 @@
 {
     public function __construct(
         private int $rows,
-        private Loader $loader,
+        private Loader|Sink $sink,
     ) {}
 
     public function run(): void
     {
         data_frame(BenchmarkConfig::builder())
             ->read(from_floe(Datasets::orders($this->rows)->floe()))
             ->batchSize(1000)
-            ->write($this->loader)
+            ->write($this->sink)
             ->run();
     }
 }

benchmarks/suites/Transformation/NestedTransformationBench.php

@@ -22,31 +22,40 @@ public function warm(array $params): void
         Datasets::orders((int) $params['rows'])->floe();
     }
 
+    #[Bench\ParamProviders('rows')]
+    #[Bench\Groups(['transformation'])]
+    public function bench_bare_branch(array $params): void
+    {
+        $sink = to_branch(ref('order_id')->isNotNull(), new NoopLoader());
+
+        (new NestedTransformationScenario((int) $params['rows'], $sink))->run();
+    }
+
     #[Bench\ParamProviders('rows')]
     #[Bench\Groups(['transformation'])]
     public function bench_blocking_transformation(array $params): void
     {
-        $loader = to_transformation(new SortByCreatedAt(), new NoopLoader());
+        $sink = to_transformation(new SortByCreatedAt(), new NoopLoader());
 
-        (new NestedTransformationScenario((int) $params['rows'], $loader))->run();
+        (new NestedTransformationScenario((int) $params['rows'], $sink))->run();
     }
 
     #[Bench\ParamProviders('rows')]
     #[Bench\Groups(['transformation'])]
     public function bench_branch_with_transformation(array $params): void
     {
-        $loader = to_branch(ref('order_id')->isNotNull(), new NoopLoader())->withTransformation(new SortByCreatedAt());
+        $sink = to_branch(ref('order_id')->isNotNull(), new NoopLoader())->withTransformation(new SortByCreatedAt());
 
-        (new NestedTransformationScenario((int) $params['rows'], $loader))->run();
+        (new NestedTransformationScenario((int) $params['rows'], $sink))->run();
     }
 
     #[Bench\ParamProviders('rows')]
     #[Bench\Groups(['transformation'])]
     public function bench_streaming_transformation(array $params): void
     {
-        $loader = to_transformation(select('order_id'), new NoopLoader());
+        $sink = to_transformation(select('order_id'), new NoopLoader());
 
-        (new NestedTransformationScenario((int) $params['rows'], $loader))->run();
+        (new NestedTransformationScenario((int) $params['rows'], $sink))->run();
     }
 
     public function rows(): Generator

documentation/components/adapters/doctrine.md

@@ -123,8 +123,8 @@ data_frame()
 
 ## Transactional Loading
 
-`to_dbal_transaction()` wraps one or more loaders so every delivery happens inside a transaction: each batch of rows
-is loaded in its own transaction, and if any loader throws, the open transaction is rolled back:
+`to_dbal_transaction()` writes one or more sinks inside transactions: each batch of rows is written in its own
+transaction, and if any sink throws, the open transaction is rolled back:
 
 ```php
 use function Flow\ETL\DSL\{data_frame, from_array};
@@ -140,29 +140,32 @@ data_frame()
     ->run();
 ```
 
-Atomicity requires every wrapped loader to use the same connection as the wrapper - pass one live `Connection` to
-`to_dbal_transaction()` and to every wrapped loader. A loader built from array params (like
+A plain loader child is a bare sink root; a `to_transformation(...)` child delivers inside the same transaction.
+Every child's loader must use the same connection as the transaction - pass one live `Connection` to
+`to_dbal_transaction()` and to every child. A loader built from array params (like
 `to_dbal_table_insert(['url' => $url], 'users')`) opens its own connection and escapes the transaction.
 
-Wrapped `to_transformation()` / `to_branch(...)->withTransformation(...)` steps with blocking operations (`sortBy()`,
+Sinks with blocking operations (`to_transformation()` / `to_branch(...)->withTransformation(...)` with `sortBy()`,
 `aggregate()`, `groupBy()->aggregate()`, `pivot()`, window functions, `collect()`, `join()` - see
-[transformations](../core/transformations.md)) buffer the stream and deliver it when the pipeline closes the loader;
+[transformations](../core/transformations.md)) buffer the stream and deliver it when the run ends;
 `to_dbal_transaction()` opens one final transaction around that delivery - the whole drained stream commits
-atomically, a failure during it rolls back.
+atomically, a failure during it rolls back and surfaces from `run()`.
 
-Do not place `write_with_retries()` inside the wrapper: on databases that abort the transaction after a failed
-statement (PostgreSQL), every retry attempt fails too. Wrap the transaction instead -
-`write_with_retries(to_dbal_transaction(...))` gives each attempt a fresh transaction (see
-[retry](../core/retry.md)).
+Inside the transaction every sink runs throw-only: a failure rolls the batch back first, and only then the frame's
+`onError()` handler decides whether the run continues.
 
-Use `withIsolationLevel()` to set the transaction isolation level; it applies to every transaction the wrapper opens,
-including the final one:
+To set the isolation level, build the transaction yourself; it applies to every transaction opened, including the
+final one, and the previous level is restored when each one ends:
 
 ```php
 use Doctrine\DBAL\TransactionIsolationLevel;
+use Flow\ETL\Adapter\Doctrine\DbalTransaction;
+use Flow\ETL\Sink\Transactional;
 
-to_dbal_transaction($connection, to_dbal_table_insert($connection, 'users'))
-    ->withIsolationLevel(TransactionIsolationLevel::SERIALIZABLE);
+new Transactional(
+    DbalTransaction::fromConnection($connection)->withIsolationLevel(TransactionIsolationLevel::SERIALIZABLE),
+    to_dbal_table_insert($connection, 'users'),
+);
 ```
 
 ## Extractor - DbalQuery

documentation/components/adapters/postgresql.md

@@ -390,12 +390,10 @@ df()
 
 ### Transactional Loading
 
-`to_pgsql_transaction()` wraps one or more loaders so every delivery happens inside a transaction: each batch of rows
-is loaded in its own transaction, and if any loader throws, the open transaction is rolled back:
+`to_pgsql_transaction()` writes one or more sinks inside transactions: each batch of rows is written in its own
+transaction, and if any sink throws, the open transaction is rolled back:
 
 ```php
-use Flow\PostgreSql\QueryBuilder\Transaction\IsolationLevel;
-
 use function Flow\ETL\Adapter\PostgreSql\{to_pgsql_table, to_pgsql_transaction};
 
 df()
@@ -408,32 +406,37 @@ df()
     ->run();
 ```
 
-Wrapped `to_transformation()` / `to_branch(...)->withTransformation(...)` steps with blocking operations (`sortBy()`,
+Sinks with blocking operations (`to_transformation()` / `to_branch(...)->withTransformation(...)` with `sortBy()`,
 `aggregate()`, `groupBy()->aggregate()`, `pivot()`, window functions, `collect()`, `join()` - see
-[transformations](../core/transformations.md)) buffer the stream and deliver it when the pipeline closes the loader;
+[transformations](../core/transformations.md)) buffer the stream and deliver it when the run ends;
 `to_pgsql_transaction()` opens one final transaction around that delivery - the whole drained stream commits
-atomically, a failure during it rolls back. Every wrapped loader must use the same `Client` instance as the wrapper -
-a loader holding its own `Client` escapes the transaction.
+atomically, a failure during it rolls back and surfaces from `run()`. A plain loader child is a bare sink root; a
+`to_transformation(...)` child delivers inside the same transaction. Every sink's loader must use the same `Client`
+instance as the transaction - a loader holding its own `Client` escapes it.
 
-Do not place `write_with_retries()` inside the wrapper: after a failed statement PostgreSQL aborts the whole
-transaction, so every retry attempt fails too. Wrap the transaction instead -
-`write_with_retries(to_pgsql_transaction(...))` gives each attempt a fresh transaction (see
-[retry](../core/retry.md)).
+Inside the transaction every sink runs throw-only: a failure rolls the batch back first, and only then the frame's
+`onError()` handler decides whether the run continues.
 
-Use `withIsolationLevel()` to set the transaction isolation level; it applies to every transaction the wrapper opens,
-including the final one:
+To set the isolation level, build the transaction yourself; it applies to every transaction opened, including the
+final one:
 
 ```php
-to_pgsql_transaction($client, to_pgsql_table($client, 'users'))
-    ->withIsolationLevel(IsolationLevel::SERIALIZABLE);
+use Flow\ETL\Adapter\PostgreSql\PostgreSqlTransaction;
+use Flow\ETL\Sink\Transactional;
+use Flow\PostgreSql\QueryBuilder\Transaction\IsolationLevel;
+
+new Transactional(
+    (new PostgreSqlTransaction($client))->withIsolationLevel(IsolationLevel::SERIALIZABLE),
+    to_pgsql_table($client, 'users'),
+);
 ```
 
 ## Loader DSL Functions Reference
 
 | Function                                       | Description                                               |
 |------------------------------------------------|-----------------------------------------------------------|
 | `to_pgsql_table($client, $table)`              | Create a PostgreSQL loader for a table                    |
-| `to_pgsql_transaction($client, ...$loaders)`   | Run multiple loaders, every delivery inside a transaction |
+| `to_pgsql_transaction($client, ...$sinks)`     | Write sinks, every delivery inside a transaction          |
 | `pgsql_insert_options(...)`                    | Configure insert behavior (conflicts, upsert)             |
 | `pgsql_update_options($primaryKeys)`           | Configure update behavior (primary key columns)           |
 | `pgsql_delete_options($primaryKeys)`           | Configure delete behavior (primary key columns)           |
@@ -502,16 +505,16 @@ $schema = schema(
 );
 ```
 
-| Metadata                          | Effect on the generated column                             |
-|-----------------------------------|------------------------------------------------------------|
-| `PostgreSqlMetadata::type($name)` | Force a specific PostgreSQL type, bypassing the type map   |
-| `PostgreSqlMetadata::length($n)`  | Emit `varchar($n)`                                         |
-| `PostgreSqlMetadata::precision($p)` / `::scale($s)` | Emit `numeric($p, $s)`                   |
-| `PostgreSqlMetadata::default($v)` | Set a column `DEFAULT`                                      |
-| `PostgreSqlMetadata::primaryKey($name)` | Include the column in the table primary key          |
+| Metadata                                            | Effect on the generated column                                                        |
+|-----------------------------------------------------|---------------------------------------------------------------------------------------|
+| `PostgreSqlMetadata::type($name)`                   | Force a specific PostgreSQL type, bypassing the type map                              |
+| `PostgreSqlMetadata::length($n)`                    | Emit `varchar($n)`                                                                    |
+| `PostgreSqlMetadata::precision($p)` / `::scale($s)` | Emit `numeric($p, $s)`                                                                |
+| `PostgreSqlMetadata::default($v)`                   | Set a column `DEFAULT`                                                                |
+| `PostgreSqlMetadata::primaryKey($name)`             | Include the column in the table primary key                                           |
 | `PostgreSqlMetadata::indexUnique($name, $position)` | Include the column in a named `UNIQUE` constraint, optionally at an explicit position |
-| `PostgreSqlMetadata::index($name, $position)` | Include the column in a named index, optionally at an explicit position |
-| `PostgreSqlMetadata::identity($generation)` | Make the column an identity column               |
+| `PostgreSqlMetadata::index($name, $position)`       | Include the column in a named index, optionally at an explicit position               |
+| `PostgreSqlMetadata::identity($generation)`         | Make the column an identity column                                                    |
 | `PostgreSqlMetadata::generated($expr)` | Make the column a generated column                    |
 
 Columns sharing the same primary key, unique constraint, or index name are grouped together, so composite keys are