perf: fused native Floe read and write path

documentation/components/extensions/flow-php-ext.md

@@ -58,10 +58,12 @@ implementation automatically:
 
 - **`Flow\Floe\RustFloeEncoderNative`** - the Floe ROW frame-body codec:
   `encode(list<TypedRowValues>, schemaBody)` returns the encoded frame bodies,
-  `decode(list<string>, schemaBody)` returns `list<Flow\ETL\Row\RawRowValues>`. The userland wrapper
-  `Flow\Floe\NativeFloeEncoder` carries the `Flow\ETL\Row\Encoder` interface, and
-  `Flow\Floe\AdaptiveFloeEncoder` - built by every writer/reader - selects it over
-  `Flow\Floe\PhpFloeEncoder` when the extension is loaded.
+  `decode(list<string>, schemaBody)` returns `list<Flow\ETL\Row\RawRowValues>`, and
+  `decodeRows(list<string>, schemaBody, Schema)` decodes and casts straight into `Flow\ETL\Rows` in one pass, and
+  `encodeFrames(Rows, schemaBody, Schema)` turns `Rows` straight into one string of complete ROW frames - what the
+  reader and the writer use when the configured hydrator is the native one. The userland wrapper
+  `Flow\Floe\NativeFloeEncoder` carries the `Flow\Floe\FloeEncoder` interface, and `FloeEngine::adaptive` - the
+  engine every writer/reader defaults to - builds it over `Flow\Floe\PhpFloeEncoder` when the extension is loaded.
 - **`Flow\ETL\Row\RustRowHydratorNative`** - the native `hydrate`/`cast`/`dehydrate` behind
   `Flow\ETL\Row\NativeRowHydrator`, which `Flow\ETL\Row\AdaptiveRowHydrator` (the config default)
   selects when the extension is loaded - used by adapter loaders and raw-scalar extractors such as

documentation/installation/docker.md

@@ -75,7 +75,7 @@ that Flow detects and uses automatically:
 
 | Extension  | Package                                                                       | Effect when loaded                                                                                                                                     |
 |------------|-------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `flow_php` | [flow-php/flow-php-ext](/documentation/components/extensions/flow-php-ext.md) | `AdaptiveRowHydrator` and `AdaptiveFloeEncoder` run native, fusing every Floe read/write and every raw-scalar hydration into one native call per batch |
+| `flow_php` | [flow-php/flow-php-ext](/documentation/components/extensions/flow-php-ext.md) | `AdaptiveRowHydrator` and the adaptive Floe engine run native, fusing every Floe read/write and every raw-scalar hydration into one native call per batch |
 | `arrow`    | [flow-php/arrow-ext](/documentation/components/extensions/arrow-ext.md)       | `AdaptiveParquetEngine` selects `ArrowParquetEngine`, so Parquet reads and writes run native                                                           |
 | `pg_query` | [flow-php/pg-query-ext](/documentation/components/extensions/pg-query-ext.md) | `Flow\PostgreSql\Parser` becomes usable at all - SQL parsing, normalization and AST manipulation                                                       |
 | `protobuf` | `pecl/protobuf`                                                               | `Flow\PostgreSql\Parser` decodes the parse tree in C instead of pure PHP - measured ~69x faster end to end                                             |

documentation/upgrading.md

@@ -356,6 +356,13 @@ final class MyExtractor implements Extractor
 | `sql_query_tables('DROP TABLE a, s.b')` - `[]`                                       | `[a, s.b]` |
 | `sql_query_tables("COMMENT ON COLUMN s.t.c IS 'x'")` - `[]`                          | `[s.t]`    |
 
+### 35) `flow-php/etl` - `AdaptiveFloeEncoder` removed, Floe engines build a `FloeEncoder`
+
+| Before                                   | After                                                                                                                                |
+|------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------|
+| `Flow\Floe\AdaptiveFloeEncoder`          | removed - `FloeEngine::adaptive->encoder($schema)` returns `NativeFloeEncoder` when the extension supports it, else `PhpFloeEncoder` |
+| `FloeEngine::encoder(): Encoder<string>` | `FloeEngine::encoder(): Flow\Floe\FloeEncoder` (`Encoder<string>` plus `decodeRows()` / `encodeFrames()`)                            |
+
 ---
 
 ## Upgrading from 0.43.x to 0.44.x

