diff --git a/documentation/components/core/floe.md b/documentation/components/core/floe.md index 5720649815..77be3c897b 100644 --- a/documentation/components/core/floe.md +++ b/documentation/components/core/floe.md @@ -41,7 +41,7 @@ data_frame() When the [`flow_php` extension](/documentation/components/extensions/flow-php-ext.md) is loaded, the strict read and the write fuse frame-split + value decode/encode + hydrate/dehydrate into one native call per batch. It is transparent — the on-disk format is unchanged and the rows are byte-for-byte -identical to the pure-PHP engine; the salvage path (`FloeStreamReader::recover()`) always stays pure PHP. +identical to the pure-PHP engine. ## Save Modes @@ -200,8 +200,8 @@ guidance live in the [caching documentation](/documentation/components/core/cach ## On-Disk Layout A `.floe` file is a fixed 6-byte header, a stream of length-prefixed frames, and a JSON footer that a -reader can locate from the last 8 bytes without scanning the body. All multi-byte integers are -**little-endian**. +reader can locate from the last 8 bytes without scanning the body. The schema lives **only in the +footer** — there is no inline schema frame. All multi-byte integers are **little-endian**. ``` ┌──────────────────────────────────────────────────────────┐ @@ -209,7 +209,6 @@ reader can locate from the last 8 bytes without scanning the body. All multi-byt ├──────────────────────────────────────────────────────────┤ │ FRAMES (repeated, in write order) │ │ PARTITIONS (0x03) emitted when combination changes │ -│ SCHEMA (0x01) written once, before first ROW │ │ ROW (0x02) one frame per row │ │ ROW (0x02) │ │ … │ @@ -232,7 +231,7 @@ reader can locate from the last 8 bytes without scanning the body. All multi-byt ### Frame envelope -Every frame — `SCHEMA`, `ROW`, `PARTITIONS`, `FOOTER` — shares the same envelope: a 1-byte type, a +Every frame — `ROW`, `PARTITIONS`, `FOOTER` — shares the same envelope: a 1-byte type, a 4-byte little-endian body length, then the body. ``` @@ -240,26 +239,35 @@ Every frame — `SCHEMA`, `ROW`, `PARTITIONS`, `FOOTER` — shares the same enve │ type │ body length │ body │ │ 1 byte │ 4 bytes (LE) │ bytes │ └────────┴───────────────┴───────────────────────┘ - 0x01 SCHEMA 0x02 ROW 0x03 PARTITIONS 0x06 FOOTER + 0x02 ROW 0x03 PARTITIONS 0x06 FOOTER ``` -- **SCHEMA (`0x01`)** — body is the JSON schema for the file. A file carries **exactly one schema**, - fixed at session start (explicit, or the first batch's union), written once before the first section. - Rows narrower than the schema ride it (see the ROW absent flag). A later batch that introduces a new - column or an incompatible type throws `IncompatibleSchemaException` — schema evolution lives only in - `merge_floe()`. +The file's schema is not a frame — it carries **exactly one schema**, fixed at session start (explicit, +or the first batch's union) and stored only in the footer (see `schema` below). A row narrower than the +schema rides it (see the ROW absent flag). A later batch that introduces a new column or an incompatible +type throws `IncompatibleSchemaException` — schema evolution lives only in `merge_floe()`. + - **ROW (`0x02`)** — one row encoded **by column** against the file schema: for each schema - column, in order, a one-byte presence flag — `0x01` present (followed by the value encoded per the - column's schema type, no per-value tag), `0x00` null, `0x02` null-from-null, or `0x03` **absent** (the - row has no such column). Only dynamically typed values (mixed/union columns, dynamic map keys) carry a - one-byte type tag (`NULL`, `INTEGER`, `FLOAT`, `BOOLEAN`, `STRING`, `ARRAY`, `DATETIME`, `UUID`, - `JSON`). A row whose columns exactly match the section produces the same bytes as a plain positional - encode. One frame per row. + column, in order, a one-byte presence flag — + - `0x01` **present** — followed by the value encoded per the column's schema type, no per-value tag. + - `0x00` **null** — the value is null. + - `0x02` **null with metadata** — a null value carrying per-value metadata that diverges from the + column's schema metadata: a metadata block (4-byte little-endian JSON length + JSON), no value. + - `0x03` **absent** — the row has no such column. + - `0x04` **present with metadata** — a divergent-metadata block (4-byte length + JSON) followed by + the encoded value. + + The two metadata flags (`0x02`, `0x04`) appear only when a row's per-value metadata differs from the + column's schema metadata; otherwise every value uses `0x00`/`0x01`. Only dynamically typed values + (mixed/union columns, dynamic map keys) carry a one-byte type tag (`NULL`, `INTEGER`, `FLOAT`, + `BOOLEAN`, `STRING`, `ARRAY`, `DATETIME`, `UUID`, `JSON`). A row whose columns exactly match the + section, with no divergent metadata, produces the same bytes as a plain positional encode. One frame + per row. - **PARTITIONS (`0x03`)** — the partition key/value pairs for the section that follows: a 4-byte count followed by repeated `[nameLen(4), name, valueLen(4), value]`. Written at the start of every section - whose combination differs from the previous one (mirror of the `SCHEMA`-frame dedup); the reader - starts at the empty combination, so an unpartitioned first section emits none and a later change back - to unpartitioned emits a `count=0` frame. + whose combination differs from the previous one; the reader starts at the empty combination, so an + unpartitioned first section emits none and a later change back to unpartitioned emits a `count=0` + frame. - **FOOTER (`0x06`)** — the footer JSON followed by the trailer (below). ### Footer @@ -285,10 +293,10 @@ scanning rows**: - **`partitions`** is the deduplicated, order-preserving table of partition combinations, indexed by `partitionsId`; the unpartitioned combination is an empty object at its own id. A file can hold many combinations (one per section). -- **`schema`** is the file's single schema, available from the footer alone. On a seamless read, a row - narrower than it (an absent column) is padded with null entries so every yielded row conforms; - `recover()` reads rows exactly as written, absent columns omitted. `merge_floe()` re-encodes drifted - sources to their union, so a merged file is still one schema. +- **`schema`** is the file's single schema and the sole place it is stored, so every read decodes + against it without scanning the body. On a read, a row narrower than it (an absent column) is padded + with null entries so every yielded row conforms. `merge_floe()` re-encodes drifted sources to their + union, so a merged file is still one schema. ### Trailer (last 8 bytes) diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/EnumEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/EnumEntry.php index 9e00eb234f..e927bc5bf8 100644 --- a/src/core/etl/src/Flow/ETL/Row/Entry/EnumEntry.php +++ b/src/core/etl/src/Flow/ETL/Row/Entry/EnumEntry.php @@ -4,6 +4,7 @@ namespace Flow\ETL\Row\Entry; +use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Row\Entry; use Flow\ETL\Row\Reference; use Flow\ETL\Schema\Definition\EnumDefinition; @@ -35,6 +36,10 @@ public function __construct( private readonly ?UnitEnum $value, ?Metadata $metadata = null, ) { + if ('' === $name) { + throw InvalidArgumentException::because('Entry name cannot be empty'); + } + $this->definition = self::buildDefinition($this->name, $this->value, $metadata); } diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/HTMLElementEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/HTMLElementEntry.php index 8d33320a68..d23e7d9853 100644 --- a/src/core/etl/src/Flow/ETL/Row/Entry/HTMLElementEntry.php +++ b/src/core/etl/src/Flow/ETL/Row/Entry/HTMLElementEntry.php @@ -6,6 +6,7 @@ use Dom\HTMLDocument; use Dom\HTMLElement; +use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Exception\RuntimeException; use Flow\ETL\Row\Entry; use Flow\ETL\Row\Reference; @@ -39,6 +40,10 @@ public function __construct( private readonly ?HTMLElement $value, ?Metadata $metadata = null, ) { + if ('' === $name) { + throw InvalidArgumentException::because('Entry name cannot be empty'); + } + $this->definition = new HTMLElementDefinition( $this->name, $this->value === null, diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/HTMLEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/HTMLEntry.php index 6901acca31..e016eb0925 100644 --- a/src/core/etl/src/Flow/ETL/Row/Entry/HTMLEntry.php +++ b/src/core/etl/src/Flow/ETL/Row/Entry/HTMLEntry.php @@ -5,6 +5,7 @@ namespace Flow\ETL\Row\Entry; use Dom\HTMLDocument; +use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Exception\RuntimeException; use Flow\ETL\Row\Entry; use Flow\ETL\Row\Reference; @@ -37,6 +38,10 @@ public function __construct( private readonly ?HTMLDocument $value, ?Metadata $metadata = null, ) { + if ('' === $name) { + throw InvalidArgumentException::because('Entry name cannot be empty'); + } + $this->definition = new HTMLDefinition($this->name, null === $this->value, $metadata ?: Metadata::empty()); } diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/XMLElementEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/XMLElementEntry.php index a790e73f0c..a19c984222 100644 --- a/src/core/etl/src/Flow/ETL/Row/Entry/XMLElementEntry.php +++ b/src/core/etl/src/Flow/ETL/Row/Entry/XMLElementEntry.php @@ -44,6 +44,10 @@ public function __construct( private readonly DOMElement|Element|null $value, ?Metadata $metadata = null, ) { + if ('' === $name) { + throw InvalidArgumentException::because('Entry name cannot be empty'); + } + $this->definition = new XMLElementDefinition( $this->name, $this->value === null, diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/XMLEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/XMLEntry.php index d0c7dc1bb6..0f6afcc04f 100644 --- a/src/core/etl/src/Flow/ETL/Row/Entry/XMLEntry.php +++ b/src/core/etl/src/Flow/ETL/Row/Entry/XMLEntry.php @@ -42,6 +42,10 @@ public function __construct( private readonly DOMDocument|XMLDocument|null $value, ?Metadata $metadata = null, ) { + if ('' === $name) { + throw InvalidArgumentException::because('Entry name cannot be empty'); + } + $this->definition = new XMLDefinition($this->name, $this->value === null, $metadata ?: Metadata::empty()); } diff --git a/src/core/etl/src/Flow/Floe/FloeMerger.php b/src/core/etl/src/Flow/Floe/FloeMerger.php index b7d0e5f343..c59f00103f 100644 --- a/src/core/etl/src/Flow/Floe/FloeMerger.php +++ b/src/core/etl/src/Flow/Floe/FloeMerger.php @@ -59,11 +59,11 @@ public function merge(array $sources, Path $dest, bool $compact = false, ?Metada } /** - * Splicing copies source frames verbatim, so it is only valid when every - * SCHEMA frame it copies is structurally identical to the merged footer - * schema - column order included, since row bytes are laid out in schema - * order. Sources that merely reconcile (isSame is order-insensitive) are - * re-encoded by the compact path instead. + * Splicing copies source row frames verbatim, so it is only valid when every + * source's schema is structurally identical to the merged footer schema - + * column order included, since row bytes are laid out in schema order. + * Sources that merely reconcile (isSame is order-insensitive) are re-encoded + * by the compact path instead. * * @param array $layouts */ diff --git a/src/core/etl/src/Flow/Floe/FloeStreamReader.php b/src/core/etl/src/Flow/Floe/FloeStreamReader.php index 9ff7c832d4..792a72a2d0 100644 --- a/src/core/etl/src/Flow/Floe/FloeStreamReader.php +++ b/src/core/etl/src/Flow/Floe/FloeStreamReader.php @@ -18,10 +18,8 @@ use Flow\Floe\Exception\FloeException; use Flow\Serializer\Exception\SerializationException; use Generator; -use Throwable; use function count; -use function Flow\ETL\DSL\schema_from_json; use function max; use function ord; use function sprintf; @@ -84,162 +82,9 @@ public function metadata(): Metadata } /** - * Salvages complete frames from the beginning of the file sequentially - no - * footer, no ranged reads. Rows are yielded as they were written (no - * padding); reading ends silently at the first truncated or invalid frame. - * - * @param int<1, max> $batchSize - * - * @throws FloeException - * - * @return \Generator - */ - public function recover(int $batchSize = 1000): Generator - { - $frames = (new FrameReader($this->source, $this->codec->id(), $this->chunkSize))->frames(lenient: true); - - $schema = null; - $partitions = []; - $batch = []; - /** @var list $pending */ - $pending = []; - - foreach ($frames as [$frameType, $frameBody]) { - if ($frameType === Format::FRAME_ROW) { - if ($schema === null) { - break; - } - - try { - $pending[] = $this->codec->decode($frameBody); - } catch (Throwable) { - break; - } - - if (count($pending) === $batchSize) { - [$ready, $stop] = $this->flushRecoverPending($schema, $pending, $batch, $batchSize, $partitions); - $pending = []; - - foreach ($ready as $rows) { - yield $rows; - } - - if ($stop) { - break; - } - } - } elseif ($frameType === Format::FRAME_SCHEMA) { - if ($schema !== null && $pending !== []) { - [$ready, $stop] = $this->flushRecoverPending($schema, $pending, $batch, $batchSize, $partitions); - $pending = []; - - foreach ($ready as $rows) { - yield $rows; - } - - if ($stop) { - break; - } - } - - try { - $schema = $this->decodeSchema($frameBody); - } catch (Throwable) { - break; - } - } elseif ($frameType === Format::FRAME_PARTITIONS) { - if ($schema !== null && $pending !== []) { - [$ready, $stop] = $this->flushRecoverPending($schema, $pending, $batch, $batchSize, $partitions); - $pending = []; - - foreach ($ready as $rows) { - yield $rows; - } - - if ($stop) { - break; - } - } - - if ($batch !== []) { - yield $this->batch($batch, $partitions); - $batch = []; - } - - $partitions = self::decodePartitions($frameBody); - } elseif ($frameType !== Format::FRAME_FOOTER) { - break; - } - } - - if ($schema !== null && $pending !== []) { - [$ready] = $this->flushRecoverPending($schema, $pending, $batch, $batchSize, $partitions); - - foreach ($ready as $rows) { - yield $rows; - } - } - - if ($batch !== []) { - yield $this->batch($batch, $partitions); - } - } - - /** - * A failed batch is replayed row by row so the prefix before the first - * corrupt body still surfaces; the true stop flag ends the recover read. - * - * @param list $pending - * @param array $batch - * @param int<1, max> $batchSize - * @param array $partitions - * - * @return array{0: array, 1: bool} ready batches + stop flag - */ - private function flushRecoverPending( - Schema $schema, - array $pending, - array &$batch, - int $batchSize, - array $partitions, - ): array { - $stop = false; - - try { - $rows = $this->hydrator->hydrate($this->encoder($schema)->decode($pending), $schema)->all(); - } catch (Throwable) { - $rows = []; - - foreach ($pending as $body) { - try { - $rows[] = $this->hydrator->hydrate($this->encoder($schema)->decode([$body]), $schema)->first(); - } catch (Throwable) { - break; - } - } - - $stop = true; - } - - $ready = []; - - foreach ($rows as $row) { - $batch[] = $row; - - if (count($batch) === $batchSize) { - $ready[] = $this->batch($batch, $partitions); - $batch = []; - } - } - - return [$ready, $stop]; - } - - /** - * Hot path: frames are walked in-buffer, unlike recover() which keeps the - * reusable FrameReader. offset skips whole leading sections using the footer, - * then the remaining rows inside the start section; limit stops the read - * after that many rows. + * Hot path: frames are walked in-buffer. offset skips whole leading sections + * using the footer, then the remaining rows inside the start section; limit + * stops the read after that many rows. * * @param int<1, max> $batchSize * @param bool $conform pad/reorder every batch to the merged file schema (default); false yields rows verbatim per section (the raw read used by spill buckets) @@ -294,7 +139,6 @@ public function rows(int $batchSize = 1000, int $offset = 0, ?int $limit = null, $buffer, Format::HEADER_LENGTH, $fileSchema, - $footer->schemaBody(), 0, [], $conform ? RowPadding::forFileSchema($fileSchema, $this->schemaDecoder) : null, @@ -344,9 +188,7 @@ public function tail(int $count, int $batchSize = 1000): Generator /** * The single strict frame-walk shared by rows() and rowsFromOffset(); they * differ only in how it is seeded - start position, skip and partitions. The - * file carries exactly one schema, so decode state comes from the footer and - * inline SCHEMA frames are only validated against it. recover() keeps its - * own lenient FrameReader walk. + * file carries exactly one schema, so decode state comes from the footer. * * @param \Generator $chunks * @param array $partitions @@ -361,7 +203,6 @@ private function walk( string $buffer, int $position, Schema $schema, - string $schemaBody, int $skip, array $partitions, ?RowPadding $padding, @@ -424,12 +265,6 @@ private function walk( return; } } - } elseif ($frameType === Format::FRAME_SCHEMA) { - if (substr($buffer, $position, $frameLength) !== $schemaBody) { - throw new FloeException('Floe found a schema frame that does not match the file schema'); - } - - $position = $frameEnd; } elseif ($frameType === Format::FRAME_PARTITIONS) { $frameBody = substr($buffer, $position, $frameLength); @@ -642,8 +477,8 @@ private function chunksFrom(SourceStream $source, int $startByte, int $size): Ge /** * Offset pushdown: seeks to the start section's byte offset, takes the - * schema from the footer (the file's single SCHEMA frame sits before the - * seek point) and skips the remaining rows inside the start section. + * schema from the footer and skips the remaining rows inside the start + * section. * * @param int<1, max> $batchSize * @param int<1, max> $offset @@ -695,7 +530,6 @@ private function rowsFromOffset(int $batchSize, int $offset, ?int $limit, bool $ '', 0, $fileSchema, - $footer->schemaBody(), $offset - $cumulative, $partitions, $conform ? RowPadding::forFileSchema($fileSchema, $this->schemaDecoder) : null, @@ -725,16 +559,4 @@ private static function decodePartitions(string $body): array return $partitions; } - - /** - * @throws FloeException - */ - private function decodeSchema(string $schemaBody): Schema - { - try { - return schema_from_json($schemaBody); - } catch (Throwable $e) { - throw new FloeException('Floe failed to decode schema JSON: ' . $e->getMessage(), 0, $e); - } - } } diff --git a/src/core/etl/src/Flow/Floe/FloeStreamWriter.php b/src/core/etl/src/Flow/Floe/FloeStreamWriter.php index 15e2c375f1..f20946f61f 100644 --- a/src/core/etl/src/Flow/Floe/FloeStreamWriter.php +++ b/src/core/etl/src/Flow/Floe/FloeStreamWriter.php @@ -16,16 +16,12 @@ use Flow\Floe\Exception\FloeException; use Flow\Floe\Exception\IncompatibleSchemaException; use Flow\Types\Type\Native\NullType; -use JsonException; use function count; use function Flow\Types\DSL\type_equals; use function implode; -use function json_encode; use function sprintf; -use const JSON_THROW_ON_ERROR; - final class FloeStreamWriter { private Metadata $metadata; @@ -56,12 +52,8 @@ final class FloeStreamWriter private bool $sectionOpen = false; - private bool $schemaFrameWritten = false; - private Schema $sessionSchema; - private ?string $sessionSchemaBody = null; - /** * @var null|Encoder */ @@ -118,18 +110,13 @@ public function resume(DestinationStream $stream, Footer $footer, int $fileLengt $this->metadata = $footer->metadata->merge($metadata ?? Metadata::empty()); $this->sections = $footer->sections; - if ($footer->schema !== []) { - if ($this->sessionSchema->normalize() !== $footer->schema()->normalize()) { - throw new IncompatibleSchemaException( - 'Floe append schema does not match the existing file schema. ' - . 'Align the pipeline with DataFrame::match($schema) before appending.', - ); - } - - $this->sessionSchemaBody = $footer->schemaBody(); + if ($footer->schema !== [] && $this->sessionSchema->normalize() !== $footer->schema()->normalize()) { + throw new IncompatibleSchemaException( + 'Floe append schema does not match the existing file schema. ' + . 'Align the pipeline with DataFrame::match($schema) before appending.', + ); } - $this->schemaFrameWritten = $footer->sections !== []; $this->lastSectionPartitionsId = $lastSection?->partitionsId; $this->partitions = $footer->partitions; $this->totalRows = $footer->totalRows; @@ -228,22 +215,9 @@ private function openSession(): void return; } - $this->sessionSchemaBody ??= self::encodeSchemaBody($this->sessionSchema); $this->sessionEncoder = $this->engine->encoder($this->sessionSchema); } - /** - * @throws FloeException - */ - private static function encodeSchemaBody(Schema $schema): string - { - try { - return json_encode($schema->normalize(), JSON_THROW_ON_ERROR); - } catch (JsonException $e) { - throw new FloeException('Floe failed to encode schema as JSON: ' . $e->getMessage(), 0, $e); - } - } - /** * @throws IncompatibleSchemaException */ @@ -380,13 +354,6 @@ private function startSection(): void $this->frameWriter()->partitions($this->partitions[$partitionsId]); } - if (!$this->schemaFrameWritten) { - $this->frameWriter()->schema( - $this->sessionSchemaBody ?? throw new FloeException('Floe writer has no session schema body'), - ); - $this->schemaFrameWritten = true; - } - $this->sectionOpen = true; $this->sectionPartitionsId = $partitionsId; $this->sectionRowCount = 0; diff --git a/src/core/etl/src/Flow/Floe/Footer.php b/src/core/etl/src/Flow/Floe/Footer.php index 9832b0bd20..f382f3e351 100644 --- a/src/core/etl/src/Flow/Floe/Footer.php +++ b/src/core/etl/src/Flow/Floe/Footer.php @@ -228,6 +228,10 @@ public function toJson(): string // metadata is a map: an empty one must encode as a JSON object ({}), not a list ([]) $data['metadata'] = (object) $data['metadata']; - return json_encode($data, JSON_THROW_ON_ERROR); + try { + return json_encode($data, JSON_THROW_ON_ERROR); + } catch (JsonException $e) { + throw new FloeException('Floe failed to encode schema as JSON: ' . $e->getMessage(), 0, $e); + } } } diff --git a/src/core/etl/src/Flow/Floe/Format.php b/src/core/etl/src/Flow/Floe/Format.php index a8fe35b69e..ccced274d1 100644 --- a/src/core/etl/src/Flow/Floe/Format.php +++ b/src/core/etl/src/Flow/Floe/Format.php @@ -33,8 +33,6 @@ final class Format public const int FRAME_HEADER_LENGTH = 5; - public const int FRAME_SCHEMA = 0x01; - public const int FRAME_ROW = 0x02; public const int FRAME_PARTITIONS = 0x03; diff --git a/src/core/etl/src/Flow/Floe/FrameWriter.php b/src/core/etl/src/Flow/Floe/FrameWriter.php index 0890fcf5c2..79c55df29f 100644 --- a/src/core/etl/src/Flow/Floe/FrameWriter.php +++ b/src/core/etl/src/Flow/Floe/FrameWriter.php @@ -39,13 +39,6 @@ public function row(string $body): void $this->flushIfFull(); } - public function schema(string $schemaBody): void - { - $this->buffer .= Format::frame(Format::FRAME_SCHEMA, $schemaBody); - $this->position += Format::FRAME_HEADER_LENGTH + strlen($schemaBody); - $this->flushIfFull(); - } - /** * @param array $combo */ diff --git a/src/core/etl/tests/Flow/ETL/Tests/Double/ArrayCache.php b/src/core/etl/tests/Flow/ETL/Tests/Double/ArrayCache.php index d32e12cbf3..e71fc8d2f0 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Double/ArrayCache.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Double/ArrayCache.php @@ -9,6 +9,12 @@ use function array_key_exists; +/** + * Parameters are intentionally untyped: psr/simple-cache 1.0 declares CacheInterface without + * parameter types, so narrowing them here is an LSP violation that PHP reports as a fatal error + * whenever the lowest supported version is installed. Untyped parameters stay compatible with + * 1.0, 2.0 and 3.0 alike. + */ final class ArrayCache implements CacheInterface { /** @@ -23,14 +29,20 @@ public function clear(): bool return true; } - public function delete(string $key): bool + /** + * @param string $key + */ + public function delete($key): bool { unset($this->values[$key]); return true; } - public function deleteMultiple(iterable $keys): bool + /** + * @param iterable $keys + */ + public function deleteMultiple($keys): bool { foreach ($keys as $key) { $this->delete($key); @@ -39,12 +51,20 @@ public function deleteMultiple(iterable $keys): bool return true; } - public function get(string $key, mixed $default = null): mixed + /** + * @param string $key + */ + public function get($key, mixed $default = null): mixed { return array_key_exists($key, $this->values) ? $this->values[$key] : $default; } - public function getMultiple(iterable $keys, mixed $default = null): iterable + /** + * @param iterable $keys + * + * @return iterable + */ + public function getMultiple($keys, mixed $default = null): iterable { $result = []; @@ -55,19 +75,30 @@ public function getMultiple(iterable $keys, mixed $default = null): iterable return $result; } - public function has(string $key): bool + /** + * @param string $key + */ + public function has($key): bool { return array_key_exists($key, $this->values); } - public function set(string $key, mixed $value, int|DateInterval|null $ttl = null): bool + /** + * @param string $key + * @param null|DateInterval|int $ttl + */ + public function set($key, mixed $value, $ttl = null): bool { $this->values[$key] = $value; return true; } - public function setMultiple(iterable $values, int|DateInterval|null $ttl = null): bool + /** + * @param iterable $values + * @param null|DateInterval|int $ttl + */ + public function setMultiple($values, $ttl = null): bool { // @mago-ignore analysis:mixed-assignment // @mago-ignore analysis:mixed-assignment diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/EnumEntryTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/EnumEntryTest.php index 6693e6597b..2896d4a1fb 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/EnumEntryTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/EnumEntryTest.php @@ -67,6 +67,13 @@ public function test_rename_preserves_metadata(): void static::assertTrue($renamedEntry->definition()->metadata()->isEqual($metadata)); } + public function test_prevents_from_creating_entry_with_empty_entry_name(): void + { + $this->expectExceptionMessage('Entry name cannot be empty'); + + enum_entry('', BasicEnum::one); + } + public function test_to_string(): void { static::assertSame('one', enum_entry('enum', BasicEnum::one)->toString()); diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/HTMLElementEntryTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/HTMLElementEntryTest.php new file mode 100644 index 0000000000..2bf2d82167 --- /dev/null +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/HTMLElementEntryTest.php @@ -0,0 +1,25 @@ +expectExceptionMessage('Entry name cannot be empty'); + + html_element_entry('', null); + } + + public function test_creating_entry_with_null_value(): void + { + static::assertNull(html_element_entry('element', null)->value()); + static::assertSame('element', html_element_entry('element', null)->name()); + } +} diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/HTMLEntryTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/HTMLEntryTest.php index c7d3d5f8b0..8b774ff9ac 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/HTMLEntryTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/HTMLEntryTest.php @@ -221,4 +221,11 @@ private function assertHtml(string $expected, string $html, bool $equals): void self::assertNotEquals($expected, $html); } } + + public function test_prevents_from_creating_entry_with_empty_entry_name(): void + { + $this->expectExceptionMessage('Entry name cannot be empty'); + + html_entry('', null); + } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/XMLElementEntryTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/XMLElementEntryTest.php index 6942c0f585..e033c43cbd 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/XMLElementEntryTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/XMLElementEntryTest.php @@ -91,4 +91,11 @@ public function test_serialization(): void static::assertInstanceOf(DOMElement::class, $entry->value()); static::assertEquals($element->attributes, $entry->value()->attributes); } + + public function test_prevents_from_creating_entry_with_empty_entry_name(): void + { + $this->expectExceptionMessage('Entry name cannot be empty'); + + xml_element_entry('', null); + } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/XMLEntryTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/XMLEntryTest.php index 374d0b17dd..db6ff624a4 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/XMLEntryTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Row/Entry/XMLEntryTest.php @@ -205,4 +205,11 @@ public function test_serialization(): void static::assertTrue($entry->isEqual($unserialized)); } + + public function test_prevents_from_creating_entry_with_empty_entry_name(): void + { + $this->expectExceptionMessage('Entry name cannot be empty'); + + xml_entry('', 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 88ea622792..2769be66fa 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Context/FloeStreamReaderContext.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Context/FloeStreamReaderContext.php @@ -129,8 +129,8 @@ public static function write(Filesystem $filesystem, Path $path, Rows $rows, arr /** * Writes a complete file then rewrites it with the FOOTER frame stripped - - * a crashed writer that flushed complete frames but never closed. recover() - * salvages such a file; strict rows() throws on the missing trailer. + * a crashed writer that flushed complete frames but never closed. Strict + * rows() throws on the missing trailer. */ public static function writeWithoutFooter(Filesystem $filesystem, Path $path, Rows $rows): void { diff --git a/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeReaderExtensionParityTest.php b/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeReaderExtensionParityTest.php index 89d9eccd82..e93ab2c61f 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeReaderExtensionParityTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeReaderExtensionParityTest.php @@ -4,22 +4,16 @@ namespace Flow\Floe\Tests\Integration; -use Flow\ETL\Row\PhpRowHydrator; use Flow\ETL\Rows; use Flow\ETL\Tests\FlowIntegrationTestCase; -use Flow\Filesystem\Partition; use Flow\Floe\Codec; use Flow\Floe\Codec\NoopCodec; use Flow\Floe\FloeMerger; use Flow\Floe\FloeStreamWriter; use Flow\Floe\FloeWriter; -use Flow\Floe\Format; use Flow\Floe\NativeFloeEncoder; use Flow\Floe\Options; -use Flow\Floe\PhpFloeEncoder; use Flow\Floe\Tests\Context\FloeEngineContext; -use Flow\Floe\Tests\Context\FloeSchemaContext; -use Flow\Floe\Tests\Context\FloeStreamReaderContext; use Flow\Floe\Tests\Double\PrefixingCodecStub; use Flow\Floe\Tests\Mother\RowsMother; use Override; @@ -28,10 +22,8 @@ use function Flow\ETL\DSL\int_entry; use function Flow\ETL\DSL\row; use function Flow\ETL\DSL\rows; -use function Flow\ETL\DSL\schema_from_json; use function Flow\ETL\DSL\str_entry; use function iterator_to_array; -use function pack; /** * With the flow_php extension loaded, FloeReader hydrates ROW frame bodies @@ -99,111 +91,6 @@ public function test_extension_and_pure_php_seek_offset_identically(): void ); } - #[DataProvider('rows_datasets')] - public function test_extension_and_pure_php_recover_identical_rows(Rows $rows): void - { - $path = $this->cacheDir->suffix('parity-recover.floe'); - - $writer = new FloeWriter($this->fs(), FloeStreamWriter::unionSchema($rows)); - $writer->create($path); - $writer->write($rows); - $writer->close(); - - static::assertEquals( - iterator_to_array(FloeEngineContext::phpReader($this->fs())->read($path)->recover()), - iterator_to_array(FloeEngineContext::nativeReader($this->fs())->read($path)->recover()), - ); - } - - public function test_extension_and_pure_php_recover_a_torn_file_identically(): void - { - $path = $this->cacheDir->suffix('parity-recover-torn.floe'); - - FloeStreamReaderContext::writeWithoutFooter( - $this->fs(), - $path, - rows(row(int_entry('id', 1)), row(int_entry('id', 2), str_entry('email', 'x')), row(int_entry('id', 3))), - ); - - static::assertEquals( - iterator_to_array(FloeEngineContext::phpReader($this->fs())->read($path)->recover()), - iterator_to_array(FloeEngineContext::nativeReader($this->fs())->read($path)->recover()), - ); - } - - public function test_extension_and_pure_php_salvage_rows_before_a_corrupt_row_identically(): void - { - $path = $this->cacheDir->suffix('parity-recover-corrupt.floe'); - - $schemaBody = FloeSchemaContext::schemaBody(row(int_entry('id', 1))->schema()); - $encoder = new PhpFloeEncoder(schema_from_json($schemaBody)); - $hydrator = new PhpRowHydrator(); - - $stream = $this->fs()->writeTo($path); - $stream->append( - Format::header(0x00) - . Format::frame(Format::FRAME_SCHEMA, $schemaBody) - . Format::frame( - Format::FRAME_ROW, - $encoder->encode($hydrator->dehydrate(rows(row(int_entry('id', 1)))))[0], - ) - . Format::frame( - Format::FRAME_ROW, - $encoder->encode($hydrator->dehydrate(rows(row(int_entry('id', 2)))))[0], - ) - . Format::frame(Format::FRAME_ROW, "\xEE"), - ); - $stream->close(); - - $pure = iterator_to_array(FloeEngineContext::phpReader($this->fs())->read($path)->recover()); - - static::assertEquals( - $pure, - iterator_to_array(FloeEngineContext::nativeReader($this->fs())->read($path)->recover()), - ); - static::assertCount(1, $pure); - static::assertCount(2, $pure[0]->all()); - } - - public function test_extension_and_pure_php_recover_rows_around_a_late_partitions_frame_identically(): void - { - $path = $this->cacheDir->suffix('parity-recover-late-partitions.floe'); - - $schemaBody = FloeSchemaContext::schemaBody(row(int_entry('id', 1))->schema()); - $encoder = new PhpFloeEncoder(schema_from_json($schemaBody)); - $hydrator = new PhpRowHydrator(); - $partitionsBody = pack('V', 1) . pack('V', 1) . 'g' . pack('V', 1) . 'a'; - - $stream = $this->fs()->writeTo($path); - $stream->append( - Format::header(0x00) - . Format::frame(Format::FRAME_SCHEMA, $schemaBody) - . Format::frame( - Format::FRAME_ROW, - $encoder->encode($hydrator->dehydrate(rows(row(int_entry('id', 1)))))[0], - ) - . Format::frame(Format::FRAME_PARTITIONS, $partitionsBody) - . Format::frame( - Format::FRAME_ROW, - $encoder->encode($hydrator->dehydrate(rows(row(int_entry('id', 2)))))[0], - ), - ); - $stream->close(); - - $pure = iterator_to_array(FloeEngineContext::phpReader($this->fs())->read($path)->recover()); - - static::assertEquals( - $pure, - iterator_to_array(FloeEngineContext::nativeReader($this->fs())->read($path)->recover()), - ); - // the partition change splits the rows into two batches so no batch mixes combinations - static::assertCount(2, $pure); - static::assertCount(1, $pure[0]->all()); - static::assertSame([], $pure[0]->partitions()->toArray()); - static::assertCount(1, $pure[1]->all()); - static::assertEquals([new Partition('g', 'a')], $pure[1]->partitions()->toArray()); - } - /** * @return array */ diff --git a/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeStreamReaderTest.php b/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeStreamReaderTest.php index f49e49fd42..10d25ae1c4 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeStreamReaderTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeStreamReaderTest.php @@ -17,8 +17,6 @@ use function Flow\ETL\DSL\str_entry; use function iterator_to_array; use function str_pad; -use function strlen; -use function substr; final class FloeStreamReaderTest extends FlowIntegrationTestCase { @@ -103,34 +101,6 @@ public function test_file_larger_than_compaction_threshold_round_trips(): void static::assertSame($written, $read); } - public function test_recover_salvages_truncated_copy_of_real_file(): void - { - $path = $this->cacheDir->suffix('original.floe'); - - $data = rows(row(int_entry('id', 1)), row(int_entry('id', 2)), row(int_entry('id', 3))); - $writer = new FloeWriter($this->fs(), FloeStreamWriter::unionSchema($data)); - $writer->create($path); - $writer->write($data); - $writer->close(); - - $content = $this->fs()->readFrom($path)->content(); - $truncatedPath = $this->cacheDir->suffix('truncated.floe'); - $stream = $this->fs()->writeTo($truncatedPath); - $stream->append(substr($content, 0, strlen($content) - 250)); - $stream->close(); - - $salvaged = 0; - - foreach ((new FloeReader($this->fs())) - ->read($truncatedPath) - ->recover() as $batch) { - $salvaged += $batch->count(); - } - - static::assertGreaterThan(0, $salvaged); - static::assertLessThanOrEqual(3, $salvaged); - } - public function test_round_trip_of_all_entry_types_through_local_filesystem(): void { $rows = RowsMother::withAllEntryTypes(); diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeReaderTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeReaderTest.php index 9213d24d8b..147f6a6832 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeReaderTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeReaderTest.php @@ -38,7 +38,6 @@ use function iterator_to_array; use function pack; use function strlen; -use function substr; final class FloeReaderTest extends TestCase { @@ -438,13 +437,10 @@ public function test_corrupt_row_body_throws_wrapped_exception(): void { $filesystem = memory_filesystem(); $path = path('memory://corrupt-row.floe'); - $schemaBody = FloeSchemaContext::schemaBody(row(int_entry('id', 1))->schema()); $footerJson = FooterMother::footer(schema: row(int_entry('id', 1))->schema()->normalize())->toJson(); $stream = $filesystem->writeTo($path); $stream->append( - Format::header(0x00) - . Format::frame(Format::FRAME_SCHEMA, $schemaBody) - . Format::frame(Format::FRAME_ROW, "\xEE") + Format::header(0x00) . Format::frame(Format::FRAME_ROW, "\xEE") . Format::frame(Format::FRAME_FOOTER, $footerJson . Format::trailer(strlen($footerJson))), ); $stream->close(); @@ -520,7 +516,6 @@ public function test_empty_file_yields_no_batches(): void $reader = (new FloeReader($filesystem))->read($path); static::assertSame([], iterator_to_array($reader->rows())); - static::assertSame([], iterator_to_array($reader->recover())); static::assertSame(0, $reader->totalRows()); static::assertCount(0, $reader->schema()->definitions()); } @@ -744,36 +739,6 @@ public function test_offset_read_into_a_later_combination_primes_the_right_parti static::assertEquals([new Partition('country', 'US')], iterator_to_array($batches[0]->partitions())); } - public function test_recover_tracks_per_section_partitions(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://recover-multi-combination.floe'); - - $batchPL = Rows::partitioned([row(int_entry('id', 1), str_entry('country', 'PL'))], [new Partition( - 'country', - 'PL', - )]); - $batchUS = Rows::partitioned([row(int_entry('id', 2), str_entry('country', 'US'))], [new Partition( - 'country', - 'US', - )]); - $writer = new FloeWriter($filesystem, FloeStreamWriter::unionSchema($batchPL->merge($batchUS))); - $writer->create($path); - $writer->write($batchPL); - $writer->write($batchUS); - $writer->close(); - - $combos = []; - - foreach ((new FloeReader($filesystem)) - ->read($path) - ->recover() as $batch) { - $combos[] = array_map(static fn(Partition $p): string => $p->value, $batch->partitions()->toArray()); - } - - static::assertSame([['PL'], ['US']], $combos); - } - public function test_reader_with_mismatched_codec_flags_throws(): void { $filesystem = memory_filesystem(); @@ -790,231 +755,6 @@ public function test_reader_with_mismatched_codec_flags_throws(): void ->footer(); } - public function test_recover_chunks_batches(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://recover-batches.floe'); - - $data = rows(row(int_entry('id', 1)), row(int_entry('id', 2)), row(int_entry('id', 3))); - $writer = new FloeWriter($filesystem, FloeStreamWriter::unionSchema($data)); - $writer->create($path); - $writer->write($data); - $writer->close(); - - static::assertCount( - 2, - iterator_to_array( - (new FloeReader($filesystem)) - ->read($path) - ->recover(2), - ), - ); - } - - public function test_recover_stops_at_corrupt_row_body(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://recover-corrupt-row.floe'); - $schemaBody = FloeSchemaContext::schemaBody(row(int_entry('id', 1))->schema()); - $stream = $filesystem->writeTo($path); - $stream->append( - Format::header(0x00) . Format::frame(Format::FRAME_SCHEMA, $schemaBody) - . Format::frame(Format::FRAME_ROW, "\xEE"), - ); - $stream->close(); - - static::assertSame( - [], - iterator_to_array( - (new FloeReader($filesystem)) - ->read($path) - ->recover(), - ), - ); - } - - public function test_recover_stops_at_corrupt_schema_body(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://recover-corrupt-schema.floe'); - $stream = $filesystem->writeTo($path); - $stream->append(Format::header(0x00) . Format::frame(Format::FRAME_SCHEMA, '{broken')); - $stream->close(); - - static::assertSame( - [], - iterator_to_array( - (new FloeReader($filesystem)) - ->read($path) - ->recover(), - ), - ); - } - - public function test_recover_stops_at_unknown_frame_type(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://recover-unknown.floe'); - - FloeStreamReaderContext::writeWithoutFooter($filesystem, $path, rows(row(int_entry('id', 1)))); - - $filesystem->appendTo($path)->append(Format::frame(0x55, 'mystery'))->close(); - - $batches = iterator_to_array( - (new FloeReader($filesystem)) - ->read($path) - ->recover(), - ); - - static::assertCount(1, $batches); - static::assertCount(1, $batches[0]->all()); - } - - public function test_recover_reads_partitions_from_frame(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://recover-partitions.floe'); - - FloeStreamReaderContext::writeWithoutFooter( - $filesystem, - $path, - Rows::partitioned([row(int_entry('id', 1), str_entry('country', 'PL'))], [new Partition('country', 'PL')]), - ); - - $batches = iterator_to_array( - (new FloeReader($filesystem)) - ->read($path) - ->recover(), - ); - - static::assertCount(1, $batches); - static::assertEquals([new Partition('country', 'PL')], iterator_to_array($batches[0]->partitions())); - } - - public function test_recover_salvages_rows_from_file_missing_close(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://torn.floe'); - - FloeStreamReaderContext::writeWithoutFooter( - $filesystem, - $path, - rows(row(int_entry('id', 1)), row(int_entry('id', 2))), - ); - - $reader = (new FloeReader($filesystem))->read($path); - $batches = iterator_to_array($reader->recover()); - - static::assertCount(1, $batches); - static::assertCount(2, $batches[0]->all()); - } - - public function test_recover_salvages_complete_frames_from_truncated_tail(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://complete.floe'); - - $data = rows(row(int_entry('id', 1)), row(int_entry('id', 2)), row(int_entry('id', 3))); - $writer = new FloeWriter($filesystem, FloeStreamWriter::unionSchema($data)); - $writer->create($path); - $writer->write($data); - $writer->close(); - - $content = $filesystem->readFrom($path)->content(); - $truncated = path('memory://truncated.floe'); - $stream = $filesystem->writeTo($truncated); - $stream->append(substr($content, 0, strlen($content) - 192)); - $stream->close(); - - $salvaged = iterator_to_array( - (new FloeReader($filesystem)) - ->read($truncated) - ->recover(), - ); - - static::assertNotSame([], $salvaged); - static::assertLessThanOrEqual(3, count($salvaged[0]->all())); - } - - public function test_recover_yields_rows_as_written_without_padding(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://recover-evolved.floe'); - - // a multi-schema file is what FloeMerger produces (schema evolution lives in merge, not the writer) - $baseFile = path('memory://recover-base.floe'); - $evolvedFile = path('memory://recover-new.floe'); - - $baseRows = rows(row(int_entry('id', 1))); - $writer = new FloeWriter($filesystem, FloeStreamWriter::unionSchema($baseRows)); - $writer->create($baseFile); - $writer->write($baseRows); - $writer->close(); - - $evolvedRows = rows(row(int_entry('id', 2), str_entry('email', null))); - $writer = new FloeWriter($filesystem, FloeStreamWriter::unionSchema($evolvedRows)); - $writer->create($evolvedFile); - $writer->write($evolvedRows); - $writer->close(); - - (new FloeMerger($filesystem))->merge([$baseFile, $evolvedFile], $path); - - $rows = []; - - foreach ((new FloeReader($filesystem)) - ->read($path) - ->recover() as $batch) { - foreach ($batch->all() as $row) { - $rows[] = $row->entries()->names(); - } - } - - static::assertSame([['id'], ['id', 'email']], $rows); - } - - public function test_recover_stops_at_row_frame_before_schema_frame(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://recover-row-first.floe'); - $stream = $filesystem->writeTo($path); - $stream->append(Format::header(0x00) . Format::frame(Format::FRAME_ROW, 'row-bytes')); - $stream->close(); - - static::assertSame( - [], - iterator_to_array( - (new FloeReader($filesystem)) - ->read($path) - ->recover(), - ), - ); - } - - public function test_recover_stops_at_row_body_longer_than_hydrated_content(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://recover-long-row.floe'); - $schemaBody = FloeSchemaContext::schemaBody(row(int_entry('id', 1))->schema()); - $rowBody = (new PhpFloeEncoder(schema_from_json( - $schemaBody, - )))->encode((new AdaptiveRowHydrator())->dehydrate(rows(row(int_entry('id', 1)))))[0]; - $stream = $filesystem->writeTo($path); - $stream->append( - Format::header(0x00) . Format::frame(Format::FRAME_SCHEMA, $schemaBody) - . Format::frame(Format::FRAME_ROW, $rowBody . 'extra-bytes'), - ); - $stream->close(); - - static::assertSame( - [], - iterator_to_array( - (new FloeReader($filesystem)) - ->read($path) - ->recover(), - ), - ); - } - public function test_rows_through_a_custom_identity_codec(): void { $filesystem = memory_filesystem(); @@ -1051,9 +791,7 @@ public function test_rows_through_a_custom_identity_codec_with_long_row_body_thr $footerJson = FooterMother::footer(schema: row(int_entry('id', 1))->schema()->normalize())->toJson(); $stream = $filesystem->writeTo($path); $stream->append( - Format::header(0x00) - . Format::frame(Format::FRAME_SCHEMA, $schemaBody) - . Format::frame(Format::FRAME_ROW, $rowBody . 'extra-bytes') + Format::header(0x00) . Format::frame(Format::FRAME_ROW, $rowBody . 'extra-bytes') . Format::frame(Format::FRAME_FOOTER, $footerJson . Format::trailer(strlen($footerJson))), ); $stream->close(); @@ -1090,29 +828,6 @@ public function test_row_frame_not_described_by_the_footer_schema_throws(): void ); } - public function test_schema_frame_not_matching_the_footer_schema_throws(): void - { - $filesystem = memory_filesystem(); - $path = path('memory://drifted-schema-frame.floe'); - $schemaBody = FloeSchemaContext::schemaBody(row(int_entry('id', 1))->schema()); - $footerJson = FooterMother::footer(schema: row(str_entry('name', 'a'))->schema()->normalize())->toJson(); - $stream = $filesystem->writeTo($path); - $stream->append( - Format::header(0x00) . Format::frame(Format::FRAME_SCHEMA, $schemaBody) - . Format::frame(Format::FRAME_FOOTER, $footerJson . Format::trailer(strlen($footerJson))), - ); - $stream->close(); - - $this->expectException(FloeException::class); - $this->expectExceptionMessage('Floe found a schema frame that does not match the file schema'); - - iterator_to_array( - (new FloeReader($filesystem)) - ->read($path) - ->rows(), - ); - } - public function test_row_body_longer_than_hydrated_content_throws(): void { $filesystem = memory_filesystem(); @@ -1124,9 +839,7 @@ public function test_row_body_longer_than_hydrated_content_throws(): void $footerJson = FooterMother::footer(schema: row(int_entry('id', 1))->schema()->normalize())->toJson(); $stream = $filesystem->writeTo($path); $stream->append( - Format::header(0x00) - . Format::frame(Format::FRAME_SCHEMA, $schemaBody) - . Format::frame(Format::FRAME_ROW, $rowBody . 'extra-bytes') + Format::header(0x00) . Format::frame(Format::FRAME_ROW, $rowBody . 'extra-bytes') . Format::frame(Format::FRAME_FOOTER, $footerJson . Format::trailer(strlen($footerJson))), ); $stream->close(); 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 8a9efabf97..a457c2b99d 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeStreamWriterTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeStreamWriterTest.php @@ -91,7 +91,7 @@ public function test_create_on_stream_round_trips(): void static::assertSame(3, $footer->totalRows); static::assertSame(['source' => 'stream'], $footer->metadata->normalize()); static::assertSame( - [Format::FRAME_SCHEMA, Format::FRAME_ROW, Format::FRAME_ROW, Format::FRAME_ROW, Format::FRAME_FOOTER], + [Format::FRAME_ROW, Format::FRAME_ROW, Format::FRAME_ROW, Format::FRAME_FOOTER], FloeStreamReaderContext::frameTypes($filesystem, $path), ); } @@ -250,7 +250,10 @@ public function test_writing_a_column_name_with_invalid_utf8_throws(): void $this->expectException(FloeException::class); $this->expectExceptionMessage('failed to encode schema as JSON'); + // the schema is stored only in the footer, so the pure PHP engine rejects it when the footer + // is written, while the native engine rejects it earlier, when it encodes the batch $writer->write($badRows); + $writer->close(); } public function test_validation_off_is_byte_identical_to_validation_on_for_a_fitting_batch(): void diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeWriterTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeWriterTest.php index 8a007722c6..260fc7dcfa 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeWriterTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeWriterTest.php @@ -308,7 +308,7 @@ public function test_append_with_new_nullable_column_throws(): void $writer->write(rows(row(int_entry('id', 2), str_entry('email', null)))); } - public function test_append_with_same_schema_starts_section_without_schema_frame(): void + public function test_append_with_same_schema_adds_a_section_reusing_the_footer_schema(): void { $filesystem = memory_filesystem(); $path = path('memory://same-schema.floe'); @@ -324,7 +324,7 @@ public function test_append_with_same_schema_starts_section_without_schema_frame $writer->close(); static::assertSame( - [Format::FRAME_SCHEMA, Format::FRAME_ROW, Format::FRAME_FOOTER, Format::FRAME_ROW, Format::FRAME_FOOTER], + [Format::FRAME_ROW, Format::FRAME_FOOTER, Format::FRAME_ROW, Format::FRAME_FOOTER], FloeStreamReaderContext::frameTypes($filesystem, $path), ); @@ -387,7 +387,7 @@ public function test_create_with_non_noop_codec_throws(): void )); } - public function test_two_writes_with_the_same_schema_in_one_session_emit_a_single_schema_frame(): void + public function test_two_writes_with_the_same_schema_in_one_session_share_one_section(): void { $filesystem = memory_filesystem(); $path = path('memory://continuation.floe'); @@ -403,7 +403,7 @@ public function test_two_writes_with_the_same_schema_in_one_session_emit_a_singl static::assertSame(2, $footer->totalRows); static::assertCount(1, $footer->sections); static::assertSame( - [Format::FRAME_SCHEMA, Format::FRAME_ROW, Format::FRAME_ROW, Format::FRAME_FOOTER], + [Format::FRAME_ROW, Format::FRAME_ROW, Format::FRAME_FOOTER], FloeStreamReaderContext::frameTypes($filesystem, $path), ); } @@ -438,7 +438,7 @@ public function test_partitioned_rows_write_partitions_frame_after_header(): voi $writer->close(); static::assertSame( - [Format::FRAME_PARTITIONS, Format::FRAME_SCHEMA, Format::FRAME_ROW, Format::FRAME_FOOTER], + [Format::FRAME_PARTITIONS, Format::FRAME_ROW, Format::FRAME_FOOTER], FloeStreamReaderContext::frameTypes($filesystem, $path), ); static::assertSame([['country' => 'PL']], FloeStreamReaderContext::footer($filesystem, $path)->partitions); @@ -470,7 +470,6 @@ public function test_a_partition_change_starts_a_new_section_with_its_own_partit static::assertSame( [ Format::FRAME_PARTITIONS, - Format::FRAME_SCHEMA, Format::FRAME_ROW, Format::FRAME_PARTITIONS, Format::FRAME_ROW, @@ -504,7 +503,6 @@ public function test_consecutive_writes_with_the_same_partition_reuse_the_id_and static::assertSame( [ Format::FRAME_PARTITIONS, - Format::FRAME_SCHEMA, Format::FRAME_ROW, Format::FRAME_ROW, Format::FRAME_FOOTER, @@ -535,7 +533,6 @@ public function test_a_change_back_to_unpartitioned_emits_an_empty_partitions_fr static::assertSame( [ Format::FRAME_PARTITIONS, - Format::FRAME_SCHEMA, Format::FRAME_ROW, Format::FRAME_PARTITIONS, Format::FRAME_ROW, @@ -643,7 +640,7 @@ public function test_section_offsets_point_at_frames(): void static::assertSame(Format::HEADER_LENGTH, $footer->sections[0]->offset); foreach ($footer->sections as $section) { - static::assertSame(Format::FRAME_SCHEMA, ord($source->read(1, $section->offset))); + static::assertSame(Format::FRAME_ROW, ord($source->read(1, $section->offset))); } } diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/FooterTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/FooterTest.php index 5b497eb5a5..5ee20478e2 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/FooterTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/FooterTest.php @@ -299,4 +299,19 @@ public function test_reconstruct_rows_preserves_non_alphabetical_partition_order ); static::assertEquals($value, $result); } + + public function test_encoding_a_schema_that_is_not_valid_utf8_throws_a_floe_exception(): void + { + $footer = FooterMother::footer([[ + 'ref' => "bad\xFFname", + 'type' => ['type' => 'string'], + 'nullable' => false, + 'metadata' => [], + ]]); + + $this->expectException(FloeException::class); + $this->expectExceptionMessage('Floe failed to encode schema as JSON'); + + $footer->toJson(); + } } diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/FrameReaderTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/FrameReaderTest.php index 5fbdee3a21..cc3e7844dd 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/FrameReaderTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/FrameReaderTest.php @@ -61,15 +61,15 @@ public function test_frames_are_yielded_in_order_with_bodies(): void { $bytes = Format::header(0x00) - . Format::frame(Format::FRAME_SCHEMA, 'schema') + . Format::frame(Format::FRAME_PARTITIONS, 'partitions') . Format::frame(Format::FRAME_ROW, 'row-1') . Format::frame(Format::FRAME_FOOTER, 'footer'); static::assertSame( [ - [Format::FRAME_SCHEMA, 'schema'], - [Format::FRAME_ROW, 'row-1'], - [Format::FRAME_FOOTER, 'footer'], + [Format::FRAME_PARTITIONS, 'partitions'], + [Format::FRAME_ROW, 'row-1'], + [Format::FRAME_FOOTER, 'footer'], ], iterator_to_array(FrameReaderMother::overBytes($bytes)->frames()), ); diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/FrameWriterTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/FrameWriterTest.php index c3d87df9c9..220e1c99f5 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/FrameWriterTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/FrameWriterTest.php @@ -40,17 +40,6 @@ public function test_row_frame_is_type_length_prefixed_body(): void static::assertSame(Format::FRAME_HEADER_LENGTH + 3, $writer->position()); } - public function test_schema_frame_matches_format_frame(): void - { - $sink = new StringDestinationStream(path('memory://frame.floe')); - $writer = new FrameWriter($sink, 0x00); - - $writer->schema('{"x":1}'); - $writer->flush(); - - static::assertSame(Format::frame(Format::FRAME_SCHEMA, '{"x":1}'), $sink->content()); - } - public function test_empty_combination_emits_count_zero_partitions_frame(): void { $sink = new StringDestinationStream(path('memory://frame.floe')); diff --git a/src/extension/flow-php-ext/tests/phpt/010_no_leaks.phpt b/src/extension/flow-php-ext/tests/phpt/010_no_leaks.phpt index 259a6f685e..a55c675d52 100644 --- a/src/extension/flow-php-ext/tests/phpt/010_no_leaks.phpt +++ b/src/extension/flow-php-ext/tests/phpt/010_no_leaks.phpt @@ -8,7 +8,6 @@ require __DIR__ . '/bootstrap.php'; use Flow\ETL\Row\PhpRowHydrator; use Flow\ETL\Rows; -use Flow\Floe\Format; use Flow\Floe\RustFloeEncoderNative; use function Flow\ETL\DSL\{row, rows, int_entry, str_entry, float_entry, bool_entry, datetime_entry, time_entry, uuid_entry, list_entry, map_entry, structure_entry, xml_entry, json_entry, schema_from_json}; @@ -38,7 +37,7 @@ $schemaBody = null; $rowBodies = []; foreach ($frames as $frame) { - if ($frame['type'] === Format::FRAME_SCHEMA) { + if ($frame['type'] === SCHEMA_ENTRY) { $schemaBody = $frame['body']; } else { $rowBodies[] = $frame['body']; diff --git a/src/extension/flow-php-ext/tests/phpt/017_rows_decoder.phpt b/src/extension/flow-php-ext/tests/phpt/017_rows_decoder.phpt index 51a32cf652..b7327989d0 100644 --- a/src/extension/flow-php-ext/tests/phpt/017_rows_decoder.phpt +++ b/src/extension/flow-php-ext/tests/phpt/017_rows_decoder.phpt @@ -6,7 +6,6 @@ RowValues pipeline decodes streamed frame bodies identically to the pure-PHP pip encode($hydrator->dehydrate($decoded), $schemaBody); var_dump($bodies === $reBodies); $frames = array_merge( - [['type' => Format::FRAME_SCHEMA, 'body' => $schemaBody]], + [['type' => SCHEMA_ENTRY, 'body' => $schemaBody]], array_map(fn($body) => ['type' => Format::FRAME_ROW, 'body' => $body], $bodies), ); assert_rows_identical(php_decode_frames($frames), $decoded->all()); diff --git a/src/extension/flow-php-ext/tests/phpt/bootstrap.php b/src/extension/flow-php-ext/tests/phpt/bootstrap.php index 375b63b393..53cb4a13e1 100644 --- a/src/extension/flow-php-ext/tests/phpt/bootstrap.php +++ b/src/extension/flow-php-ext/tests/phpt/bootstrap.php @@ -12,10 +12,17 @@ use Flow\Floe\NativeFloeEncoder; use Flow\Floe\PhpFloeEncoder; +/** + * In-memory marker for the schema-carrying entry in the reference frame lists + * below. The Floe format stores the schema in the footer, not in a frame; this + * scaffold threads the schema through the frame list to drive the encoders. + */ +const SCHEMA_ENTRY = 0x01; + /** * Pure-PHP reference framing built from the Floe encoder primitives: a write - * session carries one schema, so the whole Rows is framed as a single SCHEMA - * frame (its union) followed by one ROW frame per row. + * session carries one schema, so the whole Rows is framed as a single schema + * entry (its union) followed by one ROW frame per row. * * @return array */ @@ -29,7 +36,7 @@ function php_frames(Rows $rows): array $schemaBody = json_encode($rows->schema()->normalize(), JSON_THROW_ON_ERROR); $encoder = new PhpFloeEncoder(Flow\ETL\DSL\schema_from_json($schemaBody)); - $frames = [['type' => Format::FRAME_SCHEMA, 'body' => $schemaBody]]; + $frames = [['type' => SCHEMA_ENTRY, 'body' => $schemaBody]]; foreach ($rows->all() as $row) { $frames[] = ['type' => Format::FRAME_ROW, 'body' => $encoder->encode($hydrator->dehydrate(new Rows($row)))[0]]; @@ -55,7 +62,7 @@ function ext_frames(Rows $rows): array $schemaBody = json_encode($rows->schema()->normalize(), JSON_THROW_ON_ERROR); $encoder = new NativeFloeEncoder(Flow\ETL\DSL\schema_from_json($schemaBody)); - $frames = [['type' => Format::FRAME_SCHEMA, 'body' => $schemaBody]]; + $frames = [['type' => SCHEMA_ENTRY, 'body' => $schemaBody]]; foreach ($rows->all() as $row) { $frames[] = ['type' => Format::FRAME_ROW, 'body' => $encoder->encode($hydrator->dehydrate(new Rows($row)))[0]]; @@ -80,11 +87,11 @@ function php_decode_frames(array $frames): array $rows = []; foreach ($frames as $frame) { - if ($frame['type'] === Format::FRAME_SCHEMA) { + if ($frame['type'] === SCHEMA_ENTRY) { $schema = Flow\ETL\DSL\schema_from_json($frame['body']); $encoder = new PhpFloeEncoder($schema); } elseif ($schema === null || $encoder === null) { - throw new RuntimeException('row frame before any schema frame'); + throw new RuntimeException('row frame before any schema entry'); } else { foreach ($hydrator->hydrate($encoder->decode([$frame['body']]), $schema)->all() as $row) { $rows[] = $row; @@ -97,7 +104,7 @@ function php_decode_frames(array $frames): array /** * Drives the native two-layer pipeline over frame bodies, rebinding the schema - * per SCHEMA frame (mirrors what FloeStreamReader does on the hot path). + * per schema entry (mirrors how FloeStreamReader sources the schema). * * @param array $frames * @@ -111,11 +118,11 @@ function ext_decode_frames(array $frames): array $rows = []; foreach ($frames as $frame) { - if ($frame['type'] === Format::FRAME_SCHEMA) { + if ($frame['type'] === SCHEMA_ENTRY) { $schema = Flow\ETL\DSL\schema_from_json($frame['body']); $encoder = new NativeFloeEncoder($schema); } elseif ($schema === null || $encoder === null) { - throw new RuntimeException('row frame before any schema frame'); + throw new RuntimeException('row frame before any schema entry'); } else { foreach ($hydrator->hydrate($encoder->decode([$frame['body']]), $schema)->all() as $row) { $rows[] = $row; diff --git a/web/landing/src/Flow/Website/Service/FlowConfigFactory.php b/web/landing/src/Flow/Website/Service/FlowConfigFactory.php index df9a0fba50..8f9e8f723d 100644 --- a/web/landing/src/Flow/Website/Service/FlowConfigFactory.php +++ b/web/landing/src/Flow/Website/Service/FlowConfigFactory.php @@ -75,7 +75,7 @@ private function cache(string $directoryName, int $ttl): PSRSimpleCache { return new PSRSimpleCache(new Psr16Cache( new FilesystemAdapter( - 'flow-website', + 'flow-contributors', $ttl, directory: type_string()->assert($this->parameters->get('kernel.cache_dir')) . '/' . ltrim($directoryName, '/'),