diff --git a/documentation/components/core/partitioning.md b/documentation/components/core/partitioning.md index 583d6e3726..a5b2f03f14 100644 --- a/documentation/components/core/partitioning.md +++ b/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 +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 +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 diff --git a/documentation/upgrading.md b/documentation/upgrading.md index 0fdeeed2f8..25d0db31f2 100644 --- a/documentation/upgrading.md +++ b/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 diff --git a/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVTest.php b/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVTest.php index 0a9dba9656..95f75241a3 100644 --- a/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVTest.php +++ b/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'); + } } diff --git a/src/core/etl/src/Flow/ETL/Extractor/PathPartitionsExtractor.php b/src/core/etl/src/Flow/ETL/Extractor/PathPartitionsExtractor.php index 86a0ef5571..f9bf6937a0 100644 --- a/src/core/etl/src/Flow/ETL/Extractor/PathPartitionsExtractor.php +++ b/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()), diff --git a/src/core/etl/src/Flow/ETL/Filesystem/FilesystemStreams.php b/src/core/etl/src/Flow/ETL/Filesystem/FilesystemStreams.php index aa70f7a1aa..12e722b93b 100644 --- a/src/core/etl/src/Flow/ETL/Filesystem/FilesystemStreams.php +++ b/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(); diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/Fixtures/Partitioning/placeholders/.gitignore b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/Fixtures/Partitioning/placeholders/.gitignore new file mode 100644 index 0000000000..d6b7ef32c8 --- /dev/null +++ b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/Fixtures/Partitioning/placeholders/.gitignore @@ -0,0 +1,2 @@ +* +!.gitignore 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 eb4e36e5c0..1d925d1e35 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 @@ -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() diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/Filesystem/FilesystemStreams/Partitioned/PartitionPlaceholdersTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/Filesystem/FilesystemStreams/Partitioned/PartitionPlaceholdersTest.php new file mode 100644 index 0000000000..0bead535b1 --- /dev/null +++ b/src/core/etl/tests/Flow/ETL/Tests/Integration/Filesystem/FilesystemStreams/Partitioned/PartitionPlaceholdersTest.php @@ -0,0 +1,160 @@ +cleanFiles(); + } + + public function test_append_mode_randomizes_existing_placeholder_file(): void + { + $streams = new FilesystemStreams($this->fstab()); + $streams->setMode(append()); + + $this->setupFiles([ + __FUNCTION__ => [ + '123456-PL.csv' => 'existing content', + ], + ]); + $file = $this->getPath(__FUNCTION__ . '/{order-name}.csv'); + + $fileStream = $streams->writeTo($file, partitions: [new Partition('order-name', '123456-PL')]); + $fileStream->append('new content'); + $streams->closeStreams($file); + + $files = iterator_to_array($this->fs()->list(path($this->getPath(__FUNCTION__)->path() . '/*.csv'))); + + static::assertCount(2, $files); + + foreach ($files as $streamFile) { + static::assertStringStartsWith('123456-PL', $streamFile->path->basename()); + static::assertStringEndsWith('.csv', $streamFile->path->basename()); + } + } + + public function test_list_attaches_partitions_from_placeholders(): void + { + $streams = new FilesystemStreams($this->fstab()); + + $this->setupFiles([ + __FUNCTION__ => [ + 'order-year=2024' => [ + '123456-PL.csv' => 'file content', + ], + ], + ]); + + $streamsList = iterator_to_array($streams->list( + $this->getPath(__FUNCTION__ . '/order-year=*/{order-name}.csv'), + new OnlyFiles(), + )); + + static::assertCount(1, $streamsList); + + $partitions = $streamsList[0]->path()->partitions(); + + static::assertCount(2, $partitions); + static::assertSame('2024', $partitions->get('order-year')->value); + static::assertSame('123456-PL', $partitions->get('order-name')->value); + } + + public function test_overwrite_mode_replaces_existing_placeholder_file(): void + { + $streams = new FilesystemStreams($this->fstab()); + $streams->setMode(overwrite()); + + $this->setupFiles([ + __FUNCTION__ => [ + '123456-PL.csv' => 'existing content', + ], + ]); + $file = $this->getPath(__FUNCTION__ . '/{order-name}.csv'); + + $fileStream = $streams->writeTo($file, partitions: [new Partition('order-name', '123456-PL')]); + $fileStream->append('new content'); + $streams->closeStreams($file); + + $files = iterator_to_array($this->fs()->list(path($this->getPath(__FUNCTION__)->path() . '/*.csv'))); + + static::assertCount(1, $files); + static::assertSame('123456-PL.csv', $files[0]->path->basename()); + static::assertSame('new content', file_get_contents($files[0]->path->path())); + } + + public function test_write_to_placeholder_path_consuming_all_partitions(): void + { + $streams = new FilesystemStreams($this->fstab()); + + $this->setupFiles([__FUNCTION__ => []]); + $file = $this->getPath(__FUNCTION__ . '/{order-name}.csv'); + + $fileStream = $streams->writeTo($file, partitions: [new Partition('order-name', '123456-PL')]); + $fileStream->append('file content'); + $streams->closeStreams($file); + + $files = iterator_to_array($this->fs()->list(path($this->getPath(__FUNCTION__)->path() . '/*.csv'))); + + static::assertCount(1, $files); + static::assertSame('123456-PL.csv', $files[0]->path->basename()); + static::assertSame('file content', file_get_contents($files[0]->path->path())); + } + + public function test_write_to_placeholder_path_with_remaining_partitions(): void + { + $streams = new FilesystemStreams($this->fstab()); + + $this->setupFiles([__FUNCTION__ => []]); + $file = $this->getPath(__FUNCTION__ . '/{order-name}.csv'); + + $fileStream = $streams->writeTo($file, partitions: [ + new Partition('order-year', '2024'), + new Partition('order-name', '123456-PL'), + ]); + $fileStream->append('file content'); + $streams->closeStreams($file); + + $files = iterator_to_array($this->fs()->list(path($this->getPath(__FUNCTION__)->path() . '/**/*.csv'))); + + static::assertCount(1, $files); + static::assertSame('123456-PL.csv', $files[0]->path->basename()); + static::assertStringContainsString('order-year=2024', $files[0]->path->path()); + } + + public function test_write_to_placeholder_path_without_partitions(): void + { + $streams = new FilesystemStreams($this->fstab()); + + $this->setupFiles([__FUNCTION__ => []]); + $file = $this->getPath(__FUNCTION__ . '/{order-name}.csv'); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('contains partition placeholders but rows are not partitioned'); + + $streams->writeTo($file); + } + + protected function streams(): FilesystemStreams + { + return new FilesystemStreams($this->fstab()); + } +} diff --git a/src/lib/filesystem/src/Flow/Filesystem/Local/NativeLocalFilesystem.php b/src/lib/filesystem/src/Flow/Filesystem/Local/NativeLocalFilesystem.php index 40f4dedcde..b2f3039874 100644 --- a/src/lib/filesystem/src/Flow/Filesystem/Local/NativeLocalFilesystem.php +++ b/src/lib/filesystem/src/Flow/Filesystem/Local/NativeLocalFilesystem.php @@ -103,7 +103,7 @@ public function list(Path $path, Filter $pathFilter = new OnlyFiles()): Generato return; } - foreach (new GlobIterator($path->path()) as $filePath) { + foreach (new GlobIterator($path->glob()) as $filePath) { $filePath = type_string()->assert($filePath); $status = self::statFor(path_real($filePath, $path->options()), $filePath); @@ -182,7 +182,7 @@ public function rm(Path $path): bool $deletedCount = 0; - foreach ($this->matchChildFirst($path->path()) as $filePath) { + foreach ($this->matchChildFirst($path->glob()) as $filePath) { $filePath = type_string()->assert($filePath); if (is_dir($filePath)) { @@ -211,7 +211,7 @@ public function status(Path $path): ?FileStatus return self::statFor($path, $path->path()); } - foreach (new GlobIterator($path->path()) as $filePath) { + foreach (new GlobIterator($path->glob()) as $filePath) { $filePath = type_string()->assert($filePath); if (file_exists($filePath)) { diff --git a/src/lib/filesystem/src/Flow/Filesystem/Partition.php b/src/lib/filesystem/src/Flow/Filesystem/Partition.php index 197f6411e2..b5449b1f97 100644 --- a/src/lib/filesystem/src/Flow/Filesystem/Partition.php +++ b/src/lib/filesystem/src/Flow/Filesystem/Partition.php @@ -29,7 +29,7 @@ final class Partition /** * @var array */ - private static array $forbiddenCharacters = ['/', '\\', '=', ':', '>', '<', '|', '"', '?', '*']; + private static array $forbiddenCharacters = ['/', '\\', '=', ':', '>', '<', '|', '"', '?', '*', '{', '}']; public function __construct( public readonly string $name, @@ -43,7 +43,7 @@ public function __construct( throw new InvalidArgumentException("Partition value can't be empty"); } - $regex = '/^([^\/\\\=:><|"?*]+)$/'; + $regex = '/^([^\/\\\=:><|"?*{}]+)$/'; if (!preg_match($regex, $this->name)) { throw new InvalidArgumentException( @@ -80,7 +80,7 @@ public static function fromArray(array $data): array public static function fromUri(string $uri): Partitions { - $regex = '/^([^\/\\\=:><|"?*]+)=([^\/\\\=:><|"?*]+)$/'; + $regex = '/^([^\/\\\=:><|"?*{}]+)=([^\/\\\=:><|"?*{}]+)$/'; $partitions = []; $matches = []; diff --git a/src/lib/filesystem/src/Flow/Filesystem/Path.php b/src/lib/filesystem/src/Flow/Filesystem/Path.php index 466ae0a3d2..358a6e0f1a 100644 --- a/src/lib/filesystem/src/Flow/Filesystem/Path.php +++ b/src/lib/filesystem/src/Flow/Filesystem/Path.php @@ -21,6 +21,7 @@ { public function __construct( private WindowsPath|UnixPath $implementation, + private Partitions $partitions = new Partitions(), ) {} /** @@ -75,6 +76,18 @@ public function extension(): string|false return $this->implementation->extension(); } + /** + * Extracts partitions from a concrete path by matching it against partition placeholders in this path. + */ + public function extractPlaceholderPartitions(self $path): Partitions + { + if ($this->implementation instanceof UnixPath) { + return $this->implementation->extractPlaceholderPartitions(type_instance_of(UnixPath::class)->assert($path->implementation)); + } + + return $this->implementation->extractPlaceholderPartitions(type_instance_of(WindowsPath::class)->assert($path->implementation)); + } + public function filename(): string { return $this->implementation->filename(); @@ -87,6 +100,14 @@ public function getOption( return $this->implementation->options()->get($option, $default); } + /** + * Path with partition placeholders replaced by glob wildcards, suitable for glob-based listing. + */ + public function glob(): string + { + return $this->implementation->glob(); + } + public function hasOption(string|Option $option): bool { return $this->implementation->options()->has($option); @@ -133,9 +154,21 @@ public function parentDirectory(): self return new self($this->implementation->parentDirectory()); } + /** + * @return array + */ + public function partitionPlaceholders(): array + { + return $this->implementation->partitionPlaceholders(); + } + public function partitions(): Partitions { - return $this->implementation->partitions(); + if (!$this->partitions->count()) { + return $this->implementation->partitions(); + } + + return new Partitions(...$this->implementation->partitions()->toArray(), ...$this->partitions->toArray()); } /** @@ -214,4 +247,13 @@ public function uri(): string { return $this->implementation->uri(); } + + /** + * Attach explicit partitions to the path, merged with partitions parsed from the path itself. + * Used to carry partitions extracted from partition placeholders on concrete, listed paths. + */ + public function withPartitions(Partitions $partitions): self + { + return new self($this->implementation, $partitions); + } } diff --git a/src/lib/filesystem/src/Flow/Filesystem/Path/Filter/PlaceholderPartitions.php b/src/lib/filesystem/src/Flow/Filesystem/Path/Filter/PlaceholderPartitions.php new file mode 100644 index 0000000000..2e0d7b836e --- /dev/null +++ b/src/lib/filesystem/src/Flow/Filesystem/Path/Filter/PlaceholderPartitions.php @@ -0,0 +1,40 @@ +pattern->extractPlaceholderPartitions($status->path); + + if (!$partitions->count()) { + return $this->filter->accept($status); + } + + return $this->filter->accept( + new FileStatus( + $status->path->withPartitions($partitions), + $status->isFile(), + $status->size, + $status->lastModifiedAt, + ), + ); + } +} diff --git a/src/lib/filesystem/src/Flow/Filesystem/Path/UnixPath.php b/src/lib/filesystem/src/Flow/Filesystem/Path/UnixPath.php index d18f250607..0c1b296df9 100644 --- a/src/lib/filesystem/src/Flow/Filesystem/Path/UnixPath.php +++ b/src/lib/filesystem/src/Flow/Filesystem/Path/UnixPath.php @@ -4,6 +4,7 @@ namespace Flow\Filesystem\Path; +use Flow\ETL\Exception\InvalidArgumentException as EtlInvalidArgumentException; use Flow\Filesystem\Exception\InvalidArgumentException; use Flow\Filesystem\Exception\RuntimeException; use Flow\Filesystem\Partition; @@ -14,6 +15,8 @@ use function array_map; use function array_pop; use function array_slice; +use function array_unique; +use function array_values; use function count; use function explode; use function function_exists; @@ -29,6 +32,7 @@ use function posix_getpwuid; use function posix_getuid; use function preg_match; +use function preg_match_all; use function preg_quote; use function preg_replace; use function random_int; @@ -45,6 +49,10 @@ final readonly class UnixPath { + private const string PARTITION_PLACEHOLDER_PATTERN = '/\{([^\/\\\\{}*?\[\]]+)\}/'; + + private const string PLACEHOLDER_SENTINEL = "\x01"; + private Options $options; private string $path; @@ -147,16 +155,49 @@ public static function realpath(string $path, array|Options $options = []): self public function addPartitions(Partition $partition, Partition ...$partitions): self { - if ($this->isPattern()) { + if ($this->isPathPattern(preg_replace(self::PARTITION_PLACEHOLDER_PATTERN, '', $this->path) ?? $this->path)) { throw new InvalidArgumentException("Can't add partitions to path pattern."); } - $pathInfo = pathinfo($this->path); + $partitions = [$partition, ...$partitions]; + $partitionNames = array_map(static fn(Partition $p) => $p->name, $partitions); + $path = $this->path; + + foreach ($this->partitionPlaceholders() as $placeholder) { + $placeholderPartition = null; + + foreach ($partitions as $index => $nextPartition) { + if ($nextPartition->name === $placeholder) { + $placeholderPartition = $nextPartition; + unset($partitions[$index]); + + break; + } + } + + if ($placeholderPartition === null) { + throw new InvalidArgumentException( + 'Path partition placeholder {' + . $placeholder + . "} does not match any partition, available partitions: '" + . implode("', '", $partitionNames) + . "'", + ); + } + + $path = str_replace('{' . $placeholder . '}', $placeholderPartition->value, $path); + } + + if (count($partitions) === 0) { + return new self($this->protocol . '://' . $path, $this->options); + } + + $pathInfo = pathinfo($path); $dirname = $pathInfo['dirname'] ?? ''; $basename = $pathInfo['basename']; $partitionsString = implode('/', array_map( static fn(Partition $p) => $p->name . '=' . $p->value, - [$partition, ...$partitions], + array_values($partitions), )); return match ($dirname) { @@ -198,11 +239,72 @@ public function extension(): string|false return ($extension = pathinfo($this->path, PATHINFO_EXTENSION)) === '' ? false : strtolower($extension); } + /** + * Extracts partitions from a concrete path by matching it against partition placeholders in this path. + * Values that would not be valid partitions (e.g. produced by files not written through partitioning) are skipped. + */ + public function extractPlaceholderPartitions(self $path): Partitions + { + $matches = []; + preg_match_all(self::PARTITION_PLACEHOLDER_PATTERN, $this->path, $matches); + $names = array_values($matches[1]); + + if ([] === $names || $path->isPattern()) { + return new Partitions(); + } + + $rx = preg_quote( + preg_replace(self::PARTITION_PLACEHOLDER_PATTERN, self::PLACEHOLDER_SENTINEL, $this->path) ?? $this->path, + null, + ); + $rx = str_replace('\\*\\*', '(?:.*)?', $rx); + $rx = str_replace('\\*', '[^/]*', $rx); + $rx = strtr($rx, ['\\?' => '[^/]', '\\[' => '[', '\\]' => ']']); + $rx = str_replace(self::PLACEHOLDER_SENTINEL, '([^/]+)', $rx); + + $valueMatches = []; + + if (!preg_match('{^' . $rx . '$}', $path->path, $valueMatches)) { + return new Partitions(); + } + + $values = []; + + foreach ($names as $index => $name) { + $value = $valueMatches[$index + 1]; + + if (array_key_exists($name, $values) && $values[$name] !== $value) { + return new Partitions(); + } + + $values[$name] = $value; + } + + $partitionsList = []; + + foreach ($values as $name => $value) { + try { + $partitionsList[] = new Partition($name, $value); + } catch (EtlInvalidArgumentException) { + } + } + + return new Partitions(...$partitionsList); + } + public function filename(): string { return pathinfo($this->path, PATHINFO_FILENAME); } + /** + * Path with partition placeholders replaced by glob wildcards, suitable for glob-based listing. + */ + public function glob(): string + { + return preg_replace(self::PARTITION_PLACEHOLDER_PATTERN, '*', $this->path) ?? $this->path; + } + public function isEqual(self $path): bool { return $this->path === $path->path; @@ -245,6 +347,17 @@ public function parentDirectory(): self }; } + /** + * @return array + */ + public function partitionPlaceholders(): array + { + $matches = []; + preg_match_all(self::PARTITION_PLACEHOLDER_PATTERN, $this->path, $matches); + + return array_values(array_unique($matches[1])); + } + public function partitions(): Partitions { if ($this->isPattern()) { @@ -418,10 +531,14 @@ private function fnmatch(string $pattern, string $filename, int $flags = 0): boo } } - $rx = preg_quote($pattern, null); + $rx = preg_quote( + preg_replace(self::PARTITION_PLACEHOLDER_PATTERN, self::PLACEHOLDER_SENTINEL, $pattern) ?? $pattern, + null, + ); $rx = str_replace('\\*\\*', '(.*)?', $rx); $rx = str_replace('\\*', '[^/]*', $rx); $rx = strtr($rx, ['\\?' => '[^/]', '\\[' => '[', '\\]' => ']']); + $rx = str_replace(self::PLACEHOLDER_SENTINEL, '[^/]+', $rx); $rx = '{^' . $rx . '$}' . ($flags & 16 ? 'i' : ''); return (bool) preg_match($rx, $filename); diff --git a/src/lib/filesystem/src/Flow/Filesystem/Path/WindowsPath.php b/src/lib/filesystem/src/Flow/Filesystem/Path/WindowsPath.php index e97b102cd5..0fc6f35a3a 100644 --- a/src/lib/filesystem/src/Flow/Filesystem/Path/WindowsPath.php +++ b/src/lib/filesystem/src/Flow/Filesystem/Path/WindowsPath.php @@ -4,6 +4,7 @@ namespace Flow\Filesystem\Path; +use Flow\ETL\Exception\InvalidArgumentException as EtlInvalidArgumentException; use Flow\Filesystem\Exception\InvalidArgumentException; use Flow\Filesystem\Exception\RuntimeException; use Flow\Filesystem\Partition; @@ -14,6 +15,8 @@ use function array_map; use function array_pop; use function array_slice; +use function array_unique; +use function array_values; use function count; use function explode; use function Flow\Types\DSL\type_string; @@ -26,6 +29,7 @@ use function parse_url; use function pathinfo; use function preg_match; +use function preg_match_all; use function preg_quote; use function preg_replace; use function random_int; @@ -42,6 +46,10 @@ final readonly class WindowsPath { + private const string PARTITION_PLACEHOLDER_PATTERN = '/\{([^\/\\\\{}*?\[\]]+)\}/'; + + private const string PLACEHOLDER_SENTINEL = "\x01"; + private Options $options; private string $path; @@ -139,16 +147,49 @@ public static function realpath(string $path, array|Options $options = []): self public function addPartitions(Partition $partition, Partition ...$partitions): self { - if ($this->isPattern()) { + if ($this->isPathPattern(preg_replace(self::PARTITION_PLACEHOLDER_PATTERN, '', $this->path) ?? $this->path)) { throw new InvalidArgumentException("Can't add partitions to path pattern."); } - $pathInfo = pathinfo($this->path); + $partitions = [$partition, ...$partitions]; + $partitionNames = array_map(static fn(Partition $p) => $p->name, $partitions); + $path = $this->path; + + foreach ($this->partitionPlaceholders() as $placeholder) { + $placeholderPartition = null; + + foreach ($partitions as $index => $nextPartition) { + if ($nextPartition->name === $placeholder) { + $placeholderPartition = $nextPartition; + unset($partitions[$index]); + + break; + } + } + + if ($placeholderPartition === null) { + throw new InvalidArgumentException( + 'Path partition placeholder {' + . $placeholder + . "} does not match any partition, available partitions: '" + . implode("', '", $partitionNames) + . "'", + ); + } + + $path = str_replace('{' . $placeholder . '}', $placeholderPartition->value, $path); + } + + if (count($partitions) === 0) { + return new self($this->protocol . '://' . $path, $this->options); + } + + $pathInfo = pathinfo($path); $dirname = $pathInfo['dirname'] ?? ''; $basename = $pathInfo['basename']; $partitionsString = implode('/', array_map( static fn(Partition $p) => $p->name . '=' . $p->value, - [$partition, ...$partitions], + array_values($partitions), )); return match ($dirname) { @@ -198,11 +239,72 @@ public function extension(): string|false return ($extension = pathinfo($this->path, PATHINFO_EXTENSION)) === '' ? false : strtolower($extension); } + /** + * Extracts partitions from a concrete path by matching it against partition placeholders in this path. + * Values that would not be valid partitions (e.g. produced by files not written through partitioning) are skipped. + */ + public function extractPlaceholderPartitions(self $path): Partitions + { + $matches = []; + preg_match_all(self::PARTITION_PLACEHOLDER_PATTERN, $this->path, $matches); + $names = array_values($matches[1]); + + if ([] === $names || $path->isPattern()) { + return new Partitions(); + } + + $rx = preg_quote( + preg_replace(self::PARTITION_PLACEHOLDER_PATTERN, self::PLACEHOLDER_SENTINEL, $this->path) ?? $this->path, + null, + ); + $rx = str_replace('\\*\\*', '(?:.*)?', $rx); + $rx = str_replace('\\*', '[^/]*', $rx); + $rx = strtr($rx, ['\\?' => '[^/]', '\\[' => '[', '\\]' => ']']); + $rx = str_replace(self::PLACEHOLDER_SENTINEL, '([^/]+)', $rx); + + $valueMatches = []; + + if (!preg_match('{^' . $rx . '$}', $path->path, $valueMatches)) { + return new Partitions(); + } + + $values = []; + + foreach ($names as $index => $name) { + $value = $valueMatches[$index + 1]; + + if (array_key_exists($name, $values) && $values[$name] !== $value) { + return new Partitions(); + } + + $values[$name] = $value; + } + + $partitionsList = []; + + foreach ($values as $name => $value) { + try { + $partitionsList[] = new Partition($name, $value); + } catch (EtlInvalidArgumentException) { + } + } + + return new Partitions(...$partitionsList); + } + public function filename(): string { return pathinfo($this->path, PATHINFO_FILENAME); } + /** + * Path with partition placeholders replaced by glob wildcards, suitable for glob-based listing. + */ + public function glob(): string + { + return preg_replace(self::PARTITION_PLACEHOLDER_PATTERN, '*', $this->path) ?? $this->path; + } + public function isEqual(self $path): bool { return $this->path === $path->path; @@ -250,6 +352,17 @@ public function parentDirectory(): self }; } + /** + * @return array + */ + public function partitionPlaceholders(): array + { + $matches = []; + preg_match_all(self::PARTITION_PLACEHOLDER_PATTERN, $this->path, $matches); + + return array_values(array_unique($matches[1])); + } + public function partitions(): Partitions { if ($this->isPattern()) { @@ -450,10 +563,14 @@ private function fnmatch(string $pattern, string $filename, int $flags = 0): boo } } - $rx = preg_quote($pattern, null); + $rx = preg_quote( + preg_replace(self::PARTITION_PLACEHOLDER_PATTERN, self::PLACEHOLDER_SENTINEL, $pattern) ?? $pattern, + null, + ); $rx = str_replace('\\*\\*', '(.*)?', $rx); $rx = str_replace('\\*', '[^/]*', $rx); $rx = strtr($rx, ['\\?' => '[^/]', '\\[' => '[', '\\]' => ']']); + $rx = str_replace(self::PLACEHOLDER_SENTINEL, '[^/]+', $rx); $rx = '{^' . $rx . '$}' . ($flags & 16 ? 'i' : ''); return (bool) preg_match($rx, $filename); diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/PartitionsCollectingFilter.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/PartitionsCollectingFilter.php new file mode 100644 index 0000000000..81d2e4038e --- /dev/null +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/PartitionsCollectingFilter.php @@ -0,0 +1,24 @@ + + */ + public array $partitionsList = []; + + public function accept(FileStatus $status): bool + { + $this->partitionsList[] = $status->path->partitions(); + + return true; + } +} diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/Local/MemoryFilesystemTest.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/Local/MemoryFilesystemTest.php index d961c4dbb9..969ba850f8 100644 --- a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/Local/MemoryFilesystemTest.php +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/Local/MemoryFilesystemTest.php @@ -129,6 +129,22 @@ public function test_file_status_on_pattern(): void static::assertSame('memory://var/some_path_to/file.txt', $patternStatus->path->uri()); } + public function test_list_files_matching_partition_placeholder_pattern(): void + { + $fs = memory_filesystem(); + + $fs->writeTo(path('memory:///var/placeholders/order-year=2024/123456-PL.csv'))->append('data'); + $fs->writeTo(path('memory:///var/placeholders/order-year=2024/789-DE.csv'))->append('data'); + $fs->writeTo(path('memory:///var/placeholders/order-year=2024/ignored.txt'))->append('data'); + + $files = iterator_to_array($fs->list(path('memory:///var/placeholders/order-year=2024/{order-name}.csv'))); + + $basenames = array_map(static fn(FileStatus $status) => $status->path->basename(), $files); + sort($basenames); + + static::assertSame(['123456-PL.csv', '789-DE.csv'], $basenames); + } + public function test_list_rejects_mismatched_scheme(): void { $this->expectException(InvalidSchemeException::class); diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/NativeLocalFilesystemTest.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/NativeLocalFilesystemTest.php index 777eb51a4b..f9379f28a0 100644 --- a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/NativeLocalFilesystemTest.php +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/NativeLocalFilesystemTest.php @@ -170,6 +170,22 @@ public function test_file_status_on_root_folder(): void static::assertTrue($status->isDirectory()); } + public function test_list_files_matching_partition_placeholder_pattern(): void + { + $fs = native_local_filesystem(); + + $fs->writeTo(path(__DIR__ . '/var/placeholders/order-year=2024/123456-PL.csv'))->append('data'); + $fs->writeTo(path(__DIR__ . '/var/placeholders/order-year=2024/789-DE.csv'))->append('data'); + $fs->writeTo(path(__DIR__ . '/var/placeholders/order-year=2024/ignored.txt'))->append('data'); + + $files = iterator_to_array($fs->list(path(__DIR__ . '/var/placeholders/order-year=2024/{order-name}.csv'))); + + $basenames = array_map(static fn(FileStatus $status) => $status->path->basename(), $files); + sort($basenames); + + static::assertSame(['123456-PL.csv', '789-DE.csv'], $basenames); + } + public function test_list_rejects_mismatched_scheme(): void { $this->expectException(InvalidSchemeException::class); 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 b64c72098f..48c8717b5e 100644 --- a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PartitionTest.php +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PartitionTest.php @@ -34,6 +34,8 @@ public static function provider_forbidden_characters_values(): array ['namaccept( + new FileStatus(path('/output/order-year=2024/123456-PL.csv'), isFile: true), + )); + static::assertCount(1, $collectingFilter->partitionsList); + static::assertSame('2024', $collectingFilter->partitionsList[0]->get('order-year')->value); + static::assertSame('123456-PL', $collectingFilter->partitionsList[0]->get('order-name')->value); + } + + public function test_delegates_original_file_status_when_no_partitions_extracted(): void + { + $collectingFilter = new PartitionsCollectingFilter(); + $filter = new PlaceholderPartitions(path('/output/{order-name}.csv'), $collectingFilter); + + static::assertTrue($filter->accept(new FileStatus(path('/other/123456-PL.csv'), isFile: true))); + static::assertCount(1, $collectingFilter->partitionsList); + static::assertCount(0, $collectingFilter->partitionsList[0]); + } + + public function test_delegates_to_decorated_filter_decision(): void + { + $filter = new PlaceholderPartitions(path('/output/{order-name}.csv'), new class implements Filter { + public function accept(FileStatus $status): bool + { + return false; + } + }); + + static::assertFalse($filter->accept(new FileStatus(path('/output/123456-PL.csv'), isFile: true))); + + static::assertTrue((new PlaceholderPartitions(path('/output/{order-name}.csv'), new KeepAll()))->accept( + new FileStatus(path('/output/123456-PL.csv'), isFile: true), + )); + } +} diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/UnixPathTest.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/UnixPathTest.php index a2a42c9729..0adf9e61e1 100644 --- a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/UnixPathTest.php +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/UnixPathTest.php @@ -68,6 +68,65 @@ public function test_absolute_path_handling(): void static::assertEquals('txt', $path->extension()); } + public function test_add_partitions_consuming_all_partitions_into_placeholders(): void + { + $path = new UnixPath('/output/{year}_{month}.csv'); + + static::assertEquals( + '/output/2024_03.csv', + $path->addPartitions(partition('year', '2024'), partition('month', '03'))->path(), + ); + } + + public function test_add_partitions_with_placeholder_and_remaining_partitions(): void + { + $path = new UnixPath('/output/{order-name}.csv'); + + static::assertEquals( + '/output/order-year=2024/123456-PL.csv', + $path->addPartitions(partition('order-year', '2024'), partition('order-name', '123456-PL'))->path(), + ); + } + + public function test_add_partitions_with_placeholder_in_directory(): void + { + $path = new UnixPath('/output/{year}/file.csv'); + + static::assertEquals( + '/output/2024/country=PL/file.csv', + $path->addPartitions(partition('year', '2024'), partition('country', 'PL'))->path(), + ); + } + + public function test_add_partitions_with_placeholder_next_to_glob_pattern(): void + { + $path = new UnixPath('/output/*/{order-name}.csv'); + + $this->expectException(InvalidArgumentException::class); + $this->expectExceptionMessage("Can't add partitions to path pattern."); + + $path->addPartitions(partition('order-name', '123456-PL')); + } + + public function test_add_partitions_with_repeated_placeholder(): void + { + $path = new UnixPath('/output/{name}/{name}.csv'); + + static::assertEquals('/output/abc/abc.csv', $path->addPartitions(partition('name', 'abc'))->path()); + } + + public function test_add_partitions_with_unresolved_placeholder(): void + { + $path = new UnixPath('/output/{order-name}.csv'); + + $this->expectException(InvalidArgumentException::class); + $this->expectExceptionMessage( + "Path partition placeholder {order-name} does not match any partition, available partitions: 'order-year'", + ); + + $path->addPartitions(partition('order-year', '2024')); + } + public function test_basename_operations(): void { $path = new UnixPath('/path/to/file.txt'); @@ -163,6 +222,55 @@ public function test_extension_with_no_extension_returns_false(): void static::assertFalse($path->extension()); } + public function test_extract_placeholder_partitions(): void + { + $pattern = new UnixPath('/output/order-year=2024/{order-name}.csv'); + + $partitions = $pattern->extractPlaceholderPartitions(new UnixPath('/output/order-year=2024/123456-PL.csv')); + + static::assertCount(1, $partitions); + static::assertEquals('123456-PL', $partitions->get('order-name')->value); + } + + public function test_extract_placeholder_partitions_from_not_matching_path(): void + { + $pattern = new UnixPath('/output/{order-name}.csv'); + + static::assertCount(0, $pattern->extractPlaceholderPartitions(new UnixPath('/other/123456-PL.csv'))); + } + + public function test_extract_placeholder_partitions_from_path_without_placeholders(): void + { + $pattern = new UnixPath('/output/file.csv'); + + static::assertCount(0, $pattern->extractPlaceholderPartitions(new UnixPath('/output/file.csv'))); + } + + public function test_extract_placeholder_partitions_skips_invalid_partition_values(): void + { + $pattern = new UnixPath('/output/{order-name}.csv'); + + static::assertCount(0, $pattern->extractPlaceholderPartitions(new UnixPath('/output/foo=bar.csv'))); + } + + public function test_extract_placeholder_partitions_with_conflicting_values_of_repeated_placeholder(): void + { + $pattern = new UnixPath('/output/{name}/{name}.csv'); + + static::assertCount(0, $pattern->extractPlaceholderPartitions(new UnixPath('/output/a/b.csv'))); + static::assertCount(1, $pattern->extractPlaceholderPartitions(new UnixPath('/output/a/a.csv'))); + } + + public function test_extract_placeholder_partitions_with_glob_and_placeholder(): void + { + $pattern = new UnixPath('/output/order-year=*/{order-name}.csv'); + + $partitions = $pattern->extractPlaceholderPartitions(new UnixPath('/output/order-year=2024/123456-PL.csv')); + + static::assertCount(1, $partitions); + static::assertEquals('123456-PL', $partitions->get('order-name')->value); + } + public function test_fnmatch_with_hidden_files(): void { $pattern = new UnixPath('/*'); @@ -173,6 +281,13 @@ public function test_fnmatch_with_hidden_files(): void static::assertTrue($pattern->matches($hidden)); } + public function test_glob(): void + { + static::assertEquals('/output/*.csv', (new UnixPath('/output/{order-name}.csv'))->glob()); + static::assertEquals('/output/*_*/file.csv', (new UnixPath('/output/{year}_{month}/file.csv'))->glob()); + static::assertEquals('/output/file.csv', (new UnixPath('/output/file.csv'))->glob()); + } + public function test_is_equal(): void { $path1 = new UnixPath('/path/to/file.txt'); @@ -211,6 +326,15 @@ public function test_matches_pattern_against_pattern_returns_false(): void static::assertFalse($pattern1->matches($pattern2)); } + public function test_matches_with_placeholders(): void + { + $pattern = new UnixPath('/output/{order-name}.csv'); + + static::assertTrue($pattern->matches(new UnixPath('/output/123456-PL.csv'))); + static::assertFalse($pattern->matches(new UnixPath('/output/nested/123456-PL.csv'))); + static::assertFalse($pattern->matches(new UnixPath('/output/123456-PL.json'))); + } + public function test_options_from_array(): void { $path = new UnixPath('/file.txt', ['option1' => 'value1', 'option2' => 'value2']); @@ -244,6 +368,18 @@ public function test_parent_directory_edge_cases(): void static::assertEquals('/', $path2->parentDirectory()->path()); } + public function test_partition_placeholders(): void + { + static::assertEquals([], (new UnixPath('/path/to/file.csv'))->partitionPlaceholders()); + static::assertEquals(['order-name'], (new UnixPath('/path/to/{order-name}.csv'))->partitionPlaceholders()); + static::assertEquals( + ['year', 'month'], + (new UnixPath('/path/{year}_{month}/file.csv'))->partitionPlaceholders(), + ); + static::assertEquals(['name'], (new UnixPath('/path/{name}/{name}.csv'))->partitionPlaceholders()); + static::assertEquals([], (new UnixPath('/path/*/file.csv'))->partitionPlaceholders()); + } + public function test_partitions_extraction(): void { $path = new UnixPath('/path/country=US/region=west/file.txt'); diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/WindowsPathTest.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/WindowsPathTest.php index 1f25c06cc2..144c1be8d1 100644 --- a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/WindowsPathTest.php +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/WindowsPathTest.php @@ -74,6 +74,38 @@ public function test_add_partitions_complex_drive_path(): void static::assertEquals('C:/some/path/year=2023/file.txt', $partitioned->path()); } + public function test_add_partitions_consuming_all_partitions_into_placeholders(): void + { + $path = new WindowsPath('C:/output/{year}_{month}.csv'); + + static::assertEquals( + 'C:/output/2024_03.csv', + $path->addPartitions(partition('year', '2024'), partition('month', '03'))->path(), + ); + } + + public function test_add_partitions_with_placeholder_and_remaining_partitions(): void + { + $path = new WindowsPath('C:/output/{order-name}.csv'); + + static::assertEquals( + 'C:/output/order-year=2024/123456-PL.csv', + $path->addPartitions(partition('order-year', '2024'), partition('order-name', '123456-PL'))->path(), + ); + } + + public function test_add_partitions_with_unresolved_placeholder(): void + { + $path = new WindowsPath('C:/output/{order-name}.csv'); + + $this->expectException(InvalidArgumentException::class); + $this->expectExceptionMessage( + "Path partition placeholder {order-name} does not match any partition, available partitions: 'order-year'", + ); + + $path->addPartitions(partition('order-year', '2024')); + } + public function test_basename_operations(): void { $path = new WindowsPath('/path/to/file.txt'); @@ -195,6 +227,25 @@ public function test_extension_with_no_extension_returns_false(): void static::assertFalse($path->extension()); } + public function test_extract_placeholder_partitions(): void + { + $pattern = new WindowsPath('C:/output/order-year=2024/{order-name}.csv'); + + $partitions = $pattern->extractPlaceholderPartitions( + new WindowsPath('C:/output/order-year=2024/123456-PL.csv'), + ); + + static::assertCount(1, $partitions); + static::assertEquals('123456-PL', $partitions->get('order-name')->value); + } + + public function test_extract_placeholder_partitions_from_not_matching_path(): void + { + $pattern = new WindowsPath('C:/output/{order-name}.csv'); + + static::assertCount(0, $pattern->extractPlaceholderPartitions(new WindowsPath('C:/other/123456-PL.csv'))); + } + public function test_fnmatch_with_hidden_files(): void { $pattern = new WindowsPath('/*'); @@ -205,6 +256,12 @@ public function test_fnmatch_with_hidden_files(): void static::assertTrue($pattern->matches($hidden)); } + public function test_glob(): void + { + static::assertEquals('C:/output/*.csv', (new WindowsPath('C:/output/{order-name}.csv'))->glob()); + static::assertEquals('C:/output/file.csv', (new WindowsPath('C:/output/file.csv'))->glob()); + } + public function test_is_equal(): void { $path1 = new WindowsPath('/path/to/file.txt'); @@ -243,6 +300,14 @@ public function test_matches_pattern_against_pattern_returns_false(): void static::assertFalse($pattern1->matches($pattern2)); } + public function test_matches_with_placeholders(): void + { + $pattern = new WindowsPath('C:/output/{order-name}.csv'); + + static::assertTrue($pattern->matches(new WindowsPath('C:/output/123456-PL.csv'))); + static::assertFalse($pattern->matches(new WindowsPath('C:/output/nested/123456-PL.csv'))); + } + public function test_options_from_array(): void { $path = new WindowsPath('/file.txt', ['option1' => 'value1', 'option2' => 'value2']); @@ -283,6 +348,16 @@ public function test_partition_logic(string $input, array $partitionData, string static::assertEquals($expected, $result->path()); } + public function test_partition_placeholders(): void + { + static::assertEquals([], (new WindowsPath('C:/path/to/file.csv'))->partitionPlaceholders()); + static::assertEquals(['order-name'], (new WindowsPath('C:/path/to/{order-name}.csv'))->partitionPlaceholders()); + static::assertEquals( + ['year', 'month'], + (new WindowsPath('C:/path/{year}_{month}/file.csv'))->partitionPlaceholders(), + ); + } + public function test_partitions_extraction(): void { $path = new WindowsPath('/path/country=US/region=west/city=LA/file.txt'); diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PathTest.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PathTest.php index 02f2c3dd27..f05003a207 100644 --- a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PathTest.php +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PathTest.php @@ -149,6 +149,16 @@ public function test_extension_uppercase(): void static::assertSame('php', path('/var/file/code.PhP')->extension()); } + public function test_extract_placeholder_partitions(): void + { + $pattern = path('/output/order-year=2024/{order-name}.csv'); + + static::assertEquals( + partitions(partition('order-name', '123456-PL')), + $pattern->extractPlaceholderPartitions(path('/output/order-year=2024/123456-PL.csv')), + ); + } + public function test_file_prefix(): void { $path = path('flow-file://var/dir/file.csv', []); @@ -179,6 +189,12 @@ public function test_finding_static_part_of_the_path(string $staticPart, string static::assertEquals(path($staticPart), path($uri)->staticPart()); } + public function test_glob(): void + { + static::assertSame('/output/*.csv', path('/output/{order-name}.csv')->glob()); + static::assertSame('/output/file.csv', path('/output/file.csv')->glob()); + } + public function test_local_file(): void { static::assertNull(path(__FILE__)->context()->resource()); @@ -205,6 +221,12 @@ public function test_parsing_path(string $uri, string $schema, string $parsedUri static::assertEquals($parsedUri, path($uri)->uri()); } + public function test_partition_placeholders(): void + { + static::assertEquals(['order-name'], path('/output/{order-name}.csv')->partitionPlaceholders()); + static::assertEquals([], path('/output/file.csv')->partitionPlaceholders()); + } + #[DataProvider('paths_with_partitions')] public function test_partitions_in_path(string $uri, Partitions $partitions): void { @@ -318,4 +340,17 @@ public function test_suffix(): void static::assertSame('flow-file://var/dir/test.csv', $path->suffix('/test.csv')->uri()); } + + public function test_with_partitions(): void + { + $path = path('/output/order-year=2024/123456-PL.csv')->withPartitions(partitions(partition( + 'order-name', + '123456-PL', + ))); + + static::assertEquals( + partitions(partition('order-year', '2024'), partition('order-name', '123456-PL')), + $path->partitions(), + ); + } } diff --git a/web/landing/content/examples/topics/partitioning/partition_placeholders/_meta.yaml b/web/landing/content/examples/topics/partitioning/partition_placeholders/_meta.yaml new file mode 100644 index 0000000000..8608dd76b4 --- /dev/null +++ b/web/landing/content/examples/topics/partitioning/partition_placeholders/_meta.yaml @@ -0,0 +1,2 @@ +priority: 3 +hidden: false diff --git a/web/landing/content/examples/topics/partitioning/partition_placeholders/code.php b/web/landing/content/examples/topics/partitioning/partition_placeholders/code.php new file mode 100644 index 0000000000..594d418f28 --- /dev/null +++ b/web/landing/content/examples/topics/partitioning/partition_placeholders/code.php @@ -0,0 +1,27 @@ +read(from_array( + [ + ['id' => 1, 'color' => 'red', 'sku' => 'PRODUCT01'], + ['id' => 2, 'color' => 'red', 'sku' => 'PRODUCT02'], + ['id' => 3, 'color' => 'green', 'sku' => 'PRODUCT01'], + ['id' => 4, 'color' => 'blue', 'sku' => 'PRODUCT02'], + ] + )) + ->partitionBy(ref('color'), ref('sku')) + ->mode(overwrite()) + ->write(to_csv(__DIR__ . '/output/{color}/{sku}.csv')) + ->run(); + +data_frame() + ->read(from_csv(__DIR__ . '/output/{color}/{sku}.csv')) + ->write(to_output(truncate: false)) + ->run(); diff --git a/web/landing/content/examples/topics/partitioning/partition_placeholders/description.md b/web/landing/content/examples/topics/partitioning/partition_placeholders/description.md new file mode 100644 index 0000000000..695ca13a9f --- /dev/null +++ b/web/landing/content/examples/topics/partitioning/partition_placeholders/description.md @@ -0,0 +1,14 @@ +Partition into a flat directory structure using `{column}` placeholders in the destination path instead of Hive-style `column=value` directories. Every partition consumed by a placeholder becomes part of the file or directory name; remaining partitions still create `column=value` directories. + +```bash +output +├── blue +│ └── PRODUCT02.csv +├── green +│ └── PRODUCT01.csv +└── red + ├── PRODUCT01.csv + └── PRODUCT02.csv +``` + +Reading with the same placeholder pattern recreates the partitions from the path, including support for partition pruning with `filterPartitions()`. Keep in mind that this layout is not self-describing - a plain glob like `output/**/*.csv` will read the data but won't recognize any partitions. diff --git a/web/landing/content/examples/topics/partitioning/partition_placeholders/documentation.md b/web/landing/content/examples/topics/partitioning/partition_placeholders/documentation.md new file mode 100644 index 0000000000..17ac40fac7 --- /dev/null +++ b/web/landing/content/examples/topics/partitioning/partition_placeholders/documentation.md @@ -0,0 +1,2 @@ +- [Partitioning](/documentation/components/core/partitioning) +- [Save Mode](/documentation/components/core/save-mode)