From e3cb7e7704ed9026ccd98550746d355c96f5e613 Mon Sep 17 00:00:00 2001 From: Norbert Orzechowicz Date: Tue, 20 Oct 2026 14:35:14 -0500 Subject: [PATCH] fix(flow-php/etl): empty sources without a declared schema fail at the source - empty sources throw InferredSchemaException from schema() - reading an empty source under a declared schema yields 0 rows - parquet and floe accept withSchema(), files read by name - from_cache() without a cache is not derivable - upgrading notes and declared schema docs --- documentation/components/adapters/parquet.md | 25 ++- documentation/components/core/core.md | 14 ++ documentation/components/core/floe.md | 3 +- documentation/components/core/schema.md | 4 +- documentation/upgrading.md | 9 ++ .../src/Flow/ETL/Adapter/CSV/CSVExtractor.php | 18 ++- .../Tests/Integration/CSVExtractorTest.php | 71 +++++++-- .../CSV/Tests/Unit/CSVExtractorTest.php | 2 +- .../Flow/ETL/Adapter/Excel/ExcelExtractor.php | 18 ++- .../Tests/Integration/ExcelExtractorTest.php | 54 ++++++- .../GoogleSheet/GoogleSheetExtractor.php | 8 + .../Integration/GoogleSheetExtractorTest.php | 13 +- .../Tests/Unit/GoogleSheetExtractorTest.php | 9 +- .../JSON/JSONMachine/JsonExtractor.php | 18 ++- .../JSONMachine/JsonExtractorTest.php | 100 +++++++++++- .../JSONMachine/JsonLinesExtractorTest.php | 33 ++-- .../ETL/Adapter/Parquet/ParquetExtractor.php | 46 ++++-- .../Adapter/Parquet/ParquetFileBatches.php | 6 + .../Integration/ParquetExtractorTest.php | 143 +++++++++++++++++- .../Tests/Unit/ParquetExtractorTest.php | 15 +- .../Integration/Loader/XMLLoaderTest.php | 4 +- .../Response/FlowBufferedResponseTest.php | 2 +- .../Response/FlowStreamedResponseTest.php | 4 +- .../ETL/Exception/InferredSchemaException.php | 13 ++ src/core/etl/src/Flow/ETL/Extractor.php | 2 + .../src/Flow/ETL/Extractor/ArrayExtractor.php | 14 +- .../src/Flow/ETL/Extractor/CacheExtractor.php | 19 ++- .../src/Flow/ETL/Extractor/ChainExtractor.php | 10 +- .../Flow/ETL/Extractor/MemoryExtractor.php | 14 +- .../src/Flow/ETL/Extractor/RowsExtractor.php | 14 +- .../Flow/ETL/Extractor/SequenceExtractor.php | 14 +- src/core/etl/src/Flow/Floe/FloeExtractor.php | 47 ++++-- .../etl/src/Flow/Floe/FloeFileBatches.php | 6 + .../Integration/DataFrame/EmptySourceTest.php | 75 +++++++++ .../Extractor/CacheExtractorTest.php | 23 ++- .../Extractor/FileSourceFilesystemTest.php | 6 +- .../Exception/InferredSchemaExceptionTest.php | 9 ++ .../Flow/ETL/Tests/Unit/ExecutorTest.php | 4 +- .../Unit/Extractor/ArrayExtractorTest.php | 35 +++++ .../Unit/Extractor/ChainExtractorTest.php | 39 +++++ .../Unit/Extractor/MemoryExtractorTest.php | 29 +++- .../Unit/Extractor/RowsExtractorTest.php | 22 +++ .../Unit/Extractor/SequenceExtractorTest.php | 47 ++++++ .../ETL/Tests/Unit/Join/JoinStepsTest.php | 2 +- .../Unit/Processor/HashJoinProcessorTest.php | 2 +- .../Tests/Integration/FloeExtractorTest.php | 114 ++++++++++++++ .../Floe/Tests/Unit/FloeExtractorTest.php | 12 +- 47 files changed, 1079 insertions(+), 112 deletions(-) create mode 100644 src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/EmptySourceTest.php diff --git a/documentation/components/adapters/parquet.md b/documentation/components/adapters/parquet.md index 0625dda533..ea956d03d6 100644 --- a/documentation/components/adapters/parquet.md +++ b/documentation/components/adapters/parquet.md @@ -56,8 +56,7 @@ Flow `float` columns are written as Parquet `DOUBLE`. ## Schema -The file footer is the schema: `from_parquet(...)->withSchema()` throws. Project with `columns:`, change types after -reading: +The file footer is the schema. Project with `columns:`, change types after reading: ```php read(from_parquet(__DIR__ . '/orders/**/*.parquet')->withSchema(schema( + int_schema('id'), + str_schema('name', nullable: true), + ))) + ->select('id') + ->write(to_output()) + ->run(); +``` + +- A glob that matches no file reads 0 rows; without `withSchema()` it throws `InferredSchemaException`. +- Columns match by name: an undeclared file column is not read, a declared nullable column a file lacks is `null`, a + declared non-nullable one fails. + ## Filter pushdown A `filter()` directly over `from_parquet()` moves into the reader, which returns only matching rows: row groups whose diff --git a/documentation/components/core/core.md b/documentation/components/core/core.md index 751dc846be..e97b4a1380 100644 --- a/documentation/components/core/core.md +++ b/documentation/components/core/core.md @@ -236,6 +236,20 @@ $dataFrame = data_frame() > **Note**: Flow PHP supports many data sources through specialized adapters. See individual adapter documentation for specific extractor usage (CSV, JSON, Parquet, databases, APIs, etc.). +An empty source has no column to infer a schema from and throws `InferredSchemaException`. Declare the schema to read it: + +```php +read(from_array([], schema(int_schema('id'), str_schema('name')))) + ->select('id') + ->write(to_output()) + ->run(); +``` + ## Memory Management Best Practices 1. **Prefer Generator Methods**: Use `get()`, `getEachAsArray()` over `fetch()` for large datasets diff --git a/documentation/components/core/floe.md b/documentation/components/core/floe.md index 2ea723b4ca..ce04cafd09 100644 --- a/documentation/components/core/floe.md +++ b/documentation/components/core/floe.md @@ -41,7 +41,8 @@ data_frame() Batches hold at most `withBatchSize()` rows, never spanning two BATCH frames. -The footer is the schema: `from_floe(...)->withSchema()` throws. Change a column's type or zone after reading: +The footer is the schema; `withSchema()` reads every file by name against a declared one, as +[Parquet](/documentation/components/adapters/parquet.md) does. Change a column's type or zone after reading: ```php withSchema(...)` to read an empty source | +| `df()->read(from_array([]))->fetch()` returned 0 rows | throws `InferredSchemaException`; `from_array([], schema: $schema)` returns 0 rows | +| `from_parquet(...)->withSchema()`, `from_floe(...)->withSchema()` threw `InvalidArgumentException` | read every file by name against the declared schema: undeclared columns are not read, a missing nullable column is null | +| `from_cache($id)` without a cache instance described an empty schema | `SchemaNotDerivableException`; the plan runs unbound | + --- ## Upgrading from 0.45.x to 0.46.x diff --git a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php index c65fff7748..405cf7335b 100644 --- a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php +++ b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php @@ -5,6 +5,7 @@ namespace Flow\ETL\Adapter\CSV; use Flow\ETL\Cardinality; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Extractor; use Flow\ETL\Extractor\BatchableExtractor; @@ -101,7 +102,7 @@ public function isRepeatable(): bool public function scan(FlowContext $context, FileScan $scan): Generator { $fileColumns = $this->fileColumns($this->filesystem, $this->path); - $schema = $this->schema(); + $schema = $this->describe(); $sources = iterator_to_array($this->sourceFiles($this->filesystem, $this->path, $scan->pathFilter), false); yield from (new FileReadLoop($fileColumns, $schema, $scan->conjuncts))->read( @@ -134,6 +135,21 @@ public function inferSchema(SchemaInferenceBuilder $builder): static } public function schema(): Schema + { + $schema = $this->describe(); + $fileColumns = $this->fileColumns($this->filesystem, $this->path); + + if ($this->schema === null && $fileColumns->withoutTail($schema)->count() === 0) { + throw InferredSchemaException::nothingToInferFrom('from_csv()', sprintf( + 'no file matching "%s" carries a column', + $this->path->uri(), + )); + } + + return $schema; + } + + private function describe(): Schema { $fileColumns = $this->fileColumns($this->filesystem, $this->path); diff --git a/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVExtractorTest.php b/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVExtractorTest.php index 0eb6bc9d0d..60407a16a1 100644 --- a/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVExtractorTest.php +++ b/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVExtractorTest.php @@ -71,6 +71,39 @@ final class CSVExtractorTest extends FlowTestCase { use OperatingSystem; + public function test_an_empty_glob_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_csv() has nothing to infer a schema from: no file matching "memory://empty/*.csv" carries a column. ' + . 'Declare the schema with ->withSchema(...).', + ); + + from_csv(path('memory://empty/*.csv'), filesystem: memory_filesystem())->schema(); + } + + public function test_an_empty_glob_with_metadata_columns_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_csv() has nothing to infer a schema from: no file matching "memory://empty/*.csv" carries a column. ' + . 'Declare the schema with ->withSchema(...).', + ); + + from_csv(path('memory://empty/*.csv'), filesystem: memory_filesystem())->withMetadataColumns(true)->schema(); + } + + public function test_an_empty_glob_with_a_declared_schema_answers_it_and_extracts_nothing(): void + { + $extractor = from_csv( + path('memory://empty/*.csv'), + filesystem: memory_filesystem(), + )->withSchema(schema(int_schema('id'))); + + static::assertEquals(schema(int_schema('id')), $extractor->schema()); + static::assertSame([], iterator_to_array($extractor->extract(flow_context()), false)); + } + public function test_a_hive_null_partition_value_reads_back_as_null(): void { $dir = __DIR__ . '/var/test_a_hive_null_partition_value_reads_back_as_null'; @@ -484,17 +517,32 @@ public function test_a_header_only_file_infers_every_column_as_string(): void static::assertSame([], iterator_to_array($extractor->extract(flow_context(config())))); } - public function test_an_empty_file_infers_an_empty_schema(): void + public function test_an_empty_file_refuses_to_infer_a_schema_and_extracts_nothing(): void { $extractor = from_csv(CSVFixtureContext::path('empty.csv')); - static::assertSame([], $extractor->schema()->definitions()); static::assertSame([], iterator_to_array($extractor->extract(flow_context(config())))); + + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_csv() has nothing to infer a schema from: no file matching "' + . path_real(CSVFixtureContext::path('empty.csv'))->uri() + . '" carries a column. Declare the schema with ->withSchema(...).', + ); + + $extractor->schema(); } - public function test_a_glob_of_only_empty_files_infers_an_empty_schema(): void + public function test_a_glob_of_only_empty_files_refuses_to_infer_a_schema(): void { - static::assertSame([], from_csv(CSVFixtureContext::path('two_empty/*.csv'))->schema()->definitions()); + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_csv() has nothing to infer a schema from: no file matching "' + . path_real(CSVFixtureContext::path('two_empty/*.csv'))->uri() + . '" carries a column. Declare the schema with ->withSchema(...).', + ); + + from_csv(CSVFixtureContext::path('two_empty/*.csv'))->schema(); } public function test_an_empty_file_in_a_glob_is_skipped(): void @@ -1224,17 +1272,12 @@ public function test_record_counts_match_fgetcsv_for_every_fixture(string $fixtu fclose($handle); - static::assertSame( + self::assertExtractedRowsCount( max(0, $records - 1), - df() - ->read( - from_csv($path) - ->withSeparator($dialect->separator) - ->withEnclosure($dialect->enclosure) - ->withEscape($dialect->escape), - ) - ->fetch() - ->count(), + from_csv($path) + ->withSeparator($dialect->separator) + ->withEnclosure($dialect->enclosure) + ->withEscape($dialect->escape), ); } diff --git a/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Unit/CSVExtractorTest.php b/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Unit/CSVExtractorTest.php index 47f52924f7..db49fa4978 100644 --- a/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Unit/CSVExtractorTest.php +++ b/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Unit/CSVExtractorTest.php @@ -123,7 +123,7 @@ public function test_a_narrow_first_row_does_not_skew_the_estimate(): void public function test_an_empty_file_declares_zero_rows_exactly(): void { $extractor = from_csv(CSVFixtureContext::path('empty.csv')); - $extractor->schema(); + self::assertExtractedRowsCount(0, $extractor); static::assertEquals(Cardinality::exact(0), $extractor->statistics()->rows); static::assertEquals(Cardinality::unknown(), $extractor->statistics()->size); diff --git a/src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/ExcelExtractor.php b/src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/ExcelExtractor.php index bae5e9de48..c67f7c41a3 100644 --- a/src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/ExcelExtractor.php +++ b/src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/ExcelExtractor.php @@ -5,6 +5,7 @@ namespace Flow\ETL\Adapter\Excel; use Flow\ETL\Cardinality; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Extractor; use Flow\ETL\Extractor\BatchableExtractor; @@ -125,7 +126,7 @@ public function isRepeatable(): bool public function scan(FlowContext $context, FileScan $scan): Generator { $fileColumns = $this->fileColumns($this->filesystem, $this->path); - $schema = $this->schema(); + $schema = $this->describe(); // only the first extract() after an inference reads on from its sample; every later one parses afresh $batches = new ExcelFileBatches( new WorkbookReader($this->readOptions, new ExcelFormatDetector($this->filesystem)), @@ -163,6 +164,21 @@ public function inferSchema(SchemaInferenceBuilder $builder): static } public function schema(): Schema + { + $schema = $this->describe(); + $fileColumns = $this->fileColumns($this->filesystem, $this->path); + + if ($this->schema === null && $fileColumns->withoutTail($schema)->count() === 0) { + throw InferredSchemaException::nothingToInferFrom('from_excel()', sprintf( + 'no file matching "%s" carries a column', + $this->path->uri(), + )); + } + + return $schema; + } + + private function describe(): Schema { $fileColumns = $this->fileColumns($this->filesystem, $this->path); diff --git a/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelExtractorTest.php b/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelExtractorTest.php index 57a8c1db70..9f539dfd84 100644 --- a/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelExtractorTest.php +++ b/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelExtractorTest.php @@ -76,6 +76,38 @@ public static function provide_nullable_fixtures(): iterable yield 'xlsx' => [__DIR__ . '/../Fixtures/nullable_fixture.xlsx']; } + public function test_an_empty_glob_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_excel() has nothing to infer a schema from: no file matching "' + . path_real(ExcelFixtureContext::file('*.no-match'))->uri() + . '" carries a column. Declare the schema with ->withSchema(...).', + ); + + from_excel(ExcelFixtureContext::file('*.no-match'))->schema(); + } + + public function test_an_empty_glob_with_metadata_columns_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_excel() has nothing to infer a schema from: no file matching "' + . path_real(ExcelFixtureContext::file('*.no-match'))->uri() + . '" carries a column. Declare the schema with ->withSchema(...).', + ); + + from_excel(ExcelFixtureContext::file('*.no-match'))->withMetadataColumns(true)->schema(); + } + + public function test_an_empty_glob_with_a_declared_schema_answers_it_and_extracts_nothing(): void + { + $extractor = from_excel(ExcelFixtureContext::file('*.no-match'))->withSchema(schema(int_schema('id'))); + + static::assertEquals(schema(int_schema('id')), $extractor->schema()); + static::assertSame([], iterator_to_array($extractor->extract(flow_context()), false)); + } + #[DataProvider('provide_fixtures')] public function test_extract_excel_file(string $fixtureName): void { @@ -463,11 +495,9 @@ public function test_cross_stream_infers_one_schema_and_the_partition_column(): ); } - #[TestWith(['empty_sheet.xlsx', []])] - #[TestWith(['empty_sheet.ods', []])] #[TestWith(['header_only.xlsx', ['id', 'name']])] #[TestWith(['header_only.ods', ['id', 'name']])] - public function test_header_only_and_empty_sheets(string $fixture, array $names): void + public function test_header_only_sheets(string $fixture, array $names): void { $extractor = from_excel(ExcelFixtureContext::file($fixture)); @@ -475,6 +505,24 @@ public function test_header_only_and_empty_sheets(string $fixture, array $names) static::assertCount(0, df()->extract($extractor)->fetch()->toArray()); } + #[TestWith(['empty_sheet.xlsx'])] + #[TestWith(['empty_sheet.ods'])] + public function test_an_empty_sheet_refuses_to_infer_a_schema_and_extracts_nothing(string $fixture): void + { + $extractor = from_excel(ExcelFixtureContext::file($fixture)); + + static::assertSame([], iterator_to_array($extractor->extract(flow_context()), false)); + + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_excel() has nothing to infer a schema from: no file matching "' + . path_real(ExcelFixtureContext::file($fixture))->uri() + . '" carries a column. Declare the schema with ->withSchema(...).', + ); + + $extractor->schema(); + } + /** * Guards the three tests below: OpenSpout only writes a shared-strings folder when it picks the file-based * cache, which depends on the runner's memory_limit and the fixture's unique-string count. If it ever picks diff --git a/src/adapter/etl-adapter-google-sheet/src/Flow/ETL/Adapter/GoogleSheet/GoogleSheetExtractor.php b/src/adapter/etl-adapter-google-sheet/src/Flow/ETL/Adapter/GoogleSheet/GoogleSheetExtractor.php index 66fc196333..546cd0ef87 100644 --- a/src/adapter/etl-adapter-google-sheet/src/Flow/ETL/Adapter/GoogleSheet/GoogleSheetExtractor.php +++ b/src/adapter/etl-adapter-google-sheet/src/Flow/ETL/Adapter/GoogleSheet/GoogleSheetExtractor.php @@ -197,6 +197,14 @@ public function schema(): Schema $this->gridRowCount = $sampler->gridRowCount(); } + if ($this->schema === null && $base->count() === 0) { + throw InferredSchemaException::nothingToInferFrom('from_google_sheet()', sprintf( + 'spreadsheet "%s" sheet "%s" carries no column', + $this->spreadsheetId, + $this->columnRange->sheetName, + )); + } + return $this->addMetadataColumns ? $base->add(str_schema('_spread_sheet_id'), str_schema('_sheet_name')) : $base; diff --git a/src/adapter/etl-adapter-google-sheet/tests/Flow/ETL/Adapter/GoogleSheet/Tests/Integration/GoogleSheetExtractorTest.php b/src/adapter/etl-adapter-google-sheet/tests/Flow/ETL/Adapter/GoogleSheet/Tests/Integration/GoogleSheetExtractorTest.php index c2be75e2cc..cfd97a33bb 100644 --- a/src/adapter/etl-adapter-google-sheet/tests/Flow/ETL/Adapter/GoogleSheet/Tests/Integration/GoogleSheetExtractorTest.php +++ b/src/adapter/etl-adapter-google-sheet/tests/Flow/ETL/Adapter/GoogleSheet/Tests/Integration/GoogleSheetExtractorTest.php @@ -282,7 +282,7 @@ public function test_a_header_only_sheet_infers_nullable_strings_and_yields_noth static::assertCount(0, df()->extract($extractor)->fetch()); } - public function test_an_empty_sheet_infers_an_empty_schema(): void + public function test_an_empty_sheet_refuses_to_infer_a_schema_and_extracts_nothing(): void { $extractor = from_google_sheet( $this->context->sheets( @@ -293,8 +293,15 @@ public function test_an_empty_sheet_infers_an_empty_schema(): void 'Sheet', ); - static::assertCount(0, $extractor->schema()->definitions()); - static::assertCount(0, df()->extract($extractor)->fetch()); + static::assertCount(0, iterator_to_array($extractor->extract(flow_context()))); + + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_google_sheet() has nothing to infer a schema from: spreadsheet "1234567890" sheet "Sheet" carries no ' + . 'column. Declare the schema with ->withSchema(...).', + ); + + $extractor->schema(); } public function test_unformatted_values_keep_the_types_the_api_gave_them(): void diff --git a/src/adapter/etl-adapter-google-sheet/tests/Flow/ETL/Adapter/GoogleSheet/Tests/Unit/GoogleSheetExtractorTest.php b/src/adapter/etl-adapter-google-sheet/tests/Flow/ETL/Adapter/GoogleSheet/Tests/Unit/GoogleSheetExtractorTest.php index 6ae6e9b50d..4f3f9b95fc 100644 --- a/src/adapter/etl-adapter-google-sheet/tests/Flow/ETL/Adapter/GoogleSheet/Tests/Unit/GoogleSheetExtractorTest.php +++ b/src/adapter/etl-adapter-google-sheet/tests/Flow/ETL/Adapter/GoogleSheet/Tests/Unit/GoogleSheetExtractorTest.php @@ -525,8 +525,15 @@ public function test_works_for_no_data(): void $values, ))->withRowsPerPage(20); - static::assertCount(0, $extractor->schema()->definitions()); static::assertCount(0, iterator_to_array($extractor->extract(flow_context()))); + + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_google_sheet() has nothing to infer a schema from: spreadsheet "spread-id" sheet "sheet" carries no ' + . 'column. Declare the schema with ->withSchema(...).', + ); + + $extractor->schema(); } public function test_extract_without_a_header_row_infers_generated_names(): void diff --git a/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonExtractor.php b/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonExtractor.php index ca88a7caac..1e0ac67949 100644 --- a/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonExtractor.php +++ b/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonExtractor.php @@ -5,6 +5,7 @@ namespace Flow\ETL\Adapter\JSON\JSONMachine; use Flow\ETL\Cardinality; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Extractor; use Flow\ETL\Extractor\BatchableExtractor; @@ -98,7 +99,7 @@ public function isRepeatable(): bool public function scan(FlowContext $context, FileScan $scan): Generator { $fileColumns = $this->fileColumns($this->filesystem, $this->path); - $schema = $this->schema(); + $schema = $this->describe(); $sources = iterator_to_array($this->sourceFiles($this->filesystem, $this->path, $scan->pathFilter), false); yield from (new FileReadLoop($fileColumns, $schema, $scan->conjuncts))->read( @@ -132,6 +133,21 @@ public function inferSchema(SchemaInferenceBuilder $builder): static } public function schema(): Schema + { + $schema = $this->describe(); + $fileColumns = $this->fileColumns($this->filesystem, $this->path); + + if ($this->schema === null && $fileColumns->withoutTail($schema)->count() === 0) { + throw InferredSchemaException::nothingToInferFrom( + $this->format === JsonFormat::Lines ? 'from_json_lines()' : 'from_json()', + sprintf('no file matching "%s" carries a column', $this->path->uri()), + ); + } + + return $schema; + } + + private function describe(): Schema { $fileColumns = $this->fileColumns($this->filesystem, $this->path); diff --git a/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JSONMachine/JsonExtractorTest.php b/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JSONMachine/JsonExtractorTest.php index 21e5a22b8c..033ba8dc0c 100644 --- a/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JSONMachine/JsonExtractorTest.php +++ b/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JSONMachine/JsonExtractorTest.php @@ -10,6 +10,7 @@ use Flow\ETL\Adapter\JSON\Tests\Context\JsonFixtureContext; use Flow\ETL\Cardinality; use Flow\ETL\Config; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Exception\SchemaMismatchException; use Flow\ETL\Extractor\File\FileScan; @@ -30,6 +31,7 @@ use function array_sum; use function count; use function Flow\ETL\Adapter\JSON\from_json; +use function Flow\ETL\Adapter\JSON\from_json_lines; use function Flow\ETL\Adapter\JSON\to_json; use function Flow\ETL\DSL\config; use function Flow\ETL\DSL\config_builder; @@ -55,6 +57,74 @@ final class JsonExtractorTest extends FlowTestCase { + public function test_an_empty_glob_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_json() has nothing to infer a schema from: no file matching "memory://empty/*.json" carries a column. ' + . 'Declare the schema with ->withSchema(...).', + ); + + from_json(path('memory://empty/*.json'), filesystem: memory_filesystem())->schema(); + } + + public function test_an_empty_glob_with_metadata_columns_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_json() has nothing to infer a schema from: no file matching "memory://empty/*.json" carries a column. ' + . 'Declare the schema with ->withSchema(...).', + ); + + from_json(path('memory://empty/*.json'), filesystem: memory_filesystem())->withMetadataColumns(true)->schema(); + } + + public function test_an_empty_glob_with_a_declared_schema_answers_it_and_extracts_nothing(): void + { + $extractor = from_json( + path('memory://empty/*.json'), + filesystem: memory_filesystem(), + )->withSchema(schema(int_schema('id'))); + + static::assertEquals(schema(int_schema('id')), $extractor->schema()); + static::assertSame([], iterator_to_array($extractor->extract(flow_context()), false)); + } + + public function test_an_empty_lines_glob_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_json_lines() has nothing to infer a schema from: no file matching "memory://empty/*.jsonl" carries a column. ' + . 'Declare the schema with ->withSchema(...).', + ); + + from_json_lines(path('memory://empty/*.jsonl'), filesystem: memory_filesystem())->schema(); + } + + public function test_an_empty_lines_glob_with_metadata_columns_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_json_lines() has nothing to infer a schema from: no file matching "memory://empty/*.jsonl" carries a column. ' + . 'Declare the schema with ->withSchema(...).', + ); + + from_json_lines(path('memory://empty/*.jsonl'), filesystem: memory_filesystem()) + ->withMetadataColumns(true) + ->schema(); + } + + public function test_an_empty_lines_glob_with_a_declared_schema_answers_it_and_extracts_nothing(): void + { + $extractor = from_json_lines( + path('memory://empty/*.jsonl'), + filesystem: memory_filesystem(), + )->withSchema(schema(int_schema('id'))); + + static::assertEquals(schema(int_schema('id')), $extractor->schema()); + static::assertSame([], iterator_to_array($extractor->extract(flow_context()), false)); + } + public function test_extract_does_not_mutate_user_provided_schema(): void { $schema = schema(int_schema('id'), str_schema('value')); @@ -464,12 +534,20 @@ public function test_a_key_missing_past_the_sample_is_null(): void #[TestWith(['empty_array.json'])] #[TestWith(['array_of_empty_object.json'])] #[TestWith(['empty.json'])] - public function test_an_empty_document_infers_an_empty_schema(string $fixture): void + public function test_an_empty_document_refuses_to_infer_a_schema_and_extracts_nothing(string $fixture): void { $extractor = from_json(JsonFixtureContext::path($fixture)); - static::assertSame([], $extractor->schema()->definitions()); static::assertSame([], iterator_to_array($extractor->extract(flow_context(config())), false)); + + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_json() has nothing to infer a schema from: no file matching "' + . path_real(JsonFixtureContext::path($fixture))->uri() + . '" carries a column. Declare the schema with ->withSchema(...).', + ); + + $extractor->schema(); } /** @@ -492,14 +570,22 @@ public function test_an_empty_document_in_a_glob_is_skipped(string $fixture, int static::assertSame([['id' => 1, 'name' => 'a']], df(config())->read($extractor)->fetch()->toArray()); } - public function test_an_empty_listing_infers_an_empty_schema(): void + public function test_an_empty_listing_refuses_to_infer_a_schema_without_opening_a_file(): void { $counting = new CountingFilesystem(new NativeLocalFilesystem()); - static::assertSame( - [], - from_json(JsonFixtureContext::path('no_such_dir/*.json'), filesystem: $counting)->schema()->definitions(), - ); + try { + from_json(JsonFixtureContext::path('no_such_dir/*.json'), filesystem: $counting)->schema(); + static::fail('an empty listing must refuse to infer a schema'); + } catch (InferredSchemaException $e) { + static::assertSame( + 'from_json() has nothing to infer a schema from: no file matching "' + . path_real(JsonFixtureContext::path('no_such_dir/*.json'))->uri() + . '" carries a column. Declare the schema with ->withSchema(...).', + $e->getMessage(), + ); + } + static::assertSame(0, $counting->readFromCalls); } diff --git a/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JSONMachine/JsonLinesExtractorTest.php b/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JSONMachine/JsonLinesExtractorTest.php index a7d0723814..97bd979d58 100644 --- a/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JSONMachine/JsonLinesExtractorTest.php +++ b/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JSONMachine/JsonLinesExtractorTest.php @@ -9,6 +9,7 @@ use Flow\ETL\Adapter\JSON\Tests\Context\JsonFixtureContext; use Flow\ETL\Cardinality; use Flow\ETL\Config; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Exception\SchemaMismatchException; use Flow\ETL\Extractor\File\FileScan; @@ -458,12 +459,20 @@ public function test_a_key_missing_past_the_sample_is_null(): void ); } - public function test_an_empty_document_infers_an_empty_schema(): void + public function test_an_empty_document_refuses_to_infer_a_schema_and_extracts_nothing(): void { $extractor = from_json_lines(JsonFixtureContext::path('empty.jsonl')); - static::assertSame([], $extractor->schema()->definitions()); static::assertSame([], iterator_to_array($extractor->extract(flow_context(config())), false)); + + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_json_lines() has nothing to infer a schema from: no file matching "' + . path_real(JsonFixtureContext::path('empty.jsonl'))->uri() + . '" carries a column. Declare the schema with ->withSchema(...).', + ); + + $extractor->schema(); } /** @@ -503,16 +512,22 @@ public function test_a_blank_line_is_skipped_by_schema_and_read(string $filesyst static::assertSame([['id' => 1], ['id' => 2]], df(config())->read($extractor)->fetch()->toArray()); } - public function test_an_empty_listing_infers_an_empty_schema(): void + public function test_an_empty_listing_refuses_to_infer_a_schema_without_opening_a_file(): void { $counting = new CountingFilesystem(new NativeLocalFilesystem()); - static::assertSame( - [], - from_json_lines(JsonFixtureContext::path('no_such_dir/*.jsonl'), filesystem: $counting) - ->schema() - ->definitions(), - ); + try { + from_json_lines(JsonFixtureContext::path('no_such_dir/*.jsonl'), filesystem: $counting)->schema(); + static::fail('an empty listing must refuse to infer a schema'); + } catch (InferredSchemaException $e) { + static::assertSame( + 'from_json_lines() has nothing to infer a schema from: no file matching "' + . path_real(JsonFixtureContext::path('no_such_dir/*.jsonl'))->uri() + . '" carries a column. Declare the schema with ->withSchema(...).', + $e->getMessage(), + ); + } + static::assertSame(0, $counting->readFromCalls); } diff --git a/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetExtractor.php b/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetExtractor.php index e5706a4437..9cffac7658 100644 --- a/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetExtractor.php +++ b/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetExtractor.php @@ -4,6 +4,7 @@ namespace Flow\ETL\Adapter\Parquet; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Extractor; use Flow\ETL\Extractor\BatchableExtractor; @@ -73,6 +74,8 @@ final class ParquetExtractor implements private Options $options; + private ?Schema $schema = null; + private SchemaConverter $schemaConverter; private ?Statistics $statistics = null; @@ -117,16 +120,23 @@ public function isRepeatable(): bool public function scan(FlowContext $context, FileScan $scan): Generator { $sources = iterator_to_array($this->sourceFiles($this->filesystem, $this->path, $scan->pathFilter), false); - // every file is read under the first file's schema, or the union of all of them + + if ($sources === []) { + $this->closeFirstFile(); + + return; + } + + // every file is read under the declared schema, the first file's schema, or the union of all of them $target = $this->schema(); $fileColumns = $this->fileColumns($this->filesystem, $this->path); $batches = new ParquetFileBatches( $this->opener(), $fileColumns, $this->firstFile, - $this->unionByName - ? null - : new DerivedSchema($this->derivedSchema($this->schemaFiles(), $this->unionByName), $this->derivedFrom), + $this->schema === null && !$this->unionByName + ? new DerivedSchema($this->derivedSchema($this->schemaFiles(), $this->unionByName), $this->derivedFrom) + : null, $fileColumns->withoutTail($target), $scan->conjuncts, ); @@ -147,10 +157,24 @@ public function scan(FlowContext $context, FileScan $scan): Generator public function schema(): Schema { - return $this->fileColumns($this->filesystem, $this->path)->declare($this->derivedSchema( - $this->schemaFiles(), - $this->unionByName, - )); + $fileColumns = $this->fileColumns($this->filesystem, $this->path); + + if ($this->schema !== null) { + return $fileColumns->declare( + $this->columns === [] ? $this->schema : $this->schema->keep(...$this->columns), + ); + } + + $derived = $this->derivedSchema($this->schemaFiles(), $this->unionByName); + + if ($fileColumns->withoutTail($derived)->count() === 0) { + throw InferredSchemaException::nothingToInferFrom('from_parquet()', sprintf( + 'no file matching "%s" carries a column', + $this->path->uri(), + )); + } + + return $fileColumns->declare($derived); } public function statistics(): Statistics @@ -325,8 +349,8 @@ private function opener(): ParquetSourceFileOpener public function withSchema(Schema $schema): static { - throw new InvalidArgumentException( - 'Parquet is a self-describing format and does not accept a schema; declaring one is not supported yet.', - ); + $this->schema = $schema; + + return $this; } } diff --git a/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetFileBatches.php b/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetFileBatches.php index 54e707c23f..b0316d08b5 100644 --- a/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetFileBatches.php +++ b/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetFileBatches.php @@ -21,6 +21,8 @@ use Flow\Parquet\Reader\ReadPlanner; use Generator; +use function array_diff; + final class ParquetFileBatches implements OffsetSkippingFileBatches { /** @@ -84,6 +86,10 @@ public function batches( $this->derived?->refuseDivergence($file->source()->uri(), $file->schema()); $fileBody = $this->fileColumns->withoutTail($file->schema()); + $fileBody = $fileBody->gracefulRemove(...array_diff( + $fileBody->references()->names(), + $body->references()->names(), + )); $kept = []; foreach ($this->conjuncts as $conjunct) { diff --git a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetExtractorTest.php b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetExtractorTest.php index d94f898d99..bdc5a39189 100644 --- a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetExtractorTest.php +++ b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetExtractorTest.php @@ -9,6 +9,7 @@ use Flow\ETL\Adapter\Parquet\Tests\Context\ParquetFilesContext; use Flow\ETL\Cardinality; use Flow\ETL\Exception\InferredSchemaException; +use Flow\ETL\Exception\SchemaMismatchException; use Flow\ETL\Extractor\File\FileScan; use Flow\ETL\Extractor\Signal; use Flow\ETL\Plan\Stage; @@ -16,7 +17,7 @@ use Flow\ETL\Tests\Context\ExtractedRows; use Flow\ETL\Tests\Double\CountingFilesystem; use Flow\ETL\Tests\Double\RecordingFilesystem; -use Flow\ETL\Tests\FlowTestCase; +use Flow\ETL\Tests\FlowIntegrationTestCase; use Flow\Filesystem\Local\NativeLocalFilesystem; use Flow\Filesystem\Tests\Double\RejectingFilter; use Flow\Parquet\Binary\ByteOrder; @@ -66,7 +67,7 @@ use function Flow\Types\DSL\type_uuid; use function iterator_to_array; -final class ParquetExtractorTest extends FlowTestCase +final class ParquetExtractorTest extends FlowIntegrationTestCase { /** * @return Generator @@ -77,6 +78,144 @@ public static function engines(): Generator yield 'php engine' => [new PhpParquetEngine()]; } + public function test_a_declared_schema_reads_files_by_name(): void + { + df() + ->read(from_array([['id' => 1, 'name' => 'one']])) + ->write(to_parquet($this->cacheDir->suffix('by-name.parquet'))) + ->run(); + + static::assertSame( + [['id' => 1]], + df() + ->read(from_parquet( + $this->cacheDir->suffix('by-name.parquet'), + filesystem: $this->fs(), + )->withSchema(schema(int_schema('id')))) + ->fetch() + ->toArray(), + ); + } + + public function test_a_declared_nullable_column_a_file_lacks_reads_as_null(): void + { + df() + ->read(from_array([['id' => 1]])) + ->write(to_parquet($this->cacheDir->suffix('nullable.parquet'))) + ->run(); + + static::assertSame( + [['id' => 1, 'name' => null]], + df() + ->read(from_parquet( + $this->cacheDir->suffix('nullable.parquet'), + filesystem: $this->fs(), + )->withSchema(schema(int_schema('id'), str_schema('name', nullable: true)))) + ->fetch() + ->toArray(), + ); + } + + public function test_a_declared_non_nullable_column_a_file_lacks_fails(): void + { + df() + ->read(from_array([['id' => 1]])) + ->write(to_parquet($this->cacheDir->suffix('not-null.parquet'))) + ->run(); + + $this->expectException(SchemaMismatchException::class); + + df() + ->read(from_parquet( + $this->cacheDir->suffix('not-null.parquet'), + filesystem: $this->fs(), + )->withSchema(schema(int_schema('id'), str_schema('name')))) + ->fetch(); + } + + public function test_a_declared_schema_skips_the_divergence_check(): void + { + df() + ->read(from_array([['id' => 1, 'a' => 'x']])) + ->write(to_parquet($this->cacheDir->suffix('diverging/a.parquet'))) + ->run(); + df() + ->read(from_array([['id' => 2, 'b' => 'y']])) + ->write(to_parquet($this->cacheDir->suffix('diverging/b.parquet'))) + ->run(); + + static::assertSame( + [['id' => 1, 'a' => 'x', 'b' => null], ['id' => 2, 'a' => null, 'b' => 'y']], + df() + ->read(from_parquet( + $this->cacheDir->suffix('diverging/*.parquet'), + filesystem: $this->fs(), + )->withSchema(schema( + int_schema('id', nullable: true), + str_schema('a', nullable: true), + str_schema('b', nullable: true), + ))) + ->sortBy(ref('id')) + ->fetch() + ->toArray(), + ); + } + + public function test_with_columns_narrows_a_declared_schema(): void + { + static::assertEquals( + schema(int_schema('id')), + from_parquet($this->cacheDir->suffix('narrowed/*.parquet'), filesystem: $this->fs()) + ->withSchema(schema(int_schema('id'), str_schema('name'))) + ->withColumns(['id']) + ->schema(), + ); + } + + public function test_a_read_whose_path_filter_rejects_every_file_closes_the_file_the_schema_kept_open(): void + { + $filesystem = new CountingFilesystem(new NativeLocalFilesystem()); + $extractor = from_parquet(path(__DIR__ . '/Fixtures/orders_1k.parquet'), filesystem: $filesystem); + $extractor->schema(); + + iterator_to_array( + $extractor->scan(flow_context(config()), new FileScan(pathFilter: new RejectingFilter())), + false, + ); + + static::assertCount(1, $filesystem->openedStreams); + static::assertSame(1, $filesystem->closedStreams()); + } + + public function test_a_glob_without_files_refuses_to_infer(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_parquet() has nothing to infer a schema from: no file matching "' + . $this->cacheDir->suffix('**/*.parquet')->uri() + . '" carries a column. Declare the schema with ->withSchema(...).', + ); + + df() + ->read(from_parquet($this->cacheDir->suffix('**/*.parquet'), filesystem: $this->fs())) + ->select('id') + ->fetch(); + } + + public function test_a_glob_without_files_reads_nothing_under_a_declared_schema(): void + { + static::assertCount( + 0, + df() + ->read(from_parquet( + $this->cacheDir->suffix('**/*.parquet'), + filesystem: $this->fs(), + )->withSchema(schema(int_schema('id')))) + ->select('id') + ->fetch(), + ); + } + #[DataProvider('engines')] public function test_a_body_predicate_skips_row_groups(?ParquetEngine $engine): void { diff --git a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Unit/ParquetExtractorTest.php b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Unit/ParquetExtractorTest.php index f4883a2542..fca7ac752f 100644 --- a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Unit/ParquetExtractorTest.php +++ b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Unit/ParquetExtractorTest.php @@ -7,7 +7,6 @@ use Closure; use Flow\ETL\Adapter\Parquet\ParquetExtractor; use Flow\ETL\Cardinality; -use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Tests\Double\CountingFilesystem; use Flow\ETL\Tests\Double\SpyBackend; use Flow\ETL\Tests\FlowTestCase; @@ -189,16 +188,14 @@ public function test_schema_then_statistics_reads_the_footer_once(): void static::assertSame(1, $filesystem->readFromCalls); } - public function test_with_schema_is_refused(): void + public function test_a_declared_schema_answers_without_opening_a_file(): void { - $this->expectException(InvalidArgumentException::class); - $this->expectExceptionMessage( - 'Parquet is a self-describing format and does not accept a schema; declaring one is not supported yet.', + static::assertEquals( + schema(int_schema('id')), + from_parquet(path('memory://no-match/*.parquet'), filesystem: memory_filesystem()) + ->withSchema(schema(int_schema('id'))) + ->schema(), ); - - from_parquet(__DIR__ . '/../Integration/Fixtures/Pagination/01_1000.parquet')->withSchema(schema(int_schema( - 'id', - ))); } public function test_extract_builds_through_the_config_backend(): void diff --git a/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/Loader/XMLLoaderTest.php b/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/Loader/XMLLoaderTest.php index 8d358534eb..3872b76f4c 100644 --- a/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/Loader/XMLLoaderTest.php +++ b/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/Loader/XMLLoaderTest.php @@ -16,10 +16,12 @@ use function Flow\ETL\DSL\df; use function Flow\ETL\DSL\from_array; use function Flow\ETL\DSL\from_sequence_number; +use function Flow\ETL\DSL\int_schema; use function Flow\ETL\DSL\lit; use function Flow\ETL\DSL\overwrite; use function Flow\ETL\DSL\partition_by; use function Flow\ETL\DSL\ref; +use function Flow\ETL\DSL\schema; use function Flow\ETL\DSL\select; use function Flow\ETL\DSL\to_transformation; use function Flow\Filesystem\DSL\memory_filesystem; @@ -93,7 +95,7 @@ public function test_transformation_loader_writes_all_batches_to_xml(): void public function test_writing_empty_rows(): void { df() - ->read(from_array([])) + ->read(from_array([], schema(int_schema('id')))) ->write(to_xml($path = $this->cacheDir->suffix('test_xml_loader.xml'))) ->run(); diff --git a/src/bridge/symfony/http-foundation/tests/Flow/Bridge/Symfony/HttpFoundation/Tests/Unit/Response/FlowBufferedResponseTest.php b/src/bridge/symfony/http-foundation/tests/Flow/Bridge/Symfony/HttpFoundation/Tests/Unit/Response/FlowBufferedResponseTest.php index 18a3230f87..b4b1113c66 100644 --- a/src/bridge/symfony/http-foundation/tests/Flow/Bridge/Symfony/HttpFoundation/Tests/Unit/Response/FlowBufferedResponseTest.php +++ b/src/bridge/symfony/http-foundation/tests/Flow/Bridge/Symfony/HttpFoundation/Tests/Unit/Response/FlowBufferedResponseTest.php @@ -18,7 +18,7 @@ final class FlowBufferedResponseTest extends FlowTestCase { public function test_response_from_empty_dataset(): void { - $response = http_stream_open(from_array([]))->response(http_json_output()); + $response = http_stream_open(from_array([], schema(int_schema('id'))))->response(http_json_output()); static::assertEquals('', $response->getContent()); static::assertEquals(204, $response->getStatusCode()); diff --git a/src/bridge/symfony/http-foundation/tests/Flow/Bridge/Symfony/HttpFoundation/Tests/Unit/Response/FlowStreamedResponseTest.php b/src/bridge/symfony/http-foundation/tests/Flow/Bridge/Symfony/HttpFoundation/Tests/Unit/Response/FlowStreamedResponseTest.php index b1a6d3b150..ef90f2f843 100644 --- a/src/bridge/symfony/http-foundation/tests/Flow/Bridge/Symfony/HttpFoundation/Tests/Unit/Response/FlowStreamedResponseTest.php +++ b/src/bridge/symfony/http-foundation/tests/Flow/Bridge/Symfony/HttpFoundation/Tests/Unit/Response/FlowStreamedResponseTest.php @@ -9,12 +9,14 @@ use function Flow\Bridge\Symfony\HttpFoundation\http_json_output; use function Flow\Bridge\Symfony\HttpFoundation\http_stream_open; use function Flow\ETL\DSL\from_array; +use function Flow\ETL\DSL\int_schema; +use function Flow\ETL\DSL\schema; final class FlowStreamedResponseTest extends TestCase { public function test_response_from_empty_dataset(): void { - $response = http_stream_open(from_array([]))->streamedResponse(http_json_output()); + $response = http_stream_open(from_array([], schema(int_schema('id'))))->streamedResponse(http_json_output()); static::assertEquals('', $response->getContent()); static::assertEquals(200, $response->getStatusCode()); diff --git a/src/core/etl/src/Flow/ETL/Exception/InferredSchemaException.php b/src/core/etl/src/Flow/ETL/Exception/InferredSchemaException.php index 743bf6c26d..1e81a9d0b0 100644 --- a/src/core/etl/src/Flow/ETL/Exception/InferredSchemaException.php +++ b/src/core/etl/src/Flow/ETL/Exception/InferredSchemaException.php @@ -53,6 +53,19 @@ public static function filesDiverge(string $source, string $readFrom, Validation )); } + /** + * @param string $source the DSL call that built the source, e.g. 'from_parquet()' + * @param string $reason why nothing was there, e.g. 'no row carries a column' + */ + public static function nothingToInferFrom(string $source, string $reason): self + { + return new self(sprintf( + '%s has nothing to infer a schema from: %s. Declare the schema with ->withSchema(...).', + $source, + $reason, + )); + } + /** * @param int $row - the row's 0-based position in the whole source, not in the batch the gate refused */ diff --git a/src/core/etl/src/Flow/ETL/Extractor.php b/src/core/etl/src/Flow/ETL/Extractor.php index 2cb20bb0be..775c2d24ba 100644 --- a/src/core/etl/src/Flow/ETL/Extractor.php +++ b/src/core/etl/src/Flow/ETL/Extractor.php @@ -4,6 +4,7 @@ namespace Flow\ETL; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Exception\SchemaNotDerivableException; use Flow\ETL\Extractor\Signal; use Flow\ETL\Extractor\Statistics; @@ -35,6 +36,7 @@ public function extract(FlowContext $context, ?int $limit = null): Generator; * Called once per RUN - every run plans afresh - so it MUST be idempotent and cheap on repeat: memoise * what it sniffs, as CSVExtractor and ArrayExtractor do. * + * @throws InferredSchemaException when no schema was declared and the source has no column to infer one from * @throws SchemaNotDerivableException when the source cannot describe what it will produce */ public function schema(): Schema; diff --git a/src/core/etl/src/Flow/ETL/Extractor/ArrayExtractor.php b/src/core/etl/src/Flow/ETL/Extractor/ArrayExtractor.php index 72fcd2079c..7d7f2e1ea5 100644 --- a/src/core/etl/src/Flow/ETL/Extractor/ArrayExtractor.php +++ b/src/core/etl/src/Flow/ETL/Extractor/ArrayExtractor.php @@ -5,6 +5,7 @@ namespace Flow\ETL\Extractor; use Flow\ETL\Cardinality; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Exception\InvalidLogicException; use Flow\ETL\Extractor; use Flow\ETL\Extractor\Memory\InferredRows; @@ -71,7 +72,7 @@ public function isRepeatable(): bool public function extract(FlowContext $context, ?int $limit = null): Generator { $batchSize = $this->batchSize(); - $schema = $this->schema(); + $schema = $this->describe(); $buffer = []; // a one-shot dataset is typed from every row it spills, so only an array's bounded sample can be outrun @@ -123,6 +124,17 @@ public function inferSchema(SchemaInferenceBuilder $builder): static } public function schema(): Schema + { + $schema = $this->describe(); + + if ($this->schema === null && $schema->count() === 0) { + throw InferredSchemaException::nothingToInferFrom('from_array()', 'no row carries a column'); + } + + return $schema; + } + + private function describe(): Schema { if ($this->schema !== null) { return $this->schema; diff --git a/src/core/etl/src/Flow/ETL/Extractor/CacheExtractor.php b/src/core/etl/src/Flow/ETL/Extractor/CacheExtractor.php index a2488e42ab..0cb7bdf371 100644 --- a/src/core/etl/src/Flow/ETL/Extractor/CacheExtractor.php +++ b/src/core/etl/src/Flow/ETL/Extractor/CacheExtractor.php @@ -7,12 +7,16 @@ use Flow\ETL\Cache; use Flow\ETL\Cache\CacheIndex; use Flow\ETL\Cardinality; +use Flow\ETL\Exception\InferredSchemaException; +use Flow\ETL\Exception\SchemaNotDerivableException; use Flow\ETL\Extractor; use Flow\ETL\FlowContext; use Flow\ETL\Rows; use Flow\ETL\Schema; use Generator; +use function sprintf; + final class CacheExtractor implements Extractor, RewindableExtractor { private ?Schema $schema = null; @@ -120,10 +124,17 @@ public function schema(): Schema return $this->fallbackExtractor->schema(); } - // A cache entry that does not exist yet has no columns. That is an answer, not an - // unanswerable question, and threading a cache in from the FlowContext to look harder - // would break the rule the comment above states. - return new Schema(); + if ($this->cache !== null) { + throw InferredSchemaException::nothingToInferFrom('from_cache()', sprintf( + 'the cache holds no entry "%s" and no fallback extractor is set', + $this->id, + )); + } + + throw SchemaNotDerivableException::extractor( + self::class, + 'no cache was handed to from_cache(), and the one in the FlowContext is known only when the pipeline runs', + ); } /** diff --git a/src/core/etl/src/Flow/ETL/Extractor/ChainExtractor.php b/src/core/etl/src/Flow/ETL/Extractor/ChainExtractor.php index df3a827f23..20e8cc0f39 100644 --- a/src/core/etl/src/Flow/ETL/Extractor/ChainExtractor.php +++ b/src/core/etl/src/Flow/ETL/Extractor/ChainExtractor.php @@ -5,6 +5,7 @@ namespace Flow\ETL\Extractor; use Flow\ETL\Cardinality; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Extractor; use Flow\ETL\FlowContext; use Flow\ETL\Rows; @@ -40,11 +41,12 @@ public function extract(FlowContext $context, ?int $limit = null): Generator // Every child's batches are projected onto the folded shape, so each yields the union // rather than only the columns its own source happens to carry. A child that cannot // describe itself makes the chain unable to promise one shape, so the fold is allowed to - // throw. - $schema = $this->schema(); + // throw - on the first batch, so a chain that yields nothing reads under any declared parent. + $schema = null; foreach ($this->extractors as $extractor) { foreach ($extractor->extract($context) as $rows) { + $schema ??= $this->schema(); $signal = yield $rows->matchTo($schema, $context->backend()); if ($signal === Signal::STOP) { @@ -65,6 +67,10 @@ public function schema(): Schema return $this->schema; } + if ($this->extractors === []) { + throw InferredSchemaException::nothingToInferFrom('from_all()', 'it chains no extractor'); + } + $schema = new Schema(); foreach ($this->extractors as $extractor) { diff --git a/src/core/etl/src/Flow/ETL/Extractor/MemoryExtractor.php b/src/core/etl/src/Flow/ETL/Extractor/MemoryExtractor.php index 1161a8b1a1..86758f8a09 100644 --- a/src/core/etl/src/Flow/ETL/Extractor/MemoryExtractor.php +++ b/src/core/etl/src/Flow/ETL/Extractor/MemoryExtractor.php @@ -4,6 +4,7 @@ namespace Flow\ETL\Extractor; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Extractor; use Flow\ETL\Extractor\Memory\InferredRows; use Flow\ETL\Extractor\Memory\InMemoryRows; @@ -45,7 +46,7 @@ public function isRepeatable(): bool public function extract(FlowContext $context, ?int $limit = null): Generator { $batchSize = $this->batchSize(); - $schema = $this->schema(); + $schema = $this->describe(); $buffer = []; $batches = new InferredRows('from_memory()', $this->schema === null ? $this->inference : null); @@ -79,6 +80,17 @@ public function inferSchema(SchemaInferenceBuilder $builder): static } public function schema(): Schema + { + $schema = $this->describe(); + + if ($this->schema === null && $schema->count() === 0) { + throw InferredSchemaException::nothingToInferFrom('from_memory()', 'no row carries a column'); + } + + return $schema; + } + + private function describe(): Schema { if ($this->schema !== null) { return $this->schema; diff --git a/src/core/etl/src/Flow/ETL/Extractor/RowsExtractor.php b/src/core/etl/src/Flow/ETL/Extractor/RowsExtractor.php index 89d1539c66..05eae4fd23 100644 --- a/src/core/etl/src/Flow/ETL/Extractor/RowsExtractor.php +++ b/src/core/etl/src/Flow/ETL/Extractor/RowsExtractor.php @@ -5,6 +5,7 @@ namespace Flow\ETL\Extractor; use Flow\ETL\Cardinality; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Extractor; use Flow\ETL\FlowContext; use Flow\ETL\Rows; @@ -37,7 +38,7 @@ public function isRepeatable(): bool */ public function extract(FlowContext $context, ?int $limit = null): Generator { - $schema = $this->schema(); + $schema = $this->describe(); foreach ($this->rows as $rows) { $signal = yield $rows->matchTo($schema, $context->backend()); @@ -49,6 +50,17 @@ public function extract(FlowContext $context, ?int $limit = null): Generator } public function schema(): Schema + { + $schema = $this->describe(); + + if ($this->schema === null && $schema->count() === 0) { + throw InferredSchemaException::nothingToInferFrom('from_rows()', 'no row carries a column'); + } + + return $schema; + } + + private function describe(): Schema { if ($this->schema !== null) { return $this->schema; diff --git a/src/core/etl/src/Flow/ETL/Extractor/SequenceExtractor.php b/src/core/etl/src/Flow/ETL/Extractor/SequenceExtractor.php index 4ef18782bf..255e5c2031 100644 --- a/src/core/etl/src/Flow/ETL/Extractor/SequenceExtractor.php +++ b/src/core/etl/src/Flow/ETL/Extractor/SequenceExtractor.php @@ -4,6 +4,7 @@ namespace Flow\ETL\Extractor; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Extractor; use Flow\ETL\Extractor\Memory\InferredRows; use Flow\ETL\Extractor\Memory\InMemoryRows; @@ -49,7 +50,7 @@ public function isRepeatable(): bool public function extract(FlowContext $context, ?int $limit = null): Generator { $batchSize = $this->batchSize(); - $schema = $this->schema(); + $schema = $this->describe(); $buffer = []; $batches = new InferredRows('the sequence', $this->schema === null ? $this->inference : null); @@ -84,6 +85,17 @@ public function inferSchema(SchemaInferenceBuilder $builder): static } public function schema(): Schema + { + $schema = $this->describe(); + + if ($this->schema === null && $schema->count() === 0) { + throw InferredSchemaException::nothingToInferFrom('the sequence', 'it generates no item'); + } + + return $schema; + } + + private function describe(): Schema { if ($this->schema !== null) { return $this->schema; diff --git a/src/core/etl/src/Flow/Floe/FloeExtractor.php b/src/core/etl/src/Flow/Floe/FloeExtractor.php index d65e2f3ffb..766f74eb62 100644 --- a/src/core/etl/src/Flow/Floe/FloeExtractor.php +++ b/src/core/etl/src/Flow/Floe/FloeExtractor.php @@ -5,6 +5,7 @@ namespace Flow\Floe; use Flow\ETL\Column\PhpBackend; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Extractor; use Flow\ETL\Extractor\BatchableExtractor; @@ -46,6 +47,8 @@ final class FloeExtractor implements private ?int $offset = null; + private ?Schema $schema = null; + private ?Statistics $statistics = null; private bool $unionByName = false; @@ -81,18 +84,24 @@ public function isRepeatable(): bool */ public function scan(FlowContext $context, FileScan $scan): Generator { + $sources = iterator_to_array($this->sourceFiles($this->filesystem, $this->path, $scan->pathFilter), false); + + if ($sources === []) { + return; + } + $fileColumns = $this->fileColumns($this->filesystem, $this->path); yield from (new FileReadLoop($fileColumns, $this->schema(), $scan->conjuncts))->read( - iterator_to_array($this->sourceFiles($this->filesystem, $this->path, $scan->pathFilter), false), + $sources, new FloeFileBatches( $this->filesystem, $this->codec, $this->chunkSize, $fileColumns, - $this->unionByName - ? null - : new DerivedSchema($this->derivedSchema($this->files(), $this->unionByName), $this->derivedFrom), + $this->schema === null && !$this->unionByName + ? new DerivedSchema($this->derivedSchema($this->files(), $this->unionByName), $this->derivedFrom) + : null, ), $this->batchSize(), $context, @@ -100,16 +109,24 @@ public function scan(FlowContext $context, FileScan $scan): Generator ); } - /** - * Footer-only source schema (two ranged reads per file, no row scan). One file unless - * unionByName() asks for the fold, and memoised, so repeated calls cost nothing. - */ public function schema(): Schema { - return $this->fileColumns($this->filesystem, $this->path)->declare($this->derivedSchema( - $this->files(), - $this->unionByName, - )); + $fileColumns = $this->fileColumns($this->filesystem, $this->path); + + if ($this->schema !== null) { + return $fileColumns->declare($this->schema); + } + + $derived = $this->derivedSchema($this->files(), $this->unionByName); + + if ($fileColumns->withoutTail($derived)->count() === 0) { + throw InferredSchemaException::nothingToInferFrom('from_floe()', sprintf( + 'no file matching "%s" carries a column', + $this->path->uri(), + )); + } + + return $fileColumns->declare($derived); } public function statistics(): Statistics @@ -184,8 +201,8 @@ private function files(): Generator public function withSchema(Schema $schema): static { - throw new InvalidArgumentException( - 'Floe is a self-describing format and does not accept a schema; declaring one is not supported yet.', - ); + $this->schema = $schema; + + return $this; } } diff --git a/src/core/etl/src/Flow/Floe/FloeFileBatches.php b/src/core/etl/src/Flow/Floe/FloeFileBatches.php index 0f6d2954c0..0a017f7644 100644 --- a/src/core/etl/src/Flow/Floe/FloeFileBatches.php +++ b/src/core/etl/src/Flow/Floe/FloeFileBatches.php @@ -16,6 +16,8 @@ use Flow\Filesystem\Filesystem; use Generator; +use function array_diff; + final readonly class FloeFileBatches implements OffsetSkippingFileBatches { /** @@ -60,6 +62,10 @@ public function batches( $this->derived?->refuseDivergence($file->source()->uri(), $file->schema()); $fileBody = $this->fileColumns->withoutTail($file->schema()); + $fileBody = $fileBody->gracefulRemove(...array_diff( + $fileBody->references()->names(), + $body->references()->names(), + )); foreach ($file->reader->rows($batchSize, $window->offset, $window->limit) as $rows) { if (!$fileBody->isSame($file->schema())) { diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/EmptySourceTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/EmptySourceTest.php new file mode 100644 index 0000000000..013b294aa2 --- /dev/null +++ b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/EmptySourceTest.php @@ -0,0 +1,75 @@ +expectException(InferredSchemaException::class); + $this->expectExceptionMessage('from_all() has nothing to infer a schema from: it chains no extractor. Declare the schema with ' + . '->withSchema(...).'); + + df() + ->read(from_all()) + ->filter(ref('response_status_code')->equals(lit(200))) + ->fetch(); + } + + public function test_from_all_without_extractors_reads_nothing_under_a_declared_schema(): void + { + static::assertCount( + 0, + df() + ->read(from_all()->withSchema(schema(int_schema('response_status_code')))) + ->filter(ref('response_status_code')->equals(lit(200))) + ->fetch(), + ); + } + + public function test_from_all_over_an_empty_array_reads_nothing_under_a_declared_schema(): void + { + static::assertCount( + 0, + df() + ->read(from_all(from_array([]))->withSchema(schema(int_schema('response_status_code')))) + ->filter(ref('response_status_code')->equals(lit(200))) + ->fetch(), + ); + } + + public function test_from_array_without_rows_refuses_to_infer(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_array() has nothing to infer a schema from: no row carries a column. Declare the schema with ' + . '->withSchema(...).', + ); + + df()->read(from_array([]))->withEntry('status', ref('response_status_code'))->fetch(); + } + + public function test_a_missing_column_in_non_empty_input_still_fails(): void + { + $this->expectException(SchemaDefinitionNotFoundException::class); + + df() + ->read(from_array([['id' => 1]])) + ->select('response_status_code') + ->fetch(); + } +} diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/Extractor/CacheExtractorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/Extractor/CacheExtractorTest.php index 32cf5f4fc6..e8e9804163 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Integration/Extractor/CacheExtractorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Integration/Extractor/CacheExtractorTest.php @@ -10,6 +10,8 @@ use Flow\ETL\Cardinality; use Flow\ETL\Column\AdaptiveBackend; use Flow\ETL\Column\PhpBackend; +use Flow\ETL\Exception\InferredSchemaException; +use Flow\ETL\Exception\SchemaNotDerivableException; use Flow\ETL\Extractor\Signal; use Flow\ETL\Rows; use Flow\ETL\Tests\Double\CountingCache; @@ -299,14 +301,27 @@ public function test_fallback_extractor(): void * A cache entry that has not been written yet has no columns, which is an answer. Reaching into * the FlowContext for a cache would make describing a source need a running pipeline. */ - public function test_schema_returns_empty_schema_on_cache_id_miss(): void + public function test_schema_is_not_derivable_without_a_cache_instance(): void { - static::assertEquals(schema(), from_cache('non_existing_cache_key')->schema()); + $this->expectException(SchemaNotDerivableException::class); + $this->expectExceptionMessage( + 'Flow\\ETL\\Extractor\\CacheExtractor cannot describe what it will produce before producing it: no cache was ' + . 'handed to from_cache(), and the one in the FlowContext is known only when the pipeline runs. Declare the ' + . 'schema with ->withSchema().', + ); + + from_cache('non_existing_cache_key')->schema(); } - public function test_schema_returns_empty_schema_when_the_injected_cache_does_not_hold_the_id(): void + public function test_schema_refuses_to_infer_when_the_injected_cache_does_not_hold_the_id(): void { - static::assertEquals(schema(), from_cache('non_existing_cache_key', cache: new InMemoryCache())->schema()); + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_cache() has nothing to infer a schema from: the cache holds no entry "non_existing_cache_key" and ' + . 'no fallback extractor is set. Declare the schema with ->withSchema(...).', + ); + + from_cache('non_existing_cache_key', cache: new InMemoryCache())->schema(); } public function test_a_cache_hit_is_not_matched_to_the_fallback_extractors_schema(): void diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/Extractor/FileSourceFilesystemTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/Extractor/FileSourceFilesystemTest.php index 1a00c0a944..1d3d5762b1 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Integration/Extractor/FileSourceFilesystemTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Integration/Extractor/FileSourceFilesystemTest.php @@ -26,6 +26,8 @@ use function Flow\ETL\DSL\files; use function Flow\ETL\DSL\from_array; use function Flow\ETL\DSL\from_path_partitions; +use function Flow\ETL\DSL\int_schema; +use function Flow\ETL\DSL\schema; use function Flow\ETL\DSL\to_array; use function Flow\Filesystem\DSL\memory_filesystem; use function Flow\Filesystem\DSL\path; @@ -211,7 +213,9 @@ public function test_a_shared_memory_filesystem_round_trips_but_two_calls_do_not $unpaired = []; df() - ->read(from_csv(path('memory://s1.csv'), filesystem: memory_filesystem())) + ->read(from_csv(path('memory://s1.csv'), filesystem: memory_filesystem())->withSchema(schema(int_schema( + 'id', + )))) ->write(to_array($unpaired)) ->run(); diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Exception/InferredSchemaExceptionTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Exception/InferredSchemaExceptionTest.php index 6b024e7166..bf4954081d 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Exception/InferredSchemaExceptionTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Exception/InferredSchemaExceptionTest.php @@ -15,6 +15,15 @@ final class InferredSchemaExceptionTest extends FlowTestCase { + public function test_nothing_to_infer_from_names_the_source_the_reason_and_the_fix(): void + { + static::assertSame( + 'from_array() has nothing to infer a schema from: no row carries a column. Declare the schema with ' + . '->withSchema(...).', + InferredSchemaException::nothingToInferFrom('from_array()', 'no row carries a column')->getMessage(), + ); + } + public function test_columns_diverge_names_the_source_both_diffs_the_knobs_and_both_remedies(): void { $exception = InferredSchemaException::columnsDiverge( diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/ExecutorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/ExecutorTest.php index 2b4f76aa06..4514ebf5c7 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/ExecutorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/ExecutorTest.php @@ -65,7 +65,7 @@ public function test_merge_concatenates_the_batches_once(): void } })(); - $merged = (new Executor())->merge($batches, PhysicalPlanMother::reading(from_array([]))); + $merged = (new Executor())->merge($batches, PhysicalPlanMother::reading(from_array([], $schema))); static::assertSame([1, 2, 3, 4, 5], $merged->reduceToArray('id')); static::assertSame( @@ -85,7 +85,7 @@ public function test_merge_of_empty_batches_keeps_the_last_batch(): void ['other'], array_keys( (new Executor()) - ->merge($batches, PhysicalPlanMother::reading(from_array([]))) + ->merge($batches, PhysicalPlanMother::reading(from_array([], schema(int_schema('other'))))) ->schema() ->definitions(), ), diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/ArrayExtractorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/ArrayExtractorTest.php index 6537e21283..0008b13004 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/ArrayExtractorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/ArrayExtractorTest.php @@ -28,6 +28,7 @@ use function Flow\ETL\DSL\config_builder; use function Flow\ETL\DSL\data_frame; use function Flow\ETL\DSL\execution_context; +use function Flow\ETL\DSL\flow_context; use function Flow\ETL\DSL\from_array; use function Flow\ETL\DSL\infer_schema; use function Flow\ETL\DSL\int_schema; @@ -40,6 +41,40 @@ final class ArrayExtractorTest extends FlowTestCase { + public function test_an_empty_source_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_array() has nothing to infer a schema from: no row carries a column. Declare the schema with ->withSchema(...).', + ); + + from_array([])->schema(); + } + + public function test_an_empty_source_with_a_declared_schema_answers_it(): void + { + static::assertEquals(schema(int_schema('id')), from_array([])->withSchema(schema(int_schema('id')))->schema()); + } + + public function test_an_empty_source_extracts_nothing_without_a_declared_schema(): void + { + static::assertSame([], iterator_to_array(from_array([])->extract(flow_context()), false)); + } + + public function test_an_empty_generator_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_array() has nothing to infer a schema from: no row carries a column. Declare the schema with ' + . '->withSchema(...).', + ); + + /** @var callable(): Generator> $generator */ + $generator = static fn(): Generator => yield from []; + + from_array($generator())->schema(); + } + public function test_a_declared_schema_writes_no_spill(): void { /** @var callable(): Generator> $generator */ diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/ChainExtractorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/ChainExtractorTest.php index 80658384f0..ee4862f331 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/ChainExtractorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/ChainExtractorTest.php @@ -5,6 +5,7 @@ namespace Flow\ETL\Tests\Unit\Extractor; use Flow\ETL\Cardinality; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Extractor; use Flow\ETL\Extractor\BatchableExtractor; use Flow\ETL\Extractor\Statistics; @@ -31,6 +32,44 @@ final class ChainExtractorTest extends FlowTestCase { + public function test_a_chain_without_extractors_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage('from_all() has nothing to infer a schema from: it chains no extractor. Declare the schema with ' + . '->withSchema(...).'); + + from_all()->schema(); + } + + public function test_a_chain_without_extractors_answers_a_declared_schema(): void + { + static::assertEquals(schema(int_schema('id')), from_all()->withSchema(schema(int_schema('id')))->schema()); + } + + public function test_an_empty_nested_chain_reads_nothing_under_a_declared_schema(): void + { + static::assertSame( + [], + iterator_to_array( + from_all(from_all(), from_all(from_array([]))) + ->withSchema(schema(int_schema('id'))) + ->extract(flow_context()), + false, + ), + ); + } + + public function test_a_declared_chain_schema_reads_an_empty_child_without_asking_it(): void + { + static::assertSame( + [], + iterator_to_array( + from_all(from_array([]))->withSchema(schema(int_schema('id')))->extract(flow_context()), + false, + ), + ); + } + public function test_a_chain_containing_a_data_frame_source_folds_its_schema(): void { static::assertEquals( diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/MemoryExtractorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/MemoryExtractorTest.php index c79d2a680f..512b7e4e56 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/MemoryExtractorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/MemoryExtractorTest.php @@ -27,6 +27,29 @@ final class MemoryExtractorTest extends FlowTestCase { + public function test_an_empty_source_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_memory() has nothing to infer a schema from: no row carries a column. Declare the schema with ->withSchema(...).', + ); + + from_memory(new ArrayMemory())->schema(); + } + + public function test_an_empty_source_with_a_declared_schema_answers_it(): void + { + static::assertEquals( + schema(int_schema('id')), + from_memory(new ArrayMemory())->withSchema(schema(int_schema('id')))->schema(), + ); + } + + public function test_an_empty_source_extracts_nothing_without_a_declared_schema(): void + { + static::assertSame([], iterator_to_array(from_memory(new ArrayMemory())->extract(flow_context()), false)); + } + public function test_declared_schema_is_used_to_hydrate_extracted_rows(): void { $memory = new ArrayMemory(); @@ -89,7 +112,11 @@ public function test_schema_before_save_does_not_freeze_empty(): void $memory = new ArrayMemory(); $extractor = from_memory($memory); - static::assertSame([], $extractor->schema()->references()->names()); + try { + $extractor->schema(); + static::fail('an empty memory must refuse to infer a schema'); + } catch (InferredSchemaException) { + } $memory->save([['number' => 1, 'name' => 'one']]); diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/RowsExtractorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/RowsExtractorTest.php index 64e3a5fef4..aa9afb9d34 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/RowsExtractorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/RowsExtractorTest.php @@ -5,6 +5,7 @@ namespace Flow\ETL\Tests\Unit\Extractor; use Flow\ETL\Cardinality; +use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Extractor\Statistics; use Flow\ETL\Tests\FlowTestCase; @@ -14,9 +15,30 @@ use function Flow\ETL\DSL\int_schema; use function Flow\ETL\DSL\schema; use function Flow\ETL\DSL\str_schema; +use function iterator_to_array; final class RowsExtractorTest extends FlowTestCase { + public function test_an_empty_source_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_rows() has nothing to infer a schema from: no row carries a column. Declare the schema with ->withSchema(...).', + ); + + from_rows()->schema(); + } + + public function test_an_empty_source_with_a_declared_schema_answers_it(): void + { + static::assertEquals(schema(int_schema('id')), from_rows()->withSchema(schema(int_schema('id')))->schema()); + } + + public function test_an_empty_source_extracts_nothing_without_a_declared_schema(): void + { + static::assertSame([], iterator_to_array(from_rows()->extract(flow_context()), false)); + } + public function test_process_extractor(): void { $rows = array_to_rows( diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/SequenceExtractorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/SequenceExtractorTest.php index 81730d3827..53918e07e2 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/SequenceExtractorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Extractor/SequenceExtractorTest.php @@ -24,12 +24,59 @@ use function Flow\ETL\DSL\from_sequence_date_period_recurrences; use function Flow\ETL\DSL\from_sequence_number; use function Flow\ETL\DSL\infer_schema; +use function Flow\ETL\DSL\int_schema; use function Flow\ETL\DSL\schema; use function Flow\ETL\DSL\str_schema; use function iterator_to_array; final class SequenceExtractorTest extends FlowTestCase { + public function test_an_empty_source_refuses_to_infer_a_schema(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'the sequence has nothing to infer a schema from: it generates no item. Declare the schema with ->withSchema(...).', + ); + + from_sequence_date_period( + 'id', + new DateTimeImmutable('2024-01-01'), + new DateInterval('P1D'), + new DateTimeImmutable('2024-01-01'), + )->schema(); + } + + public function test_an_empty_source_with_a_declared_schema_answers_it(): void + { + static::assertEquals( + schema(int_schema('id')), + from_sequence_date_period( + 'id', + new DateTimeImmutable('2024-01-01'), + new DateInterval('P1D'), + new DateTimeImmutable('2024-01-01'), + ) + ->withSchema(schema(int_schema('id'))) + ->schema(), + ); + } + + public function test_an_empty_source_extracts_nothing_without_a_declared_schema(): void + { + static::assertSame( + [], + iterator_to_array( + from_sequence_date_period( + 'id', + new DateTimeImmutable('2024-01-01'), + new DateInterval('P1D'), + new DateTimeImmutable('2024-01-01'), + )->extract(flow_context()), + false, + ), + ); + } + public function test_declared_schema_is_used_to_hydrate_extracted_rows(): void { self::assertExtractedRowsEquals( diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Join/JoinStepsTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Join/JoinStepsTest.php index 3cf2a3e7e9..ae893cdf26 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Join/JoinStepsTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Join/JoinStepsTest.php @@ -23,7 +23,7 @@ final class JoinStepsTest extends FlowTestCase public function test_builds_a_hash_join_processor(): void { $steps = JoinSteps::of( - PhysicalPlanMother::reading(from_rows(rows(schema()))), + PhysicalPlanMother::reading(from_rows(rows(schema()))->withSchema(schema(int_schema('user_id')))), Expression::on(['id' => 'user_id']), Join::left, config_builder()->build(), diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Processor/HashJoinProcessorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Processor/HashJoinProcessorTest.php index 23829a8c54..2e5f1d900a 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Processor/HashJoinProcessorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Processor/HashJoinProcessorTest.php @@ -241,7 +241,7 @@ public function test_handles_empty_left_side(): void public function test_handles_empty_right_side(): void { $processor = HashJoinProcessorMother::grace( - PhysicalPlanMother::reading(from_rows(rows(schema()))), + PhysicalPlanMother::reading(from_rows(rows(schema()))->withSchema(schema(int_schema('user_id')))), Expression::on(['id' => 'user_id']), Join::inner, ); diff --git a/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeExtractorTest.php b/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeExtractorTest.php index 0e6f80a785..02c825fba4 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeExtractorTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeExtractorTest.php @@ -7,6 +7,8 @@ use DateTimeImmutable; use DateTimeZone; use Flow\ETL\Cardinality; +use Flow\ETL\Exception\InferredSchemaException; +use Flow\ETL\Exception\SchemaMismatchException; use Flow\ETL\Function\Between\Boundary; use Flow\ETL\Rows; use Flow\ETL\Tests\Context\ExtractedRows; @@ -39,6 +41,118 @@ final class FloeExtractorTest extends FlowIntegrationTestCase { + public function test_a_declared_schema_reads_files_by_name(): void + { + df() + ->read(from_array([['id' => 1, 'name' => 'one']])) + ->write(to_floe($this->cacheDir->suffix('by-name.floe'))) + ->run(); + + static::assertSame( + [['id' => 1]], + df() + ->read(from_floe( + $this->cacheDir->suffix('by-name.floe'), + filesystem: $this->fs(), + )->withSchema(schema(int_schema('id')))) + ->fetch() + ->toArray(), + ); + } + + public function test_a_declared_nullable_column_a_file_lacks_reads_as_null(): void + { + df() + ->read(from_array([['id' => 1]])) + ->write(to_floe($this->cacheDir->suffix('nullable.floe'))) + ->run(); + + static::assertSame( + [['id' => 1, 'name' => null]], + df() + ->read(from_floe($this->cacheDir->suffix('nullable.floe'), filesystem: $this->fs())->withSchema(schema( + int_schema('id'), + str_schema('name', nullable: true), + ))) + ->fetch() + ->toArray(), + ); + } + + public function test_a_declared_non_nullable_column_a_file_lacks_fails(): void + { + df() + ->read(from_array([['id' => 1]])) + ->write(to_floe($this->cacheDir->suffix('not-null.floe'))) + ->run(); + + $this->expectException(SchemaMismatchException::class); + + df() + ->read(from_floe($this->cacheDir->suffix('not-null.floe'), filesystem: $this->fs())->withSchema(schema( + int_schema('id'), + str_schema('name'), + ))) + ->fetch(); + } + + public function test_a_declared_schema_skips_the_divergence_check(): void + { + df() + ->read(from_array([['id' => 1, 'a' => 'x']])) + ->write(to_floe($this->cacheDir->suffix('diverging/a.floe'))) + ->run(); + df() + ->read(from_array([['id' => 2, 'b' => 'y']])) + ->write(to_floe($this->cacheDir->suffix('diverging/b.floe'))) + ->run(); + + static::assertSame( + [['id' => 1, 'a' => 'x', 'b' => null], ['id' => 2, 'a' => null, 'b' => 'y']], + df() + ->read(from_floe( + $this->cacheDir->suffix('diverging/*.floe'), + filesystem: $this->fs(), + )->withSchema(schema( + int_schema('id', nullable: true), + str_schema('a', nullable: true), + str_schema('b', nullable: true), + ))) + ->sortBy(ref('id')) + ->fetch() + ->toArray(), + ); + } + + public function test_a_glob_without_files_refuses_to_infer(): void + { + $this->expectException(InferredSchemaException::class); + $this->expectExceptionMessage( + 'from_floe() has nothing to infer a schema from: no file matching "' + . $this->cacheDir->suffix('**/*.floe')->uri() + . '" carries a column. Declare the schema with ->withSchema(...).', + ); + + df() + ->read(from_floe($this->cacheDir->suffix('**/*.floe'), filesystem: $this->fs())) + ->select('id') + ->fetch(); + } + + public function test_a_glob_without_files_reads_nothing_under_a_declared_schema(): void + { + static::assertCount( + 0, + df() + ->read(from_floe( + $this->cacheDir->suffix('**/*.floe'), + filesystem: $this->fs(), + )->withSchema(schema(int_schema('id')))) + ->select('id') + ->fetch(), + ); + } + public function test_a_glob_declares_an_approximate_row_count(): void { FloeStreamReaderContext::write( diff --git a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeExtractorTest.php b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeExtractorTest.php index 3510f1bdda..aac9e6cd40 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeExtractorTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Unit/FloeExtractorTest.php @@ -292,14 +292,14 @@ public function test_extract_yields_the_metadata_column_schema_promises(): void static::assertSame($path->uri(), $batches[0]->column('_input_file_uri')->value(0)); } - public function test_with_schema_is_refused(): void + public function test_a_declared_schema_answers_without_opening_a_file(): void { - $this->expectException(InvalidArgumentException::class); - $this->expectExceptionMessage( - 'Floe is a self-describing format and does not accept a schema; declaring one is not supported yet.', + static::assertEquals( + schema(int_schema('id')), + from_floe(path('memory://no-match/*.floe'), filesystem: memory_filesystem()) + ->withSchema(schema(int_schema('id'))) + ->schema(), ); - - from_floe(path('memory://x.floe'), filesystem: memory_filesystem())->withSchema(schema(int_schema('id'))); } public function test_extract_builds_through_the_config_backend(): void