From 7b07270cfa3633fc1fa7e9ceb5e3b6c484432fb0 Mon Sep 17 00:00:00 2001 From: Norbert Orzechowicz Date: Wed, 7 Oct 2026 17:16:46 -0500 Subject: [PATCH] fix: partitioned reads keep their values and datetime filters work - datetime and float partition values written in full, read back exactly - parquet and floe type a partition column the file holds like the read - string operands cast to a datetime / date counterpart in comparisons - comparisons share a ComparisonFunction contract, between() checks types at bind - RoundTripPrecision moved to flow-php/types --- documentation/components/core/partitioning.md | 43 ++++++++- documentation/upgrading.md | 43 +++++++++ .../Tests/Integration/CSVExtractorTest.php | 28 ++++++ .../Tests/Integration/ExcelExtractorTest.php | 26 ++++++ .../Flow/ETL/Adapter/JSON/PhpJsonEncoder.php | 2 +- .../JSONMachine/JsonExtractorTest.php | 29 ++++++ .../ETL/Adapter/Parquet/ParquetExtractor.php | 2 +- .../ParquetExtractorFooterTest.php | 61 +++++++++++++ .../Integration/ParquetExtractorTest.php | 60 +++++++++++++ .../Parquet/Tests/Integration/ParquetTest.php | 61 +++++++++++++ .../Tests/Integration/TextExtractorTest.php | 26 ++++++ .../Integration/XMLParserExtractorTest.php | 29 ++++++ .../etl/src/Flow/ETL/Column/TextValues.php | 1 + .../Flow/ETL/Extractor/File/FileColumns.php | 3 +- .../etl/src/Flow/ETL/Function/Between.php | 26 +++++- .../Flow/ETL/Function/ComparisonFunction.php | 18 ++++ src/core/etl/src/Flow/ETL/Function/Equals.php | 18 +++- .../etl/src/Flow/ETL/Function/GreaterThan.php | 18 +++- .../Flow/ETL/Function/GreaterThanEqual.php | 18 +++- src/core/etl/src/Flow/ETL/Function/IsIn.php | 18 +++- .../etl/src/Flow/ETL/Function/LessThan.php | 18 +++- .../src/Flow/ETL/Function/LessThanEqual.php | 18 +++- .../etl/src/Flow/ETL/Function/NotEquals.php | 18 +++- .../Flow/ETL/Function/ReferenceResolver.php | 18 ++-- .../ETL/Function/TemporalStringCoercion.php | 89 +++++++++++++++++++ src/core/etl/src/Flow/Floe/FloeExtractor.php | 2 +- .../DataFrame/PartitioningTest.php | 10 +-- .../TemporalStringComparisonTest.php | 77 ++++++++++++++++ .../ETL/Tests/Unit/Function/BetweenTest.php | 15 ++++ .../Unit/Function/ReferenceResolverTest.php | 66 ++++++++++++++ .../Function/TemporalStringCoercionTest.php | 82 +++++++++++++++++ .../Unit/Loader/File/PartitionRouterTest.php | 2 +- .../Tests/Unit/Loader/RowPartitionsTest.php | 6 +- .../Rule/PushFilterIntoSourceTest.php | 24 ++--- .../Tests/Integration/FloeExtractorTest.php | 57 ++++++++++++ .../src/Flow/Filesystem/Partition.php | 29 +++++- .../Filesystem/Tests/Unit/PartitionTest.php | 86 +++++++++++++++++- .../Flow/Types/Type}/RoundTripPrecision.php | 4 +- .../Unit/Type}/RoundTripPrecisionTest.php | 4 +- 39 files changed, 1097 insertions(+), 58 deletions(-) create mode 100644 src/core/etl/src/Flow/ETL/Function/ComparisonFunction.php create mode 100644 src/core/etl/src/Flow/ETL/Function/TemporalStringCoercion.php create mode 100644 src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/TemporalStringComparisonTest.php create mode 100644 src/core/etl/tests/Flow/ETL/Tests/Unit/Function/TemporalStringCoercionTest.php rename src/{core/etl/src/Flow/ETL/Column => lib/types/src/Flow/Types/Type}/RoundTripPrecision.php (91%) rename src/{core/etl/tests/Flow/ETL/Tests/Unit/Column => lib/types/tests/Flow/Types/Tests/Unit/Type}/RoundTripPrecisionTest.php (96%) diff --git a/documentation/components/core/partitioning.md b/documentation/components/core/partitioning.md index 7b004e4e4..186cc6841 100644 --- a/documentation/components/core/partitioning.md +++ b/documentation/components/core/partitioning.md @@ -125,6 +125,44 @@ as-is. Appended files get randomized suffixes (`sales_a1b2c3.csv`), which become Aim for balanced partition sizes: `date` or `department` over a unique `id`, which would create one tiny partition per row. +## Partition Values + +Each value is written in full, so a typed read gets it back unchanged: + +| Column type | Value | Directory | +|-------------|----------------------------------|-------------------------------------------------| +| `datetime` | `2026-10-02 13:00:00 UTC` | `date=2026-10-02T13%3A00%3A00%2B00%3A00` | +| `datetime` | `2026-10-02 13:00:00.123456 UTC` | `date=2026-10-02T13%3A00%3A00.123456%2B00%3A00` | +| `date` | `2026-10-02` | `date=2026-10-02` | +| `float` | `1.0E-9` | `x=1.0e-9` | + +A `datetime` column makes one directory per distinct value. For one directory per day, partition by a `date`: + +```php +read(from_array([ + ['created_at' => new DateTimeImmutable('2024-01-01 09:00:00'), 'amount' => 100], + ['created_at' => new DateTimeImmutable('2024-01-01 13:00:00'), 'amount' => 200], + ])) + ->withEntry('day', ref('created_at')->cast(type_date())) + ->write(to_parquet(__DIR__ . '/output/events.parquet') + ->partitionBy(partition_by('day')) + ->saveMode(overwrite())) + ->run(); +``` + +```text +output/ +└── day=2024-01-01/ + └── events.parquet +``` + ## Save Modes with Partitioning The save mode decides how existing partition directories are handled. @@ -223,7 +261,10 @@ data_frame() Partition values from the path become regular columns. Given a schema that does not declare a partition column, it is appended as a **string** column - declare it (`int_schema('date')`) to read -it as another type. +it as another type. A file that also holds the partition column (`partition_by(...)->writeColumns()`) types +it from the file schema (Parquet, Floe), so `filter(ref('date')->between(lit($from), lit($to)))` with +`DateTimeImmutable` bounds prunes. String bounds (`lit('2026-10-02')`) are cast to the column's `date` / +`datetime` type, so they prune too. ### Partition Pruning diff --git a/documentation/upgrading.md b/documentation/upgrading.md index 4860a4a17..4f678e750 100644 --- a/documentation/upgrading.md +++ b/documentation/upgrading.md @@ -98,6 +98,49 @@ Page by scalar columns. Pass at least one key. +### 11) `flow-php/filesystem`, `flow-php/etl` - `datetime` and `float` partition values keep their full value + +| Before | After | +|----------------------------------------------------------------------------------------------------------------------|------------------------------------------| +| `partition_by('date')` over `datetime` `2026-10-02 13:00:00 UTC` - `date=2026-10-02` | `date=2026-10-02T13%3A00%3A00%2B00%3A00` | +| reading it back as `datetime` (`writeColumns()` or `partition_types(date: type_datetime())`) - `2026-10-02 00:00:00` | `2026-10-02 13:00:00` | +| `partition_by('x')` over `float` `1.0E-9` - `x=0.000000`, read back `0.0` | `x=1.0e-9`, read back `1.0E-9` | +| `from_path_partitions()` / `files()` / an undeclared partition column - `'2026-10-02'` | `'2026-10-02T13:00:00+00:00'` | + +Directories written before keep reading as midnight. Writing into a dataset that has them adds a second directory per +day, and `overwrite()` replaces only the new one - rewrite the dataset, or partition by a `date` column to keep +`date=2026-10-02`: + +```php +$df->withEntry('date', ref('date')->cast(type_date())) +``` + +### 12) `flow-php/etl-adapter-parquet`, `flow-php/etl` - a partition column the file holds is typed from the file in a pushed filter + +| Before | After | +|------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------------| +| `from_parquet()` / `from_floe()` over `writeColumns()` data - `partitionSchema()` types `date` `string` | the file's type (`datetime`, `date`, `integer`, ...) | +| `filter(ref('date')->between(lit(DateTime), lit(DateTime)))` - throws `Boundary::compare requires value and bounds to be of a comparable type` | returns the matching rows, directories outside the range are skipped | +| same filter with `lit('2026-10-02')` bounds | returns the same rows - see `### 14)` | + +### 13) `flow-php/etl` - `RoundTripPrecision` moved to `flow-php/types` + +| Before | After | +|--------------------------------------------------------|-----------------------------------------| +| `Flow\ETL\Column\RoundTripPrecision` | `Flow\Types\Type\RoundTripPrecision` | +| `force()` throws `Flow\ETL\Exception\RuntimeException` | `Flow\Types\Exception\RuntimeException` | + +### 14) `flow-php/etl` - a string compared with a `datetime` / `date` is cast to it + +| Before | After | +|---------------------------------------------------------------------------------------------|------------------------------------------------------------------| +| `ref('ts')->between(lit('2026-10-02'), lit('2026-10-03'))` - throws `Boundary::compare ...` | rows between those dates | +| `ref('ts')->equals(lit('2026-10-02'))` - throws `Can't compare '(datetime == string)'` | compares as `datetime` | +| `ref('ts')->equals(ref('s'))`, `s` a string column - throws `Can't compare` | `s` cast per row; a non-date value throws `Cast function failed` | +| `ref('a')->between(lit(1), lit($dateTime))` - throws at evaluation | throws `Can't compare` when the plan binds | + +`same()` / `notSame()` stay strict. + --- ## Upgrading from 0.44.x to 0.45.x 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 1993e6a0f..501557b90 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 @@ -5,6 +5,7 @@ namespace Flow\ETL\Adapter\CSV\Tests\Integration; use Closure; +use DateTimeImmutable; use Flow\ETL\Adapter\CSV\CSVExtractor; use Flow\ETL\Adapter\CSV\Tests\Context\CSVFixtureContext; use Flow\ETL\Config; @@ -37,6 +38,7 @@ use function Flow\ETL\Adapter\CSV\from_csv; use function Flow\ETL\Adapter\CSV\to_csv; use function Flow\ETL\DSL\config; +use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\df; use function Flow\ETL\DSL\flow_context; use function Flow\ETL\DSL\from_array; @@ -50,7 +52,9 @@ use function Flow\ETL\DSL\schema_metadata; use function Flow\ETL\DSL\schema_to_ascii; use function Flow\ETL\DSL\str_schema; +use function Flow\Filesystem\DSL\memory_filesystem; use function Flow\Filesystem\DSL\native_local_filesystem; +use function Flow\Filesystem\DSL\path; use function Flow\Filesystem\DSL\path_real; use function Flow\Types\DSL\type_boolean; use function Flow\Types\DSL\type_html; @@ -1297,4 +1301,28 @@ public function test_one_extractor_read_twice_interleaved_gives_each_read_every_ count(array_keys($filesystem->calls, 'closeSource', true)), ); } + + public function test_partition_schema_is_the_partition_block_of_the_read_schema(): void + { + $memory = memory_filesystem(); + + df() + ->read(from_array( + [ + ['date' => new DateTimeImmutable('2026-10-01 00:00:00 UTC'), 'text' => 'a'], + ['date' => new DateTimeImmutable('2026-10-02 00:00:00 UTC'), 'text' => 'b'], + ], + schema(datetime_schema('date'), str_schema('text')), + )) + ->write( + to_csv(path('memory://var/csv/file.csv'), filesystem: $memory)->partitionBy( + partition_by('date')->writeColumns(), + ), + ) + ->run(); + + $extractor = from_csv(path('memory://var/csv/**/*.csv'), filesystem: $memory); + + static::assertEquals($extractor->schema()->keep('date'), $extractor->partitionSchema()); + } } 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 e4ad96e05..b5190be47 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 @@ -29,6 +29,7 @@ use function array_keys; use function Flow\ETL\Adapter\Excel\DSL\from_excel; use function Flow\ETL\Adapter\Excel\DSL\is_valid_excel_sheet_name; +use function Flow\ETL\Adapter\Excel\DSL\to_excel; use function Flow\ETL\DSL\array_to_rows; use function Flow\ETL\DSL\config; use function Flow\ETL\DSL\config_builder; @@ -36,9 +37,12 @@ use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\df; use function Flow\ETL\DSL\flow_context; +use function Flow\ETL\DSL\from_array; use function Flow\ETL\DSL\from_rows; use function Flow\ETL\DSL\infer_schema; use function Flow\ETL\DSL\int_schema; +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\str_schema; @@ -958,4 +962,26 @@ public function test_one_extractor_read_twice_interleaved_gives_each_read_every_ static::assertGreaterThan(0, $first->count()); static::assertSame($first->toArray(), $second->toArray()); } + + public function test_partition_schema_is_the_partition_block_of_the_read_schema(): void + { + df() + ->read(from_array( + [ + ['date' => new DateTimeImmutable('2026-10-01 00:00:00 UTC'), 'text' => 'a'], + ['date' => new DateTimeImmutable('2026-10-02 00:00:00 UTC'), 'text' => 'b'], + ], + schema(datetime_schema('date'), str_schema('text')), + )) + ->write( + to_excel(__DIR__ . '/var/partition_schema/file.xlsx') + ->saveMode(overwrite()) + ->partitionBy(partition_by('date')->writeColumns()), + ) + ->run(); + + $extractor = from_excel(__DIR__ . '/var/partition_schema/**/*.xlsx'); + + static::assertEquals($extractor->schema()->keep('date'), $extractor->partitionSchema()); + } } diff --git a/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/PhpJsonEncoder.php b/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/PhpJsonEncoder.php index 9769f277e..5434a7148 100644 --- a/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/PhpJsonEncoder.php +++ b/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/PhpJsonEncoder.php @@ -6,11 +6,11 @@ use DateTimeInterface; use Flow\ETL\Column\Column; -use Flow\ETL\Column\RoundTripPrecision; use Flow\ETL\Column\TextValues; use Flow\ETL\Exception\RuntimeException; use Flow\ETL\Rows; use Flow\Types\Type; +use Flow\Types\Type\RoundTripPrecision; use JsonException; use function array_combine; 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 89297df69..2475b0459 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 @@ -5,6 +5,7 @@ namespace Flow\ETL\Adapter\JSON\Tests\Integration\JSONMachine; use Closure; +use DateTimeImmutable; use Flow\ETL\Adapter\JSON\JSONMachine\JsonExtractor; use Flow\ETL\Adapter\JSON\Tests\Context\JsonFixtureContext; use Flow\ETL\Cardinality; @@ -28,13 +29,17 @@ use function array_sum; use function count; use function Flow\ETL\Adapter\JSON\from_json; +use function Flow\ETL\Adapter\JSON\to_json; use function Flow\ETL\DSL\config; use function Flow\ETL\DSL\config_builder; use function Flow\ETL\DSL\data_frame; +use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\df; 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; +use function Flow\ETL\DSL\partition_by; use function Flow\ETL\DSL\partition_types; use function Flow\ETL\DSL\schema; use function Flow\ETL\DSL\schema_to_ascii; @@ -724,4 +729,28 @@ public function test_one_extractor_read_twice_interleaved_gives_each_read_every_ count(array_keys($filesystem->calls, 'closeSource', true)), ); } + + public function test_partition_schema_is_the_partition_block_of_the_read_schema(): void + { + $memory = memory_filesystem(); + + df() + ->read(from_array( + [ + ['date' => new DateTimeImmutable('2026-10-01 00:00:00 UTC'), 'text' => 'a'], + ['date' => new DateTimeImmutable('2026-10-02 00:00:00 UTC'), 'text' => 'b'], + ], + schema(datetime_schema('date'), str_schema('text')), + )) + ->write( + to_json(path('memory://var/json/file.json'), filesystem: $memory)->partitionBy( + partition_by('date')->writeColumns(), + ), + ) + ->run(); + + $extractor = from_json(path('memory://var/json/**/*.json'), filesystem: $memory); + + static::assertEquals($extractor->schema()->keep('date'), $extractor->partitionSchema()); + } } 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 fbfe8cb34..1f1f37441 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 @@ -168,7 +168,7 @@ public function unionByName(bool $union = true): self public function partitionSchema(): Schema { - return $this->fileColumns($this->filesystem, $this->path)->partitions(new Schema()); + return $this->fileColumns($this->filesystem, $this->path)->partitions($this->schema()); } public function source(): Path diff --git a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetExtractorFooterTest.php b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetExtractorFooterTest.php index 1a3e1d46e..8a52068fe 100644 --- a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetExtractorFooterTest.php +++ b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetExtractorFooterTest.php @@ -4,21 +4,32 @@ namespace Flow\ETL\Adapter\Parquet\Tests\Integration; +use DateTimeImmutable; use Flow\ETL\Adapter\Parquet\Tests\Context\ParquetFilesContext; +use Flow\ETL\Function\Between\Boundary; use Flow\ETL\Rows; use Flow\ETL\Tests\Double\CountingFilesystem; use Flow\ETL\Tests\Double\KeepPaths; use Flow\ETL\Tests\FlowTestCase; use Flow\Filesystem\Local\NativeLocalFilesystem; +use Generator; +use PHPUnit\Framework\Attributes\DataProvider; use function array_map; use function array_sum; use function array_values; use function Flow\ETL\Adapter\Parquet\from_parquet; +use function Flow\ETL\Adapter\Parquet\to_parquet; use function Flow\ETL\DSL\data_frame; +use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\flow_context; +use function Flow\ETL\DSL\from_array; use function Flow\ETL\DSL\lit; +use function Flow\ETL\DSL\partition_by; use function Flow\ETL\DSL\ref; +use function Flow\ETL\DSL\schema; +use function Flow\ETL\DSL\str_schema; +use function Flow\Filesystem\DSL\memory_filesystem; use function Flow\Filesystem\DSL\path; use function gc_collect_cycles; use function iterator_to_array; @@ -132,4 +143,54 @@ public function test_a_partition_filter_excluding_the_first_file_closes_the_kept ); static::assertSame(2, $filesystem->closedStreams()); } + + public static function day_range_bounds(): Generator + { + yield 'datetime bounds' => [ + new DateTimeImmutable('2026-10-02 00:00:00 UTC'), + new DateTimeImmutable('2026-10-03 00:00:00 UTC'), + ]; + yield 'string bounds' => ['2026-10-02', '2026-10-03']; + } + + #[DataProvider('day_range_bounds')] + public function test_a_datetime_range_over_a_partition_column_the_file_holds_opens_only_the_matching_folders( + DateTimeImmutable|string $from, + DateTimeImmutable|string $to, + ): void { + $filesystem = new CountingFilesystem(memory_filesystem()); + + data_frame() + ->read(from_array( + [ + ['id' => 'a', 'date' => new DateTimeImmutable('2026-10-01 00:00:00 UTC')], + ['id' => 'b', 'date' => new DateTimeImmutable('2026-10-02 00:00:00 UTC')], + ['id' => 'c', 'date' => new DateTimeImmutable('2026-10-03 00:00:00 UTC')], + ['id' => 'd', 'date' => new DateTimeImmutable('2026-10-04 00:00:00 UTC')], + ], + schema(str_schema('id'), datetime_schema('date')), + )) + ->write( + to_parquet(path('memory://var/pruned/file.parquet'), filesystem: $filesystem)->partitionBy( + partition_by('date')->writeColumns(), + ), + ) + ->run(); + + $rows = data_frame() + ->read(from_parquet(path('memory://var/pruned/**/*.parquet'), filesystem: $filesystem)) + ->filter(ref('date')->between(lit($from), lit($to), Boundary::INCLUSIVE)) + ->fetch(); + + static::assertSame(['b', 'c'], $rows->reduceToArray('id')); + // the first listed file is opened once for the footer + static::assertSame( + [ + '/var/pruned/date=2026-10-01T00%3A00%3A00%2B00%3A00/file.parquet' => 1, + '/var/pruned/date=2026-10-02T00%3A00%3A00%2B00%3A00/file.parquet' => 1, + '/var/pruned/date=2026-10-03T00%3A00%3A00%2B00%3A00/file.parquet' => 1, + ], + ParquetFilesContext::opensPerFile($filesystem), + ); + } } 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 14bb23ac2..b3735eaeb 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 @@ -11,6 +11,7 @@ use Flow\ETL\Exception\InferredSchemaException; use Flow\ETL\Extractor\Signal; use Flow\ETL\Plan\Stage; +use Flow\ETL\Schema\Definition; use Flow\ETL\Tests\Context\ExtractedRows; use Flow\ETL\Tests\Double\CountingFilesystem; use Flow\ETL\Tests\Double\RecordingFilesystem; @@ -23,6 +24,8 @@ use Flow\Parquet\Reader; use Flow\Types\Value\Json; use Flow\Types\Value\Uuid; +use Generator; +use PHPUnit\Framework\Attributes\DataProvider; use function array_keys; use function count; @@ -31,13 +34,17 @@ use function Flow\ETL\DSL\analyze; use function Flow\ETL\DSL\array_to_rows; use function Flow\ETL\DSL\config; +use function Flow\ETL\DSL\date_schema; +use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\df; use function Flow\ETL\DSL\flow_context; +use function Flow\ETL\DSL\from_array; use function Flow\ETL\DSL\from_rows; use function Flow\ETL\DSL\from_sequence_number; use function Flow\ETL\DSL\int_schema; use function Flow\ETL\DSL\list_schema; use function Flow\ETL\DSL\map_schema; +use function Flow\ETL\DSL\partition_by; use function Flow\ETL\DSL\partition_types; use function Flow\ETL\DSL\schema; use function Flow\ETL\DSL\str_schema; @@ -562,4 +569,57 @@ public function test_one_extractor_read_twice_interleaved_gives_each_read_every_ count(array_keys($filesystem->calls, 'closeSource', true)), ); } + + public static function partition_columns_the_file_holds(): Generator + { + yield 'datetime' => [datetime_schema('date'), new DateTimeImmutable('2026-10-02 13:00:00 UTC')]; + yield 'date' => [date_schema('day'), new DateTimeImmutable('2026-10-02')]; + yield 'integer' => [int_schema('year'), 2026]; + } + + /** + * @param Definition $definition + */ + #[DataProvider('partition_columns_the_file_holds')] + public function test_partition_schema_types_a_partition_column_the_file_holds_like_the_read( + Definition $definition, + mixed $value, + ): void { + $memory = memory_filesystem(); + $name = $definition->entry()->name(); + + df() + ->read(from_array([['id' => 'a', $name => $value]], schema(str_schema('id'), $definition))) + ->write( + to_parquet(path('memory://var/typed/file.parquet'), filesystem: $memory)->partitionBy( + partition_by($name)->writeColumns(), + ), + ) + ->run(); + + $extractor = from_parquet(path('memory://var/typed/**/*.parquet'), filesystem: $memory); + + static::assertEquals($extractor->schema()->keep($name), $extractor->partitionSchema()); + } + + public function test_partition_schema_types_a_path_only_partition_column_as_a_string(): void + { + $memory = memory_filesystem(); + + df() + ->read(from_array( + [['id' => 'a', 'date' => new DateTimeImmutable('2026-10-02 13:00:00 UTC')]], + schema(str_schema('id'), datetime_schema('date')), + )) + ->write(to_parquet( + path('memory://var/path_only/file.parquet'), + filesystem: $memory, + )->partitionBy(partition_by('date'))) + ->run(); + + static::assertEquals( + schema(str_schema('date')), + from_parquet(path('memory://var/path_only/**/*.parquet'), filesystem: $memory)->partitionSchema(), + ); + } } diff --git a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetTest.php b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetTest.php index b7fcaae8c..46348ffe8 100644 --- a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetTest.php +++ b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetTest.php @@ -28,6 +28,7 @@ use Generator; use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\Attributes\RequiresPhpExtension; +use PHPUnit\Framework\Attributes\TestWith; use Ramsey\Uuid\Uuid; use function array_column; @@ -62,6 +63,7 @@ use function Flow\Filesystem\DSL\memory_filesystem; use function Flow\Filesystem\DSL\path; use function Flow\Types\DSL\type_datetime; +use function Flow\Types\DSL\type_float; use function Flow\Types\DSL\type_integer; use function Flow\Types\DSL\type_json; use function Flow\Types\DSL\type_list; @@ -768,6 +770,65 @@ public function test_write_columns_keeps_the_partition_column_in_the_file(?Parqu ); } + #[TestWith([true])] + #[TestWith([false])] + public function test_a_datetime_partition_reads_back_the_written_time(bool $writeColumns): void + { + $memory = memory_filesystem(); + $by = partition_by('date'); + + data_frame() + ->read(from_array( + [ + ['id' => 'a', 'date' => new DateTimeImmutable('2026-10-02 09:00:00 UTC')], + ['id' => 'b', 'date' => new DateTimeImmutable('2026-10-02 13:00:00.123456 UTC')], + ], + schema(str_schema('id'), datetime_schema('date')), + )) + ->write(to_parquet(path('memory://var/times/file.parquet'), filesystem: $memory)->partitionBy( + $writeColumns ? $by->writeColumns() : $by, + )) + ->run(); + + static::assertEquals( + [ + ['id' => 'a', 'date' => new DateTimeImmutable('2026-10-02 09:00:00 UTC')], + ['id' => 'b', 'date' => new DateTimeImmutable('2026-10-02 13:00:00.123456 UTC')], + ], + data_frame() + ->read(from_parquet(path('memory://var/times/**/*.parquet'), filesystem: $memory)->partitionTypes( + partition_types(date: type_datetime()), + )) + ->fetch() + ->toArray(), + ); + } + + public function test_a_float_partition_reads_back_the_written_value(): void + { + $memory = memory_filesystem(); + + data_frame() + ->read(from_array( + [['id' => 'a', 'x' => 0.0], ['id' => 'b', 'x' => 1.0E-9]], + schema(str_schema('id'), float_schema('x')), + )) + ->write(to_parquet(path('memory://var/floats/file.parquet'), filesystem: $memory)->partitionBy(partition_by( + 'x', + ))) + ->run(); + + static::assertSame( + [['id' => 'a', 'x' => 0.0], ['id' => 'b', 'x' => 1.0E-9]], + data_frame() + ->read(from_parquet(path('memory://var/floats/**/*.parquet'), filesystem: $memory)->partitionTypes( + partition_types(x: type_float()), + )) + ->fetch() + ->toArray(), + ); + } + public function test_xml_nodes_written_before_utf8_node_documents_read_back_and_load_as_their_text(): void { $memory = memory_filesystem(); diff --git a/src/adapter/etl-adapter-text/tests/Flow/ETL/Adapter/Text/Tests/Integration/TextExtractorTest.php b/src/adapter/etl-adapter-text/tests/Flow/ETL/Adapter/Text/Tests/Integration/TextExtractorTest.php index 3a26ef315..66513a4c6 100644 --- a/src/adapter/etl-adapter-text/tests/Flow/ETL/Adapter/Text/Tests/Integration/TextExtractorTest.php +++ b/src/adapter/etl-adapter-text/tests/Flow/ETL/Adapter/Text/Tests/Integration/TextExtractorTest.php @@ -4,6 +4,7 @@ namespace Flow\ETL\Adapter\Text\Tests\Integration; +use DateTimeImmutable; use Flow\ETL\Adapter\Text\TextExtractor; use Flow\ETL\Cardinality; use Flow\ETL\Exception\InvalidArgumentException; @@ -21,12 +22,17 @@ use function array_values; use function count; use function Flow\ETL\Adapter\Text\from_text; +use function Flow\ETL\Adapter\Text\to_text; use function Flow\ETL\DSL\config; use function Flow\ETL\DSL\config_builder; use function Flow\ETL\DSL\data_frame; +use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\flow_context; +use function Flow\ETL\DSL\from_array; +use function Flow\ETL\DSL\partition_by; use function Flow\ETL\DSL\schema; use function Flow\ETL\DSL\str_schema; +use function Flow\Filesystem\DSL\memory_filesystem; use function Flow\Filesystem\DSL\native_local_filesystem; use function Flow\Filesystem\DSL\path; use function Flow\Filesystem\DSL\path_real; @@ -247,4 +253,24 @@ public function test_one_extractor_read_twice_interleaved_gives_each_read_every_ count(array_keys($filesystem->calls, 'closeSource', true)), ); } + + public function test_partition_schema_is_the_partition_block_of_the_read_schema(): void + { + $memory = memory_filesystem(); + + data_frame() + ->read(from_array( + [ + ['date' => new DateTimeImmutable('2026-10-01 00:00:00 UTC'), 'text' => 'a'], + ['date' => new DateTimeImmutable('2026-10-02 00:00:00 UTC'), 'text' => 'b'], + ], + schema(datetime_schema('date'), str_schema('text')), + )) + ->write(to_text(path('memory://var/text/file.txt'), filesystem: $memory)->partitionBy(partition_by('date'))) + ->run(); + + $extractor = from_text(path('memory://var/text/**/*.txt'), filesystem: $memory); + + static::assertEquals($extractor->schema()->keep('date'), $extractor->partitionSchema()); + } } diff --git a/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/XMLParserExtractorTest.php b/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/XMLParserExtractorTest.php index 0efe23ef0..f61c618cf 100644 --- a/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/XMLParserExtractorTest.php +++ b/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/XMLParserExtractorTest.php @@ -4,6 +4,7 @@ namespace Flow\ETL\Adapter\XML\Tests\Integration; +use DateTimeImmutable; use Flow\ETL\Adapter\XML\XMLParserExtractor; use Flow\ETL\Cardinality; use Flow\ETL\Config; @@ -22,10 +23,14 @@ use function array_keys; use function file_get_contents; use function Flow\ETL\Adapter\XML\from_xml; +use function Flow\ETL\Adapter\XML\to_xml; use function Flow\ETL\DSL\config; use function Flow\ETL\DSL\config_builder; +use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\df; use function Flow\ETL\DSL\flow_context; +use function Flow\ETL\DSL\from_array; +use function Flow\ETL\DSL\partition_by; use function Flow\ETL\DSL\ref; use function Flow\ETL\DSL\schema; use function Flow\ETL\DSL\str_schema; @@ -517,4 +522,28 @@ public function test_extract_builds_through_the_config_backend(): void static::assertGreaterThanOrEqual(1, $backend->builders()); } + + public function test_partition_schema_is_the_partition_block_of_the_read_schema(): void + { + $memory = memory_filesystem(); + + df() + ->read(from_array( + [ + ['date' => new DateTimeImmutable('2026-10-01 00:00:00 UTC'), 'text' => 'a'], + ['date' => new DateTimeImmutable('2026-10-02 00:00:00 UTC'), 'text' => 'b'], + ], + schema(datetime_schema('date'), str_schema('text')), + )) + ->write( + to_xml(path('memory://var/xml/file.xml'), filesystem: $memory)->partitionBy( + partition_by('date')->writeColumns(), + ), + ) + ->run(); + + $extractor = from_xml(path('memory://var/xml/**/*.xml'), 'rows/row', filesystem: $memory); + + static::assertEquals($extractor->schema()->keep('date'), $extractor->partitionSchema()); + } } diff --git a/src/core/etl/src/Flow/ETL/Column/TextValues.php b/src/core/etl/src/Flow/ETL/Column/TextValues.php index 76138623d..10be62d68 100644 --- a/src/core/etl/src/Flow/ETL/Column/TextValues.php +++ b/src/core/etl/src/Flow/ETL/Column/TextValues.php @@ -29,6 +29,7 @@ use Flow\Types\Type\Native\FloatType; use Flow\Types\Type\Native\IntegerType; use Flow\Types\Type\NonFiniteFloat; +use Flow\Types\Type\RoundTripPrecision; use JsonException; use function array_keys; diff --git a/src/core/etl/src/Flow/ETL/Extractor/File/FileColumns.php b/src/core/etl/src/Flow/ETL/Extractor/File/FileColumns.php index 96f1aca71..ab4042b23 100644 --- a/src/core/etl/src/Flow/ETL/Extractor/File/FileColumns.php +++ b/src/core/etl/src/Flow/ETL/Extractor/File/FileColumns.php @@ -40,7 +40,8 @@ public function declare(Schema $schema): Schema } /** - * $declared carries a withSchema()d partition column's type; keep() drops the body columns it brings. + * $declared carries the type the read gives a partition column - a withSchema()d or a file footer type; + * keep() drops the body columns it brings. */ public function partitions(Schema $declared): Schema { diff --git a/src/core/etl/src/Flow/ETL/Function/Between.php b/src/core/etl/src/Flow/ETL/Function/Between.php index 29b09dbfd..89f9dd2e8 100644 --- a/src/core/etl/src/Flow/ETL/Function/Between.php +++ b/src/core/etl/src/Flow/ETL/Function/Between.php @@ -14,14 +14,16 @@ use Flow\ETL\Function\Evaluation\ResultColumn; use Flow\ETL\Rows; use Flow\Types\Type; +use Flow\Types\Type\Comparison\Operator; use Flow\Types\Type\Nullability; +use Flow\Types\Type\ValueComparator; use function array_map; use function Flow\ETL\DSL\lit; use function Flow\Types\DSL\type_bare; use function Flow\Types\DSL\type_boolean; -final class Between implements ScalarFunction +final class Between implements ComparisonFunction { use ScalarFunctionChain; @@ -42,6 +44,14 @@ public function __construct( $this->boundary = $boundary instanceof ScalarFunction ? $boundary : lit($boundary); } + /** + * @return list + */ + public function operands(): array + { + return [$this->value, $this->lowerBoundRef, $this->upperBoundRef]; + } + /** * @return list */ @@ -59,11 +69,25 @@ public function withChildren(array $children): static return new self($children[0], $children[1], $children[2], $children[3]); } + /** + * @param list $operands + */ + public function withOperands(array $operands): static + { + return new self($operands[0], $operands[1], $operands[2], $this->boundary); + } + /** * @return Type */ public function returns(): Type { + (new ValueComparator())->assertAllTypesComparable([ + $this->value->returns(), + $this->lowerBoundRef->returns(), + $this->upperBoundRef->returns(), + ], Operator::GREATER_THAN_EQUAL); + return (new Nullability())->any( type_boolean(), $this->value->returns(), diff --git a/src/core/etl/src/Flow/ETL/Function/ComparisonFunction.php b/src/core/etl/src/Flow/ETL/Function/ComparisonFunction.php new file mode 100644 index 000000000..83a5e0e47 --- /dev/null +++ b/src/core/etl/src/Flow/ETL/Function/ComparisonFunction.php @@ -0,0 +1,18 @@ + operands compared with each other, in children() order + */ + public function operands(): array; + + /** + * @param list $operands same order as operands() + */ + public function withOperands(array $operands): static; +} diff --git a/src/core/etl/src/Flow/ETL/Function/Equals.php b/src/core/etl/src/Flow/ETL/Function/Equals.php index 16f3ab93a..b2156dc60 100644 --- a/src/core/etl/src/Flow/ETL/Function/Equals.php +++ b/src/core/etl/src/Flow/ETL/Function/Equals.php @@ -22,7 +22,7 @@ use function Flow\Types\DSL\type_bare; use function Flow\Types\DSL\type_boolean; -final class Equals implements ScalarFunction +final class Equals implements ComparisonFunction { use ScalarFunctionChain; @@ -35,6 +35,14 @@ public function __construct(mixed $left, mixed $right) $this->right = $right instanceof ScalarFunction ? $right : lit($right); } + /** + * @return list + */ + public function operands(): array + { + return [$this->left, $this->right]; + } + /** * @return list */ @@ -52,6 +60,14 @@ public function withChildren(array $children): static return new self($children[0], $children[1]); } + /** + * @param list $operands + */ + public function withOperands(array $operands): static + { + return new self($operands[0], $operands[1]); + } + /** * @return Type */ diff --git a/src/core/etl/src/Flow/ETL/Function/GreaterThan.php b/src/core/etl/src/Flow/ETL/Function/GreaterThan.php index 14fd765b0..d87f2fd5c 100644 --- a/src/core/etl/src/Flow/ETL/Function/GreaterThan.php +++ b/src/core/etl/src/Flow/ETL/Function/GreaterThan.php @@ -21,7 +21,7 @@ use function Flow\Types\DSL\type_bare; use function Flow\Types\DSL\type_boolean; -final class GreaterThan implements ScalarFunction +final class GreaterThan implements ComparisonFunction { use ScalarFunctionChain; @@ -34,6 +34,14 @@ public function __construct(mixed $left, mixed $right) $this->right = $right instanceof ScalarFunction ? $right : lit($right); } + /** + * @return list + */ + public function operands(): array + { + return [$this->left, $this->right]; + } + /** * @return list */ @@ -51,6 +59,14 @@ public function withChildren(array $children): static return new self($children[0], $children[1]); } + /** + * @param list $operands + */ + public function withOperands(array $operands): static + { + return new self($operands[0], $operands[1]); + } + /** * @return Type */ diff --git a/src/core/etl/src/Flow/ETL/Function/GreaterThanEqual.php b/src/core/etl/src/Flow/ETL/Function/GreaterThanEqual.php index ef0a84a05..41d7f760d 100644 --- a/src/core/etl/src/Flow/ETL/Function/GreaterThanEqual.php +++ b/src/core/etl/src/Flow/ETL/Function/GreaterThanEqual.php @@ -21,7 +21,7 @@ use function Flow\Types\DSL\type_bare; use function Flow\Types\DSL\type_boolean; -final class GreaterThanEqual implements ScalarFunction +final class GreaterThanEqual implements ComparisonFunction { use ScalarFunctionChain; @@ -34,6 +34,14 @@ public function __construct(mixed $left, mixed $right) $this->right = $right instanceof ScalarFunction ? $right : lit($right); } + /** + * @return list + */ + public function operands(): array + { + return [$this->left, $this->right]; + } + /** * @return list */ @@ -51,6 +59,14 @@ public function withChildren(array $children): static return new self($children[0], $children[1]); } + /** + * @param list $operands + */ + public function withOperands(array $operands): static + { + return new self($operands[0], $operands[1]); + } + /** * @return Type */ diff --git a/src/core/etl/src/Flow/ETL/Function/IsIn.php b/src/core/etl/src/Flow/ETL/Function/IsIn.php index 02851f2e5..646f8f6be 100644 --- a/src/core/etl/src/Flow/ETL/Function/IsIn.php +++ b/src/core/etl/src/Flow/ETL/Function/IsIn.php @@ -25,7 +25,7 @@ use function get_debug_type; use function in_array; -final class IsIn implements ScalarFunction +final class IsIn implements ComparisonFunction { use ScalarFunctionChain; @@ -41,6 +41,14 @@ public function __construct(ScalarFunction|array $haystack, mixed $needle) $this->needle = $needle instanceof ScalarFunction ? $needle : lit($needle); } + /** + * @return list + */ + public function operands(): array + { + return [$this->haystack, $this->needle]; + } + /** * @return list */ @@ -58,6 +66,14 @@ public function withChildren(array $children): static return new self($children[0], $children[1]); } + /** + * @param list $operands + */ + public function withOperands(array $operands): static + { + return new self($operands[0], $operands[1]); + } + /** * @return Type */ diff --git a/src/core/etl/src/Flow/ETL/Function/LessThan.php b/src/core/etl/src/Flow/ETL/Function/LessThan.php index f8a228c61..c644f4c1c 100644 --- a/src/core/etl/src/Flow/ETL/Function/LessThan.php +++ b/src/core/etl/src/Flow/ETL/Function/LessThan.php @@ -21,7 +21,7 @@ use function Flow\Types\DSL\type_bare; use function Flow\Types\DSL\type_boolean; -final class LessThan implements ScalarFunction +final class LessThan implements ComparisonFunction { use ScalarFunctionChain; @@ -34,6 +34,14 @@ public function __construct(mixed $left, mixed $right) $this->right = $right instanceof ScalarFunction ? $right : lit($right); } + /** + * @return list + */ + public function operands(): array + { + return [$this->left, $this->right]; + } + /** * @return list */ @@ -51,6 +59,14 @@ public function withChildren(array $children): static return new self($children[0], $children[1]); } + /** + * @param list $operands + */ + public function withOperands(array $operands): static + { + return new self($operands[0], $operands[1]); + } + /** * @return Type */ diff --git a/src/core/etl/src/Flow/ETL/Function/LessThanEqual.php b/src/core/etl/src/Flow/ETL/Function/LessThanEqual.php index de7af8e23..c94d5de93 100644 --- a/src/core/etl/src/Flow/ETL/Function/LessThanEqual.php +++ b/src/core/etl/src/Flow/ETL/Function/LessThanEqual.php @@ -21,7 +21,7 @@ use function Flow\Types\DSL\type_bare; use function Flow\Types\DSL\type_boolean; -final class LessThanEqual implements ScalarFunction +final class LessThanEqual implements ComparisonFunction { use ScalarFunctionChain; @@ -34,6 +34,14 @@ public function __construct(mixed $left, mixed $right) $this->right = $right instanceof ScalarFunction ? $right : lit($right); } + /** + * @return list + */ + public function operands(): array + { + return [$this->left, $this->right]; + } + /** * @return list */ @@ -51,6 +59,14 @@ public function withChildren(array $children): static return new self($children[0], $children[1]); } + /** + * @param list $operands + */ + public function withOperands(array $operands): static + { + return new self($operands[0], $operands[1]); + } + /** * @return Type */ diff --git a/src/core/etl/src/Flow/ETL/Function/NotEquals.php b/src/core/etl/src/Flow/ETL/Function/NotEquals.php index 75b702ff3..99630aa2b 100644 --- a/src/core/etl/src/Flow/ETL/Function/NotEquals.php +++ b/src/core/etl/src/Flow/ETL/Function/NotEquals.php @@ -15,7 +15,7 @@ use function Flow\ETL\DSL\lit; use function Flow\Types\DSL\type_boolean; -final class NotEquals implements ScalarFunction +final class NotEquals implements ComparisonFunction { use ScalarFunctionChain; @@ -28,6 +28,14 @@ public function __construct(mixed $left, mixed $right) $this->right = $right instanceof ScalarFunction ? $right : lit($right); } + /** + * @return list + */ + public function operands(): array + { + return [$this->left, $this->right]; + } + /** * @return list */ @@ -45,6 +53,14 @@ public function withChildren(array $children): static return new self($children[0], $children[1]); } + /** + * @param list $operands + */ + public function withOperands(array $operands): static + { + return new self($operands[0], $operands[1]); + } + /** * @return Type */ diff --git a/src/core/etl/src/Flow/ETL/Function/ReferenceResolver.php b/src/core/etl/src/Flow/ETL/Function/ReferenceResolver.php index 31b10d0ff..1778b69aa 100644 --- a/src/core/etl/src/Flow/ETL/Function/ReferenceResolver.php +++ b/src/core/etl/src/Flow/ETL/Function/ReferenceResolver.php @@ -16,12 +16,12 @@ * its Type and nullability. Never mutates $function. * Total: a reference the schema does not know is left unresolved, not thrown on. * - * One pass, not a fixpoint - this resolver inserts nothing, so a post-order pass over a finite - * tree converges in one traversal by construction. The day a coercion pass inserts Cast nodes - * into this tree, the pass count must be re-decided. + * One pass, not a fixpoint - the only nodes it inserts are the Casts TemporalStringCoercion wraps around + * operands already resolved, and a Cast adds no reference, so a post-order pass over a finite tree + * converges in one traversal. * - * Sound: the only non-leaf return is $function->withChildren(), typed static; a root that is - * itself a reference leaf resolves to its Reference sibling class, never to a foreign node. + * Sound: the only non-leaf returns are withChildren() and withOperands(), both typed static; + * a root that is itself a reference leaf resolves to its Reference sibling class, never to a foreign node. * * @template T of FunctionTree * @@ -51,13 +51,11 @@ public function resolve(FunctionTree $function, Schema $schema): FunctionTree $resolved[] = $this->resolve($child, $schema); } - if ($resolved === $children) { - return $function; - } - // @mago-ignore analysis:less-specific-return-statement - withChildren(): static preserves // the node's class, which the analyzer cannot unify with T. - return $function->withChildren($resolved); + return (new TemporalStringCoercion())->coerce( + $resolved === $children ? $function : $function->withChildren($resolved), + ); } /** diff --git a/src/core/etl/src/Flow/ETL/Function/TemporalStringCoercion.php b/src/core/etl/src/Flow/ETL/Function/TemporalStringCoercion.php new file mode 100644 index 000000000..137c25e49 --- /dev/null +++ b/src/core/etl/src/Flow/ETL/Function/TemporalStringCoercion.php @@ -0,0 +1,89 @@ +operands(); + $types = []; + + try { + foreach ($operands as $operand) { + if (!$operand->resolved()) { + return $function; + } + + $types[] = $operand->returns(); + } + } catch (SchemaNotDerivableException) { + return $function; + } + + $temporal = null; + + foreach ($types as $type) { + $bare = type_bare($type); + $candidate = $bare instanceof ListType ? type_bare($bare->element()) : $bare; + + if ($candidate instanceof DateTimeType || $candidate instanceof DateType) { + $temporal = $candidate; + + break; + } + } + + if ($temporal === null) { + return $function; + } + + $coerced = false; + + foreach ($types as $i => $type) { + $bare = type_bare($type); + + if ($bare instanceof StringType) { + $operands[$i] = new Cast($operands[$i], $this->nullableAs($temporal, $type)); + $coerced = true; + } elseif ($bare instanceof ListType && type_bare($bare->element()) instanceof StringType) { + $operands[$i] = new Cast($operands[$i], $this->nullableAs( + type_list($this->nullableAs($temporal, $bare->element())), + $type, + )); + $coerced = true; + } + } + + return $coerced ? $function->withOperands($operands) : $function; + } + + /** + * @param Type $type + * @param Type $operand + * + * @return Type + */ + private function nullableAs(Type $type, Type $operand): Type + { + return type_is_nullable($operand) ? type_optional($type) : $type; + } +} diff --git a/src/core/etl/src/Flow/Floe/FloeExtractor.php b/src/core/etl/src/Flow/Floe/FloeExtractor.php index b612fe975..1aa43bd42 100644 --- a/src/core/etl/src/Flow/Floe/FloeExtractor.php +++ b/src/core/etl/src/Flow/Floe/FloeExtractor.php @@ -130,7 +130,7 @@ public function unionByName(bool $union = true): self public function partitionSchema(): Schema { - return $this->fileColumns($this->filesystem, $this->path)->partitions(new Schema()); + return $this->fileColumns($this->filesystem, $this->path)->partitions($this->schema()); } public function source(): Path diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/PartitioningTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/PartitioningTest.php index 421acac53..b62bd6a0d 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/PartitioningTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/PartitioningTest.php @@ -5,10 +5,10 @@ namespace Flow\ETL\Tests\Integration\DataFrame; use DateTimeImmutable; +use Flow\ETL\Exception\EvaluationException; use Flow\ETL\Extractor; use Flow\ETL\Rows; use Flow\ETL\Tests\FlowIntegrationTestCase; -use Flow\Types\Exception\InvalidArgumentException; use Generator; use PHPUnit\Framework\Attributes\DataProvider; @@ -100,12 +100,10 @@ public static function file_listings(): Generator yield 'files' => [files($glob)]; } - public function test_a_partition_filter_with_an_incomparable_literal_is_refused_at_bind(): void + public function test_a_partition_value_that_is_not_a_date_fails_a_date_filter(): void { - $this->expectException(InvalidArgumentException::class); - $this->expectExceptionMessage( - "Can't compare '(string == date)' due to data type mismatch - an explicit cast is required.", - ); + $this->expectException(EvaluationException::class); + $this->expectExceptionMessage('Cast function failed: Can\'t cast "string" into "date" type (row 0)'); df() ->read(from_text(__DIR__ diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/TemporalStringComparisonTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/TemporalStringComparisonTest.php new file mode 100644 index 000000000..57567b55a --- /dev/null +++ b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/TemporalStringComparisonTest.php @@ -0,0 +1,77 @@ +read(from_array( + [ + ['id' => 'a', 'ts' => new DateTimeImmutable('2026-10-01 00:00:00 UTC')], + ['id' => 'b', 'ts' => new DateTimeImmutable('2026-10-02 09:00:00 UTC')], + ['id' => 'c', 'ts' => new DateTimeImmutable('2026-10-03 00:00:00 UTC')], + ], + schema(str_schema('id'), datetime_schema('ts')), + )) + ->filter(ref('ts')->between(lit('2026-10-02'), lit('2026-10-03'), Boundary::INCLUSIVE)) + ->fetch() + ->reduceToArray('id'), + ); + } + + public function test_a_string_column_is_compared_as_a_datetime(): void + { + static::assertSame( + ['b', 'c'], + df() + ->read(from_array( + [ + ['id' => 'a', 'ts' => new DateTimeImmutable('2026-10-01 00:00:00 UTC'), 's' => null], + [ + 'id' => 'b', + 'ts' => new DateTimeImmutable('2026-10-02 09:00:00 UTC'), + 's' => '2026-10-02 09:00:00', + ], + ['id' => 'c', 'ts' => new DateTimeImmutable('2026-10-03 00:00:00 UTC'), 's' => '2026-10-03'], + ], + schema(str_schema('id'), datetime_schema('ts'), str_schema('s', nullable: true)), + )) + ->filter(ref('ts')->equals(ref('s'))) + ->fetch() + ->reduceToArray('id'), + ); + } + + public function test_a_string_that_is_not_a_date_fails_the_evaluation(): void + { + $this->expectException(EvaluationException::class); + $this->expectExceptionMessage('Cast function failed: Can\'t cast "string" into "datetime" type (row 0)'); + + df() + ->read(from_array( + [['id' => 'a', 'ts' => new DateTimeImmutable('2026-10-01 00:00:00 UTC')]], + schema(str_schema('id'), datetime_schema('ts')), + )) + ->filter(ref('ts')->greaterThan(lit('not a date'))) + ->fetch(); + } +} diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/BetweenTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/BetweenTest.php index 3bbcb9b6d..17cf7fa38 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/BetweenTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/BetweenTest.php @@ -4,10 +4,12 @@ namespace Flow\ETL\Tests\Unit\Function; +use DateTimeImmutable; use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Function\Between\Boundary; use Flow\ETL\Tests\Context\FunctionContext; use Flow\ETL\Tests\FlowTestCase; +use Flow\Types\Exception\InvalidArgumentException as TypesInvalidArgumentException; use function Flow\ETL\DSL\between; use function Flow\ETL\DSL\flow_context; @@ -150,4 +152,17 @@ public function test_between_with_invalid_boundary_in_strict_mode(): void schema(int_schema('value')), ); } + + public function test_between_refuses_incomparable_bounds_at_bind(): void + { + $this->expectException(TypesInvalidArgumentException::class); + $this->expectExceptionMessage( + "Can't compare '(date >= integer)' due to data type mismatch - an explicit cast is required.", + ); + + ref('a') + ->resolve(int_schema('a')) + ->between(lit(1), lit(new DateTimeImmutable('2024-01-01'))) + ->returns(); + } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/ReferenceResolverTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/ReferenceResolverTest.php index 18154c384..e73f86c95 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/ReferenceResolverTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/ReferenceResolverTest.php @@ -5,12 +5,17 @@ namespace Flow\ETL\Tests\Unit\Function; use Flow\ETL\Exception\SchemaDefinitionNotFoundException; +use Flow\ETL\Function\Between\Boundary; +use Flow\ETL\Function\Literal; use Flow\ETL\Function\ReferenceResolver; use Flow\ETL\Function\ScalarFunction; use Flow\ETL\Row\ResolvedReference; use Flow\ETL\Row\UnresolvedReference; use Flow\ETL\Tests\FlowTestCase; +use Generator; +use PHPUnit\Framework\Attributes\DataProvider; +use function Flow\ETL\DSL\date_schema; use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\int_schema; use function Flow\ETL\DSL\list_schema; @@ -24,6 +29,67 @@ final class ReferenceResolverTest extends FlowTestCase { + public static function temporal_comparisons_with_a_string(): Generator + { + yield 'equals' => [ref('ts')->equals(lit('2026-10-02')), 'boolean']; + yield 'not equals' => [ref('ts')->notEquals(lit('2026-10-02')), 'boolean']; + yield 'greater than' => [ref('ts')->greaterThan(lit('2026-10-02')), 'boolean']; + yield 'greater than equal' => [ref('ts')->greaterThanEqual(lit('2026-10-02')), 'boolean']; + yield 'less than' => [ref('ts')->lessThan(lit('2026-10-02')), 'boolean']; + yield 'less than equal' => [ref('ts')->lessThanEqual(lit('2026-10-02')), 'boolean']; + yield 'string on the left' => [lit('2026-10-02')->lessThan(ref('ts')), 'boolean']; + yield 'date column' => [ref('d')->equals(lit('2026-10-02')), 'boolean']; + yield 'string column' => [ref('ts')->equals(ref('s')), 'boolean']; + yield 'nullable string column' => [ref('ts')->equals(ref('n')), '?boolean']; + yield 'between' => [ref('ts')->between(lit('2026-10-02'), lit('2026-10-03'), Boundary::INCLUSIVE), 'boolean']; + yield 'is in' => [ref('ts')->isIn(lit(['2026-10-02'])), 'boolean']; + } + + #[DataProvider('temporal_comparisons_with_a_string')] + public function test_a_string_compared_with_a_temporal_operand_is_cast_to_it( + ScalarFunction $predicate, + string $expected, + ): void { + static::assertSame( + $expected, + (new ReferenceResolver()) + ->resolve($predicate, schema( + datetime_schema('ts'), + date_schema('d'), + str_schema('s'), + str_schema('n', nullable: true), + )) + ->returns() + ->toString(), + ); + } + + public function test_same_keeps_a_string_and_a_datetime_strict(): void + { + /** @var ScalarFunction $resolved */ + $resolved = (new ReferenceResolver())->resolve( + ref('ts')->same(lit('2026-10-02')), + schema(datetime_schema('ts')), + ); + + static::assertInstanceOf(Literal::class, $resolved->children()[1]); + } + + public function test_a_coerced_comparison_resolves_to_itself(): void + { + $schema = schema(datetime_schema('ts'), str_schema('s')); + $resolved = (new ReferenceResolver())->resolve(ref('ts')->equals(ref('s')), $schema); + + static::assertSame($resolved, (new ReferenceResolver())->resolve($resolved, $schema)); + } + + public function test_a_comparison_over_an_unknown_column_is_left_as_it_is(): void + { + $predicate = ref('missing')->equals(lit('2026-10-02')); + + static::assertSame($predicate, (new ReferenceResolver())->resolve($predicate, schema(datetime_schema('ts')))); + } + public function test_resolves_a_leaf_from_a_hand_built_schema(): void { /** @var ResolvedReference $resolved */ diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/TemporalStringCoercionTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/TemporalStringCoercionTest.php new file mode 100644 index 000000000..fde23b04c --- /dev/null +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/TemporalStringCoercionTest.php @@ -0,0 +1,82 @@ +resolve(datetime_schema('ts')); + + yield 'string literal on the right' => [$ts->equals(lit('2026-10-02')), ['datetime', 'datetime']]; + yield 'string literal on the left' => [lit('2026-10-02')->lessThan($ts), ['datetime', 'datetime']]; + yield 'date column' => [ref('d')->resolve(date_schema('d'))->equals(lit('2026-10-02')), ['date', 'date']]; + yield 'nullable string column' => [ + $ts->equals(ref('n')->resolve(str_schema('n', nullable: true))), + ['datetime', '?datetime'], + ]; + yield 'between bounds' => [ + $ts->between(lit('2026-10-02'), lit('2026-10-03'), Boundary::INCLUSIVE), + ['datetime', 'datetime', 'datetime'], + ]; + yield 'is in a list of strings' => [$ts->isIn(lit(['2026-10-02'])), ['list', 'datetime']]; + yield 'string is in a list of dates' => [ + lit('2026-10-02')->isIn(lit([new DateTimeImmutable('2026-10-02')])), + ['list', 'date'], + ]; + } + + public static function untouched(): Generator + { + $s = ref('s')->resolve(str_schema('s')); + + yield 'two strings' => [$s->equals(lit('2026-10-02'))]; + yield 'integer and string' => [ref('i')->resolve(int_schema('i'))->equals(lit('1'))]; + yield 'same stays strict' => [ref('ts')->resolve(datetime_schema('ts'))->same(lit('2026-10-02'))]; + yield 'unresolved operand' => [ref('ts')->equals(lit('2026-10-02'))]; + yield 'operand without a derivable schema' => [ + array_get($s, 'key')->equals(ref('ts')->resolve(datetime_schema('ts'))), + ]; + yield 'not a comparison' => [ref('ts')->resolve(datetime_schema('ts'))->isNull()]; + } + + /** + * @param list $expected + */ + #[DataProvider('coerced')] + public function test_a_string_operand_is_cast_to_the_temporal_operand( + ScalarFunction $predicate, + array $expected, + ): void { + static::assertSame($expected, array_map(static fn(ScalarFunction $operand): string => $operand + ->returns() + ->toString(), type_instance_of(ComparisonFunction::class)->assert((new TemporalStringCoercion())->coerce($predicate))->operands())); + } + + #[DataProvider('untouched')] + public function test_the_function_is_returned_as_it_is(ScalarFunction $predicate): void + { + static::assertSame($predicate, (new TemporalStringCoercion())->coerce($predicate)); + } +} diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Loader/File/PartitionRouterTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Loader/File/PartitionRouterTest.php index 2cd3624da..6b1d5aa2b 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Loader/File/PartitionRouterTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Loader/File/PartitionRouterTest.php @@ -127,6 +127,6 @@ public function test_equal_datetime_instants_route_to_one_partition_and_keep_the static::assertCount(1, $groups); static::assertSame([1, 2], $groups[0][1]->reduceToArray(ref('id'))); - static::assertSame('2026-01-02', $groups[0][0]->get('at')->value); + static::assertSame('2026-01-02T10:00:00+00:00', $groups[0][0]->get('at')->value); } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Loader/RowPartitionsTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Loader/RowPartitionsTest.php index 368ae9965..feeac0d9c 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Loader/RowPartitionsTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Loader/RowPartitionsTest.php @@ -36,13 +36,13 @@ public static function scalar_columns(): array { return [ 'int' => [int_schema('v'), 7, '7'], - 'float' => [float_schema('v'), 1.5, '1.500000'], + 'float' => [float_schema('v'), 1.5, '1.5'], 'string' => [str_schema('v'), 'eu', 'eu'], 'bool' => [bool_schema('v'), true, 'true'], - 'datetime truncates to a day' => [ + 'datetime' => [ datetime_schema('v'), new DateTimeImmutable('2024-01-01 21:30:00'), - '2024-01-01', + '2024-01-01T21:30:00+00:00', ], ]; } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Optimizer/Rule/PushFilterIntoSourceTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Optimizer/Rule/PushFilterIntoSourceTest.php index 5c0c37cb0..7c55ab75b 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Optimizer/Rule/PushFilterIntoSourceTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Optimizer/Rule/PushFilterIntoSourceTest.php @@ -5,13 +5,13 @@ namespace Flow\ETL\Tests\Unit\Optimizer\Rule; use DateTimeImmutable; +use Flow\ETL\Exception\EvaluationException; use Flow\ETL\Memory\ArrayMemory; use Flow\ETL\Optimizer; use Flow\ETL\Optimizer\Rule\PushFilterIntoSource; use Flow\ETL\Plan\Node\Filter; use Flow\ETL\Plan\Node\Read; use Flow\ETL\Plan\Stage; -use Flow\ETL\Planner; use Flow\ETL\Tests\Double\FixedRandomValueGenerator; use Flow\ETL\Tests\Double\RecordingExtractor; use Flow\ETL\Tests\Double\RecordingFileExtractor; @@ -21,7 +21,6 @@ use Flow\ETL\WithEntry; use Flow\Filesystem\Path\Filter\Filters; use Flow\Filesystem\Path\Filter\OnlyFiles; -use Flow\Types\Exception\InvalidArgumentException; use function Flow\ETL\DSL\array_to_rows; use function Flow\ETL\DSL\df; @@ -357,22 +356,15 @@ public function test_a_filter_inside_a_joined_frame_is_pushed_with_that_frames_c static::assertFalse($pathFilter->accept(PartitionedSourceMother::file('year=2024/month=07'))); } - public function test_an_incomparable_partition_predicate_fails_the_plan(): void + public function test_a_string_partition_compared_with_a_date_is_cast_and_a_non_date_value_fails(): void { - $this->expectException(InvalidArgumentException::class); - $this->expectExceptionMessage( - "Can't compare '(string == date)' due to data type mismatch - an explicit cast is required.", - ); + $this->expectException(EvaluationException::class); + $this->expectExceptionMessage('Cast function failed: Can\'t cast "string" into "date" type (row 0)'); - (new Planner(Optimizer::default()))->plan( - NodeMother::plan( - new Filter( - new Read(PartitionedSourceMother::yearMonth()), - ref('month')->equals(lit(new DateTimeImmutable('2024-01-01'))), - ), - ), - NodeMother::context(), - ); + df() + ->read(PartitionedSourceMother::yearMonth()) + ->filter(ref('month')->equals(lit(new DateTimeImmutable('2024-01-01')))) + ->fetch(); } public function test_the_extractor_instance_is_never_mutated(): void 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 382123a4a..0e6f80a78 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,7 @@ use DateTimeImmutable; use DateTimeZone; use Flow\ETL\Cardinality; +use Flow\ETL\Function\Between\Boundary; use Flow\ETL\Rows; use Flow\ETL\Tests\Context\ExtractedRows; use Flow\ETL\Tests\Double\CountingFilesystem; @@ -22,8 +23,11 @@ use function Flow\ETL\DSL\flow_context; use function Flow\ETL\DSL\from_array; use function Flow\ETL\DSL\int_schema; +use function Flow\ETL\DSL\lit; +use function Flow\ETL\DSL\partition_by; use function Flow\ETL\DSL\ref; use function Flow\ETL\DSL\schema; +use function Flow\ETL\DSL\str_schema; use function Flow\ETL\DSL\to_timezone; use function Flow\Filesystem\DSL\memory_filesystem; use function Flow\Filesystem\DSL\path; @@ -269,4 +273,57 @@ public function test_one_extractor_read_twice_interleaved_gives_each_read_every_ static::assertSame($first->toArray(), $second->toArray()); static::assertSame($filesystem->readFromCalls, $filesystem->closedStreams()); } + + public function test_partition_schema_types_a_partition_column_the_file_holds_like_the_read(): void + { + $memory = memory_filesystem(); + + df() + ->read(from_array( + [['id' => 'a', 'date' => new DateTimeImmutable('2026-10-02 13:00:00 UTC')]], + schema(str_schema('id'), datetime_schema('date')), + )) + ->write( + to_floe(path('memory://var/typed/file.floe'), filesystem: $memory)->partitionBy( + partition_by('date')->writeColumns(), + ), + ) + ->run(); + + $extractor = from_floe(path('memory://var/typed/**/*.floe'), filesystem: $memory); + + static::assertEquals($extractor->schema()->keep('date'), $extractor->partitionSchema()); + } + + public function test_a_datetime_range_over_a_partition_column_the_file_holds_returns_the_matching_rows(): void + { + $memory = memory_filesystem(); + + df() + ->read(from_array( + [ + ['id' => 'a', 'date' => new DateTimeImmutable('2026-10-01 00:00:00 UTC')], + ['id' => 'b', 'date' => new DateTimeImmutable('2026-10-02 00:00:00 UTC')], + ['id' => 'c', 'date' => new DateTimeImmutable('2026-10-03 00:00:00 UTC')], + ], + schema(str_schema('id'), datetime_schema('date')), + )) + ->write( + to_floe(path('memory://var/range/file.floe'), filesystem: $memory)->partitionBy( + partition_by('date')->writeColumns(), + ), + ) + ->run(); + + $rows = df() + ->read(from_floe(path('memory://var/range/**/*.floe'), filesystem: $memory)) + ->filter(ref('date')->between( + lit(new DateTimeImmutable('2026-10-02 00:00:00 UTC')), + lit(new DateTimeImmutable('2026-10-03 00:00:00 UTC')), + Boundary::INCLUSIVE, + )) + ->fetch(); + + static::assertSame(['b', 'c'], $rows->reduceToArray('id')); + } } diff --git a/src/lib/filesystem/src/Flow/Filesystem/Partition.php b/src/lib/filesystem/src/Flow/Filesystem/Partition.php index cdecf34f0..5c767c653 100644 --- a/src/lib/filesystem/src/Flow/Filesystem/Partition.php +++ b/src/lib/filesystem/src/Flow/Filesystem/Partition.php @@ -4,6 +4,7 @@ namespace Flow\Filesystem; +use DateTimeInterface; use Flow\Filesystem\Exception\InvalidArgumentException; use Flow\Types\Type; use Flow\Types\Type\Logical\DateTimeType; @@ -15,18 +16,25 @@ use Flow\Types\Type\Logical\StructureType; use Flow\Types\Type\Logical\XMLElementType; use Flow\Types\Type\Logical\XMLType; +use Flow\Types\Type\Native\FloatType; +use Flow\Types\Type\NonFiniteFloat; +use Flow\Types\Type\RoundTripPrecision; use Flow\Types\Type\TypedValueFormatter; use function array_filter; use function array_keys; use function explode; use function Flow\Types\DSL\type_string; +use function json_encode; use function preg_match; use function rawurldecode; use function rawurlencode; use function sprintf; use function strlen; +use const JSON_PRESERVE_ZERO_FRACTION; +use const JSON_THROW_ON_ERROR; + final class Partition { /** @@ -112,7 +120,26 @@ public static function fromUri(string $uri): Partitions public static function fromValue(string $name, Type $type, mixed $value): string { if ($type instanceof DateTimeType) { - return $type->cast($type->assert($value))->format('Y-m-d'); + $datetime = $type->cast($type->assert($value)); + + return $datetime->format($datetime->format('u') === '000000' ? DateTimeInterface::ATOM : 'Y-m-d\TH:i:s.uP'); + } + + if ($type instanceof FloatType) { + $float = $type->assert($value); + $precision = new RoundTripPrecision(); + $precision->force(); + + try { + return ( + NonFiniteFloat::text($float) ?? json_encode( + $float, + JSON_PRESERVE_ZERO_FRACTION | JSON_THROW_ON_ERROR, + ) + ); + } finally { + $precision->restore(); + } } if ( diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PartitionTest.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PartitionTest.php index c663a7acd..a53f72fb1 100644 --- a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PartitionTest.php +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PartitionTest.php @@ -7,13 +7,20 @@ use DateTimeImmutable; use Flow\Filesystem\Exception\InvalidArgumentException; use Flow\Filesystem\Partition; +use Flow\Types\Type; +use Generator; use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\Attributes\RequiresPhp; use PHPUnit\Framework\TestCase; +use function Flow\Types\DSL\type_date; use function Flow\Types\DSL\type_datetime; +use function Flow\Types\DSL\type_float; use function Flow\Types\DSL\type_html; +use function Flow\Types\DSL\type_integer; use function Flow\Types\DSL\type_xml; +use function ini_get; +use function ini_set; final class PartitionTest extends TestCase { @@ -43,9 +50,65 @@ public static function provider_reserved_characters_values(): array ]; } + public static function float_values(): Generator + { + yield 'shortest round trip' => [0.1 + 0.2, '0.30000000000000004']; + yield 'small' => [1.0E-9, '1.0e-9']; + yield 'zero fraction' => [1.0, '1.0']; + yield 'infinity' => [INF, 'INF']; + yield 'negative infinity' => [-INF, '-INF']; + yield 'not a number' => [NAN, 'NAN']; + } + + public static function round_trips(): Generator + { + yield 'datetime' => [type_datetime(), new DateTimeImmutable('2026-10-02 13:00:00 UTC')]; + yield 'datetime in a column zone' => [ + type_datetime('America/Los_Angeles'), + new DateTimeImmutable('2023-01-01 21:51:14 PST'), + ]; + yield 'datetime microseconds' => [type_datetime(), new DateTimeImmutable('2026-10-02 13:00:00.123456 UTC')]; + yield 'first 02:30 of a DST fall-back' => [ + type_datetime('Europe/Warsaw'), + new DateTimeImmutable('2026-10-25 00:30:00 UTC'), + ]; + yield 'second 02:30 of a DST fall-back' => [ + type_datetime('Europe/Warsaw'), + new DateTimeImmutable('2026-10-25 01:30:00 UTC'), + ]; + yield 'float' => [type_float(), 0.1 + 0.2]; + yield 'small float' => [type_float(), 1.0E-9]; + yield 'date' => [type_date(), new DateTimeImmutable('2026-10-02')]; + yield 'integer' => [type_integer(), -42]; + } + + /** + * @param Type $type + */ + #[DataProvider('round_trips')] + public function test_a_partition_value_casts_back_to_the_written_value(Type $type, mixed $value): void + { + $segment = (new Partition('c', Partition::fromValue('c', $type, $value)))->segment(); + + static::assertEquals($value, $type->cast(Partition::fromSegment($segment)?->value)); + } + + public function test_a_float_partition_value_ignores_a_lossy_serialize_precision(): void + { + $previous = ini_get('serialize_precision'); + ini_set('serialize_precision', '10'); + + try { + static::assertSame('0.30000000000000004', Partition::fromValue('x', type_float(), 0.1 + 0.2)); + static::assertSame('10', ini_get('serialize_precision')); + } finally { + ini_set('serialize_precision', (string) $previous); + } + } + public function test_creating_partition_value_from_date(): void { - static::assertEquals('2023-01-01', Partition::fromValue( + static::assertEquals('2023-01-01T00:00:00+00:00', Partition::fromValue( 'date', type_datetime(), new DateTimeImmutable('2023-01-01 00:00:00 UTC'), @@ -54,22 +117,37 @@ public function test_creating_partition_value_from_date(): void public function test_creating_partition_value_from_datetime(): void { - static::assertEquals('2023-01-02', Partition::fromValue( + static::assertEquals('2023-01-02T05:51:14+00:00', Partition::fromValue( 'date', type_datetime(), new DateTimeImmutable('2023-01-01 21:51:14 PST'), )); } - public function test_partition_value_is_the_date_in_the_column_zone(): void + public function test_partition_value_carries_the_column_zone_offset(): void { - static::assertEquals('2023-01-01', Partition::fromValue( + static::assertEquals('2023-01-01T21:51:14-08:00', Partition::fromValue( 'date', type_datetime('America/Los_Angeles'), new DateTimeImmutable('2023-01-01 21:51:14 PST'), )); } + public function test_partition_value_keeps_microseconds(): void + { + static::assertSame('2026-10-02T13:00:00.123456+00:00', Partition::fromValue( + 'date', + type_datetime(), + new DateTimeImmutable('2026-10-02 13:00:00.123456 UTC'), + )); + } + + #[DataProvider('float_values')] + public function test_creating_partition_value_from_float(float $value, string $expected): void + { + static::assertSame($expected, Partition::fromValue('x', type_float(), $value)); + } + #[RequiresPhp('>= 8.4.0')] public function test_creating_partition_value_from_html(): void { diff --git a/src/core/etl/src/Flow/ETL/Column/RoundTripPrecision.php b/src/lib/types/src/Flow/Types/Type/RoundTripPrecision.php similarity index 91% rename from src/core/etl/src/Flow/ETL/Column/RoundTripPrecision.php rename to src/lib/types/src/Flow/Types/Type/RoundTripPrecision.php index 7a38f4d35..4fbd1a88f 100644 --- a/src/core/etl/src/Flow/ETL/Column/RoundTripPrecision.php +++ b/src/lib/types/src/Flow/Types/Type/RoundTripPrecision.php @@ -2,9 +2,9 @@ declare(strict_types=1); -namespace Flow\ETL\Column; +namespace Flow\Types\Type; -use Flow\ETL\Exception\RuntimeException; +use Flow\Types\Exception\RuntimeException; use function ini_get; use function ini_set; diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Column/RoundTripPrecisionTest.php b/src/lib/types/tests/Flow/Types/Tests/Unit/Type/RoundTripPrecisionTest.php similarity index 96% rename from src/core/etl/tests/Flow/ETL/Tests/Unit/Column/RoundTripPrecisionTest.php rename to src/lib/types/tests/Flow/Types/Tests/Unit/Type/RoundTripPrecisionTest.php index c071cd1f2..80f4ef196 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Column/RoundTripPrecisionTest.php +++ b/src/lib/types/tests/Flow/Types/Tests/Unit/Type/RoundTripPrecisionTest.php @@ -2,9 +2,9 @@ declare(strict_types=1); -namespace Flow\ETL\Tests\Unit\Column; +namespace Flow\Types\Tests\Unit\Type; -use Flow\ETL\Column\RoundTripPrecision; +use Flow\Types\Type\RoundTripPrecision; use PHPUnit\Framework\TestCase; use function ini_get;