feat(flow-php/etl): schema-driven Floe writes with validation option

src/core/etl/src/Flow/ArrayComparison/ArrayComparison.php

@@ -34,6 +34,14 @@ public function equals(?array $a, ?array $b): bool
             return false;
         }
 
+        if ($a === $b) {
+            return true;
+        }
+
+        if (count($a) !== count($b)) {
+            return false;
+        }
+
         return $this->valueEquals((new ArraySortByKey())($a), (new ArraySortByKey())($b));
     }
 

src/core/etl/src/Flow/ETL/Schema/Metadata.php

@@ -108,6 +108,10 @@ public function isEmpty(): bool
 
     public function isEqual(self $metadata): bool
     {
+        if ($this->map === $metadata->map) {
+            return true;
+        }
+
         return (new ArrayComparison())->equals($this->map, $metadata->map);
     }
 

src/core/etl/src/Flow/ETL/Sort/ExternalSort/BucketsCache/FilesystemBucketsCache.php

@@ -8,11 +8,14 @@
 use Flow\ETL\Hash\NativePHPHash;
 use Flow\ETL\Row;
 use Flow\ETL\Rows;
+use Flow\ETL\Schema;
 use Flow\ETL\Sort\ExternalSort\BucketsCache;
 use Flow\Filesystem\Filesystem;
 use Flow\Filesystem\Path;
 use Flow\Floe\FloeReader;
+use Flow\Floe\FloeStreamWriter;
 use Flow\Floe\FloeWriter;
+use Flow\Floe\Options;
 use Generator;
 
 use function count;
@@ -50,13 +53,19 @@ public function __construct(
      */
     public function append(string $bucketId, iterable $rows): void
     {
-        if (!isset($this->writers[$bucketId])) {
-            $writer = new FloeWriter($this->filesystem);
-            $writer->append($this->keyPath($bucketId));
-            $this->writers[$bucketId] = $writer;
-        }
+        foreach ($this->batches($rows) as $batch) {
+            if (!isset($this->writers[$bucketId])) {
+                $writer = new FloeWriter(
+                    $this->filesystem,
+                    FloeStreamWriter::unionSchema($batch),
+                    new Options(validateData: false),
+                );
+                $writer->append($this->keyPath($bucketId));
+                $this->writers[$bucketId] = $writer;
+            }
 
-        $this->write($this->writers[$bucketId], $rows);
+            $this->writers[$bucketId]->write($batch);
+        }
     }
 
     /**
@@ -93,10 +102,25 @@ public function set(string $bucketId, iterable $rows): void
     {
         $this->closeWriter($bucketId);
 
-        $writer = new FloeWriter($this->filesystem);
-        $writer->create($this->keyPath($bucketId));
+        $writer = null;
 
-        $this->write($writer, $rows);
+        foreach ($this->batches($rows) as $batch) {
+            if ($writer === null) {
+                $writer = new FloeWriter(
+                    $this->filesystem,
+                    FloeStreamWriter::unionSchema($batch),
+                    new Options(validateData: false),
+                );
+                $writer->create($this->keyPath($bucketId));
+            }
+
+            $writer->write($batch);
+        }
+
+        if ($writer === null) {
+            $writer = new FloeWriter($this->filesystem, new Schema(), new Options(validateData: false));
+            $writer->create($this->keyPath($bucketId));
+        }
 
         $writer->close();
     }
@@ -120,22 +144,24 @@ private function keyPath(string $key): Path
 
     /**
      * @param iterable<Row>|Rows $rows
+     *
+     * @return Generator<Rows>
      */
-    private function write(FloeWriter $writer, iterable $rows): void
+    private function batches(iterable $rows): Generator
     {
         $batch = [];
 
         foreach ($rows as $row) {
             $batch[] = $row;
 
             if (count($batch) >= $this->batchSize) {
-                $writer->write(new Rows(...$batch));
+                yield new Rows(...$batch);
                 $batch = [];
             }
         }
 
         if ($batch !== []) {
-            $writer->write(new Rows(...$batch));
+            yield new Rows(...$batch);
         }
     }
 }

src/core/etl/src/Flow/Floe/DSL/functions.php

@@ -15,6 +15,7 @@
 use Flow\Floe\FloeExtractor;
 use Flow\Floe\FloeLoader;
 use Flow\Floe\FloeMerger;
