Merge pull request #2696 from flow-php/empty-input-schema

fix(flow-php/etl): empty sources without a declared schema fail at the source

Author: norberttech

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
 <?php
@@ -91,6 +90,28 @@ use Flow\Parquet\Writer;
 );
 ```
 
+### Declared schema
+
+```php
+<?php
+
+use function Flow\ETL\Adapter\Parquet\from_parquet;
+use function Flow\ETL\DSL\{data_frame, int_schema, schema, str_schema, to_output};
+
+data_frame()
+    ->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

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
+<?php
+
+use function Flow\ETL\DSL\{data_frame, from_array, int_schema, schema, str_schema, to_output};
+
+data_frame()
+    ->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

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
 <?php

documentation/components/core/schema.md

@@ -210,8 +210,8 @@ The HTTP extractors accept a schema the same way - see
 [Typing the row with a schema](/documentation/components/adapters/http.md) for declaring the response body as a
 `structure`.
 
-Self-descriptive formats carry their schema in the file itself, so their extractors read it from there and have no
-`withSchema()` - Parquet and Floe in this repository.
+Self-descriptive formats carry their schema in the file itself - Parquet and Floe in this repository. Their extractors
+read it from there unless `withSchema()` declares one, which reads every file by name.
 
 ### PostgreSQL sources describe themselves, and keep `withSchema()`
 

documentation/upgrading.md

@@ -138,6 +138,15 @@ public function scan(FlowContext $context, FileScan $scan): Generator
 | Symfony HttpClient `url.full`, `url.scheme`, `server.address`, `server.port` came from the `$url` argument (`/me`, `unknown` with `base_uri`)     | they come from the effective URL, `base_uri` and the `query` option included                                                                                                                                                                                                                                                 |
 | `exception.stacktrace` was `getTraceAsString()`, with argument values when `zend.exception_ignore_args` is off                                    | `StackTrace::format()`: every frame renders `()`, whatever `zend.exception_ignore_args` is                                                                                                                                                                                                                                   |
 
+### 9) `flow-php/etl`, `flow-php/etl-adapter-*` - an empty source without a declared schema fails at the source; Parquet and Floe accept `withSchema()`
+
+| Before                                                                                                  | After                                                                                                                   |
+|---------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------|
+| an empty source (`from_all()`, `from_array([])`, a glob that matches no file) described an empty schema | `schema()` throws `InferredSchemaException`; declare the schema with `->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

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);
 

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),
         );
     }
 

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);

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);
 

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,18 +495,34 @@ 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));
 
         static::assertSame($names, $extractor->schema()->references()->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