fix: partitioned reads keep their values and datetime filters work

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
+<?php
+
+use function Flow\ETL\Adapter\Parquet\to_parquet;
+use function Flow\ETL\DSL\{data_frame, from_array, overwrite, partition_by, ref};
+use function Flow\Types\DSL\type_date;
+
+data_frame()
+    ->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
 

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

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());
+    }
 }

src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelExtractorTest.php

@@ -29,16 +29,20 @@
 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;
 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\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());
+    }
 }

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;

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());
+    }
 }

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

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