Branching Loader - aggregating / sorting / grouping transformations bug

benchmarks/src/Service/Doctrine/DoctrineWrappedWriteScenario.php

@@ -0,0 +1,58 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Benchmarks\Service\Doctrine;
+
+use Flow\Benchmarks\Datasets\Datasets;
+
+use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
+use function Flow\ETL\DSL\batches;
+use function Flow\ETL\DSL\data_frame;
+use function Flow\ETL\DSL\write_with_retries;
+use function Flow\Floe\DSL\from_floe;
+
+/**
+ * Rows are re-batched to one per Rows because a floe file arrives pre-chunked, which would hide the
+ * BatchSizeOptimization that from_csv()/from_json() sources trigger.
+ */
+final readonly class DoctrineWrappedWriteScenario
+{
+    public function __construct(
+        private int $rows,
+    ) {}
+
+    public function table(): string
+    {
+        return 'benchmark_orders_dbal_wrapped_write_' . $this->rows;
+    }
+
+    public function setUp(): void
+    {
+        Datasets::orders($this->rows)->floe();
+
+        $connection = DoctrineConnection::open();
+        DoctrineConnection::dropTable($connection, $this->table());
+        DoctrineConnection::createTable($connection, $this->table());
+        $connection->close();
+    }
+
+    public function run(): void
+    {
+        $connection = DoctrineConnection::open();
+
+        data_frame()
+            ->read(batches(from_floe(Datasets::orders($this->rows)->floe()), 1))
+            ->write(write_with_retries(to_dbal_table_insert($connection, $this->table())))
+            ->run();
+
+        $connection->close();
+    }
+
+    public function dropTable(): void
+    {
+        $connection = DoctrineConnection::open();
+        DoctrineConnection::dropTable($connection, $this->table());
+        $connection->close();
+    }
+}

benchmarks/src/Transformation/NestedTransformationScenario.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Benchmarks\Transformation;
+
+use Flow\Benchmarks\BenchmarkConfig;
+use Flow\Benchmarks\Datasets\Datasets;
+use Flow\ETL\Loader;
+
+use function Flow\ETL\DSL\data_frame;
+use function Flow\Floe\DSL\from_floe;
+
+final readonly class NestedTransformationScenario
+{
+    public function __construct(
+        private int $rows,
+        private Loader $loader,
+    ) {}
+
+    public function run(): void
+    {
+        data_frame(BenchmarkConfig::builder())
+            ->read(from_floe(Datasets::orders($this->rows)->floe()))
+            ->batchSize(1000)
+            ->write($this->loader)
+            ->run();
+    }
+}

benchmarks/suites/Service/Doctrine/DoctrineWrappedWriteBench.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Benchmarks\Service\Doctrine;
+
+use Generator;
+use PhpBench\Attributes as Bench;
+
+#[Bench\AfterClassMethods('dropWrappedWrite')]
+final class DoctrineWrappedWriteBench
+{
+    public function setUpWrappedWrite(array $params): void
+    {
+        (new DoctrineWrappedWriteScenario((int) $params['rows']))->setUp();
+    }
+
+    public static function dropWrappedWrite(): void
+    {
+        (new DoctrineWrappedWriteScenario(100_000))->dropTable();
+
+        (new DoctrineWrappedWriteScenario((int) (getenv('FLOW_BENCH_ROWS') ?: 100_000)))->dropTable();
+    }
+
+    #[Bench\ParamProviders('rows')]
+    #[Bench\Groups(['service', 'service-doctrine'])]
+    #[Bench\BeforeMethods('setUpWrappedWrite')]
+    public function bench_doctrine_wrapped_write(array $params): void
+    {
+        (new DoctrineWrappedWriteScenario((int) $params['rows']))->run();
+    }
+
+    public function rows(): Generator
+    {
+        $rows = (int) (getenv('FLOW_BENCH_ROWS') ?: 100_000);
+
+        yield number_format($rows) => ['rows' => $rows];
+    }
+}

benchmarks/suites/Transformation/NestedTransformationBench.php

