Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 49 additions & 0 deletions documentation/components/core/partitioning.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
10 changes: 10 additions & 0 deletions documentation/upgrading.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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');
}
}
13 changes: 11 additions & 2 deletions src/core/etl/src/Flow/ETL/Extractor/PathPartitionsExtractor.php
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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()),
Expand Down
38 changes: 33 additions & 5 deletions src/core/etl/src/Flow/ETL/Filesystem/FilesystemStreams.php
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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,
);
}
}

Expand All @@ -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);
Expand Down Expand Up @@ -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();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
*
!.gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
Loading
Loading