shell.nix

@@ -30,21 +30,41 @@ let
             allowUnfree = true;
         };
         overlays = [
-            # Blackfire upstream republishes the same versioned tarball with different
-            # bytes when they rebuild, invalidating the sha256 pinned in nixpkgs. Override
-            # just the CLI agent's src on macOS arm64 with the current upstream hash.
-            # Linux and other platforms keep nixpkgs' original src untouched. The PHP
-            # extension (php83.extensions.blackfire) uses a separate upstream URL and is
-            # unaffected on every platform.
             (final: prev:
-                if prev.stdenv.hostPlatform.system == "aarch64-darwin" then {
-                    blackfire = prev.blackfire.overrideAttrs (old: {
-                        src = prev.fetchurl {
-                            url = "https://packages.blackfire.io/blackfire/2.29.7/blackfire-darwin_arm64.pkg.tar.gz";
-                            sha256 = "sha256-e0oTxGFxgURMyUoTNh+NFGVoO9qGKrHNKud3IFD0fec=";
+                if prev.stdenv.hostPlatform.system == "aarch64-darwin" then
+                    let
+                        blackfire-probe-version = "2026.9.2";
+                        blackfire-probe-hashes = {
+                            "83" = "sha256-3oJtMuVKGUgpduMp7snSdGE/BH764EA7yz+N70+qFNg=";
+                            "84" = "sha256-0HOCBB9dgU9Vq5/F0iKCzumjwT81qxHElPsLGKgVhr0=";
+                            "85" = "sha256-Hi9bC/CigkA3VWFTqfE7JzBcGGJAhUwnmMndHgbFIW4=";
                         };
-                    });
-                } else {}
+                        with-blackfire-probe = php-version: php: php.override {
+                            packageOverrides = php-final: php-prev: {
+                                extensions = php-prev.extensions // {
+                                    blackfire = php-prev.extensions.blackfire.overrideAttrs (old: {
+                                        version = blackfire-probe-version;
+                                        src = prev.fetchurl {
+                                            url = "https://packages.blackfire.io/binaries/blackfire-php/${blackfire-probe-version}/blackfire-php-darwin_arm64-php-${php-version}.so";
+                                            hash = blackfire-probe-hashes.${php-version};
+                                        };
+                                    });
+                                };
+                            };
+                        };
+                    in {
+                        blackfire = prev.blackfire.overrideAttrs (old: {
+                            version = "2026.9.1";
+                            src = prev.fetchurl {
+                                url = "https://packages.blackfire.io/blackfire/2026.9.1/blackfire-darwin_arm64.pkg.tar.gz";
+                                sha256 = "sha256-xNn78U4jdABzWrSKMSSZXE5tuf/SRK8OwdhldKBBKk0=";
+                            };
+                        });
+                        php83 = with-blackfire-probe "83" prev.php83;
+                        php84 = with-blackfire-probe "84" prev.php84;
+                        php85 = with-blackfire-probe "85" prev.php85;
+                    }
+                else {}
             )
         ];
     };

src/core/etl/src/Flow/ETL/Extractor/FileConstants.php

@@ -4,6 +4,10 @@
 
 namespace Flow\ETL\Extractor;
 
+use Flow\ETL\Row;
+use Flow\ETL\Rows;
+use Flow\ETL\Schema;
+
 final readonly class FileConstants
 {
     /**
@@ -33,4 +37,24 @@ public function fill(array $row): array
 
         return $this->partitionColumns->fill($row, $this->partitionNames, $this->partitionValues);
     }
+
+    /**
+     * fill() over a batch the reader already matched to the file's body schema, adopting $declared - the schema
+     * FileColumns::declare() built over that body schema. A batch with nothing to add is returned as it is.
+     */
+    public function fillRows(Rows $rows, Schema $declared): Rows
+    {
+        if ($this->uri === null && $this->partitionNames === []) {
+            return $rows;
+        }
+
+        $filled = [];
+
+        foreach ($rows->all() as $row) {
+            $filled[] = new Row($this->fill($row->values()));
+        }
+
+        // declare() appends the tail in the order fill() writes it, so the rows need no second check
+        return Rows::trusted($declared, $filled);
+    }
 }