@@ -0,0 +1,74 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Benchmarks\Transformation;
+
+use Flow\ETL\DataFrame;
+use Flow\ETL\FlowContext;
+use Flow\ETL\Rows;
+use Flow\ETL\Transformation;
+use Generator;
+use PhpBench\Attributes as Bench;
+
+use function Flow\ETL\DSL\ref;
+use function Flow\ETL\DSL\select;
+use function Flow\ETL\DSL\to_branch;
+use function Flow\ETL\DSL\to_callable;
+use function Flow\ETL\DSL\to_transformation;
+
+final class NestedTransformationBench
+{
+    #[Bench\ParamProviders('rows')]
+    #[Bench\Groups(['transformation'])]
+    public function bench_blocking_transformation(array $params): void
+    {
+        $loader = to_transformation(
+            new class implements Transformation {
+                public function transform(DataFrame $dataFrame): DataFrame
+                {
+                    return $dataFrame->sortBy(ref('created_at'));
+                }
+            },
+            to_callable(static function (Rows $rows, FlowContext $context): void {}),
+        );
+
+        (new NestedTransformationScenario((int) $params['rows'], $loader))->run();
+    }
+
+    #[Bench\ParamProviders('rows')]
+    #[Bench\Groups(['transformation'])]
+    public function bench_branch_with_transformation(array $params): void
+    {
+        $loader = to_branch(
+            ref('order_id')->isNotNull(),
+            to_callable(static function (Rows $rows, FlowContext $context): void {}),
+        )->withTransformation(new class implements Transformation {
+            public function transform(DataFrame $dataFrame): DataFrame
+            {
+                return $dataFrame->sortBy(ref('created_at'));
+            }
+        });
+
+        (new NestedTransformationScenario((int) $params['rows'], $loader))->run();
+    }
+
+    #[Bench\ParamProviders('rows')]
+    #[Bench\Groups(['transformation'])]
+    public function bench_streaming_transformation(array $params): void
+    {
+        $loader = to_transformation(
+            select('order_id'),
+            to_callable(static function (Rows $rows, FlowContext $context): void {}),
+        );
+
+        (new NestedTransformationScenario((int) $params['rows'], $loader))->run();
+    }
+
+    public function rows(): Generator
+    {
+        $rows = (int) (getenv('FLOW_BENCH_ROWS') ?: 100_000);
+
+        yield number_format($rows) => ['rows' => $rows];
+    }
+}

bin/docs.php

@@ -67,6 +67,7 @@ public function execute(InputInterface $input, OutputInterface $output): int
             __DIR__ . '/../src/adapter/etl-adapter-http/src/Flow/ETL/Adapter/Http/DSL/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/functions.php',
+            __DIR__ . '/../src/adapter/etl-adapter-postgresql/src/Flow/ETL/Adapter/PostgreSql/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-seal/src/Flow/ETL/Adapter/Seal/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-text/src/Flow/ETL/Adapter/Text/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-xml/src/Flow/ETL/Adapter/XML/functions.php',

documentation/components/adapters/doctrine.md

@@ -121,6 +121,50 @@ data_frame()
 // XML entries are automatically converted to strings before database insertion
 ```
 
+## 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:
+
+```php
+use function Flow\ETL\DSL\{data_frame, from_array};
+use function Flow\ETL\Adapter\Doctrine\{to_dbal_table_insert, to_dbal_transaction};
+
+data_frame()
+    ->read(from_array($data))
+    ->write(to_dbal_transaction(
+        $connection,
+        to_dbal_table_insert($connection, 'users'),
+        to_dbal_table_insert($connection, 'users_audit'),
+    ))
+    ->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
+`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()`,
+`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;
+`to_dbal_transaction()` opens one final transaction around that delivery - the whole drained stream commits
+atomically, a failure during it rolls back.
+
+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)).
+
+Use `withIsolationLevel()` to set the transaction isolation level; it applies to every transaction the wrapper opens,
+including the final one:
+
+```php
+use Doctrine\DBAL\TransactionIsolationLevel;
+
+to_dbal_transaction($connection, to_dbal_table_insert($connection, 'users'))
+    ->withIsolationLevel(TransactionIsolationLevel::SERIALIZABLE);
+```
+
 ## Extractor - DbalQuery
 
 This simple but powerful extractor let you extract data from a single or multiple parametrized queries.

documentation/components/adapters/postgresql.md

@@ -394,8 +394,8 @@ df()
 
 ### Transactional Loading
 
-`to_pgsql_transaction()` wraps one or more loaders so each batch of rows is loaded inside a single transaction. If any
-loader throws, the whole batch is rolled back:
+`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:
 
 ```php
 use Flow\PostgreSql\QueryBuilder\Transaction\IsolationLevel;
@@ -412,7 +412,20 @@ df()
     ->run();
 ```
 
-Use `withIsolationLevel()` to set the transaction isolation level:
+Wrapped `to_transformation()` / `to_branch(...)->withTransformation(...)` steps with blocking operations (`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;
+`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.
+
+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)).
+
+Use `withIsolationLevel()` to set the transaction isolation level; it applies to every transaction the wrapper opens,
+including the final one:
 
 ```php
 to_pgsql_transaction($client, to_pgsql_table($client, 'users'))
@@ -421,13 +434,13 @@ to_pgsql_transaction($client, 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 within a single 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)     |
+| 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 |
+| `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)           |
 
 ## Schema Conversion