From d538129d9e043dd7eb8c361d47d1a94c868b372e Mon Sep 17 00:00:00 2001 From: Norbert Orzechowicz Date: Thu, 24 Sep 2026 15:07:14 +0200 Subject: [PATCH 1/2] perf: fused native Floe read and write path - decode frames straight into Rows in one native pass - encode Rows straight into complete ROW frames natively - skip offset rows before decoding, drop reader re-batching - build datetimes from timestamps natively on PHP 8.4+ - remove AdaptiveFloeEncoder, FloeEngine picks the encoder - fall back to PHP for extensions without the new methods --- .../components/extensions/flow-php-ext.md | 10 +- documentation/installation/docker.md | 2 +- documentation/upgrading.md | 7 + .../src/Flow/ETL/Extractor/FileConstants.php | 24 ++ .../Flow/ETL/Extractor/PartitionColumns.php | 16 -- .../Optimizer/Rule/CountFromStatistics.php | 5 - .../src/Flow/ETL/Row/AdaptiveRowHydrator.php | 5 + .../src/Flow/ETL/Row/NativeRowHydrator.php | 21 +- .../etl/src/Flow/Floe/AdaptiveFloeEncoder.php | 36 --- src/core/etl/src/Flow/Floe/FloeEncoder.php | 37 +++ src/core/etl/src/Flow/Floe/FloeEngine.php | 10 +- src/core/etl/src/Flow/Floe/FloeExtractor.php | 12 +- .../etl/src/Flow/Floe/FloeStreamReader.php | 156 ++--------- .../etl/src/Flow/Floe/FloeStreamWriter.php | 39 ++- src/core/etl/src/Flow/Floe/Format.php | 14 + src/core/etl/src/Flow/Floe/FrameWriter.php | 2 +- .../etl/src/Flow/Floe/NativeFloeEncoder.php | 54 +++- src/core/etl/src/Flow/Floe/PhpFloeEncoder.php | 18 +- .../Unit/Extractor/FileConstantsTest.php | 39 +++ .../Unit/Row/AdaptiveRowHydratorTest.php | 6 + .../Tests/Unit/Row/NativeRowHydratorTest.php | 1 - .../Tests/Context/FloeStreamReaderContext.php | 69 ++++- .../Flow/Floe/Tests/Double/SpyHydrator.php | 4 + .../Flow/Floe/Tests/Mother/RowsMother.php | 28 ++ .../Tests/Unit/AdaptiveFloeEncoderTest.php | 38 --- .../Flow/Floe/Tests/Unit/FloeEngineTest.php | 8 +- .../Floe/Tests/Unit/FloeStreamReaderTest.php | 164 ++++++++++++ .../Floe/Tests/Unit/FloeStreamWriterTest.php | 143 ++++++++++ .../tests/Flow/Floe/Tests/Unit/FormatTest.php | 9 + .../Floe/Tests/Unit/NativeFloeEncoderTest.php | 124 +++++++++ .../Floe/Tests/Unit/PhpFloeEncoderTest.php | 30 +++ src/extension/flow-php-ext/Cargo.lock | 1 + src/extension/flow-php-ext/Cargo.toml | 3 + src/extension/flow-php-ext/build.rs | 21 ++ .../php/Flow/Floe/RustFloeEncoderNative.php | 26 ++ src/extension/flow-php-ext/src/cast.rs | 246 +++++++++++++----- src/extension/flow-php-ext/src/encode.rs | 49 +++- src/extension/flow-php-ext/src/format.rs | 2 + src/extension/flow-php-ext/src/hydrate.rs | 112 +++++--- src/extension/flow-php-ext/src/lib.rs | 71 ++++- src/extension/flow-php-ext/src/values.rs | 70 +++-- .../tests/phpt/024_row_hydrator_parity.phpt | 5 + .../tests/phpt/056_fused_decode_rows.phpt | 182 +++++++++++++ .../tests/phpt/057_decode_rows_no_leaks.phpt | 98 +++++++ .../058_datetime_from_timestamp_parity.phpt | 104 ++++++++ .../tests/phpt/059_encode_frames.phpt | 137 ++++++++++ .../phpt/060_encode_frames_no_leaks.phpt | 98 +++++++ 47 files changed, 1919 insertions(+), 437 deletions(-) delete mode 100644 src/core/etl/src/Flow/Floe/AdaptiveFloeEncoder.php create mode 100644 src/core/etl/src/Flow/Floe/FloeEncoder.php delete mode 100644 src/core/etl/tests/Flow/Floe/Tests/Unit/AdaptiveFloeEncoderTest.php create mode 100644 src/extension/flow-php-ext/tests/phpt/056_fused_decode_rows.phpt create mode 100644 src/extension/flow-php-ext/tests/phpt/057_decode_rows_no_leaks.phpt create mode 100644 src/extension/flow-php-ext/tests/phpt/058_datetime_from_timestamp_parity.phpt create mode 100644 src/extension/flow-php-ext/tests/phpt/059_encode_frames.phpt create mode 100644 src/extension/flow-php-ext/tests/phpt/060_encode_frames_no_leaks.phpt diff --git a/documentation/components/extensions/flow-php-ext.md b/documentation/components/extensions/flow-php-ext.md index eff26a454..ec5844203 100644 --- a/documentation/components/extensions/flow-php-ext.md +++ b/documentation/components/extensions/flow-php-ext.md @@ -58,10 +58,12 @@ implementation automatically: - **`Flow\Floe\RustFloeEncoderNative`** - the Floe ROW frame-body codec: `encode(list, schemaBody)` returns the encoded frame bodies, - `decode(list, schemaBody)` returns `list`. 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, schemaBody)` returns `list`, and + `decodeRows(list, 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 diff --git a/documentation/installation/docker.md b/documentation/installation/docker.md index 467b37d75..63286b992 100644 --- a/documentation/installation/docker.md +++ b/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 | diff --git a/documentation/upgrading.md b/documentation/upgrading.md index 3c36a0663..24ed23d67 100644 --- a/documentation/upgrading.md +++ b/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` | `FloeEngine::encoder(): Flow\Floe\FloeEncoder` (`Encoder` plus `decodeRows()` / `encodeFrames()`) | + --- ## Upgrading from 0.43.x to 0.44.x diff --git a/src/core/etl/src/Flow/ETL/Extractor/FileConstants.php b/src/core/etl/src/Flow/ETL/Extractor/FileConstants.php index bb9849301..e478dda5e 100644 --- a/src/core/etl/src/Flow/ETL/Extractor/FileConstants.php +++ b/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); + } } diff --git a/src/core/etl/src/Flow/ETL/Extractor/PartitionColumns.php b/src/core/etl/src/Flow/ETL/Extractor/PartitionColumns.php index eae15601e..3c3bc991f 100644 --- a/src/core/etl/src/Flow/ETL/Extractor/PartitionColumns.php +++ b/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 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 $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 $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; } diff --git a/src/core/etl/src/Flow/ETL/Optimizer/Rule/CountFromStatistics.php b/src/core/etl/src/Flow/ETL/Optimizer/Rule/CountFromStatistics.php index 1ab40fce6..0122ea91c 100644 --- a/src/core/etl/src/Flow/ETL/Optimizer/Rule/CountFromStatistics.php +++ b/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; diff --git a/src/core/etl/src/Flow/ETL/Row/AdaptiveRowHydrator.php b/src/core/etl/src/Flow/ETL/Row/AdaptiveRowHydrator.php index 558b52153..91017e3e5 100644 --- a/src/core/etl/src/Flow/ETL/Row/AdaptiveRowHydrator.php +++ b/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); diff --git a/src/core/etl/src/Flow/ETL/Row/NativeRowHydrator.php b/src/core/etl/src/Flow/ETL/Row/NativeRowHydrator.php index 60a354e82..e63765e13 100644 --- a/src/core/etl/src/Flow/ETL/Row/NativeRowHydrator.php +++ b/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); } } diff --git a/src/core/etl/src/Flow/Floe/AdaptiveFloeEncoder.php b/src/core/etl/src/Flow/Floe/AdaptiveFloeEncoder.php deleted file mode 100644 index e24069b76..000000000 --- a/src/core/etl/src/Flow/Floe/AdaptiveFloeEncoder.php +++ /dev/null @@ -1,36 +0,0 @@ - - */ -final class AdaptiveFloeEncoder implements Encoder -{ - /** - * @var Encoder - */ - private readonly Encoder $delegate; - - public function __construct(Schema $schema) - { - $this->delegate = NativeFloeEncoder::isSupported() - ? new NativeFloeEncoder($schema) - : new PhpFloeEncoder($schema); - } - - public function decode(array $batch): array - { - return $this->delegate->decode($batch); - } - - public function encode(array $batch): array - { - return $this->delegate->encode($batch); - } -} diff --git a/src/core/etl/src/Flow/Floe/FloeEncoder.php b/src/core/etl/src/Flow/Floe/FloeEncoder.php new file mode 100644 index 000000000..01fcea9d2 --- /dev/null +++ b/src/core/etl/src/Flow/Floe/FloeEncoder.php @@ -0,0 +1,37 @@ + + */ +interface FloeEncoder extends Encoder +{ + /** + * `$hydrator->hydrate($this->decode($bodies), $schema)`, in one native pass where the hydrator is native too. + * + * @param list $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; +} diff --git a/src/core/etl/src/Flow/Floe/FloeEngine.php b/src/core/etl/src/Flow/Floe/FloeEngine.php index 66415c1cf..15e45f0a4 100644 --- a/src/core/etl/src/Flow/Floe/FloeEngine.php +++ b/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 - */ - 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), }; diff --git a/src/core/etl/src/Flow/Floe/FloeExtractor.php b/src/core/etl/src/Flow/Floe/FloeExtractor.php index 2de58e36e..27b8b2a9d 100644 --- a/src/core/etl/src/Flow/Floe/FloeExtractor.php +++ b/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); diff --git a/src/core/etl/src/Flow/Floe/FloeStreamReader.php b/src/core/etl/src/Flow/Floe/FloeStreamReader.php index ab3b818e9..18bfe2bde 100644 --- a/src/core/etl/src/Flow/Floe/FloeStreamReader.php +++ b/src/core/etl/src/Flow/Floe/FloeStreamReader.php @@ -4,14 +4,13 @@ namespace Flow\Floe; -use Flow\ETL\Row; use Flow\ETL\Row\AdaptiveRowHydrator; -use Flow\ETL\Row\Encoder; use Flow\ETL\Row\Hydrator; use Flow\ETL\Rows; use Flow\ETL\Schema; use Flow\ETL\Schema\Metadata; use Flow\Filesystem\SourceStream; +use Flow\Floe\Codec\NoopCodec; use Flow\Floe\Exception\ExtensionException; use Flow\Floe\Exception\FloeException; use Flow\Serializer\Exception\SerializationException; @@ -19,6 +18,7 @@ use function count; use function max; +use function min; use function ord; use function sprintf; use function strlen; @@ -27,10 +27,7 @@ final class FloeStreamReader { - /** - * @var null|Encoder - */ - private ?Encoder $encoder = null; + private ?FloeEncoder $encoder = null; private ?Footer $footer = null; @@ -59,10 +56,7 @@ public function close(): void $this->source->close(); } - /** - * @return Encoder - */ - private function encoder(Schema $schema): Encoder + private function encoder(Schema $schema): FloeEncoder { return $this->encoder ??= $this->engine->encoder($schema); } @@ -92,7 +86,7 @@ public function metadata(): Metadata * * @throws FloeException * - * @return \Generator every batch matches the merged file schema - Rows::__construct sees to that + * @return \Generator every batch carries the merged file schema */ public function rows(int $batchSize = 1000, int $offset = 0, ?int $limit = null): Generator { @@ -198,12 +192,11 @@ private function walk( ): Generator { $fill = FrameReader::chunkFiller($buffer, $position, $chunks); - $batch = []; $yielded = 0; /** @var list $pending */ $pending = []; - $stop = false; $flushThreshold = $limit !== null && $limit < $batchSize ? $limit : $batchSize; + $transforms = !$this->codec instanceof NoopCodec; try { while (true) { @@ -226,26 +219,26 @@ private function walk( $frameEnd = $position + $frameLength; if ($frameType === Format::FRAME_ROW) { - $pending[] = $this->codec->decode(substr($buffer, $position, $frameLength)); + if ($skip > 0) { + $skip--; + } else { + $body = substr($buffer, $position, $frameLength); + $pending[] = $transforms ? $this->codec->decode($body) : $body; + } + $position = $frameEnd; if (count($pending) === $flushThreshold) { - foreach ($this->emitBatch( - $schema, - $pending, - $batch, - $batchSize, - $limit, - $yielded, - $stop, - $skip, - ) as $ready) { - yield $ready; - } + yield $this->decode($schema, $pending); + $yielded += $flushThreshold; $pending = []; - if ($stop) { - return; + if ($limit !== null) { + if ($yielded === $limit) { + return; + } + + $flushThreshold = min($batchSize, $limit - $yielded); } } } elseif ($frameType === Format::FRAME_FOOTER) { @@ -261,108 +254,23 @@ private function walk( } if ($pending !== []) { - foreach ($this->emitBatch( - $schema, - $pending, - $batch, - $batchSize, - $limit, - $yielded, - $stop, - $skip, - ) as $ready) { - yield $ready; - } - - if ($stop) { - return; - } + yield $this->decode($schema, $pending); } } catch (SerializationException|ExtensionException $e) { throw new FloeException($e->getMessage(), 0, $e); } - - if ($batch !== []) { - yield $this->batch($batch); - } } /** - * Applies the per-row skip / batch-yield / limit logic to a hydrated - * batch of row frame bodies; $batch, $yielded, $stop and $skip are updated by reference. + * A batch carries the file schema even when the hydrator folded per-value metadata into a schema of its own. * * @param list $pending - * @param array $batch - * @param int<1, max> $batchSize - * - * @return array completed batches ready to yield - */ - private function emitBatch( - Schema $schema, - array $pending, - array &$batch, - int $batchSize, - ?int $limit, - int &$yielded, - bool &$stop, - int &$skip, - ): array { - return $this->emitRows( - $this->hydrator->hydrate($this->encoder($schema)->decode($pending), $schema)->all(), - $batch, - $batchSize, - $limit, - $yielded, - $stop, - $skip, - ); - } - - /** - * Per-row skip / batch-yield / limit logic over already-hydrated rows; - * $batch, $yielded, $stop and $skip are updated by reference. - * - * @param array $rows - * @param array $batch - * @param int<1, max> $batchSize - * - * @return array completed batches ready to yield */ - private function emitRows( - array $rows, - array &$batch, - int $batchSize, - ?int $limit, - int &$yielded, - bool &$stop, - int &$skip, - ): array { - $ready = []; - - foreach ($rows as $row) { - if ($skip > 0) { - $skip--; - - continue; - } - - $batch[] = $row; - - if ($limit !== null && ++$yielded >= $limit) { - $ready[] = $this->batch($batch); - $batch = []; - $stop = true; - - return $ready; - } - - if (count($batch) === $batchSize) { - $ready[] = $this->batch($batch); - $batch = []; - } - } + private function decode(Schema $schema, array $pending): Rows + { + $rows = $this->encoder($schema)->decodeRows($pending, $schema, $this->hydrator); - return $ready; + return $rows->schema() === $schema ? $rows : Rows::trusted($schema, $rows->all()); } /** @@ -391,16 +299,6 @@ public function totalRows(): int return $this->footer()->statistics->rows; } - /** - * @param array $rows - */ - private function batch(array $rows): Rows - { - // every row comes out of the hydrator, which already conformed it to the file schema - padding and the - // NOT NULL check included - return Rows::trusted($this->schema(), $rows); - } - /** * @return \Generator */ diff --git a/src/core/etl/src/Flow/Floe/FloeStreamWriter.php b/src/core/etl/src/Flow/Floe/FloeStreamWriter.php index c28a22cd1..b82b5975a 100644 --- a/src/core/etl/src/Flow/Floe/FloeStreamWriter.php +++ b/src/core/etl/src/Flow/Floe/FloeStreamWriter.php @@ -6,13 +6,13 @@ use Composer\InstalledVersions; use Flow\ETL\Row\AdaptiveRowHydrator; -use Flow\ETL\Row\Encoder; use Flow\ETL\Row\Hydrator; use Flow\ETL\Rows; use Flow\ETL\Schema; use Flow\ETL\Schema\Metadata; use Flow\ETL\Schema\Validator\EvolvingValidator; use Flow\Filesystem\DestinationStream; +use Flow\Floe\Codec\NoopCodec; use Flow\Floe\Exception\FloeException; use Flow\Floe\Exception\IncompatibleSchemaException; @@ -51,10 +51,7 @@ final class FloeStreamWriter private Schema $sessionSchema; - /** - * @var null|Encoder - */ - private ?Encoder $sessionEncoder = null; + private ?FloeEncoder $sessionEncoder = null; private ?FrameWriter $frameWriter = null; @@ -160,14 +157,22 @@ public function write(Rows $rows): void $this->openSession(); $this->assertBatchFitsSession($rows->schema()); - $typed = $this->hydrator->dehydrate( - $rows->schema()->isSame($this->sessionSchema) ? $rows : $rows->matchTo($this->sessionSchema), - ); + $matched = $rows->schema()->isSame($this->sessionSchema) ? $rows : $rows->matchTo($this->sessionSchema); - if (!$this->sectionOpen || $this->sectionRowCount >= self::SECTION_MAX_ROWS) { - $this->startSection(); + if ($this->options->codec instanceof NoopCodec) { + $frames = $this->sessionEncoder()->encodeFrames($matched, $this->hydrator); + + $this->startSectionWhenDue(); + $this->frameWriter()->raw($frames); + $this->sectionRowCount += $matched->count(); + $this->totalRows += $matched->count(); + + return; } + $typed = $this->hydrator->dehydrate($matched); + + $this->startSectionWhenDue(); $this->emitBatch($typed); } @@ -255,6 +260,16 @@ private function guardOpen(): void } } + /** + * @throws FloeException + */ + private function startSectionWhenDue(): void + { + if (!$this->sectionOpen || $this->sectionRowCount >= self::SECTION_MAX_ROWS) { + $this->startSection(); + } + } + /** * @throws FloeException */ @@ -269,10 +284,8 @@ private function startSection(): void /** * @throws FloeException - * - * @return Encoder */ - private function sessionEncoder(): Encoder + private function sessionEncoder(): FloeEncoder { return $this->sessionEncoder ?? throw new FloeException('Floe writer has no active session encoder'); } diff --git a/src/core/etl/src/Flow/Floe/Format.php b/src/core/etl/src/Flow/Floe/Format.php index de5afaf8d..bf4c32517 100644 --- a/src/core/etl/src/Flow/Floe/Format.php +++ b/src/core/etl/src/Flow/Floe/Format.php @@ -62,6 +62,20 @@ public static function frame(int $type, string $body): string return chr($type) . pack('V', strlen($body)) . $body; } + /** + * @param list $bodies + */ + public static function rowFrames(array $bodies): string + { + $frames = ''; + + foreach ($bodies as $body) { + $frames .= self::frame(self::FRAME_ROW, $body); + } + + return $frames; + } + /** * @throws FloeException */ diff --git a/src/core/etl/src/Flow/Floe/FrameWriter.php b/src/core/etl/src/Flow/Floe/FrameWriter.php index d44cc0a52..b47824cc5 100644 --- a/src/core/etl/src/Flow/Floe/FrameWriter.php +++ b/src/core/etl/src/Flow/Floe/FrameWriter.php @@ -45,7 +45,7 @@ public function footer(string $footerJson): void } /** - * Verbatim byte passthrough for the mergeSplice fast path (no re-encode). + * Complete frames the caller already built, appended as they are. */ public function raw(string $bytes): void { diff --git a/src/core/etl/src/Flow/Floe/NativeFloeEncoder.php b/src/core/etl/src/Flow/Floe/NativeFloeEncoder.php index da7a60f8d..171afbfd8 100644 --- a/src/core/etl/src/Flow/Floe/NativeFloeEncoder.php +++ b/src/core/etl/src/Flow/Floe/NativeFloeEncoder.php @@ -4,7 +4,10 @@ namespace Flow\Floe; -use Flow\ETL\Row\Encoder; +use Flow\ETL\Row\AdaptiveRowHydrator; +use Flow\ETL\Row\Hydrator; +use Flow\ETL\Row\NativeRowHydrator; +use Flow\ETL\Rows; use Flow\ETL\Schema; use Flow\Floe\Exception\ExtensionException; use Flow\Floe\Exception\FloeException; @@ -14,13 +17,11 @@ use function class_exists; use function extension_loaded; use function json_encode; +use function method_exists; use const JSON_THROW_ON_ERROR; -/** - * @implements Encoder - */ -final class NativeFloeEncoder implements Encoder +final class NativeFloeEncoder implements FloeEncoder { private readonly RustFloeEncoderNative $native; @@ -38,7 +39,13 @@ public function __construct( public static function isSupported(): bool { - return extension_loaded('flow_php') && class_exists(RustFloeEncoderNative::class, false); + // an extension older than this library lacks decodeRows()/encodeFrames() - it falls back to the PHP engine instead + return ( + extension_loaded('flow_php') + && class_exists(RustFloeEncoderNative::class, false) + && method_exists(RustFloeEncoderNative::class, 'decodeRows') + && method_exists(RustFloeEncoderNative::class, 'encodeFrames') + ); } public function decode(array $batch): array @@ -50,6 +57,32 @@ public function decode(array $batch): array } } + public function decodeRows(array $bodies, Schema $schema, Hydrator $hydrator): Rows + { + if (!self::isNativeHydrator($hydrator)) { + return $hydrator->hydrate($this->decode($bodies), $schema); + } + + try { + return $this->native->decodeRows($bodies, $this->schemaBody(), $schema); + } catch (ExtensionException $e) { + throw new FloeException($e->getMessage(), 0, $e); + } + } + + public function encodeFrames(Rows $rows, Hydrator $hydrator): string + { + if (!self::isNativeHydrator($hydrator)) { + return Format::rowFrames($this->encode($hydrator->dehydrate($rows))); + } + + try { + return $this->native->encodeFrames($rows, $this->schemaBody(), $this->schema); + } catch (ExtensionException $e) { + throw new FloeException($e->getMessage(), 0, $e); + } + } + public function encode(array $batch): array { try { @@ -59,6 +92,15 @@ public function encode(array $batch): array } } + private static function isNativeHydrator(Hydrator $hydrator): bool + { + return ( + $hydrator instanceof NativeRowHydrator + || $hydrator instanceof AdaptiveRowHydrator + && $hydrator->isNative() + ); + } + private function schemaBody(): string { if ($this->schemaBody !== null) { diff --git a/src/core/etl/src/Flow/Floe/PhpFloeEncoder.php b/src/core/etl/src/Flow/Floe/PhpFloeEncoder.php index a0704d5a6..94bf44aed 100644 --- a/src/core/etl/src/Flow/Floe/PhpFloeEncoder.php +++ b/src/core/etl/src/Flow/Floe/PhpFloeEncoder.php @@ -6,8 +6,9 @@ use Flow\ETL\Exception\ColumnMismatchException; use Flow\ETL\Exception\SchemaMismatchException; -use Flow\ETL\Row\Encoder; +use Flow\ETL\Row\Hydrator; use Flow\ETL\Row\RawRowValues; +use Flow\ETL\Rows; use Flow\ETL\Schema; use Flow\ETL\Schema\Metadata; use Flow\Floe\Exception\FloeException; @@ -21,10 +22,7 @@ use const JSON_THROW_ON_ERROR; -/** - * @implements Encoder - */ -final class PhpFloeEncoder implements Encoder +final class PhpFloeEncoder implements FloeEncoder { /** * @var null|array @@ -85,6 +83,16 @@ public function decode(array $batch): array return $decoded; } + public function decodeRows(array $bodies, Schema $schema, Hydrator $hydrator): Rows + { + return $hydrator->hydrate($this->decode($bodies), $schema); + } + + public function encodeFrames(Rows $rows, Hydrator $hydrator): string + { + return Format::rowFrames($this->encode($hydrator->dehydrate($rows))); + } + public function encode(array $batch): array { $encoders = $this->encoders ??= $this->buildEncoders(); diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/FileConstantsTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/FileConstantsTest.php index 0babed67b..4b204a152 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/FileConstantsTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/FileConstantsTest.php @@ -8,6 +8,11 @@ use Flow\ETL\Extractor\PartitionColumns; use Flow\ETL\Tests\FlowTestCase; +use function Flow\ETL\DSL\int_schema; +use function Flow\ETL\DSL\row; +use function Flow\ETL\DSL\rows; +use function Flow\ETL\DSL\schema; +use function Flow\ETL\DSL\str_schema; use function Flow\Filesystem\DSL\memory_filesystem; final class FileConstantsTest extends FlowTestCase @@ -53,4 +58,38 @@ public function test_the_uri_joins_the_body_when_metadata_columns_are_on(): void ]), ); } + + public function test_fill_rows_returns_a_batch_with_nothing_to_add_as_it_is(): void + { + $rows = rows(schema(str_schema('name')), row(['name' => 'Norbert'])); + + static::assertSame($rows, (new FileConstants( + new PartitionColumns(memory_filesystem()), + null, + [], + [], + ))->fillRows($rows, $rows->schema())); + } + + public function test_fill_rows_adds_the_constants_to_every_row_under_the_declared_schema(): void + { + $declared = schema(str_schema('name'), str_schema('_input_file_uri'), int_schema('year')); + + static::assertEquals( + rows( + $declared, + row(['name' => 'Norbert', '_input_file_uri' => 'memory://orders/data.csv', 'year' => 2024]), + row(['name' => 'Flow', '_input_file_uri' => 'memory://orders/data.csv', 'year' => 2024]), + ), + (new FileConstants( + new PartitionColumns(memory_filesystem()), + 'memory://orders/data.csv', + ['year' => false], + ['year' => 2024], + ))->fillRows( + rows(schema(str_schema('name')), row(['name' => 'Norbert']), row(['name' => 'Flow'])), + $declared, + ), + ); + } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/AdaptiveRowHydratorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/AdaptiveRowHydratorTest.php index 2c3209112..07bb3601d 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/AdaptiveRowHydratorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/AdaptiveRowHydratorTest.php @@ -6,6 +6,7 @@ use Flow\ETL\Exception\SchemaMismatchException; use Flow\ETL\Row\AdaptiveRowHydrator; +use Flow\ETL\Row\NativeRowHydrator; use Flow\ETL\Row\RawRowValues; use Flow\ETL\Tests\FlowTestCase; @@ -53,4 +54,9 @@ public function test_hydrate_throws_on_missing_required_structure_element(): voi 'id' => 1, ]])], schema(structure_schema('data', type_structure(['id' => type_integer(), 'name' => type_string()])))); } + + public function test_is_native_when_the_native_hydrator_is_supported(): void + { + static::assertSame(NativeRowHydrator::isSupported(), (new AdaptiveRowHydrator())->isNative()); + } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/NativeRowHydratorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/NativeRowHydratorTest.php index 2202d90aa..240d1fb1f 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/NativeRowHydratorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/NativeRowHydratorTest.php @@ -364,7 +364,6 @@ public static function refusing_datasets(): Generator /** @var UnionType $unmatchable */ $unmatchable = type_union(type_uuid(), type_datetime()); - // the row gate refuses this one, so it is the case that reaches NativeRowHydrator::unwrap() yield 'null in a not null column' => [ schema(int_schema('id'), str_schema('name')), [new RawRowValues(['id' => 1, 'name' => null])], diff --git a/src/core/etl/tests/Flow/Floe/Tests/Context/FloeStreamReaderContext.php b/src/core/etl/tests/Flow/Floe/Tests/Context/FloeStreamReaderContext.php index 77e61f9e3..188c81e6a 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Context/FloeStreamReaderContext.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Context/FloeStreamReaderContext.php @@ -4,18 +4,26 @@ namespace Flow\Floe\Tests\Context; +use Flow\ETL\Row\Hydrator; use Flow\ETL\Rows; +use Flow\ETL\Schema; use Flow\ETL\Schema\Metadata; use Flow\Filesystem\Filesystem; use Flow\Filesystem\Path; +use Flow\Floe\Codec; use Flow\Floe\Codec\NoopCodec; +use Flow\Floe\FloeEngine; use Flow\Floe\FloeReader; use Flow\Floe\FloeWriter; use Flow\Floe\Footer; use Flow\Floe\FooterReader; use Flow\Floe\Format; use Flow\Floe\FrameReader; +use Flow\Floe\Options; +use Flow\Floe\Section; +use Flow\Floe\Statistics; +use function count; use function Flow\ETL\DSL\rows; use function Flow\ETL\DSL\schema; use function Flow\Filesystem\DSL\memory_filesystem; @@ -132,14 +140,69 @@ public static function readAll(Filesystem $filesystem, Path $path): Rows /** * @param array $metadata */ - public static function write(Filesystem $filesystem, Path $path, Rows $rows, array $metadata = []): void - { - $writer = new FloeWriter($filesystem, $rows->schema()); + public static function write( + Filesystem $filesystem, + Path $path, + Rows $rows, + array $metadata = [], + Codec $codec = new NoopCodec(), + ): void { + $writer = new FloeWriter($filesystem, $rows->schema(), new Options(codec: $codec)); $writer->create($path, Metadata::fromArray($metadata)); $writer->write($rows); $writer->close(); } + /** + * A file holding exactly these ROW frame bodies under `$schema` - frames no FloeWriter would produce. + * + * @param list $bodies + */ + public static function writeFrames(Filesystem $filesystem, Path $path, Schema $schema, array $bodies): void + { + /** @var array> $normalized */ + $normalized = $schema->normalize(); + $footerJson = (new Footer( + Format::VERSION, + 'test', + $normalized, + [new Section(Format::HEADER_LENGTH, count($bodies))], + new Statistics(count($bodies), strlen(Format::rowFrames($bodies))), + Metadata::empty(), + ))->toJson(); + + $stream = $filesystem->writeTo($path); + $stream->append( + Format::header(0x00) . Format::rowFrames($bodies) + . Format::frame(Format::FRAME_FOOTER, $footerJson . Format::trailer(strlen($footerJson))), + ); + $stream->close(); + } + + /** + * @param list $batches + */ + public static function writeBatches( + Filesystem $filesystem, + Path $path, + Schema $schema, + array $batches, + Options $options = new Options(), + ?Hydrator $hydrator = null, + FloeEngine $engine = FloeEngine::adaptive, + ): string { + $writer = new FloeWriter($filesystem, $schema, $options, $hydrator, $engine); + $writer->create($path); + + foreach ($batches as $batch) { + $writer->write($batch); + } + + $writer->close(); + + return $filesystem->readFrom($path)->content(); + } + /** * Writes a complete file then rewrites it with the FOOTER frame stripped - * a crashed writer that flushed complete frames but never closed. Strict diff --git a/src/core/etl/tests/Flow/Floe/Tests/Double/SpyHydrator.php b/src/core/etl/tests/Flow/Floe/Tests/Double/SpyHydrator.php index d5851b048..5bf885e8c 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Double/SpyHydrator.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Double/SpyHydrator.php @@ -13,6 +13,8 @@ final class SpyHydrator implements Hydrator { public int $dehydrateCalls = 0; + public int $hydrateCalls = 0; + public function dehydrate(Rows $rows): array { $this->dehydrateCalls++; @@ -22,6 +24,8 @@ public function dehydrate(Rows $rows): array public function hydrate(array $batch, Schema $schema): Rows { + $this->hydrateCalls++; + return (new AdaptiveRowHydrator())->hydrate($batch, $schema); } } diff --git a/src/core/etl/tests/Flow/Floe/Tests/Mother/RowsMother.php b/src/core/etl/tests/Flow/Floe/Tests/Mother/RowsMother.php index 626ff37c6..4faa7ee21 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Mother/RowsMother.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Mother/RowsMother.php @@ -12,6 +12,7 @@ use Flow\ETL\Tests\Fixtures\Enum\BackedStringEnum; use Flow\ETL\Tests\Fixtures\Enum\BasicEnum; +use function array_map; use function Flow\ETL\DSL\bool_schema; use function Flow\ETL\DSL\date_schema; use function Flow\ETL\DSL\datetime_schema; @@ -41,6 +42,7 @@ use function Flow\Types\DSL\type_uuid; use function Flow\Types\DSL\type_xml; use function Flow\Types\DSL\type_xml_element; +use function range; use const PHP_INT_MAX; use const PHP_INT_MIN; @@ -52,6 +54,32 @@ public static function empty(): Rows return rows(schema()); } + public static function ids(int $from, int $to): Rows + { + return rows( + schema(int_schema('id')), + ...array_map(static fn(int $id) => row(['id' => $id]), range($from, $to)), + ); + } + + /** + * Ids 1..$count, every fourth name null, one datetime an hour apart per row. + */ + public static function numbered(int $count): Rows + { + return rows( + schema(int_schema('id'), str_schema('name', nullable: true), datetime_schema('at')), + ...array_map( + static fn(int $id) => row([ + 'id' => $id, + 'name' => ($id % 4) === 0 ? null : 'user_' . $id, + 'at' => (new DateTimeImmutable('2026-01-01 00:00:00.000001 +00:00'))->modify("+{$id} hours"), + ]), + range(1, $count), + ), + ); + } + public static function heterogeneous(): Rows { return rows( diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/AdaptiveFloeEncoderTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/AdaptiveFloeEncoderTest.php deleted file mode 100644 index d6e56d5cf..000000000 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/AdaptiveFloeEncoderTest.php +++ /dev/null @@ -1,38 +0,0 @@ - 1, 'name' => 'flow']), - row(['id' => 2, 'name' => null]), - ); - $encoder = new AdaptiveFloeEncoder(schema_from_json(FloeSchemaContext::schemaBody($data->schema()))); - - $decoded = $encoder->decode($encoder->encode((new PhpRowHydrator())->dehydrate($data))); - - static::assertEquals( - [new RawRowValues(['id' => 1, 'name' => 'flow']), new RawRowValues(['id' => 2, 'name' => null])], - $decoded, - ); - } -} diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeEngineTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeEngineTest.php index 0c84ebca1..fe6e249af 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeEngineTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeEngineTest.php @@ -5,7 +5,6 @@ namespace Flow\Floe\Tests\Unit; use Flow\ETL\Schema; -use Flow\Floe\AdaptiveFloeEncoder; use Flow\Floe\FloeEngine; use Flow\Floe\NativeFloeEncoder; use Flow\Floe\PhpFloeEncoder; @@ -19,9 +18,12 @@ final class FloeEngineTest extends TestCase { - public function test_adaptive_engine_builds_adaptive_encoder(): void + public function test_adaptive_engine_builds_the_native_encoder_when_supported_else_the_php_one(): void { - static::assertInstanceOf(AdaptiveFloeEncoder::class, FloeEngine::adaptive->encoder($this->schema())); + static::assertInstanceOf( + NativeFloeEncoder::isSupported() ? NativeFloeEncoder::class : PhpFloeEncoder::class, + FloeEngine::adaptive->encoder($this->schema()), + ); } public function test_native_engine_builds_native_encoder_when_supported(): void diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeStreamReaderTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeStreamReaderTest.php index 48fd76357..66a178f70 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeStreamReaderTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeStreamReaderTest.php @@ -4,10 +4,26 @@ namespace Flow\Floe\Tests\Unit; +use Closure; +use Flow\ETL\Row\Hydrator; +use Flow\ETL\Row\PhpRowHydrator; +use Flow\ETL\Row\TypedRowValues; +use Flow\ETL\Rows; +use Flow\ETL\Schema\Metadata; use Flow\Floe\Codec\NoopCodec; +use Flow\Floe\FloeEngine; +use Flow\Floe\FloeReader; use Flow\Floe\FloeStreamReader; use Flow\Floe\FloeWriter; +use Flow\Floe\NativeFloeEncoder; +use Flow\Floe\PhpFloeEncoder; +use Flow\Floe\Tests\Context\FloeStreamReaderContext; use Flow\Floe\Tests\Double\ClosingSpySourceStream; +use Flow\Floe\Tests\Double\PrefixingCodecStub; +use Flow\Floe\Tests\Double\SpyHydrator; +use Flow\Floe\Tests\Mother\RowsMother; +use Generator; +use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\TestCase; use function Flow\ETL\DSL\int_schema; @@ -16,9 +32,55 @@ use function Flow\ETL\DSL\schema; use function Flow\Filesystem\DSL\memory_filesystem; use function Flow\Filesystem\DSL\path; +use function Flow\Types\DSL\type_integer; +use function iterator_to_array; +use function range; +use function serialize; final class FloeStreamReaderTest extends TestCase { + /** + * @return array>}> + */ + public static function reads(): array + { + return [ + 'every row' => [ + static fn(FloeStreamReader $reader): Generator => $reader->rows(10), + [range(1, 10), range(11, 20), range(21, 25)], + ], + 'batch larger than the file' => [ + static fn(FloeStreamReader $reader): Generator => $reader->rows(100), + [range(1, 25)], + ], + 'limit below the batch size' => [ + static fn(FloeStreamReader $reader): Generator => $reader->rows(10, 0, 3), + [range(1, 3)], + ], + 'limit on a batch boundary' => [ + static fn(FloeStreamReader $reader): Generator => $reader->rows(10, 0, 20), + [range(1, 10), range(11, 20)], + ], + 'limit inside a batch' => [ + static fn(FloeStreamReader $reader): Generator => $reader->rows(10, 0, 15), + [range(1, 10), range(11, 15)], + ], + 'offset' => [ + static fn(FloeStreamReader $reader): Generator => $reader->rows(10, 4), + [range(5, 14), range(15, 24), [25]], + ], + 'offset and limit' => [ + static fn(FloeStreamReader $reader): Generator => $reader->rows(10, 4, 12), + [range(5, 14), [15, 16]], + ], + 'head' => [static fn(FloeStreamReader $reader): Generator => $reader->head(7, 5), [range(1, 5), [6, 7]]], + 'tail' => [ + static fn(FloeStreamReader $reader): Generator => $reader->tail(7, 5), + [range(19, 23), [24, 25]], + ], + ]; + } + public function test_close_closes_the_source_stream(): void { $filesystem = memory_filesystem(); @@ -36,4 +98,106 @@ public function test_close_closes_the_source_stream(): void static::assertSame(1, $source->closeCount); } + + /** + * @param Closure(FloeStreamReader): Generator $read + * @param list> $batches + */ + #[DataProvider('reads')] + public function test_fused_read_yields_the_batches_of_the_two_step_read(Closure $read, array $batches): void + { + if (!NativeFloeEncoder::isSupported()) { + static::markTestSkipped('flow_php extension with RustFloeEncoderNative is not loaded'); + } + + $filesystem = memory_filesystem(); + $path = path('memory://fused.floe'); + FloeStreamReaderContext::write($filesystem, $path, RowsMother::numbered(25)); + $spy = new SpyHydrator(); + + $twoStep = iterator_to_array($read((new FloeReader($filesystem, hydrator: $spy))->read($path))); + $fused = iterator_to_array($read((new FloeReader($filesystem))->read($path))); + + static::assertSame($batches, array_map(static fn(Rows $rows): array => $rows->reduceToArray('id'), $fused)); + static::assertEquals($twoStep, $fused); + static::assertGreaterThan(0, $spy->hydrateCalls); + } + + /** + * @param Closure(FloeStreamReader): Generator $read + * @param list> $batches + */ + #[DataProvider('reads')] + public function test_native_engine_reads_a_transforming_codec_like_the_php_engine( + Closure $read, + array $batches, + ): void { + if (!NativeFloeEncoder::isSupported()) { + static::markTestSkipped('flow_php extension with RustFloeEncoderNative is not loaded'); + } + + $filesystem = memory_filesystem(); + $codec = new PrefixingCodecStub(); + $path = path('memory://codec.floe'); + FloeStreamReaderContext::write($filesystem, $path, RowsMother::numbered(25), codec: $codec); + + $php = iterator_to_array($read((new FloeReader($filesystem, $codec, engine: FloeEngine::php))->read($path))); + $native = iterator_to_array($read((new FloeReader($filesystem, $codec, engine: FloeEngine::native))->read( + $path, + ))); + + static::assertEquals($php, $native); + static::assertSame($batches, array_map(static fn(Rows $rows): array => $rows->reduceToArray('id'), $native)); + } + + /** + * @param Closure(FloeStreamReader): Generator $read + * @param list> $batches + */ + #[DataProvider('reads')] + public function test_native_engine_reads_like_the_php_engine(Closure $read, array $batches): void + { + if (!NativeFloeEncoder::isSupported()) { + static::markTestSkipped('flow_php extension with RustFloeEncoderNative is not loaded'); + } + + $filesystem = memory_filesystem(); + $path = path('memory://engines.floe'); + FloeStreamReaderContext::write($filesystem, $path, RowsMother::numbered(25)); + + $php = iterator_to_array($read((new FloeReader($filesystem, engine: FloeEngine::php))->read($path))); + $native = iterator_to_array($read((new FloeReader($filesystem, engine: FloeEngine::native))->read($path))); + + static::assertSame(serialize($php), serialize($native)); + static::assertSame($batches, array_map(static fn(Rows $rows): array => $rows->reduceToArray('id'), $native)); + } + + /** + * @return array + */ + public static function hydrators(): array + { + return ['default' => [null], 'php' => [new PhpRowHydrator()]]; + } + + #[DataProvider('hydrators')] + public function test_batches_carry_the_file_schema_when_frames_carry_per_value_metadata(?Hydrator $hydrator): void + { + $filesystem = memory_filesystem(); + $path = path('memory://per-value-metadata.floe'); + $schema = schema(int_schema('id')); + FloeStreamReaderContext::writeFrames($filesystem, $path, $schema, (new PhpFloeEncoder($schema))->encode([ + new TypedRowValues(['id' => 1], ['id' => type_integer()], ['id' => Metadata::fromArray(['k' => 'v'])]), + ])); + + $batches = iterator_to_array( + (new FloeReader($filesystem, hydrator: $hydrator)) + ->read($path) + ->rows(), + ); + + static::assertCount(1, $batches); + static::assertSame($schema->normalize(), $batches[0]->schema()->normalize()); + static::assertSame([1], $batches[0]->reduceToArray('id')); + } } diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeStreamWriterTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeStreamWriterTest.php index 5b5f9eee3..d3f7986bd 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeStreamWriterTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeStreamWriterTest.php @@ -4,17 +4,24 @@ namespace Flow\Floe\Tests\Unit; +use DateTimeImmutable; use Flow\ETL\Schema\Metadata; use Flow\Floe\Exception\FloeException; use Flow\Floe\Exception\IncompatibleSchemaException; +use Flow\Floe\FloeEngine; use Flow\Floe\FloeStreamWriter; use Flow\Floe\FloeWriter; use Flow\Floe\Format; +use Flow\Floe\NativeFloeEncoder; use Flow\Floe\Options; use Flow\Floe\Tests\Context\FloeStreamReaderContext; use Flow\Floe\Tests\Double\CodecStub; +use Flow\Floe\Tests\Double\PrefixingCodecStub; +use Flow\Floe\Tests\Double\SpyHydrator; +use Flow\Floe\Tests\Mother\RowsMother; use PHPUnit\Framework\TestCase; +use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\int_schema; use function Flow\ETL\DSL\row; use function Flow\ETL\DSL\rows; @@ -357,4 +364,140 @@ public function test_column_absent_from_the_session_schema_throws(): void $writer->write(rows(schema(int_schema('a'), int_schema('b')), row(['a' => 1, 'b' => 2]))); } + + public function test_fused_write_is_byte_identical_to_the_two_step_write_across_sections(): void + { + if (!NativeFloeEncoder::isSupported()) { + static::markTestSkipped('flow_php extension with RustFloeEncoderNative is not loaded'); + } + + $filesystem = memory_filesystem(); + $schema = schema(int_schema('id')); + $batches = [RowsMother::ids(1, 100_000), RowsMother::ids(100_001, 100_002)]; + $spy = new SpyHydrator(); + + $twoStep = FloeStreamReaderContext::writeBatches( + $filesystem, + path('memory://two-step.floe'), + $schema, + $batches, + hydrator: $spy, + ); + $fused = FloeStreamReaderContext::writeBatches($filesystem, path('memory://fused.floe'), $schema, $batches); + + static::assertSame($twoStep, $fused); + static::assertSame(2, $spy->dehydrateCalls); + static::assertCount(2, FloeStreamReaderContext::footer($filesystem, path('memory://fused.floe'))->sections); + } + + public function test_fused_write_is_byte_identical_to_the_two_step_write_around_an_empty_batch(): void + { + if (!NativeFloeEncoder::isSupported()) { + static::markTestSkipped('flow_php extension with RustFloeEncoderNative is not loaded'); + } + + $filesystem = memory_filesystem(); + $numbered = RowsMother::numbered(5); + $batches = [rows($numbered->schema()), $numbered, rows($numbered->schema())]; + + static::assertSame( + FloeStreamReaderContext::writeBatches( + $filesystem, + path('memory://two-step.floe'), + $numbered->schema(), + $batches, + hydrator: new SpyHydrator(), + ), + FloeStreamReaderContext::writeBatches( + $filesystem, + path('memory://fused.floe'), + $numbered->schema(), + $batches, + ), + ); + } + + public function test_fused_write_is_byte_identical_to_the_two_step_write_for_a_batch_that_needs_match_to(): void + { + if (!NativeFloeEncoder::isSupported()) { + static::markTestSkipped('flow_php extension with RustFloeEncoderNative is not loaded'); + } + + $filesystem = memory_filesystem(); + $session = RowsMother::numbered(1)->schema(); + $batches = [rows( + schema(datetime_schema('at'), int_schema('id')), + row(['at' => new DateTimeImmutable('2026-01-01 00:00:00'), 'id' => 1]), + )]; + + static::assertSame( + FloeStreamReaderContext::writeBatches( + $filesystem, + path('memory://two-step.floe'), + $session, + $batches, + hydrator: new SpyHydrator(), + ), + FloeStreamReaderContext::writeBatches($filesystem, path('memory://fused.floe'), $session, $batches), + ); + } + + public function test_native_engine_writes_a_transforming_codec_like_the_php_engine(): void + { + if (!NativeFloeEncoder::isSupported()) { + static::markTestSkipped('flow_php extension with RustFloeEncoderNative is not loaded'); + } + + $filesystem = memory_filesystem(); + $options = new Options(codec: new PrefixingCodecStub()); + $numbered = RowsMother::numbered(10); + $batches = [$numbered, RowsMother::numbered(3)]; + + static::assertSame( + FloeStreamReaderContext::writeBatches( + $filesystem, + path('memory://php.floe'), + $numbered->schema(), + $batches, + $options, + engine: FloeEngine::php, + ), + FloeStreamReaderContext::writeBatches( + $filesystem, + path('memory://native.floe'), + $numbered->schema(), + $batches, + $options, + engine: FloeEngine::native, + ), + ); + } + + public function test_native_engine_writes_like_the_php_engine(): void + { + if (!NativeFloeEncoder::isSupported()) { + static::markTestSkipped('flow_php extension with RustFloeEncoderNative is not loaded'); + } + + $filesystem = memory_filesystem(); + $numbered = RowsMother::numbered(10); + $batches = [$numbered, RowsMother::numbered(3)]; + + static::assertSame( + FloeStreamReaderContext::writeBatches( + $filesystem, + path('memory://php.floe'), + $numbered->schema(), + $batches, + engine: FloeEngine::php, + ), + FloeStreamReaderContext::writeBatches( + $filesystem, + path('memory://native.floe'), + $numbered->schema(), + $batches, + engine: FloeEngine::native, + ), + ); + } } diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/FormatTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/FormatTest.php index ea19bcc10..63390952c 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/FormatTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/FormatTest.php @@ -19,6 +19,15 @@ public function test_frame_prefixes_type_and_length(): void static::assertSame(chr(Format::FRAME_ROW) . pack('V', 3) . 'abc', Format::frame(Format::FRAME_ROW, 'abc')); } + public function test_row_frames_frames_every_body_in_order(): void + { + static::assertSame( + chr(Format::FRAME_ROW) . pack('V', 1) . 'a' . chr(Format::FRAME_ROW) . pack('V', 0), + Format::rowFrames(['a', '']), + ); + static::assertSame('', Format::rowFrames([])); + } + public function test_header_starts_with_magic_version_and_flags(): void { $header = Format::header(0x00); diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/NativeFloeEncoderTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/NativeFloeEncoderTest.php index bb8252392..8332553fc 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/NativeFloeEncoderTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/NativeFloeEncoderTest.php @@ -4,14 +4,27 @@ namespace Flow\Floe\Tests\Unit; +use DateTimeImmutable; +use DateTimeZone; +use Flow\ETL\Exception\SchemaMismatchException; +use Flow\ETL\Row\AdaptiveRowHydrator; +use Flow\ETL\Row\Hydrator; +use Flow\ETL\Row\NativeRowHydrator; use Flow\ETL\Row\PhpRowHydrator; use Flow\ETL\Row\TypedRowValues; +use Flow\ETL\Rows; use Flow\ETL\Schema\Metadata; +use Flow\Floe\Exception\FloeException; +use Flow\Floe\Format; use Flow\Floe\NativeFloeEncoder; use Flow\Floe\PhpFloeEncoder; use Flow\Floe\Tests\Context\FloeSchemaContext; +use Flow\Floe\Tests\Double\SpyHydrator; +use Flow\Floe\Tests\Mother\RowsMother; +use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\TestCase; +use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\int_schema; use function Flow\ETL\DSL\row; use function Flow\ETL\DSL\rows; @@ -86,4 +99,115 @@ public function test_native_metadata_bearing_frames_match_the_php_engine(): void (new NativeFloeEncoder($schema))->encode($encoded), ); } + + /** + * @return array + */ + public static function native_hydrators(): array + { + return [ + 'native' => [new NativeRowHydrator()], + 'adaptive' => [new AdaptiveRowHydrator()], + ]; + } + + #[DataProvider('native_hydrators')] + public function test_decode_rows_matches_hydrate_of_decode(Hydrator $hydrator): void + { + $data = rows( + schema(int_schema('id'), str_schema('name', nullable: true), datetime_schema('at')), + row([ + 'id' => 1, + 'name' => 'flow', + 'at' => new DateTimeImmutable('2026-01-01 10:00:00.5', new DateTimeZone('Europe/Warsaw')), + ]), + row(['id' => 2, 'name' => null, 'at' => new DateTimeImmutable('2026-01-02T00:00:00Z')]), + ); + $schema = schema_from_json(FloeSchemaContext::schemaBody($data->schema())); + $encoder = new NativeFloeEncoder($schema); + $bodies = $encoder->encode((new PhpRowHydrator())->dehydrate($data)); + + static::assertEquals( + (new NativeRowHydrator())->hydrate($encoder->decode($bodies), $schema), + $encoder->decodeRows($bodies, $schema, $hydrator), + ); + } + + public function test_decode_rows_hands_a_non_native_hydrator_the_decoded_values(): void + { + $data = rows(schema(int_schema('id')), row(['id' => 1]), row(['id' => 2])); + $schema = schema_from_json(FloeSchemaContext::schemaBody($data->schema())); + $encoder = new NativeFloeEncoder($schema); + $bodies = $encoder->encode((new PhpRowHydrator())->dehydrate($data)); + $spy = new SpyHydrator(); + + static::assertEquals( + (new PhpRowHydrator())->hydrate($encoder->decode($bodies), $schema), + $encoder->decodeRows($bodies, $schema, $spy), + ); + static::assertSame(1, $spy->hydrateCalls); + } + + public function test_decode_rows_throws_the_schema_mismatch_the_hydrator_throws(): void + { + $data = rows(schema(int_schema('id')), row(['id' => 1])); + $schema = schema_from_json(FloeSchemaContext::schemaBody($data->schema())); + $encoder = new NativeFloeEncoder($schema); + $bodies = $encoder->encode((new PhpRowHydrator())->dehydrate($data)); + + $this->expectException(SchemaMismatchException::class); + $this->expectExceptionMessage('column "name" (row 0) declared by the schema is missing from the row'); + + $encoder->decodeRows($bodies, schema(int_schema('id'), str_schema('name')), new NativeRowHydrator()); + } + + public function test_decode_rows_turns_a_corrupt_frame_into_a_floe_exception(): void + { + $data = rows(schema(int_schema('id')), row(['id' => 1])); + $schema = schema_from_json(FloeSchemaContext::schemaBody($data->schema())); + $encoder = new NativeFloeEncoder($schema); + $bodies = $encoder->encode((new PhpRowHydrator())->dehydrate($data)); + + $this->expectException(FloeException::class); + $this->expectExceptionMessage('flow_php row frame length does not match its content'); + + $encoder->decodeRows([$bodies[0] . "\xEF"], $schema, new NativeRowHydrator()); + } + + #[DataProvider('native_hydrators')] + public function test_encode_frames_matches_framing_encode_of_dehydrate(Hydrator $hydrator): void + { + $data = RowsMother::numbered(4); + $encoder = new NativeFloeEncoder($data->schema()); + + static::assertSame( + Format::rowFrames($encoder->encode((new PhpRowHydrator())->dehydrate($data))), + $encoder->encodeFrames($data, $hydrator), + ); + } + + public function test_encode_frames_hands_a_non_native_hydrator_the_rows(): void + { + $data = RowsMother::numbered(2); + $encoder = new NativeFloeEncoder($data->schema()); + $spy = new SpyHydrator(); + + static::assertSame( + Format::rowFrames($encoder->encode((new PhpRowHydrator())->dehydrate($data))), + $encoder->encodeFrames($data, $spy), + ); + static::assertSame(1, $spy->dehydrateCalls); + } + + public function test_encode_frames_turns_a_row_without_a_declared_column_into_a_floe_exception(): void + { + $schema = schema(int_schema('id'), str_schema('name')); + + $this->expectException(FloeException::class); + $this->expectExceptionMessage('flow_php found a row that does not carry the declared column "name"'); + + (new NativeFloeEncoder($schema))->encodeFrames(Rows::trusted($schema, [row([ + 'id' => 1, + ])]), new NativeRowHydrator()); + } } diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/PhpFloeEncoderTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/PhpFloeEncoderTest.php index d9098d204..77d8d9721 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/PhpFloeEncoderTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/PhpFloeEncoderTest.php @@ -9,8 +9,10 @@ use Flow\ETL\Row\TypedRowValues; use Flow\ETL\Schema\Metadata; use Flow\Floe\Exception\FloeException; +use Flow\Floe\Format; use Flow\Floe\PhpFloeEncoder; use Flow\Floe\Tests\Context\FloeSchemaContext; +use Flow\Floe\Tests\Mother\RowsMother; use PHPUnit\Framework\TestCase; use function Flow\ETL\DSL\float_schema; @@ -176,4 +178,32 @@ public function test_unknown_value_flag_throws(): void (new PhpFloeEncoder($schema))->decode(["\xEF"]); } + + public function test_decode_rows_hydrates_the_decoded_values(): void + { + $data = rows( + schema(int_schema('id'), str_schema('name', nullable: true)), + row(['id' => 1, 'name' => 'flow']), + row(['id' => 2, 'name' => null]), + ); + $schema = schema_from_json(FloeSchemaContext::schemaBody($data->schema())); + $encoder = new PhpFloeEncoder($schema); + $bodies = $encoder->encode((new PhpRowHydrator())->dehydrate($data)); + + static::assertEquals( + (new PhpRowHydrator())->hydrate($encoder->decode($bodies), $schema), + $encoder->decodeRows($bodies, $schema, new PhpRowHydrator()), + ); + } + + public function test_encode_frames_frames_the_encoded_dehydrated_rows(): void + { + $data = RowsMother::numbered(3); + $encoder = new PhpFloeEncoder($data->schema()); + + static::assertSame( + Format::rowFrames($encoder->encode((new PhpRowHydrator())->dehydrate($data))), + $encoder->encodeFrames($data, new PhpRowHydrator()), + ); + } } diff --git a/src/extension/flow-php-ext/Cargo.lock b/src/extension/flow-php-ext/Cargo.lock index 2dd3b5705..cf93be066 100644 --- a/src/extension/flow-php-ext/Cargo.lock +++ b/src/extension/flow-php-ext/Cargo.lock @@ -446,6 +446,7 @@ name = "flow_php" version = "0.0.0" dependencies = [ "ext-php-rs", + "ext-php-rs-build", "memchr", "serde", "serde_json", diff --git a/src/extension/flow-php-ext/Cargo.toml b/src/extension/flow-php-ext/Cargo.toml index 0eb36b00b..26a2ef0c9 100644 --- a/src/extension/flow-php-ext/Cargo.toml +++ b/src/extension/flow-php-ext/Cargo.toml @@ -11,3 +11,6 @@ ext-php-rs = "0.15" memchr = "2" serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["raw_value"] } + +[build-dependencies] +ext-php-rs-build = "0.1" diff --git a/src/extension/flow-php-ext/build.rs b/src/extension/flow-php-ext/build.rs index eccfc9392..24b1b8563 100644 --- a/src/extension/flow-php-ext/build.rs +++ b/src/extension/flow-php-ext/build.rs @@ -5,6 +5,27 @@ fn main() { let version = extension_version("FLOW_PHP_EXT_VERSION"); println!("cargo:rustc-env=FLOW_PHP_EXT_VERSION={version}"); + + emit_php_version_cfg(); +} + +/// The `php84`/`php85` cfgs ext-php-rs sets for itself, for the same PHP (`PHP` first, then `PATH`): the extension +/// binary is bound to the PHP minor it is built against, so version gates are compile-time. +fn emit_php_version_cfg() { + use ext_php_rs_build::{emit_check_cfg, emit_php_cfg_flags, emit_rerun_if_env_changed, find_php, ApiVersion, PHPInfo}; + + emit_rerun_if_env_changed(); + emit_check_cfg(); + + let php = find_php().expect("cannot find the php executable"); + let info = PHPInfo::get(&php).expect("cannot read php -i"); + let version: ApiVersion = info + .zend_version() + .expect("cannot read the Zend API version") + .try_into() + .expect("unsupported Zend API version"); + + emit_php_cfg_flags(version); } fn extension_version(env_name: &str) -> String { diff --git a/src/extension/flow-php-ext/php/Flow/Floe/RustFloeEncoderNative.php b/src/extension/flow-php-ext/php/Flow/Floe/RustFloeEncoderNative.php index 01286b9e1..e9277415a 100644 --- a/src/extension/flow-php-ext/php/Flow/Floe/RustFloeEncoderNative.php +++ b/src/extension/flow-php-ext/php/Flow/Floe/RustFloeEncoderNative.php @@ -6,6 +6,7 @@ use Flow\ETL\Row\RawRowValues; use Flow\ETL\Row\TypedRowValues; +use Flow\ETL\Rows; use Flow\ETL\Schema; use Flow\Floe\Exception\ExtensionException; use RuntimeException; @@ -49,4 +50,29 @@ public function decode(array $frameBodies, string $schemaBody): array { throw new RuntimeException('flow_php extension is not loaded'); } + + /** + * `RustRowHydratorNative::hydrate($this->decode($frameBodies, $schemaBody), $schema)` in one pass. + * + * @param list $frameBodies + * @param string $schemaBody SCHEMA frame body (JSON list of normalized definitions) + * + * @throws ExtensionException + */ + public function decodeRows(array $frameBodies, string $schemaBody, Schema $schema): Rows + { + throw new RuntimeException('flow_php extension is not loaded'); + } + + /** + * `Format::frame(Format::FRAME_ROW, ...)` over `$this->encode(RustRowHydratorNative::dehydrate($rows), ...)`. + * + * @param string $schemaBody SCHEMA frame body (JSON list of normalized definitions) + * + * @throws ExtensionException + */ + public function encodeFrames(Rows $rows, string $schemaBody, Schema $schema): string + { + throw new RuntimeException('flow_php extension is not loaded'); + } } diff --git a/src/extension/flow-php-ext/src/cast.rs b/src/extension/flow-php-ext/src/cast.rs index f2f7baad2..363b0bc2d 100644 --- a/src/extension/flow-php-ext/src/cast.rs +++ b/src/extension/flow-php-ext/src/cast.rs @@ -13,9 +13,13 @@ use crate::ctx::{ }; use crate::encode::{expect_object, ht_for_each, read_slot}; use crate::exception::ext_exception; -use crate::hydrate::{build_hydrate_plan, fold_metadata_into_schema, AssemblyClasses, HydratePlan, RowValuesClass}; +use crate::format::Reader; +use crate::hydrate::{ + build_hydrate_plan, fold_metadata_into_schema, fold_pending_metadata, read_cell, AssemblyClasses, + HydrateColumn, HydratePlan, RowValuesClass, +}; use crate::json_check::json_valid; -use crate::plan::{parse_schema_json, TypeJson}; +use crate::plan::{parse_schema_json, Plan, TypeJson}; use crate::values::date_from_free_form; extern "C" { @@ -868,6 +872,96 @@ fn cast_json(value: &Zval, ctx: &mut Ctx) -> Result, PhpException> Ok(None) } +/// One value of a row against its column: a present null under NOT NULL is refused, a null passes, anything else is +/// cast natively where proven identical and otherwise through the retained PHP `Type::cast`. +fn cast_cell( + column: &HydrateColumn, + cast_column: &CastColumn, + value: &Zval, + row_index: u64, + assembly: &AssemblyClasses, + ctx: &mut Ctx, +) -> Result { + // a present null is a DIFFERENT refusal from an absence: valueDoesNotMatch, not + // missingColumn, and 027 pins the two messages apart + if value.is_null() && !column.nullable { + return Err(schema_mismatch( + assembly.schema_mismatch_ce, + assembly.value_does_not_match, + row_index, + &mut [column.base_def.shallow_clone(), null_zval()], + )?); + } + + if value.is_null() { + return Ok(null_zval()); + } + + if let Some(casted) = cast_value(&cast_column.kind, value, ctx)? { + return Ok(casted); + } + + let type_obj = cast_column + .type_zv + .object() + .ok_or_else(|| ext_exception("flow_php expected a Type object"))?; + + match call_handle_catching(cast_column.cast_fn, Some(type_obj), &mut [value.shallow_clone()]) { + Ok(casted) => Ok(casted), + Err(mut refusal) => { + if !refusal.instance_of(assembly.types_exception_ce) { + return Err(transparent_exception(&mut refusal)); + } + + // The declared type refused the value. Its own exception is dropped rather than chained, exactly as + // HydratedBatch's guard drops it - 027 compares the two hydrators' class and message byte for byte. + Err(schema_mismatch( + assembly.schema_mismatch_ce, + assembly.value_does_not_match, + row_index, + &mut [column.base_def.shallow_clone(), value.shallow_clone()], + )?) + } + } +} + +/// Builds one `Row` through its PHP constructor and appends it to the batch. +fn push_row( + rows_ht: &mut ZendHashTable, + row_values: ZBox, + assembly: &AssemblyClasses, +) -> Result<(), PhpException> { + let mut values_zv = Zval::new(); + values_zv.set_hashtable(row_values); + + let mut row = construct_with_zvals(assembly.row_ce, &mut [values_zv], "a Row")?; + let mut row_zv = Zval::new(); + row_zv.set_object(&mut row); + + rows_ht + .push(row_zv) + .map_err(|e| ext_exception(format!("flow_php failed to collect hydrated rows: {e:?}"))) +} + +/// Every non-null value was cast to its column's type, so the batch takes the shape-only door - the one +/// HydratedBatch returns through. Row::conform() returns $this for a row holding exactly the Schema's keys, in +/// order, with no null under NOT NULL - which every cast row is once no column was absent. An absence keeps the +/// PHP door: padding, missingColumn and the failing row's index are conform()'s to decide. +fn assemble_rows( + rows_ht: ZBox, + schema_zv: Zval, + every_column_present: bool, + assembly: &AssemblyClasses, +) -> Result { + let mut rows_zv = Zval::new(); + rows_zv.set_hashtable(rows_ht); + + let door = if every_column_present { assembly.rows_trusted } else { assembly.rows_conformed }; + + // conform()'s refusal surfaces as the SchemaMismatchException PHP threw, like the cast refusals above + call_handle_transparent(door, None, &mut [schema_zv, rows_zv]) +} + /// Native `PhpRowHydrator::hydrate`: raw scalars cast against a `Schema` and /// assembled into `Flow\ETL\Rows` in one pass. Column values cast natively /// where proven identical, otherwise per value through the retained PHP @@ -914,80 +1008,96 @@ pub fn cast_rows( continue; }; - // a present null is a DIFFERENT refusal from an absence: valueDoesNotMatch, not - // missingColumn, and 027 pins the two messages apart - if value.is_null() && !column.nullable { - return Err(schema_mismatch( - assembly.schema_mismatch_ce, - assembly.value_does_not_match, - row_index, - &mut [column.base_def.shallow_clone(), null_zval()], - )?); + let casted = cast_cell(column, cast_column, value, row_index, assembly, ctx)?; + ht_insert_key(&mut row_values, &key, casted); + } + + push_row(&mut rows_ht, row_values, assembly) + })?; + + assemble_rows(rows_ht, schema_zv, every_column_present, assembly) +} + +/// `cast_rows(decode(...))` in one pass: each ROW frame body is decoded straight into the cast, so no +/// `RawRowValues` is built. A Schema column takes the decoded column of the same name - the lookup `cast_rows` does +/// in `RawRowValues::values` - and a Schema column the frames do not carry is an absence, exactly as there. +pub fn decode_rows( + frame_bodies: &Zval, + decode_plan: &Plan, + schema: &Zval, + plan_slot: &mut Option, + assembly: &AssemblyClasses, + ctx: &mut Ctx, +) -> Result { + ensure_cast_plan(plan_slot, schema, ctx)?; + let plan = plan_slot.as_ref().expect("plan built above"); + + let bodies_ht = frame_bodies + .array() + .ok_or_else(|| ext_exception("flow_php expected a list of row frame bodies"))?; + + let decoded_names: Vec<&[u8]> = decode_plan.columns.iter().map(|column| column.name.as_bytes()).collect(); + // RawRowValues::values refuses a name decoded twice at the second occurrence + let duplicated = (0..decoded_names.len()).find(|&index| decoded_names[..index].contains(&decoded_names[index])); + let sources: Vec> = plan + .hydrate + .columns + .iter() + .map(|column| { + let name = column.name_zv.zend_str().map(ZendStr::as_bytes); + decoded_names.iter().position(|decoded| Some(*decoded) == name) + }) + .collect(); + + let mut decoded: Vec = (0..decoded_names.len()).map(|_| Zval::new()).collect(); + let mut metadata: Vec<(Vec, Zval)> = Vec::new(); + let mut rows_ht = ZendHashTable::with_capacity(bodies_ht.len() as u32); + let every_column_present = sources.iter().all(Option::is_some); + + ht_for_each(bodies_ht, |_, row_index, body_zv| { + let bytes = body_zv + .zend_str() + .ok_or_else(|| ext_exception("flow_php expected a row frame body to be a string"))? + .as_bytes(); + + let mut reader = Reader::new(bytes); + + for (index, column) in decode_plan.columns.iter().enumerate() { + let (value, value_metadata) = read_cell(&column.decoder, &mut reader, ctx)?; + + if let Some(value_metadata) = value_metadata { + metadata.push((column.name.as_bytes().to_vec(), value_metadata)); } - let casted = if value.is_null() { - null_zval() - } else { - match cast_value(&cast_column.kind, value, ctx)? { - Some(casted) => casted, - None => { - let type_obj = cast_column - .type_zv - .object() - .ok_or_else(|| ext_exception("flow_php expected a Type object"))?; - - match call_handle_catching( - cast_column.cast_fn, - Some(type_obj), - &mut [value.shallow_clone()], - ) { - Ok(casted) => casted, - Err(mut refusal) => { - if !refusal.instance_of(assembly.types_exception_ce) { - return Err(transparent_exception(&mut refusal)); - } - - // The declared type refused the value. Its own exception is dropped - // rather than chained, exactly as HydratedBatch's guard drops it - - // 027 compares the two hydrators' class and message byte for byte. - return Err(schema_mismatch( - assembly.schema_mismatch_ce, - assembly.value_does_not_match, - row_index, - &mut [column.base_def.shallow_clone(), value.shallow_clone()], - )?); - } - } - } - } - }; + if duplicated == Some(index) { + return Err(ext_exception(format!( + "flow_php found duplicated entry name \"{}\" in a row frame", + column.name + ))); + } - ht_insert_key(&mut row_values, &key, casted); + decoded[index] = value; } - let mut values_zv = Zval::new(); - values_zv.set_hashtable(row_values); + if !reader.is_eof() { + return Err(ext_exception("flow_php row frame length does not match its content")); + } - let mut row = construct_with_zvals(assembly.row_ce, &mut [values_zv], "a Row")?; - let mut row_zv = Zval::new(); - row_zv.set_object(&mut row); - rows_ht.push(row_zv).map_err(|e| { - ext_exception(format!("flow_php failed to collect hydrated rows: {e:?}")) - })?; + let mut row_values = ZendHashTable::with_capacity(plan.hydrate.columns.len() as u32); - Ok(()) - })?; + for ((column, cast_column), source) in plan.hydrate.columns.iter().zip(&plan.columns).zip(&sources) { + let Some(source) = source else { + continue; + }; - let mut rows_zv = Zval::new(); - rows_zv.set_hashtable(rows_ht); + let casted = cast_cell(column, cast_column, &decoded[*source], row_index, assembly, ctx)?; + ht_insert_key(&mut row_values, &column.key(), casted); + } - // every non-null value was just cast to its column's type, so the batch takes - // the shape-only door - the one HydratedBatch returns through. Row::conform() returns - // $this for a row holding exactly the Schema's keys, in order, with no null under - // NOT NULL - which every row built above is once no column was absent. An absence - // keeps the PHP door: padding, missingColumn and the failing row's index are - // conform()'s to decide. - let door = if every_column_present { assembly.rows_trusted } else { assembly.rows_conformed }; + push_row(&mut rows_ht, row_values, assembly) + })?; + + let schema_zv = fold_pending_metadata(schema, metadata, ctx)?; - call_handle(door, None, &mut [schema_zv, rows_zv], "assemble Rows") + assemble_rows(rows_ht, schema_zv, every_column_present, assembly) } diff --git a/src/extension/flow-php-ext/src/encode.rs b/src/extension/flow-php-ext/src/encode.rs index c761c9480..4544219ca 100644 --- a/src/extension/flow-php-ext/src/encode.rs +++ b/src/extension/flow-php-ext/src/encode.rs @@ -12,7 +12,7 @@ use crate::ctx::{ }; use crate::exception::ext_exception; use crate::format::{ - write_u32, VALUE_ABSENT, VALUE_NULL, VALUE_NULL_WITH_META, VALUE_PRESENT, VALUE_PRESENT_WITH_META, + write_u32, FRAME_ROW, VALUE_ABSENT, VALUE_NULL, VALUE_NULL_WITH_META, VALUE_PRESENT, VALUE_PRESENT_WITH_META, }; use crate::plan::{parse_schema_json, TypeJson}; @@ -334,18 +334,17 @@ fn schema_definitions(schema: &Zval) -> Result, PhpException> { Ok(base_defs) } -/// Encodes one `Flow\ETL\Row\TypedRowValues` (its `values` + `metadata` maps) -/// into a bare ROW frame body (no length prefix, no frame type). Byte-identical +/// Appends one `Flow\ETL\Row\TypedRowValues` (its `values` + `metadata` maps) to `out` +/// as a bare ROW frame body (no length prefix, no frame type). Byte-identical /// to `Flow\Floe\PhpFloeEncoder::encode` for the same row. pub fn encode_typed_row( plan: &EncodePlan, row_index: u64, values_ht: &ZendHashTable, metadata_ht: &ZendHashTable, + out: &mut Vec, ctx: &mut Ctx, -) -> Result, PhpException> { - let mut out = Vec::with_capacity(1024); - +) -> Result<(), PhpException> { for column in &plan.columns { let Some(value) = ht_find(values_ht, &column.name) else { return Err(ext_exception(format!( @@ -368,7 +367,7 @@ pub fn encode_typed_row( if diverges { out.push(VALUE_NULL_WITH_META); - write_len_prefixed(&mut out, &entry_metadata_json); + write_len_prefixed(out, &entry_metadata_json); } else { out.push(VALUE_NULL); } @@ -378,14 +377,46 @@ pub fn encode_typed_row( if diverges { out.push(VALUE_PRESENT_WITH_META); - write_len_prefixed(&mut out, &entry_metadata_json); + write_len_prefixed(out, &entry_metadata_json); } else { out.push(VALUE_PRESENT); } - encode_value(&column.encoder, value, &mut out, ctx)?; + encode_value(&column.encoder, value, out, ctx)?; } + Ok(()) +} + +/// `Format::frame(Format::FRAME_ROW, $body)` over `encode_typed_row` for every `Row` of a batch: the rows' own +/// values tables and the batch's one `TypedRowValues::metadata` map, complete ROW frames in one buffer. +pub fn encode_frames( + plan: &EncodePlan, + rows_ht: &ZendHashTable, + row_values_slot: u32, + metadata_ht: &ZendHashTable, + ctx: &mut Ctx, +) -> Result, PhpException> { + let mut out = Vec::with_capacity(rows_ht.len() * 256); + + ht_for_each(rows_ht, |_, row_index, row_zv| { + let values_ht = read_slot(expect_object(row_zv, "a Row")?, row_values_slot) + .array() + .ok_or_else(|| ext_exception("flow_php expected Row::values to be an array"))?; + + out.push(FRAME_ROW); + let length_at = out.len(); + out.extend_from_slice(&[0; 4]); + + encode_typed_row(plan, row_index, values_ht, metadata_ht, &mut out, ctx)?; + + let length = u32::try_from(out.len() - length_at - 4) + .map_err(|_| ext_exception("flow_php encoded a row frame longer than 4 GiB"))?; + out[length_at..length_at + 4].copy_from_slice(&length.to_le_bytes()); + + Ok(()) + })?; + Ok(out) } diff --git a/src/extension/flow-php-ext/src/format.rs b/src/extension/flow-php-ext/src/format.rs index 4510b481c..5fbaa072c 100644 --- a/src/extension/flow-php-ext/src/format.rs +++ b/src/extension/flow-php-ext/src/format.rs @@ -9,6 +9,8 @@ use crate::exception::ext_exception; #[cfg(target_endian = "big")] compile_error!("flow_php only supports little-endian targets"); +pub const FRAME_ROW: u8 = 0x02; + pub const VALUE_NULL: u8 = 0x00; pub const VALUE_PRESENT: u8 = 0x01; pub const VALUE_NULL_WITH_META: u8 = 0x02; diff --git a/src/extension/flow-php-ext/src/hydrate.rs b/src/extension/flow-php-ext/src/hydrate.rs index d7ae3caf2..767e9123c 100644 --- a/src/extension/flow-php-ext/src/hydrate.rs +++ b/src/extension/flow-php-ext/src/hydrate.rs @@ -20,7 +20,7 @@ use crate::exception::ext_exception; use crate::format::{ Reader, VALUE_NULL, VALUE_NULL_WITH_META, VALUE_PRESENT, VALUE_PRESENT_WITH_META, }; -use crate::plan::Plan; +use crate::plan::{Decoder, Plan}; use crate::values::decode_value; /// Resolved `Flow\ETL\Row\RawRowValues` class handles, shared by the binary decoder and the hydrator. @@ -67,6 +67,29 @@ fn read_metadata(reader: &mut Reader, ctx: &mut Ctx) -> Result Result<(Zval, Option), PhpException> { + Ok(match reader.u8("row value flag")? { + VALUE_PRESENT => (decode_value(decoder, reader, ctx)?, None), + VALUE_NULL => (Zval::new(), None), + VALUE_PRESENT_WITH_META => { + let metadata = read_metadata(reader, ctx)?; + + (decode_value(decoder, reader, ctx)?, Some(metadata)) + } + VALUE_NULL_WITH_META => (Zval::new(), Some(read_metadata(reader, ctx)?)), + other => { + return Err(ext_exception(format!( + "flow_php found unknown value flag 0x{other:02X}" + ))); + } + }) +} + /// Decodes one ROW frame body into a `Flow\ETL\Row\RawRowValues` object - values keyed by /// column name, absent columns omitted, diverging per-value metadata recorded in `metadata`. pub fn decode_row_values( @@ -79,29 +102,11 @@ pub fn decode_row_values( let mut metadata_ht = ZendHashTable::new(); for column in &plan.columns { - let flag = reader.u8("row value flag")?; - - let value = match flag { - VALUE_PRESENT => decode_value(&column.decoder, reader, ctx)?, - VALUE_NULL => Zval::new(), - VALUE_PRESENT_WITH_META => { - let metadata = read_metadata(reader, ctx)?; - ht_insert(&mut metadata_ht, column.name.as_bytes(), metadata); - - decode_value(&column.decoder, reader, ctx)? - } - VALUE_NULL_WITH_META => { - let metadata = read_metadata(reader, ctx)?; - ht_insert(&mut metadata_ht, column.name.as_bytes(), metadata); - - Zval::new() - } - other => { - return Err(ext_exception(format!( - "flow_php found unknown value flag 0x{other:02X}" - ))); - } - }; + let (value, metadata) = read_cell(&column.decoder, reader, ctx)?; + + if let Some(metadata) = metadata { + ht_insert(&mut metadata_ht, column.name.as_bytes(), metadata); + } if !ht_add(&mut values_ht, column.name.as_bytes(), value) { return Err(ext_exception(format!( @@ -189,7 +194,7 @@ pub(crate) fn def_dehydrate_fns( /// One column of a batch's dehydrate plan: the key, and the `type()` / /// `metadata()` zvals read once from the batch Schema. -struct DehydrateColumn { +pub(crate) struct DehydrateColumn { numeric_key: Option, name_zv: Zval, type_zv: Zval, @@ -251,15 +256,14 @@ fn dehydrate_row( Ok(row_values) } -/// Native `PhpRowHydrator::dehydrate`: turns a `Flow\ETL\Rows` into a list of -/// `Flow\ETL\Row\TypedRowValues`, value zvals moved verbatim (no casting). -pub fn dehydrate_rows( - rows: &Zval, +/// The batch behind a `Flow\ETL\Rows`: its `rows` table and one dehydrate column per Schema definition - the +/// `type()`/`metadata()` calls made once per batch, not once per cell. +pub(crate) fn dehydrate_batch<'a>( + rows: &'a Zval, rows_classes: &RowsClasses, - class: &TypedRowValuesClass, def_fn_cache: &mut DefFnCache, ctx: &mut Ctx, -) -> Result { +) -> Result<(&'a ZendHashTable, Vec), PhpException> { let rows_obj = expect_object(rows, "Rows")?; let rows_ht = read_slot(rows_obj, rows_classes.rows_slot) .array() @@ -274,7 +278,6 @@ pub fn dehydrate_rows( let metadata_map_slot = ctx.metadata_map_slot()?; - // one type()/metadata() call per column per batch, not per cell let mut columns: Vec = Vec::with_capacity(definitions_ht.len()); ht_for_each(definitions_ht, |_, _, def_zv| { @@ -317,6 +320,34 @@ pub fn dehydrate_rows( Ok(()) })?; + Ok((rows_ht, columns)) +} + +/// `TypedRowValues::metadata` as `dehydrate_row` builds it - the batch's columns with non-empty metadata, the same +/// map for every row of the batch. +pub(crate) fn batch_metadata(columns: &[DehydrateColumn]) -> ZBox { + let mut metadata_ht = ZendHashTable::new(); + + for column in columns { + if let Some(metadata) = &column.metadata_zv { + ht_insert_key(&mut metadata_ht, &column.key(), metadata.shallow_clone()); + } + } + + metadata_ht +} + +/// Native `PhpRowHydrator::dehydrate`: turns a `Flow\ETL\Rows` into a list of +/// `Flow\ETL\Row\TypedRowValues`, value zvals moved verbatim (no casting). +pub fn dehydrate_rows( + rows: &Zval, + rows_classes: &RowsClasses, + class: &TypedRowValuesClass, + def_fn_cache: &mut DefFnCache, + ctx: &mut Ctx, +) -> Result { + let (rows_ht, columns) = dehydrate_batch(rows, rows_classes, def_fn_cache, ctx)?; + let mut out = ZendHashTable::with_capacity(rows_ht.len() as u32); ht_for_each(rows_ht, |_, _, row_zv| { @@ -491,15 +522,24 @@ pub fn fold_metadata_into_schema( return Ok(()); }; - ht_for_each(metadata_ht, |key, _, metadata_zv| { - if let Some(name) = key { - pending.push((name.as_bytes().to_vec(), metadata_zv.shallow_clone())); - } + ht_for_each(metadata_ht, |key, index, metadata_zv| { + // PHP stores a numeric column name as an integer key; HydratedBatch casts it back to the name + let name = key.map_or_else(|| (index as i64).to_string().into_bytes(), |name| name.as_bytes().to_vec()); + pending.push((name, metadata_zv.shallow_clone())); Ok(()) }) })?; + fold_pending_metadata(schema, pending, ctx) +} + +/// Folds `(column name, Metadata)` pairs, in the order the rows carried them, into the Schema. +pub fn fold_pending_metadata( + schema: &Zval, + pending: Vec<(Vec, Zval)>, + ctx: &mut Ctx, +) -> Result { if pending.is_empty() { return Ok(schema.shallow_clone()); } diff --git a/src/extension/flow-php-ext/src/lib.rs b/src/extension/flow-php-ext/src/lib.rs index 8072751f5..0f2b69dba 100644 --- a/src/extension/flow-php-ext/src/lib.rs +++ b/src/extension/flow-php-ext/src/lib.rs @@ -18,7 +18,9 @@ use ext_php_rs::zend::ModuleEntry; use ext_php_rs::{info_table_end, info_table_row, info_table_start}; use crate::ctx::{ht_insert, zval_str, Ctx}; -use crate::encode::{build_encode_plan, encode_typed_row, expect_object, ht_for_each, read_slot, EncodePlan}; +use crate::encode::{ + build_encode_plan, encode_frames, encode_typed_row, expect_object, ht_for_each, read_slot, EncodePlan, +}; use crate::exception::ext_exception; use crate::format::Reader; use crate::plan::Plan; @@ -72,7 +74,8 @@ fn ensure_plan

( } /// Floe's consolidated binary codec, both directions: ROW frame bodies to/from -/// `Flow\ETL\Row\RawRowValues` (decode) and `Flow\ETL\Row\TypedRowValues` (encode). +/// `Flow\ETL\Row\RawRowValues` (decode) and `Flow\ETL\Row\TypedRowValues` (encode), +/// plus `decode_rows` / `encode_frames`, which go straight between frames and `Flow\ETL\Rows`. /// The PHP side keeps buffering/framing/sectioning and hands over bare frame /// bodies; the extension owns value encode/decode against a primed schema. #[php_class] @@ -82,6 +85,10 @@ pub struct RustFloeEncoderNative { encode_bound: Option>, decode_bound: Option>, row_values_class: hydrate::RowValuesClass, + cast_plan: Option, + assembly: hydrate::AssemblyClasses, + rows_classes: hydrate::RowsClasses, + def_dehydrate_fns: hydrate::DefFnCache, } #[php_impl] @@ -92,6 +99,10 @@ impl RustFloeEncoderNative { encode_bound: None, decode_bound: None, row_values_class: hydrate::RowValuesClass::resolve()?, + cast_plan: None, + assembly: hydrate::AssemblyClasses::resolve()?, + rows_classes: hydrate::RowsClasses::resolve()?, + def_dehydrate_fns: hydrate::DefFnCache::new(), }) } @@ -131,7 +142,8 @@ impl RustFloeEncoderNative { ext_exception("flow_php expected TypedRowValues::metadata to be an array") })?; - let body = encode_typed_row(plan, row_index, values_ht, metadata_ht, ctx)?; + let mut body = Vec::with_capacity(1024); + encode_typed_row(plan, row_index, values_ht, metadata_ht, &mut body, ctx)?; encoded.push(zval_str(&body)).map_err(|e| { ext_exception(format!("flow_php failed to collect a row body: {e:?}")) @@ -146,6 +158,35 @@ impl RustFloeEncoderNative { Ok(zv) } + /// `Rows` straight to complete ROW frames (frame type + length + body per row) in one string - see + /// `encode::encode_frames`. + pub fn encode_frames( + &mut self, + rows: &Zval, + schema_body: BinarySlice, + schema: &Zval, + ) -> PhpResult { + ensure_plan( + &mut self.encode_bound, + &mut self.ctx, + &schema_body, + |body, ctx| build_encode_plan(body, schema, ctx), + )?; + + let (rows_ht, columns) = + hydrate::dehydrate_batch(rows, &self.rows_classes, &mut self.def_dehydrate_fns, &mut self.ctx)?; + + let frames = encode_frames( + &self.encode_bound.as_ref().expect("plan bound above").plan, + rows_ht, + self.rows_classes.row_values_slot, + &hydrate::batch_metadata(&columns), + &mut self.ctx, + )?; + + Ok(zval_str(&frames)) + } + /// Decodes a list of ROW frame bodies written with the given SCHEMA frame /// body into `Flow\ETL\Row\RawRowValues` objects. pub fn decode(&mut self, frame_bodies: &Zval, schema_body: BinarySlice) -> PhpResult { @@ -193,6 +234,30 @@ impl RustFloeEncoderNative { Ok(zv) } + + /// Frame bodies decoded and cast against `schema` straight into `Flow\ETL\Rows` - see `cast::decode_rows`. + pub fn decode_rows( + &mut self, + frame_bodies: &Zval, + schema_body: BinarySlice, + schema: &Zval, + ) -> PhpResult { + ensure_plan( + &mut self.decode_bound, + &mut self.ctx, + &schema_body, + plan::build_plan, + )?; + + cast::decode_rows( + frame_bodies, + &self.decode_bound.as_ref().expect("plan bound above").plan, + schema, + &mut self.cast_plan, + &self.assembly, + &mut self.ctx, + ) + } } /// Native counterpart of `PhpRowHydrator`: `hydrate` builds `Flow\ETL\Rows` from diff --git a/src/extension/flow-php-ext/src/values.rs b/src/extension/flow-php-ext/src/values.rs index 91dbac450..20c22c7bc 100644 --- a/src/extension/flow-php-ext/src/values.rs +++ b/src/extension/flow-php-ext/src/values.rs @@ -18,6 +18,9 @@ extern "C" { fn php_date_instantiate(pce: *mut ext_php_rs::zend::ClassEntry, object: *mut Zval) -> *mut Zval; + #[cfg(php84)] + fn php_date_initialize_from_ts_long(dateobj: *mut c_void, sec: i64, usec: c_int); + fn php_date_initialize( dateobj: *mut c_void, time_str: *const c_char, @@ -30,6 +33,14 @@ extern "C" { const PHP_DATE_OBJ_STD_OFFSET: usize = std::mem::size_of::<*const c_void>(); +/// `php_date_obj_from_obj`: the `php_date_obj` a datetime `zend_object` is embedded in. +fn php_date_obj(object: &mut ZendObject) -> *mut c_void { + std::ptr::from_mut(object) + .cast::() + .wrapping_sub(PHP_DATE_OBJ_STD_OFFSET) + .cast::() +} + /// `new DateTimeImmutable($str, $timezone)` through the same C-level timelib parser, in its /// non-throwing `date_create()` flavor: `Ok(None)` on parse failure, no exception. pub(crate) fn date_from_free_form( @@ -59,10 +70,7 @@ pub(crate) fn date_from_free_form( let initialized = unsafe { php_date_initialize( - std::ptr::from_mut(datetime_obj) - .cast::() - .sub(PHP_DATE_OBJ_STD_OFFSET) - .cast::(), + php_date_obj(datetime_obj), time_str.as_mut_ptr().cast::(), time_str.len() - 1, std::ptr::null(), @@ -211,6 +219,13 @@ fn decode_datetime(reader: &mut Reader, ctx: &mut Ctx) -> Result 999_999 { + return Err(restore_failed()); + } + // the class is not stored: a datetime column always hydrates to DateTimeImmutable let fns = ctx.datetime_fns(false)?; @@ -229,29 +244,38 @@ fn decode_datetime(reader: &mut Reader, ctx: &mut Ctx) -> Result() - .sub(PHP_DATE_OBJ_STD_OFFSET) - .cast::(), - time_str.as_mut_ptr().cast::(), - time_str.len() - 1, - c"U.u".as_ptr(), - std::ptr::null_mut(), - // PHP_DATE_INIT_FORMAT - the flags createFromFormat passes. - 0x02, - ) + let dateobj = php_date_obj(datetime_obj); + + // the createFromTimestamp() initializer, exported from PHP 8.4 + #[cfg(php84)] + let initialized = { + unsafe { php_date_initialize_from_ts_long(dateobj, timestamp, microseconds as c_int) }; + + true + }; + + #[cfg(not(php84))] + let initialized = { + // timelib reads the byte AFTER the consumed input, so the buffer must be + // NUL-terminated like the zend_strings PHP hands it (length excludes the NUL) + let mut time_str = format!("{timestamp}.{microseconds:06}\0"); + + unsafe { + php_date_initialize( + dateobj, + time_str.as_mut_ptr().cast::(), + time_str.len() - 1, + c"U.u".as_ptr(), + std::ptr::null_mut(), + // PHP_DATE_INIT_FORMAT - the flags createFromFormat passes. + 0x02, + ) + } }; ensure_no_pending_exception("restore a datetime value")?; if !initialized { - return Err(ext_exception(format!( - "flow_php failed to restore datetime from timestamp \"{timestamp}\"" - ))); + return Err(restore_failed()); } let timezone = ctx.timezone(timezone_name)?.shallow_clone(); diff --git a/src/extension/flow-php-ext/tests/phpt/024_row_hydrator_parity.phpt b/src/extension/flow-php-ext/tests/phpt/024_row_hydrator_parity.phpt index 59cf028ef..13d29787e 100644 --- a/src/extension/flow-php-ext/tests/phpt/024_row_hydrator_parity.phpt +++ b/src/extension/flow-php-ext/tests/phpt/024_row_hydrator_parity.phpt @@ -63,6 +63,10 @@ $datasets = [ new RawRowValues(['id' => 2], ['id' => Metadata::fromArray(['k' => 'v2'])]), ], ], + 'numeric_name_metadata' => [ + schema(int_schema('7'), int_schema('id')), + [new RawRowValues(['7' => 1, 'id' => 1], ['7' => Metadata::fromArray(['k' => 'v']), 'id' => Metadata::fromArray(['n' => 1])])], + ], 'temporal' => [ schema(datetime_schema('at'), date_schema('d'), time_schema('t'), uuid_schema('u')), [new RawRowValues([ @@ -124,6 +128,7 @@ scalars hydrate:yes dehydrate:yes null_nonnullable hydrate:yes dehydrate:yes absent hydrate:yes dehydrate:yes metadata hydrate:yes dehydrate:yes +numeric_name_metadata hydrate:yes dehydrate:yes temporal hydrate:yes dehydrate:yes containers hydrate:yes dehydrate:yes enum_and_null hydrate:yes dehydrate:yes diff --git a/src/extension/flow-php-ext/tests/phpt/056_fused_decode_rows.phpt b/src/extension/flow-php-ext/tests/phpt/056_fused_decode_rows.phpt new file mode 100644 index 000000000..3e0183108 --- /dev/null +++ b/src/extension/flow-php-ext/tests/phpt/056_fused_decode_rows.phpt @@ -0,0 +1,182 @@ +--TEST-- +RustFloeEncoderNative::decodeRows() yields the same Rows, and refuses with the same message, as hydrate(decode()) +--SKIPIF-- + +--FILE-- + + */ +function bodies(Rows $rows, Schema $fileSchema): array +{ + $schemaBody = json_encode($fileSchema->normalize(), JSON_THROW_ON_ERROR); + + return (new RustFloeEncoderNative())->encode((new NativeRowHydrator())->dehydrate($rows), $schemaBody, $fileSchema); +} + +function outcome(callable $read): string +{ + try { + return serialize($read()); + } catch (Throwable $e) { + $previous = $e->getPrevious(); + + return get_class($e) . ': ' . $e->getMessage() . ($previous === null ? '' : ' <- ' . get_class($previous) . ': ' . $previous->getMessage()); + } +} + +/** + * @param list $bodies + */ +function compare(string $label, array $bodies, string $schemaBody, Schema $schema): void +{ + $twoStep = outcome(static fn(): Rows => (new RustRowHydratorNative())->hydrate((new RustFloeEncoderNative())->decode($bodies, $schemaBody), $schema)); + $fused = outcome(static fn(): Rows => (new RustFloeEncoderNative())->decodeRows($bodies, $schemaBody, $schema)); + + echo $label, ': ', $twoStep === $fused ? 'identical' : "FAIL\n two-step: {$twoStep}\n fused: {$fused}", "\n"; + + if (!str_starts_with($fused, 'O:')) { + echo ' ', $fused, "\n"; + } +} + +function compare_rows(string $label, Rows $rows, ?Schema $readSchema = null): void +{ + $fileSchema = $rows->schema(); + + compare( + $label, + bodies($rows, $fileSchema), + json_encode($fileSchema->normalize(), JSON_THROW_ON_ERROR), + $readSchema ?? $fileSchema, + ); +} + +compare_rows('017 rows', rows( + schema(int_schema('id'), str_schema('name', nullable: true), float_schema('price', nullable: true), datetime_schema('at', nullable: true)), + row(['id' => 1, 'name' => 'a']), + row(['id' => 2, 'name' => null]), + row(['id' => 3, 'price' => 1.5]), + row(['id' => 4, 'at' => new DateTimeImmutable('2025-01-01 00:00:00.123456', new DateTimeZone('Europe/Warsaw'))]), +)); + +compare_rows('021 rows', rows( + schema(int_schema('id'), str_schema('name', nullable: true), float_schema('price')), + ...array_map(static fn(int $i) => row(['id' => $i, 'name' => $i % 7 === 0 ? null : 'user_' . $i, 'price' => $i / 4.0]), range(1, 500)), +)); + +$containers = rows( + schema( + int_schema('id'), + datetime_schema('utc'), + datetime_schema('zoned', nullable: true), + list_schema('tags', type_list(type_string()), nullable: true), + map_schema('scores', type_map(type_string(), type_integer())), + structure_schema('address', type_structure(['street' => type_string(), 'zip' => type_integer()]), nullable: true), + bool_schema('active'), + int_schema('1', nullable: true), + ), + row(['id' => 1, 'utc' => new DateTimeImmutable('2026-03-01T10:11:12.654321Z'), 'zoned' => new DateTimeImmutable('2026-03-01 10:11:12', new DateTimeZone('+02:00')), 'tags' => ['a', 'b'], 'scores' => ['x' => 1], 'address' => ['street' => 'Main', 'zip' => 12345], 'active' => true, '1' => 7]), + row(['id' => 2, 'utc' => new DateTimeImmutable('1970-01-01T00:00:00Z'), 'zoned' => null, 'tags' => null, 'scores' => [], 'address' => null, 'active' => false, '1' => null]), + row(['id' => 3, 'utc' => new DateTimeImmutable('2026-12-31 23:59:59.000001', new DateTimeZone('Europe/Warsaw')), 'zoned' => new DateTimeImmutable('2026-06-01', new DateTimeZone('America/New_York')), 'tags' => [], 'scores' => ['y' => -2, 'z' => 3], 'address' => ['street' => '', 'zip' => 0], 'active' => true, '1' => 0]), +); + +compare_rows('nullable / datetime / list / map / structure / numeric name', $containers); + +$fileSchema = schema(int_schema('id'), str_schema('name', nullable: true), int_schema('7', nullable: true)); +$tagged = rows( + schema(int_schema('id'), str_schema('name', nullable: true, metadata: Metadata::fromArray(['source' => 'crm'])), int_schema('7', nullable: true, metadata: Metadata::fromArray(['n' => 1]))), + row(['id' => 1, 'name' => 'a', '7' => 1]), + row(['id' => 2, 'name' => null, '7' => null]), +); +compare('per-value metadata', bodies($tagged, $fileSchema), json_encode($fileSchema->normalize(), JSON_THROW_ON_ERROR), $fileSchema); + +compare_rows('empty batch', rows(schema(int_schema('id')))); +compare('no frames', [], json_encode(schema(int_schema('id'))->normalize(), JSON_THROW_ON_ERROR), schema(int_schema('id'))); + +$narrow = rows(schema(int_schema('id'), str_schema('name')), row(['id' => 1, 'name' => 'a']), row(['id' => 2, 'name' => 'b'])); +compare_rows('read schema declares a nullable column the frames lack', $narrow, schema(int_schema('id'), str_schema('name'), str_schema('extra', nullable: true))); +compare_rows('read schema declares a NOT NULL column the frames lack', $narrow, schema(int_schema('id'), str_schema('name'), str_schema('extra'))); +compare_rows('read schema drops a decoded column', $narrow, schema(int_schema('id'))); +compare_rows('read schema reorders columns', $narrow, schema(str_schema('name'), int_schema('id'))); +compare_rows('read schema refuses a decoded value', $narrow, schema(int_schema('id'), int_schema('name'))); +compare_rows('read schema refuses a decoded null', rows(schema(int_schema('id'), str_schema('name', nullable: true)), row(['id' => 1, 'name' => null])), schema(int_schema('id'), str_schema('name'))); + +$narrowSchemaBody = json_encode($narrow->schema()->normalize(), JSON_THROW_ON_ERROR); +$narrowBodies = bodies($narrow, $narrow->schema()); +compare('trailing bytes', [$narrowBodies[0] . "\xEF"], $narrowSchemaBody, $narrow->schema()); +compare('truncated body', [substr($narrowBodies[1], 0, -1)], $narrowSchemaBody, $narrow->schema()); +compare('unknown value flag', ["\x09" . substr($narrowBodies[0], 1)], $narrowSchemaBody, $narrow->schema()); + +$duplicatedSchemaBody = json_encode([...$narrow->schema()->normalize(), ...schema(int_schema('id'))->normalize()], JSON_THROW_ON_ERROR); +compare('duplicated decoded name', [$narrowBodies[0] . "\x01" . pack('q', 9)], $duplicatedSchemaBody, $narrow->schema()); + +// the two-step path decodes every body before it casts; the fused path casts each row as it decodes it - a cast +// refusal in row 1 is reported ahead of a corrupt frame in row 2 only by the fused path +$precedence = rows(schema(int_schema('id'), str_schema('name')), row(['id' => 1, 'name' => '1']), row(['id' => 2, 'name' => 'x'])); +$precedenceBody = json_encode($precedence->schema()->normalize(), JSON_THROW_ON_ERROR); +$precedenceBodies = bodies($precedence, $precedence->schema()); +$precedenceInput = [$precedenceBodies[0], $precedenceBodies[1], $precedenceBodies[0] . "\xEF"]; +$castsName = schema(int_schema('id'), int_schema('name')); +echo 'precedence two-step: ', outcome(static fn(): Rows => (new RustRowHydratorNative())->hydrate((new RustFloeEncoderNative())->decode($precedenceInput, $precedenceBody), $castsName)), "\n"; +echo 'precedence fused: ', outcome(static fn(): Rows => (new RustFloeEncoderNative())->decodeRows($precedenceInput, $precedenceBody, $castsName)), "\n"; + +$encoder = new RustFloeEncoderNative(); +$first = $encoder->decodeRows($narrowBodies, $narrowSchemaBody, $narrow->schema()); +$second = $encoder->decodeRows(bodies($containers, $containers->schema()), json_encode($containers->schema()->normalize(), JSON_THROW_ON_ERROR), $containers->schema()); +$third = $encoder->decodeRows($narrowBodies, $narrowSchemaBody, $narrow->schema()); +echo 'plans rebind across schemas: ', $first->count(), ' ', $second->count(), ' ', serialize($first) === serialize($third) ? 'identical' : 'FAIL', "\n"; +?> +--EXPECT-- +017 rows: identical +021 rows: identical +nullable / datetime / list / map / structure / numeric name: identical +per-value metadata: identical +empty batch: identical +no frames: identical +read schema declares a nullable column the frames lack: identical +read schema declares a NOT NULL column the frames lack: identical + Flow\ETL\Exception\SchemaMismatchException: Rows do not match their schema: column "extra" (row 0) declared by the schema is missing from the row <- Flow\ETL\Exception\ColumnMismatchException: Row does not match its schema: column "extra" declared by the schema is missing from the row +read schema drops a decoded column: identical +read schema reorders columns: identical +read schema refuses a decoded value: identical + Flow\ETL\Exception\SchemaMismatchException: Rows do not match their schema: column "name" (row 0): could not convert 'a' (string) to integer <- Flow\ETL\Exception\ColumnMismatchException: Row does not match its schema: column "name": could not convert 'a' (string) to integer +read schema refuses a decoded null: identical + Flow\ETL\Exception\SchemaMismatchException: Rows do not match their schema: column "name" (row 0): could not convert null to string, column is not nullable <- Flow\ETL\Exception\ColumnMismatchException: Row does not match its schema: column "name": could not convert null to string, column is not nullable +trailing bytes: identical + Flow\Floe\Exception\ExtensionException: flow_php row frame length does not match its content +truncated body: identical + Flow\Floe\Exception\ExtensionException: flow_php frame body is truncated, string value is incomplete +unknown value flag: identical + Flow\Floe\Exception\ExtensionException: flow_php found unknown value flag 0x09 +duplicated decoded name: identical + Flow\Floe\Exception\ExtensionException: flow_php found duplicated entry name "id" in a row frame +precedence two-step: Flow\Floe\Exception\ExtensionException: flow_php row frame length does not match its content +precedence fused: Flow\ETL\Exception\SchemaMismatchException: Rows do not match their schema: column "name" (row 1): could not convert 'x' (string) to integer <- Flow\ETL\Exception\ColumnMismatchException: Row does not match its schema: column "name": could not convert 'x' (string) to integer +plans rebind across schemas: 2 3 identical diff --git a/src/extension/flow-php-ext/tests/phpt/057_decode_rows_no_leaks.phpt b/src/extension/flow-php-ext/tests/phpt/057_decode_rows_no_leaks.phpt new file mode 100644 index 000000000..e25fe89d0 --- /dev/null +++ b/src/extension/flow-php-ext/tests/phpt/057_decode_rows_no_leaks.phpt @@ -0,0 +1,98 @@ +--TEST-- +repeated fused decodeRows, refusals included, does not leak memory +--SKIPIF-- + +--FILE-- + type_integer(), 'tags' => type_list(type_string())])), + json_schema('json'), +); + +$batch = []; + +for ($i = 1; $i <= 100; $i++) { + $batch[] = new RawRowValues( + [ + 'id' => $i, + 'name' => $i % 10 === 0 ? null : 'user_' . $i, // null values under a nullable declaration + 'price' => (float) $i / 100, + 'active' => $i % 3 === 0, + 'created_at' => new DateTimeImmutable('2025-01-01 00:00:00.123456', new DateTimeZone('Europe/Warsaw')), + 'duration' => new DateInterval('PT1H2M3S'), + 'uuid' => new Flow\Types\Value\Uuid('01234567-89ab-4def-8123-456789abcdef'), + 'tags' => ['x', 'user_' . $i], + 'metrics' => ['cpu' => 0.5, 'mem' => 0.25], + 'nested' => ['a' => $i, 'tags' => ['x']], + 'json' => Flow\Types\Value\Json::fromArray([$i, ['a' => true]]), + ], + $i % 5 === 0 ? ['id' => Metadata::fromArray(['batch' => $i])] : [], // exercises the clone + setMetadata path + ); +} + +$schemaBody = json_encode($schema->normalize(), JSON_THROW_ON_ERROR); +// the per-row metadata diverges from the file schema, so the bodies carry VALUE_*_WITH_META flags +$bodies = (new RustFloeEncoderNative())->encode((new RustRowHydratorNative())->dehydrate((new PhpRowHydrator())->hydrate($batch, $schema)), $schemaBody, $schema); +$refusing = schema(int_schema('id'), int_schema('name', nullable: true)); + +$cycle = static function () use ($bodies, $schemaBody, $schema, $refusing): void { + $native = new RustFloeEncoderNative(); + $native->decodeRows($bodies, $schemaBody, $schema); + + try { + $native->decodeRows($bodies, $schemaBody, $refusing); + } catch (Flow\ETL\Exception\SchemaMismatchException) { + } +}; + +for ($i = 0; $i < 10; $i++) { + $cycle(); +} +gc_collect_cycles(); +$baseline = memory_get_usage(false); + +for ($i = 0; $i < 100; $i++) { + $cycle(); +} +gc_collect_cycles(); + +var_dump(memory_get_usage(false) <= $baseline); +?> +--EXPECT-- +bool(true) diff --git a/src/extension/flow-php-ext/tests/phpt/058_datetime_from_timestamp_parity.phpt b/src/extension/flow-php-ext/tests/phpt/058_datetime_from_timestamp_parity.phpt new file mode 100644 index 000000000..3cd36c4eb --- /dev/null +++ b/src/extension/flow-php-ext/tests/phpt/058_datetime_from_timestamp_parity.phpt @@ -0,0 +1,104 @@ +--TEST-- +native datetime decode builds the same DateTimeImmutable as createFromFormat('U.u')->setTimezone(), state and behaviour +--SKIPIF-- + +--FILE-- +format('Y-m-d H:i:s.u T e P p O U I Z L N z t'), + $value->getTimestamp(), + $value->getOffset(), + $value->getTimezone()->getName(), + $value->modify('+1 month')->format('Y-m-d H:i:s.u T e'), + $value->modify('midnight')->format('Y-m-d H:i:s.u T e U'), + $value->modify('last day of next month noon')->format('Y-m-d H:i:s.u T e'), + $value->setTime(1, 2, 3, 4)->format('Y-m-d H:i:s.u T e U'), + $value->setDate(2024, 2, 29)->format('Y-m-d H:i:s.u T e U'), + $value->add(new DateInterval('P1DT25H'))->format('Y-m-d H:i:s.u T e U'), + $value->sub(new DateInterval('P1M'))->format('Y-m-d H:i:s.u T e U'), + $value->diff($fixed)->format('%R %y %m %d %h %i %s %f %a'), + $value->setTimezone(new DateTimeZone('UTC'))->format('Y-m-d H:i:s.u T e'), + $value->setTimestamp(86400)->format('Y-m-d H:i:s.u T e'), + $value == $fixed, + $value < $fixed, + ]; +} + +$timezones = ['UTC', 'Z', 'Europe/Warsaw', 'America/New_York', 'Australia/Lord_Howe', '+00:00', '+02:00', '-05:30', 'CEST', 'EST']; +$instants = [ + '1970-01-01 00:00:00.000000 UTC', + '1969-12-31 23:59:59.999999 UTC', + '1969-07-20 20:17:00.5 UTC', + '2026-03-29 00:59:59.999999 UTC', + '2026-03-29 01:00:00.000001 UTC', + '2026-10-25 00:30:00 UTC', + '2026-10-25 01:30:00 UTC', + '2038-01-19 03:14:08.000001 UTC', + '0001-01-01 00:00:00 UTC', + '9999-12-31 23:59:59.999999 UTC', + '-0044-03-15 12:00:00 UTC', +]; + +$schema = schema(datetime_schema('at')); +$schemaBody = json_encode($schema->normalize(), JSON_THROW_ON_ERROR); +$values = []; + +foreach ($timezones as $timezone) { + foreach ($instants as $instant) { + $values[] = (new DateTimeImmutable($instant))->setTimezone(new DateTimeZone($timezone)); + } +} + +$bodies = (new PhpFloeEncoder($schema))->encode((new PhpRowHydrator())->dehydrate(rows($schema, ...array_map(static fn($at) => row(['at' => $at]), $values)))); +$php = (new PhpFloeEncoder($schema))->decode($bodies); +$native = (new RustFloeEncoderNative())->decode($bodies, $schemaBody); + +$differences = 0; + +foreach ($php as $index => $expected) { + if (observe($expected->values['at']) !== observe($native[$index]->values['at'])) { + $differences++; + echo 'FAIL: ', $values[$index]->format('Y-m-d H:i:s.u e'), "\n"; + var_dump(array_diff_assoc(observe($expected->values['at']), observe($native[$index]->values['at']))); + } +} + +echo count($php), ' values, ', $differences, " differences\n"; + +$overlong = pack('P', 0) . pack('V', 1_000_000) . pack('V', 3) . 'UTC'; + +try { + (new RustFloeEncoderNative())->decode(["\x01" . $overlong], $schemaBody); + echo "FAIL: no exception\n"; +} catch (Flow\Floe\Exception\ExtensionException $e) { + echo $e->getMessage(), "\n"; +} + +try { + (new PhpFloeEncoder($schema))->decode(["\x01" . $overlong]); + echo "FAIL: no exception\n"; +} catch (Flow\Floe\Exception\FloeException $e) { + echo $e->getMessage(), "\n"; +} +?> +--EXPECT-- +110 values, 0 differences +flow_php failed to restore datetime from timestamp "0" +Floe failed to restore datetime from timestamp "0" diff --git a/src/extension/flow-php-ext/tests/phpt/059_encode_frames.phpt b/src/extension/flow-php-ext/tests/phpt/059_encode_frames.phpt new file mode 100644 index 000000000..270386d37 --- /dev/null +++ b/src/extension/flow-php-ext/tests/phpt/059_encode_frames.phpt @@ -0,0 +1,137 @@ +--TEST-- +RustFloeEncoderNative::encodeFrames() writes the bytes, and refuses with the message, of framing encode(dehydrate()) +--SKIPIF-- + +--FILE-- +getMessage(); + } +} + +function compare(string $label, Rows $rows, ?Schema $sessionSchema = null): void +{ + $sessionSchema ??= $rows->schema(); + $schemaBody = json_encode($sessionSchema->normalize(), JSON_THROW_ON_ERROR); + + $twoStep = outcome(static fn(): string => Format::rowFrames( + (new RustFloeEncoderNative())->encode((new RustRowHydratorNative())->dehydrate($rows), $schemaBody, $sessionSchema), + )); + $fused = outcome(static fn(): string => (new RustFloeEncoderNative())->encodeFrames($rows, $schemaBody, $sessionSchema)); + + echo $label, ': ', $twoStep === $fused ? 'identical' : "FAIL\n two-step: {$twoStep}\n fused: {$fused}", "\n"; + + if ($fused !== '' && !ctype_xdigit($fused)) { + echo ' ', $fused, "\n"; + } +} + +compare('017 rows', rows( + schema(int_schema('id'), str_schema('name', nullable: true), float_schema('price', nullable: true), datetime_schema('at', nullable: true)), + row(['id' => 1, 'name' => 'a']), + row(['id' => 2, 'name' => null]), + row(['id' => 3, 'price' => 1.5]), + row(['id' => 4, 'at' => new DateTimeImmutable('2025-01-01 00:00:00.123456', new DateTimeZone('Europe/Warsaw'))]), +)); + +compare('021 rows', rows( + schema(int_schema('id'), str_schema('name', nullable: true), float_schema('price')), + ...array_map(static fn(int $i) => row(['id' => $i, 'name' => $i % 7 === 0 ? null : 'user_' . $i, 'price' => $i / 4.0]), range(1, 500)), +)); + +compare('034 nullable note', rows( + schema(int_schema('id'), str_schema('name'), str_schema('note', nullable: true)), + row(['id' => 1, 'name' => 'a', 'note' => null]), +)); + +compare('nullable / datetime / list / map / structure / numeric name', rows( + schema( + int_schema('id'), + datetime_schema('zoned', nullable: true), + list_schema('tags', type_list(type_string()), nullable: true), + map_schema('scores', type_map(type_string(), type_integer())), + structure_schema('address', type_structure(['street' => type_string(), 'zip' => type_integer()]), nullable: true), + bool_schema('active'), + int_schema('1', nullable: true), + ), + row(['id' => 1, 'zoned' => new DateTimeImmutable('2026-03-01 10:11:12', new DateTimeZone('+02:00')), 'tags' => ['a', 'b'], 'scores' => ['x' => 1], 'address' => ['street' => 'Main', 'zip' => 12345], 'active' => true, '1' => 7]), + row(['id' => 2, 'zoned' => null, 'tags' => null, 'scores' => [], 'address' => null, 'active' => false, '1' => null]), +)); + +$sessionSchema = schema(int_schema('id'), str_schema('name', nullable: true), int_schema('7', nullable: true)); +compare('batch metadata diverging from the session schema', rows( + schema(int_schema('id'), str_schema('name', nullable: true, metadata: Metadata::fromArray(['source' => 'crm'])), int_schema('7', nullable: true, metadata: Metadata::fromArray(['n' => 1]))), + row(['id' => 1, 'name' => 'a', '7' => 1]), + row(['id' => 2, 'name' => null, '7' => null]), +), $sessionSchema); + +compare('session schema carrying metadata', rows( + schema(int_schema('id', metadata: Metadata::fromArray(['k' => 'v']))), + row(['id' => 1]), +)); + +compare('empty Rows', rows(schema(int_schema('id')))); + +$schema = schema(int_schema('id'), str_schema('name')); +compare('null under NOT NULL', Rows::trusted($schema, [row(['id' => 1, 'name' => 'a']), row(['id' => 2, 'name' => null])])); +compare('row without a declared column', Rows::trusted($schema, [row(['id' => 1])])); + +// the two-step path dehydrates every row before it encodes; the fused path encodes each row as it reads it - a null +// under NOT NULL in row 1 is reported ahead of a missing column in row 2 only by the fused path +$precedence = Rows::trusted($schema, [row(['id' => 1, 'name' => 'a']), row(['id' => 2, 'name' => null]), row(['id' => 3])]); +$precedenceBody = json_encode($schema->normalize(), JSON_THROW_ON_ERROR); +echo 'precedence two-step: ', outcome(static fn(): string => Format::rowFrames((new RustFloeEncoderNative())->encode((new RustRowHydratorNative())->dehydrate($precedence), $precedenceBody, $schema))), "\n"; +echo 'precedence fused: ', outcome(static fn(): string => (new RustFloeEncoderNative())->encodeFrames($precedence, $precedenceBody, $schema)), "\n"; + +$encoder = new RustFloeEncoderNative(); +$small = rows($schema, row(['id' => 1, 'name' => 'a'])); +$first = $encoder->encodeFrames($small, json_encode($schema->normalize(), JSON_THROW_ON_ERROR), $schema); +$other = schema(int_schema('id')); +$encoder->encodeFrames(rows($other, row(['id' => 9])), json_encode($other->normalize(), JSON_THROW_ON_ERROR), $other); +echo 'plans rebind across schemas: ', $first === $encoder->encodeFrames($small, json_encode($schema->normalize(), JSON_THROW_ON_ERROR), $schema) ? 'identical' : 'FAIL', "\n"; +?> +--EXPECT-- +017 rows: identical +021 rows: identical +034 nullable note: identical +nullable / datetime / list / map / structure / numeric name: identical +batch metadata diverging from the session schema: identical +session schema carrying metadata: identical +empty Rows: identical +null under NOT NULL: identical + Flow\ETL\Exception\SchemaMismatchException: Rows do not match their schema: column "name" (row 1): could not convert null to string, column is not nullable +row without a declared column: identical + Flow\Floe\Exception\ExtensionException: flow_php found a row that does not carry the declared column "name" +precedence two-step: Flow\Floe\Exception\ExtensionException: flow_php found a row that does not carry the declared column "name" +precedence fused: Flow\ETL\Exception\SchemaMismatchException: Rows do not match their schema: column "name" (row 1): could not convert null to string, column is not nullable +plans rebind across schemas: identical diff --git a/src/extension/flow-php-ext/tests/phpt/060_encode_frames_no_leaks.phpt b/src/extension/flow-php-ext/tests/phpt/060_encode_frames_no_leaks.phpt new file mode 100644 index 000000000..5a97dc7de --- /dev/null +++ b/src/extension/flow-php-ext/tests/phpt/060_encode_frames_no_leaks.phpt @@ -0,0 +1,98 @@ +--TEST-- +repeated fused encodeFrames, refusals included, does not leak memory +--SKIPIF-- + +--FILE-- + type_integer(), 'tags' => type_list(type_string())])), + json_schema('json'), +); + +$batch = []; + +for ($i = 1; $i <= 100; $i++) { + $batch[] = new RawRowValues( + [ + 'id' => $i, + 'name' => $i % 10 === 0 ? null : 'user_' . $i, // null values under a nullable declaration + 'price' => (float) $i / 100, + 'active' => $i % 3 === 0, + 'created_at' => new DateTimeImmutable('2025-01-01 00:00:00.123456', new DateTimeZone('Europe/Warsaw')), + 'duration' => new DateInterval('PT1H2M3S'), + 'uuid' => new Flow\Types\Value\Uuid('01234567-89ab-4def-8123-456789abcdef'), + 'tags' => ['x', 'user_' . $i], + 'metrics' => ['cpu' => 0.5, 'mem' => 0.25], + 'nested' => ['a' => $i, 'tags' => ['x']], + 'json' => Flow\Types\Value\Json::fromArray([$i, ['a' => true]]), + ], + $i % 5 === 0 ? ['id' => Metadata::fromArray(['batch' => $i])] : [], // exercises the clone + setMetadata path + ); +} + +$schemaBody = json_encode($schema->normalize(), JSON_THROW_ON_ERROR); +// the hydrated batch schema carries id's folded metadata, which the session schema does not - VALUE_*_WITH_META frames +$rows = (new PhpRowHydrator())->hydrate($batch, $schema); +$refused = Flow\ETL\Rows::trusted($rows->schema(), [...$rows->all(), row(['id' => 101, 'name' => 'x'])]); + +$cycle = static function () use ($rows, $refused, $schemaBody, $schema): void { + $native = new RustFloeEncoderNative(); + $native->encodeFrames($rows, $schemaBody, $schema); + + try { + $native->encodeFrames($refused, $schemaBody, $schema); + } catch (Flow\Floe\Exception\ExtensionException) { + } +}; + +for ($i = 0; $i < 10; $i++) { + $cycle(); +} +gc_collect_cycles(); +$baseline = memory_get_usage(false); + +for ($i = 0; $i < 100; $i++) { + $cycle(); +} +gc_collect_cycles(); + +var_dump(memory_get_usage(false) <= $baseline); +?> +--EXPECT-- +bool(true) From 066a039020684c635d1309b23d05a5faf1571ce9 Mon Sep 17 00:00:00 2001 From: Norbert Orzechowicz Date: Thu, 24 Sep 2026 15:25:49 +0200 Subject: [PATCH 2/2] fix(flow-php/etl): native encoder tests without the flow_php extension - data provider no longer builds native hydrators before the skip - bump Blackfire agent and PHP probe on macOS arm64 --- shell.nix | 46 +++++++++++++------ .../Floe/Tests/Unit/NativeFloeEncoderTest.php | 18 ++++---- 2 files changed, 43 insertions(+), 21 deletions(-) diff --git a/shell.nix b/shell.nix index 13a773a29..175bbdf56 100644 --- a/shell.nix +++ b/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 {} ) ]; }; diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/NativeFloeEncoderTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/NativeFloeEncoderTest.php index 8332553fc..83db5c9b0 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/NativeFloeEncoderTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/NativeFloeEncoderTest.php @@ -8,7 +8,6 @@ use DateTimeZone; use Flow\ETL\Exception\SchemaMismatchException; use Flow\ETL\Row\AdaptiveRowHydrator; -use Flow\ETL\Row\Hydrator; use Flow\ETL\Row\NativeRowHydrator; use Flow\ETL\Row\PhpRowHydrator; use Flow\ETL\Row\TypedRowValues; @@ -101,18 +100,21 @@ public function test_native_metadata_bearing_frames_match_the_php_engine(): void } /** - * @return array + * Flags, not instances: PHPUnit builds provider data before setUp() skips, and a native hydrator cannot be + * constructed without the extension. + * + * @return array */ public static function native_hydrators(): array { return [ - 'native' => [new NativeRowHydrator()], - 'adaptive' => [new AdaptiveRowHydrator()], + 'native' => [false], + 'adaptive' => [true], ]; } #[DataProvider('native_hydrators')] - public function test_decode_rows_matches_hydrate_of_decode(Hydrator $hydrator): void + public function test_decode_rows_matches_hydrate_of_decode(bool $adaptive): void { $data = rows( schema(int_schema('id'), str_schema('name', nullable: true), datetime_schema('at')), @@ -129,7 +131,7 @@ public function test_decode_rows_matches_hydrate_of_decode(Hydrator $hydrator): static::assertEquals( (new NativeRowHydrator())->hydrate($encoder->decode($bodies), $schema), - $encoder->decodeRows($bodies, $schema, $hydrator), + $encoder->decodeRows($bodies, $schema, $adaptive ? new AdaptiveRowHydrator() : new NativeRowHydrator()), ); } @@ -175,14 +177,14 @@ public function test_decode_rows_turns_a_corrupt_frame_into_a_floe_exception(): } #[DataProvider('native_hydrators')] - public function test_encode_frames_matches_framing_encode_of_dehydrate(Hydrator $hydrator): void + public function test_encode_frames_matches_framing_encode_of_dehydrate(bool $adaptive): void { $data = RowsMother::numbered(4); $encoder = new NativeFloeEncoder($data->schema()); static::assertSame( Format::rowFrames($encoder->encode((new PhpRowHydrator())->dehydrate($data))), - $encoder->encodeFrames($data, $hydrator), + $encoder->encodeFrames($data, $adaptive ? new AdaptiveRowHydrator() : new NativeRowHydrator()), ); }