src/core/etl/src/Flow/ETL/Extractor/PartitionColumns.php

@@ -24,12 +24,6 @@ public function __construct(
     ) {}
 
     /**
-     * One read yields one Schema, so a partition column that only some paths under the listing carry
-     * still has to be declared for all of them - and it is nullable when some path lacks it, or when
-     * some path carries the Hive null sentinel, which is a value the column has to be able to hold.
-     * Partition values live in the path, so both the union and its nullability are known without
-     * opening a single file.
-     *
      * @return array<string, bool> partition name => nullable
      */
     public function names(Path $path, Filter $filter): array
@@ -62,10 +56,6 @@ public function names(Path $path, Filter $filter): array
     }
 
     /**
-     * A partition column keeps the type its declared definition gives it, but never its body
-     * position: it is removed from wherever the file put it and re-appended in the partition block,
-     * so a declared read and an undeclared one emit the same column order.
-     *
      * @param array<string, bool> $names
      */
     public function declare(Schema $schema, array $names, PartitionTypes $types = new PartitionTypes()): Schema
@@ -93,10 +83,6 @@ public function declare(Schema $schema, array $names, PartitionTypes $types = ne
     }
 
     /**
-     * Inference sees one stream at a time, so it types a partition column from that stream alone -
-     * `string` where the path carries it, `?null` where it does not, and the two batches then refuse
-     * to merge. The path knows better than the values do, so here the partition definition wins.
-     *
      * @param array<string, bool> $names
      */
     public function apply(Rows $rows, array $names, PartitionTypes $types = new PartitionTypes()): Rows
