feat(flow-php/etl): partition placeholders in read/write paths

- {column} placeholders in destination paths turn partitions into
file/directory names
- placeholders in extractor paths match like wildcards and recover
partition values, incl. pruning
- Partition name and value now forbid { and } (breaking, upgrade note
added)
- save modes (overwrite/append) applied to resolved placeholder paths

Author: norberttech

documentation/components/core/partitioning.md

@@ -92,6 +92,55 @@ $dataFrame
     ->run();
 ```
 
+## Partition Placeholders
+
+By default every partition becomes a `column=value` directory. With `{column}` placeholders in the destination path, selected partitions can become part of the file (or directory) name instead:
+
+```php
+<?php
+
+data_frame()
+    ->read(from_array([
+        ['date' => '2024-01-01', 'department' => 'sales', 'amount' => 100],
+        ['date' => '2024-01-01', 'department' => 'marketing', 'amount' => 200],
+    ]))
+    ->partitionBy('date', 'department')
+    ->write(to_parquet(__DIR__ . '/output/{department}.parquet'))
+    ->run();
+```
+
+**File structure:**
+```
+output/
+├── date=2024-01-01/
+│   ├── sales.parquet
+│   └── marketing.parquet
+```
+
+Partitions consumed by placeholders are removed from the `column=value` directory chain; all remaining partitions still become directories. Placeholders can appear in any path segment and can be combined, for example `to_csv(__DIR__ . '/output/{date}/{department}_report.csv')`.
+
+Rules:
+
+- Every placeholder must match a `partitionBy()` column, otherwise the write fails.
+- A destination path with placeholders requires partitioned rows - without `partitionBy()` the write fails.
+- Save modes behave exactly like with directory partitions, applied to the resolved file path.
+
+### Reading Data Partitioned with Placeholders
+
+Placeholders work in extractor paths too - matching files like a wildcard and re-attaching the partition value from the file name:
+
+```php
+<?php
+
+data_frame()
+    ->read(from_parquet(__DIR__ . '/output/date=*/{department}.parquet'))
+    ->filterPartitions(ref('department')->equals(lit('sales'))) // partition pruning works too
+    ->write(to_output())
+    ->run();
+```
+
+Keep in mind that a placeholder matches any file in that location, and the partition value is taken from the file name as-is. Files appended to an existing location get randomized suffixes (`sales_a1b2c3.parquet`), which become part of the recovered partition value.
+
 ## Performance Considerations
 
 ### Choosing Partition Columns

documentation/upgrading.md

@@ -258,6 +258,16 @@ Keep the registered `Telemetry` instance referenced for as long as it should be
 Custom `SchemaValidator` implementations must return a `Flow\ETL\Schema\Validator\ValidationContext`
 built from the missing, mismatched (`MismatchedDefinition`), and unexpected definitions they reject.
 
+### 18) `flow-php/filesystem` - `Partition` name and value forbid `{` and `}`
+
+| Before                           | After                             |
+|----------------------------------|-----------------------------------|
+| `new Partition('na{me', 'a}b')`  | throws `InvalidArgumentException` |
+| `partitionBy()` values with `{}` | throws `InvalidArgumentException` |
+
+`{name}` in a path is now a partition placeholder resolved from `partitionBy()` columns; strip braces from partition
+values before partitioning.
+
 ---
 
 ## Upgrading from 0.40.x to 0.41.x

src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVTest.php

@@ -11,6 +11,8 @@
 use function Flow\ETL\Adapter\CSV\from_csv;
 use function Flow\ETL\Adapter\CSV\to_csv;
 use function Flow\ETL\DSL\df;
+use function Flow\ETL\DSL\from_array;
+use function Flow\ETL\DSL\lit;
 use function Flow\ETL\DSL\overwrite;
 use function Flow\ETL\DSL\ref;
 use function mkdir;
@@ -41,4 +43,65 @@ public function test_loading_csv_files(): void
             unlink($path);
         }
     }
+
+    public function test_writing_and_reading_csv_files_with_partition_placeholders(): void
+    {
+        $dir = __DIR__ . '/var/test_writing_and_reading_csv_files_with_partition_placeholders';
+
+        df()
+            ->read(from_array([
+                ['year' => '2024', 'name' => '123456-PL', 'total' => 100],
+                ['year' => '2024', 'name' => '789-DE', 'total' => 200],
+                ['year' => '2025', 'name' => '555-FR', 'total' => 300],
+            ]))
+            ->partitionBy('year', 'name')
+            ->saveMode(overwrite())
+            ->load(to_csv($dir . '/{name}.csv'))
+            ->run();
+
+        static::assertFileExists($dir . '/year=2024/123456-PL.csv');
+        static::assertFileExists($dir . '/year=2024/789-DE.csv');
+        static::assertFileExists($dir . '/year=2025/555-FR.csv');
+
+        $rows = df()
+            ->read(from_csv($dir . '/year=*/{name}.csv'))
+            ->sortBy(ref('total'))
+            ->fetch();
+
+        static::assertCount(3, $rows);
+        static::assertEquals(['2024', '2024', '2025'], $rows->reduceToArray(ref('year')));
+        static::assertEquals(['123456-PL', '789-DE', '555-FR'], $rows->reduceToArray(ref('name')));
+
+        $prunedRows = df()
+            ->read(from_csv($dir . '/year=*/{name}.csv'))
+            ->filterPartitions(ref('name')->equals(lit('789-DE')))
+            ->fetch();
+
+        static::assertCount(1, $prunedRows);
+        static::assertEquals(['789-DE'], $prunedRows->reduceToArray(ref('name')));
+    }
+
+    /**
+     * https://github.com/flow-php/flow/issues/2238
+     */
+    public function test_writing_csv_files_with_last_partition_as_file_name(): void
+    {
+        $output = __DIR__ . '/var/test_writing_csv_files_with_last_partition_as_file_name/output';
+
+        df()
+            ->read(from_array([
+                ['order-year' => '2024', 'order-month' => '03', 'order-name' => '123456-PL', 'total' => 100],
+                ['order-year' => '2024', 'order-month' => '03', 'order-name' => '789-DE', 'total' => 200],
+                ['order-year' => '2025', 'order-month' => '01', 'order-name' => '555-FR', 'total' => 300],
+            ]))
+            ->partitionBy('order-year', 'order-month', 'order-name')
+            ->saveMode(overwrite())
+            ->load(to_csv($output . '/{order-name}.csv'))
+            ->run();
+
+        static::assertFileExists($output . '/order-year=2024/order-month=03/123456-PL.csv');
+        static::assertFileExists($output . '/order-year=2024/order-month=03/789-DE.csv');
+        static::assertFileExists($output . '/order-year=2025/order-month=01/555-FR.csv');
+        static::assertFileDoesNotExist($output . '/order-year=2024/order-month=03/order-name=123456-PL');
+    }
 }

src/core/etl/src/Flow/ETL/Extractor/PathPartitionsExtractor.php

@@ -8,6 +8,7 @@
 use Flow\ETL\FlowContext;
 use Flow\Filesystem\Partition;
 use Flow\Filesystem\Path;
+use Flow\Filesystem\Path\Filter\PlaceholderPartitions;
 use Generator;
 
 use function array_map;
@@ -34,8 +35,16 @@ public function __construct(
      */
     public function extract(FlowContext $context): Generator
     {
-        foreach ($context->filesystem($this->path)->list($this->path, $this->filter()) as $fileStatus) {
-            $partitions = $fileStatus->path->partitions();
+        $hasPlaceholders = [] !== $this->path->partitionPlaceholders();
+        $filter = $hasPlaceholders ? new PlaceholderPartitions($this->path, $this->filter()) : $this->filter();
+
+        foreach ($context->filesystem($this->path)->list($this->path, $filter) as $fileStatus) {
+            $partitions = $hasPlaceholders
+                ? $fileStatus
+                    ->path
+                    ->withPartitions($this->path->extractPlaceholderPartitions($fileStatus->path))
+                    ->partitions()
+                : $fileStatus->path->partitions();
 
             $row = row(
                 string_entry('path', $fileStatus->path->uri()),

src/core/etl/src/Flow/ETL/Filesystem/FilesystemStreams.php

@@ -12,6 +12,7 @@
 use Flow\Filesystem\Partition;
 use Flow\Filesystem\Path;
 use Flow\Filesystem\Path\Filter;
+use Flow\Filesystem\Path\Filter\PlaceholderPartitions;
 use Flow\Filesystem\SourceStream;
 use Flow\Filesystem\Stream\VoidStream;
 use Generator;
@@ -58,7 +59,7 @@ public function closeStreams(Path $path): void
                     }
 
                     if ($this->saveMode === SaveMode::Overwrite) {
-                        if ($fileStream->path()->partitions()->count()) {
+                        if ($fileStream->path()->partitions()->count() || [] !== $path->partitionPlaceholders()) {
                             $filename = str_replace(self::FLOW_TMP_FILE_PREFIX, '', $fileStream->path()->filename());
 
                             $partitionFilesPattern = path(
@@ -134,9 +135,18 @@ public function isOpen(Path $path, array $partitions = []): bool
     public function list(Path $path, Filter $pathFilter): Generator
     {
         $fs = $this->fstab->for($path);
+        $hasPlaceholders = [] !== $path->partitionPlaceholders();
+
+        if ($hasPlaceholders) {
+            $pathFilter = new PlaceholderPartitions($path, $pathFilter);
+        }
 
         foreach ($fs->list($path, $pathFilter) as $file) {
-            yield $fs->readFrom($file->path);
+            yield $fs->readFrom(
+                $hasPlaceholders
+                    ? $file->path->withPartitions($path->extractPlaceholderPartitions($file->path))
+                    : $file->path,
+            );
         }
     }
 
@@ -163,10 +173,18 @@ public function listOpenStreams(Path $path): Generator
      */
     public function read(Path $path, array $partitions = []): SourceStream
     {
-        if ($path->isPattern()) {
+        $placeholders = $path->partitionPlaceholders();
+
+        if ($path->isPattern() && [] === $placeholders) {
             throw new RuntimeException("Path can't be pattern, given: " . $path->uri());
         }
 
+        if ([] !== $placeholders && !count($partitions)) {
+            throw new RuntimeException(
+                'Path "' . $path->uri() . '" contains partition placeholders but no partitions were given',
+            );
+        }
+
         $destination = count($partitions) ? $path->addPartitions(...$partitions) : $path;
 
         return $this->fstab->for($path)->readFrom($destination);
@@ -203,8 +221,18 @@ public function writeTo(Path $path, array $partitions = []): DestinationStream
             throw new RuntimeException('Stream path must have an extension, given: ' . $path->uri());
         }
 
-        if ($path->isPattern()) {
-            throw new RuntimeException("Destination path can't be patter, given:" . $path->uri());
+        $placeholders = $path->partitionPlaceholders();
+
+        if ($path->isPattern() && [] === $placeholders) {
+            throw new RuntimeException("Destination path can't be pattern, given: " . $path->uri());
+        }
+
+        if ([] !== $placeholders && !count($partitions)) {
+            throw new RuntimeException(
+                'Destination path "'
+                . $path->uri()
+                . '" contains partition placeholders but rows are not partitioned, add partitionBy() to your pipeline',
+            );
         }
 
         $pathUri = $path->uri();

src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/Fixtures/Partitioning/placeholders/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/PartitioningTest.php

@@ -216,6 +216,47 @@ static function (int $i): array {
             });
     }
 
+    public function test_partitioning_by_path_placeholders_only(): void
+    {
+        $output = __DIR__ . '/Fixtures/Partitioning/placeholders';
+
+        df()
+            ->read(from_array([
+                ['order-year' => '2024', 'order-month' => '03', 'order-name' => '123456-PL', 'text' => 'order 1'],
+                ['order-year' => '2024', 'order-month' => '03', 'order-name' => '789-DE', 'text' => 'order 2'],
+                ['order-year' => '2025', 'order-month' => '01', 'order-name' => '555-FR', 'text' => 'order 3'],
+            ]))
+            ->partitionBy('order-year', 'order-month', 'order-name')
+            ->drop('order-year', 'order-month', 'order-name')
+            ->saveMode(overwrite())
+            ->write(to_text($output . '/{order-year}/{order-month}/{order-name}.txt'))
+            ->run();
+
+        static::assertFileExists($output . '/2024/03/123456-PL.txt');
+        static::assertFileExists($output . '/2024/03/789-DE.txt');
+        static::assertFileExists($output . '/2025/01/555-FR.txt');
+
+        df()->read(from_text($output
+        . '/{order-year}/{order-month}/{order-name}.txt'))->run(function (Rows $rows): void {
+            $this->assertSame(
+                ['order-year', 'order-month', 'order-name'],
+                array_map(static fn(Partition $p) => $p->name, $rows->partitions()->toArray()),
+            );
+        });
+
+        df()->read(from_text($output . '/**/*.txt'))->run(function (Rows $rows): void {
+            $this->assertFalse($rows->isPartitioned());
+        });
+
+        $prunedRows = df()
+            ->read(from_text($output . '/{order-year}/{order-month}/{order-name}.txt'))
+            ->filterPartitions(ref('order-month')->equals(lit('01')))
+            ->fetch();
+
+        static::assertCount(1, $prunedRows);
+        static::assertSame(['555-FR'], $prunedRows->reduceToArray('order-name'));
+    }
+
     public function test_pruning_multiple_partitions(): void
     {
         $rows = df()