+use Flow\Floe\Options;
 
 use function array_map;
 use function Flow\Filesystem\DSL\native_local_filesystem;
@@ -42,10 +43,16 @@ function from_floe(
 function to_floe(
     string|Path $path,
     ?Metadata $metadata = null,
-    Codec $codec = new NoopCodec(),
+    Options $options = new Options(),
     FloeEngine $engine = FloeEngine::adaptive,
 ): FloeLoader {
-    return new FloeLoader(is_string($path) ? path_real($path) : $path, $metadata, $codec, $engine);
+    return new FloeLoader(is_string($path) ? path_real($path) : $path, $metadata, $options, $engine);
+}
+
+#[DocumentationDSL(module: Module::FLOE, type: DSLType::HELPER)]
+function floe_options(bool $validate_data = true, int $buffer_size = 65536, Codec $codec = new NoopCodec()): Options
+{
+    return new Options($validate_data, $buffer_size, $codec);
 }
 
 /**

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

@@ -13,7 +13,7 @@
 use Flow\ETL\Schema;
 use Flow\ETL\Schema\Metadata;
 use Flow\Filesystem\Path;
-use Flow\Floe\Codec\NoopCodec;
+use Flow\Floe\Exception\FloeException;
 use Throwable;
 
 use function array_key_exists;
@@ -32,7 +32,7 @@ final class FloeLoader implements Closure, FileLoader, Loader
     public function __construct(
         private readonly Path $path,
         private readonly ?Metadata $metadata = null,
-        private readonly Codec $codec = new NoopCodec(),
+        private readonly Options $options = new Options(),
         private readonly FloeEngine $engine = FloeEngine::adaptive,
     ) {}
 
@@ -78,11 +78,14 @@ public function load(Rows $rows, FlowContext $context): void
             if (!array_key_exists($uri, $this->writers)) {
                 $writer = new FloeWriter(
                     $context->filesystem($this->path),
-                    $this->codec,
+                    $this->schema ?? $this->inferredSchema ?? throw new FloeException(
+                        'Floe loader has no schema to write with',
+                    ),
+                    $this->options,
                     hydrator: $context->hydrator(),
                     engine: $this->engine,
                 );
-                $writer->createForStream($stream, $this->metadata, schema: $this->schema ?? $this->inferredSchema);
+                $writer->createForStream($stream, $this->metadata);
                 $this->writers[$uri] = $writer;
             }
 

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

@@ -146,8 +146,13 @@ private function mergeCompact(array $sources, array $layouts, ?Schema $merged, P
             $mergedMetadata = $mergedMetadata->merge($layout['footer']->metadata);
         }
 
-        $writer = new FloeWriter($this->filesystem, $this->codec, hydrator: $this->hydrator);
-        $writer->create($dest, $mergedMetadata->merge($metadata), schema: $merged);
+        $writer = new FloeWriter(
+            $this->filesystem,
+            $merged ?? new Schema(),
+            new Options(codec: $this->codec),
+            hydrator: $this->hydrator,
+        );
+        $writer->create($dest, $mergedMetadata->merge($metadata));
 
         $reader = new FloeReader($this->filesystem, $this->codec, hydrator: $this->hydrator);
 

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

@@ -38,14 +38,14 @@ public function __construct(
     public function serialize(Rows $rows, DestinationStream $destination): void
     {
         try {
-            $writer = new FloeStreamWriter(hydrator: $this->hydrator);
-            // multi-chunk payloads need the whole-value union upfront so later chunks with columns
-            // absent from the first still fit the session; a single chunk derives the identical
-            // union inside write() - skip the second O(rows) schema pass
-            $writer->create(
-                $destination,
-                schema: $rows->count() > $this->batchSize ? FloeStreamWriter::unionSchema($rows) : null,
+            // the whole-Rows union is the schema for every chunk, so no chunk can drift from it -
+            // per-write validation is a provable no-op here and is skipped
+            $writer = new FloeStreamWriter(
+                FloeStreamWriter::unionSchema($rows),
+                new Options(validateData: false),
+                hydrator: $this->hydrator,
             );
+            $writer->create($destination);
 
             foreach ($rows->count() === 0 ? [$rows] : $rows->chunks($this->batchSize) as $chunk) {
                 $writer->write($chunk);