@@ -128,8 +114,6 @@ public function apply(Rows $rows, array $names, PartitionTypes $types = new Part
     public function fill(array $row, array $names, array $values): array
     {
         foreach ($names as $name => $_) {
-            // declare() re-appends every partition column in the partition block, so a column the
-            // file data also carries must leave its body position here or the two disagree
             unset($row[$name]);
             $row[$name] = array_key_exists($name, $values) ? $values[$name] : null;
         }

src/core/etl/src/Flow/ETL/Optimizer/Rule/CountFromStatistics.php

@@ -17,11 +17,6 @@
 
 final readonly class CountFromStatistics implements Rule
 {
-    /**
-     * A count straight over a source that knows its rows exactly reads a row holding that number instead. Any node
-     * in between runs code that may fail or skip a batch, and a pushed limit or partition filter changes the rows
-     * the statistics describe, so those plans are counted by running them.
-     */
     public function apply(LogicalPlan $plan, FlowContext $context): LogicalPlan
     {
         $count = $plan->root instanceof Result ? $plan->root->children()[0] : null;

src/core/etl/src/Flow/ETL/Row/AdaptiveRowHydrator.php

@@ -16,6 +16,11 @@ public function __construct()
         $this->delegate = NativeRowHydrator::isSupported() ? new NativeRowHydrator() : new PhpRowHydrator();
     }
 
+    public function isNative(): bool
+    {
+        return $this->delegate instanceof NativeRowHydrator;
+    }
+
     public function dehydrate(Rows $rows): array
     {
         return $this->delegate->dehydrate($rows);

src/core/etl/src/Flow/ETL/Row/NativeRowHydrator.php

@@ -4,12 +4,9 @@
 
 namespace Flow\ETL\Row;
 
-use Flow\ETL\Exception\SchemaMismatchException;
 use Flow\ETL\Rows;
 use Flow\ETL\Schema;
-use Flow\Floe\Exception\ExtensionException;
 use RuntimeException;
-use Throwable;
 
 use function class_exists;
 use function extension_loaded;
@@ -39,22 +36,6 @@ public function dehydrate(Rows $rows): array
 
     public function hydrate(array $batch, Schema $schema): Rows
     {
-        try {
-            return $this->native->hydrate($batch, $schema);
-        } catch (ExtensionException $e) {
-            throw self::unwrap($e);
-        }
-    }
-
-    /**
-     * The extension turns any PHP exception raised inside it into an ExtensionException carrying the
-     * original as previous. A batch refused by the row gate must reach the caller as the same
-     * exception both hydrators throw, or the two disagree on nothing but the type.
-     */
-    private static function unwrap(ExtensionException $exception): Throwable
-    {
-        $previous = $exception->getPrevious();
-
-        return $previous instanceof SchemaMismatchException ? $previous : $exception;
+        return $this->native->hydrate($batch, $schema);
     }
 }

src/core/etl/src/Flow/Floe/FloeEncoder.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Floe;
+
+use Flow\ETL\Exception\SchemaMismatchException;
+use Flow\ETL\Row\Encoder;
+use Flow\ETL\Row\Hydrator;
+use Flow\ETL\Rows;
+use Flow\ETL\Schema;
+use Flow\Floe\Exception\FloeException;
+
+/**
+ * @extends Encoder<string>
+ */
+interface FloeEncoder extends Encoder
+{
+    /**
+     * `$hydrator->hydrate($this->decode($bodies), $schema)`, in one native pass where the hydrator is native too.
+     *
+     * @param list<string> $bodies
+     *
+     * @throws FloeException
+     * @throws SchemaMismatchException
+     */
+    public function decodeRows(array $bodies, Schema $schema, Hydrator $hydrator): Rows;
+
+    /**
+     * `Format::rowFrames($this->encode($hydrator->dehydrate($rows)))`, in one native pass where the hydrator is native
+     * too - complete ROW frames for a writer whose codec leaves bodies as they are.
+     *
+     * @throws FloeException
+     * @throws SchemaMismatchException
+     */
+    public function encodeFrames(Rows $rows, Hydrator $hydrator): string;
+}

src/core/etl/src/Flow/Floe/FloeEngine.php

@@ -4,7 +4,6 @@
 
 namespace Flow\Floe;
 
-use Flow\ETL\Row\Encoder;
 use Flow\ETL\Schema;
 
 enum FloeEngine: string
@@ -13,13 +12,12 @@ enum FloeEngine: string
     case native = 'native';
     case php = 'php';
 
-    /**
-     * @return Encoder<string>
-     */
-    public function encoder(Schema $schema): Encoder
+    public function encoder(Schema $schema): FloeEncoder
     {
         return match ($this) {
-            self::adaptive => new AdaptiveFloeEncoder($schema),
+            self::adaptive => NativeFloeEncoder::isSupported()
+                ? new NativeFloeEncoder($schema)
+                : new PhpFloeEncoder($schema),
             self::native => new NativeFloeEncoder($schema),
             self::php => new PhpFloeEncoder($schema),
         };

src/core/etl/src/Flow/Floe/FloeExtractor.php

@@ -16,9 +16,7 @@
 use Flow\ETL\Extractor\Signal;
 use Flow\ETL\Extractor\Statistics;
 use Flow\ETL\FlowContext;
-use Flow\ETL\Row;
 use Flow\ETL\Row\Hydrator;
-use Flow\ETL\Rows;
 use Flow\ETL\Schema;
 use Flow\ETL\Schema\Validator\StrictValidator;
 use Flow\Filesystem\Filesystem;
@@ -125,15 +123,7 @@ public function extract(FlowContext $context, ?int $limit = null, Filter $pathFi
                 foreach ($file->reader->rows($this->batchSize(), $fileOffset, $remaining) as $rows) {
                     // R7: the stamp stays post-hydration - FloeStreamReader::rows() yields hydrated Rows and
                     // must not learn about paths - but the constants are the shared ones, already typed
-                    $filled = [];
-
-                    foreach ($rows->all() as $row) {
-                        $filled[] = new Row($constants->fill($row->values()));
-                    }
-
-                    // the reader already matched every row against the footer schema, and the tail is
-                    // written in the order declare() emits it, so a second full check buys nothing
-                    $rows = Rows::trusted($fileSchema, $filled);
+                    $rows = $constants->fillRows($rows, $fileSchema);
 
                     if ($matchTo !== null) {
                         $rows = $rows->matchTo($matchTo);