From e122c992eefcfb82cea4f6d7e3a2bb9a5a71b249 Mon Sep 17 00:00:00 2001 From: Norbert Orzechowicz Date: Sat, 12 Sep 2026 17:37:48 +0200 Subject: [PATCH] fix: write and read back partitioned datasets - parquet: openForWrite() returns a per-file ParquetFileWriter - parquet: path-only partition columns left out of the file schema - parquet: PHP engine splits array batches into row groups - etl: a failed closure() discards every staged file - filesystem: one glob matcher for local, memory, S3 and Azure - filesystem: webmozart/glob removed --- composer.json | 3 +- composer.lock | 51 +---- documentation/components/libs/filesystem.md | 29 +++ documentation/quick-start.md | 2 +- documentation/upgrading.md | 34 +++ .../Adapter/CSV/Tests/Integration/CSVTest.php | 25 +++ .../Tests/Integration/ExcelLoaderTest.php | 32 +++ .../JSON/Tests/Integration/JsonTest.php | 24 +++ .../ETL/Adapter/Parquet/ParquetLoader.php | 44 ++-- .../Tests/Context/ParquetFilesContext.php | 31 +++ .../Parquet/Tests/Integration/ParquetTest.php | 183 ++++++++++++++++ .../Integration/Loader/XMLLoaderTest.php | 25 +++ .../Integration/AsyncAWSS3FilesystemTest.php | 36 ++++ .../Integration/AzureBlobFilesystemTest.php | 39 ++++ .../FilesystemCommandsIntegrationTest.php | 14 ++ src/core/etl/composer.json | 1 - .../etl/src/Flow/ETL/Filesystem/FilesSink.php | 68 +++--- .../etl/src/Flow/ETL/Loader/Discardable.php | 2 +- .../etl/src/Flow/ETL/Pipeline/Segment.php | 19 ++ src/core/etl/src/Flow/Floe/FloeLoader.php | 23 +- .../ETL/Tests/Context/LoaderEndingContext.php | 45 ++++ .../Tests/Double/ClosureThrowingLoader.php | 15 +- .../Filesystem/FilesSink/FilesSinkTest.php | 71 ++++++- .../Tests/Unit/Pipeline/DiscardableTest.php | 85 ++++++-- .../Tests/Integration/FloeDataFrameTest.php | 38 ++++ src/lib/filesystem/composer.json | 3 +- .../src/Flow/Filesystem/Local/GlobWalker.php | 106 +++++++++ .../Local/NativeLocalFilesystem.php | 80 +------ .../src/Flow/Filesystem/Path/GlobPattern.php | 123 +++++++++++ .../src/Flow/Filesystem/Path/UnixPath.php | 27 +-- .../src/Flow/Filesystem/Path/WindowsPath.php | 27 +-- .../Tests/Context/GlobMatrixContext.php | 58 +++++ .../Double/FailingCloseDestinationStream.php | 53 +++++ .../Tests/Double/FailingCloseFilesystem.php | 86 ++++++++ .../Double/FakeNativeLocalFilesystem.php | 21 +- .../Integration/GlobListingAgreementTest.php | 89 ++++++++ .../Integration/Local/GlobWalkerTest.php | 107 ++++++++++ .../Tests/Unit/Path/GlobPatternTest.php | 77 +++++++ .../Flow/Filesystem/Tests/Unit/PathTest.php | 4 + .../Filesystem/Tests/Unit/PathTestCase.php | 4 + .../Parquet/Engine/AdaptiveParquetEngine.php | 20 +- .../Parquet/Engine/ArrowParquetEngine.php | 100 ++------- .../Parquet/Engine/ArrowParquetFileWriter.php | 77 +++++++ .../Flow/Parquet/Engine/PhpParquetEngine.php | 201 +----------------- .../Parquet/Engine/PhpParquetFileWriter.php | 149 +++++++++++++ .../src/Flow/Parquet/ParquetEngine.php | 14 +- .../src/Flow/Parquet/ParquetFileWriter.php | 24 +++ src/lib/parquet/src/Flow/Parquet/Writer.php | 25 ++- .../Flow/Parquet/Writer/RowGroupBuilder.php | 3 + .../Tests/Integration/IO/WriterTest.php | 64 ++++++ .../Tests/Mother/ParquetFileWriterMother.php | 29 +++ .../Engine/ArrowParquetFileWriterTest.php | 140 ++++++++++++ .../Unit/Engine/PhpParquetEngineTest.php | 37 ++-- .../Unit/Engine/PhpParquetFileWriterTest.php | 143 +++++++++++++ .../Tests/Unit/Writer/RowGroupBuilderTest.php | 16 ++ 55 files changed, 2235 insertions(+), 611 deletions(-) create mode 100644 src/core/etl/tests/Flow/ETL/Tests/Context/LoaderEndingContext.php create mode 100644 src/lib/filesystem/src/Flow/Filesystem/Local/GlobWalker.php create mode 100644 src/lib/filesystem/src/Flow/Filesystem/Path/GlobPattern.php create mode 100644 src/lib/filesystem/tests/Flow/Filesystem/Tests/Context/GlobMatrixContext.php create mode 100644 src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FailingCloseDestinationStream.php create mode 100644 src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FailingCloseFilesystem.php create mode 100644 src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/GlobListingAgreementTest.php create mode 100644 src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/Local/GlobWalkerTest.php create mode 100644 src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/GlobPatternTest.php create mode 100644 src/lib/parquet/src/Flow/Parquet/Engine/ArrowParquetFileWriter.php create mode 100644 src/lib/parquet/src/Flow/Parquet/Engine/PhpParquetFileWriter.php create mode 100644 src/lib/parquet/src/Flow/Parquet/ParquetFileWriter.php create mode 100644 src/lib/parquet/tests/Flow/Parquet/Tests/Mother/ParquetFileWriterMother.php create mode 100644 src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/ArrowParquetFileWriterTest.php create mode 100644 src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/PhpParquetFileWriterTest.php diff --git a/composer.json b/composer.json index d862372610..0cdb6fc9d5 100644 --- a/composer.json +++ b/composer.json @@ -48,8 +48,7 @@ "symfony/http-kernel": "^6.4 || ^7.4 || ^8.0", "symfony/polyfill-mbstring": "^1.33", "symfony/string": "^6.4 || ^7.4 || ^8.0", - "symfony/uid": "^6.4 || ^7.4 || ^8.0", - "webmozart/glob": "^3.0 || ^4.0" + "symfony/uid": "^6.4 || ^7.4 || ^8.0" }, "require-dev": { "cmsig/seal-elasticsearch-adapter": "^0.12", diff --git a/composer.lock b/composer.lock index 9c87f01275..7cd55a77ed 100644 --- a/composer.lock +++ b/composer.lock @@ -4,7 +4,7 @@ "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies", "This file is @generated automatically" ], - "content-hash": "f0028e7c06e259e8926ed718211f34af", + "content-hash": "8f9d5377d63c3504f0337d9a88b24790", "packages": [ { "name": "async-aws/core", @@ -4380,55 +4380,6 @@ } ], "time": "2026-08-23T10:03:40+00:00" - }, - { - "name": "webmozart/glob", - "version": "4.7.0", - "source": { - "type": "git", - "url": "https://github.com/webmozarts/glob.git", - "reference": "8a2842112d6916e61e0e15e316465b611f3abc17" - }, - "dist": { - "type": "zip", - "url": "https://api.github.com/repos/webmozarts/glob/zipball/8a2842112d6916e61e0e15e316465b611f3abc17", - "reference": "8a2842112d6916e61e0e15e316465b611f3abc17", - "shasum": "" - }, - "require": { - "php": "^7.3 || ^8.0.0" - }, - "require-dev": { - "phpunit/phpunit": "^9.5", - "symfony/filesystem": "^5.3" - }, - "type": "library", - "extra": { - "branch-alias": { - "dev-master": "4.1-dev" - } - }, - "autoload": { - "psr-4": { - "Webmozart\\Glob\\": "src/" - } - }, - "notification-url": "https://packagist.org/downloads/", - "license": [ - "MIT" - ], - "authors": [ - { - "name": "Bernhard Schussek", - "email": "bschussek@gmail.com" - } - ], - "description": "A PHP implementation of Ant's glob.", - "support": { - "issues": "https://github.com/webmozarts/glob/issues", - "source": "https://github.com/webmozarts/glob/tree/4.7.0" - }, - "time": "2024-03-07T20:33:40+00:00" } ], "packages-dev": [ diff --git a/documentation/components/libs/filesystem.md b/documentation/components/libs/filesystem.md index 5afa9369d7..2ecf9eddac 100644 --- a/documentation/components/libs/filesystem.md +++ b/documentation/components/libs/filesystem.md @@ -152,6 +152,35 @@ $stream->append('3,jane,true'); $stream->close(); ``` +## Glob patterns + +Every filesystem - local, memory, S3, Azure - lists the same files for the same pattern. + +| token | matches | +|-------------------------|------------------------------------------------------------------------| +| `*` | any run of characters inside one path segment, a leading `.` included | +| `?` | one character inside one segment | +| `[abc]` `[a-z]` | one character of the set / range, never `/` | +| `[!abc]` | one character outside the set, never `/`; `^` inside `[]` is a literal | +| `**` as a whole segment | zero or more segments; as the last segment: everything below | +| `**` inside a segment | same as `*` | +| `{name}` | flow partition placeholder: one or more characters inside one segment | +| unclosed `[` | the literal `[` - also when its `]` comes after a `/` | + +```php +list(path(__DIR__ . '/data/**/*.parquet')); // all three +native_local_filesystem()->list(path(__DIR__ . '/data/**.parquet')); // data/flat.parquet +``` + +Local listing follows symlinks a pattern names; `**` never descends into a symlinked directory. There is no escape +character. + ## Cross-filesystem copy & move `FilesystemTable` coupled with the `Copy` / `Move` operations lets you copy or move files between any diff --git a/documentation/quick-start.md b/documentation/quick-start.md index cadf4194a4..630033c720 100644 --- a/documentation/quick-start.md +++ b/documentation/quick-start.md @@ -67,7 +67,7 @@ data_frame() In this example we're using the `from_csv()` function to create a new instance of the `Flow\ETL\Adapter\CSV\CSVExtractor` class. -All file-based extractors accept [glob path patterns](https://github.com/webmozarts/glob), allowing you to read multiple files at once. +All file-based extractors accept [glob path patterns](/documentation/components/libs/filesystem.md#glob-patterns), allowing you to read multiple files at once. ```php data_frame() diff --git a/documentation/upgrading.md b/documentation/upgrading.md index 38a06e69d8..06c2fafc20 100644 --- a/documentation/upgrading.md +++ b/documentation/upgrading.md @@ -1497,6 +1497,40 @@ Reinstall it with the new release: `pie install flow-php/flow-php-ext`. | `array_dot_get([], '{a}')` message `Path "{a}" does not exists ...` | `Path "a" does not exists ...` | | `array_get_collection(ref('c'), ['id'])` over `[['name' => 'a'], []]` throws `InvalidArgumentException` | `[['id' => null], ['id' => null]]` | +### 106) `flow-php/filesystem` - every filesystem matches glob patterns the same way + +Files: `data/flat.parquet`, `data/.hidden.parquet`, `data/.dir/x.parquet`, `data/date=2026-09-01/one.parquet`, +`data/id=1/date=2026-09-01/two.parquet` + +| pattern | local, before | memory / S3 / Azure, before | every filesystem, after | +|-------------------------|---------------|-----------------------------|-----------------------------------| +| `data/**.parquet` | `flat` | all five | `.hidden`, `flat` | +| `data/**/*.parquet` | all five | `.dir/x`, `one`, `two` | all five | +| `data/**` | `flat` | all five | all five | +| `data/*.parquet` | `flat` | `.hidden`, `flat` | `.hidden`, `flat` | +| `data/*/*.parquet` | `one` | `.dir/x`, `one` | `.dir/x`, `one` | +| `data/**/[!f]*.parquet` | `flat` | none | `.dir/x`, `.hidden`, `one`, `two` | +| `data/[a-g]*.parquet` | `flat` | none | `flat` | + +Recurse with `data/**/*.parquet`, not `data/**.parquet`. `webmozart/glob` is no longer installed with +`flow-php/filesystem` or `flow-php/etl`. + +### 107) `flow-php/parquet` - `ParquetEngine::openForWrite()` returns a `ParquetFileWriter` + +| Before | After | +|-------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------| +| `$engine->openForWrite($stream, $schema, $c, $o); $engine->writeBatch($rows); $engine->closeWrite();` | `$file = $engine->openForWrite($stream, $schema, $c, $o); $file->writeBatch($rows); $file->close();` | +| `ParquetEngine::closeWrite()`, `writeBatch()`, `writeRow()` | removed - on `ParquetFileWriter` | +| two `Writer`s sharing one engine overwrote each other | every `openForWrite()` returns an independent writer | +| `ArrowParquetEngine` left the destination stream open | `ParquetFileWriter::close()` closes it on every engine | + +### 108) `flow-php/etl-adapter-parquet` - path-only partition columns leave the file body + +| Before | After | +|-----------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------| +| `partitionBy(partition_by('date'))` - file body carries an all-null `date` column | file body without `date` | +| `from_parquet()` types `date` from that body column, e.g. `datetime` | `string` - declare it: `from_parquet($path)->partitionTypes(partition_types(date: type_datetime()))` | + --- ## Upgrading from 0.42.x to 0.43.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 1b9af6e497..093786cf92 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 @@ -6,6 +6,8 @@ use Flow\ETL\Tests\Double\FakeExtractor; use Flow\ETL\Tests\FlowTestCase; +use Flow\Filesystem\Exception\RuntimeException as FilesystemRuntimeException; +use Flow\Filesystem\Tests\Double\FailingCloseFilesystem; use function file_exists; use function file_get_contents; @@ -27,7 +29,10 @@ use function Flow\ETL\DSL\str_schema; use function Flow\ETL\DSL\to_transformation; use function Flow\ETL\DSL\write_with_retries; +use function Flow\Filesystem\DSL\memory_filesystem; +use function Flow\Filesystem\DSL\path; use function implode; +use function iterator_to_array; use function mkdir; use function unlink; @@ -201,4 +206,24 @@ public function test_writing_csv_files_with_last_partition_as_file_name(): void 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'); } + + public function test_a_close_that_fails_during_closure_leaves_no_file(): void + { + $memory = memory_filesystem(); + + try { + df() + ->read(from_array([['p' => 'a', 't' => 'x'], ['p' => 'b', 't' => 'y'], ['p' => 'c', 't' => 'z']])) + ->write(to_csv( + path('memory://var/staged/file.csv'), + filesystem: new FailingCloseFilesystem($memory, failingStreams: 2), + )->partitionBy(partition_by('p'))) + ->run(); + static::fail('the run was expected to throw'); + } catch (FilesystemRuntimeException $failure) { + static::assertSame('Closing "memory://var/staged/p=a/file.csv" failed', $failure->getMessage()); + } + + static::assertSame([], iterator_to_array($memory->list(path('memory://var/staged/**/*')), false)); + } } diff --git a/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelLoaderTest.php b/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelLoaderTest.php index 15b66255f8..748e687d94 100644 --- a/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelLoaderTest.php +++ b/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelLoaderTest.php @@ -11,6 +11,8 @@ use Flow\ETL\Exception\InvalidArgumentException; use Flow\ETL\Schema\Definition; use Flow\ETL\Tests\FlowTestCase; +use Flow\Filesystem\Exception\RuntimeException as FilesystemRuntimeException; +use Flow\Filesystem\Tests\Double\FailingCloseFilesystem; use OpenSpout\Common\Entity\Style\Style; use OpenSpout\Writer\ODS\Options as OdsOptions; use OpenSpout\Writer\XLSX\Options as XlsxOptions; @@ -24,12 +26,14 @@ use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\df; use function Flow\ETL\DSL\float_schema; +use function Flow\ETL\DSL\from_array; use function Flow\ETL\DSL\from_rows; use function Flow\ETL\DSL\from_sequence_number; use function Flow\ETL\DSL\int_schema; use function Flow\ETL\DSL\json_schema; use function Flow\ETL\DSL\lit; use function Flow\ETL\DSL\overwrite; +use function Flow\ETL\DSL\partition_by; use function Flow\ETL\DSL\row; use function Flow\ETL\DSL\rows; use function Flow\ETL\DSL\schema; @@ -38,8 +42,11 @@ use function Flow\ETL\DSL\time_schema; use function Flow\ETL\DSL\to_transformation; use function Flow\ETL\DSL\uuid_schema; +use function Flow\Filesystem\DSL\native_local_filesystem; +use function Flow\Filesystem\DSL\path; use function Flow\Types\DSL\type_json; use function Flow\Types\DSL\type_uuid; +use function iterator_to_array; final class ExcelLoaderTest extends FlowTestCase { @@ -551,4 +558,29 @@ public function test_with_xlsx_options(): void $rows, ); } + + public function test_a_close_that_fails_during_closure_leaves_no_file(): void + { + try { + df() + ->read(from_array([['p' => 'a', 't' => 'x'], ['p' => 'b', 't' => 'y'], ['p' => 'c', 't' => 'z']])) + ->write( + to_excel( + __DIR__ . '/var/staged/file.xlsx', + new FailingCloseFilesystem(native_local_filesystem(), failingStreams: 2), + ) + ->saveMode(overwrite()) + ->partitionBy(partition_by('p')), + ) + ->run(); + static::fail('the run was expected to throw'); + } catch (FilesystemRuntimeException $failure) { + static::assertStringStartsWith('Closing "', $failure->getMessage()); + } + + static::assertSame( + [], + iterator_to_array(native_local_filesystem()->list(path(__DIR__ . '/var/staged/**/*')), false), + ); + } } diff --git a/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JsonTest.php b/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JsonTest.php index 2e8cbc2b58..60bb1af054 100644 --- a/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JsonTest.php +++ b/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JsonTest.php @@ -8,6 +8,8 @@ use Flow\ETL\Adapter\JSON\JsonLoader; use Flow\ETL\Tests\Double\FakeExtractor; use Flow\ETL\Tests\FlowTestCase; +use Flow\Filesystem\Exception\RuntimeException as FilesystemRuntimeException; +use Flow\Filesystem\Tests\Double\FailingCloseFilesystem; use function file_exists; use function file_get_contents; @@ -31,8 +33,10 @@ use function Flow\ETL\DSL\schema; use function Flow\ETL\DSL\select; use function Flow\ETL\DSL\to_transformation; +use function Flow\Filesystem\DSL\memory_filesystem; use function Flow\Filesystem\DSL\path; use function Flow\Types\DSL\type_json; +use function iterator_to_array; use function unlink; final class JsonTest extends FlowTestCase @@ -308,4 +312,24 @@ public function test_transformation_loader_writes_parseable_json_across_batches( unlink($path); } } + + public function test_a_close_that_fails_during_closure_leaves_no_file(): void + { + $memory = memory_filesystem(); + + try { + df() + ->read(from_array([['p' => 'a', 't' => 'x'], ['p' => 'b', 't' => 'y'], ['p' => 'c', 't' => 'z']])) + ->write(to_json( + path('memory://var/staged/file.json'), + filesystem: new FailingCloseFilesystem($memory, failingStreams: 2), + )->partitionBy(partition_by('p'))) + ->run(); + static::fail('the run was expected to throw'); + } catch (FilesystemRuntimeException $failure) { + static::assertSame('Closing "memory://var/staged/p=a/file.json" failed', $failure->getMessage()); + } + + static::assertSame([], iterator_to_array($memory->list(path('memory://var/staged/**/*')), false)); + } } diff --git a/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetLoader.php b/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetLoader.php index 5d11e10774..3c6aa23d48 100644 --- a/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetLoader.php +++ b/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetLoader.php @@ -102,10 +102,12 @@ public function closure(FlowContext $context): void public function discard(FlowContext $context): void { - $this->closeWriters(); - - $this->files?->abandon(); - $this->files = null; + try { + $this->closeWriters(); + } finally { + $this->files?->abandon(); + $this->files = null; + } } public function destination(): Path @@ -120,8 +122,8 @@ public function load(Rows $rows, FlowContext $context): void ]); try { - if ($this->schema === null && $this->inferredSchema === null) { - $this->inferSchema($rows); + if ($this->schema === null) { + $this->inferredSchema ??= $rows->schema()->makeNullable(); } foreach ($this->router->route($rows) as [$partitions, $group]) { @@ -180,9 +182,20 @@ public function withSchema(Schema $schema): static private function closeWriters(): void { + $failure = null; + foreach ($this->writers as $uri => $writer) { unset($this->writers[$uri]); - $writer->close(); + + try { + $writer->close(); + } catch (Throwable $closeFailure) { + $failure ??= $closeFailure; + } + } + + if ($failure !== null) { + throw $failure; } } @@ -191,15 +204,6 @@ private function encoder(): ParquetEncoder return $this->encoder ??= new ParquetEncoder($this->converter->toParquet($this->schema())); } - private function inferSchema(Rows $rows): void - { - if ($this->inferredSchema === null) { - $this->inferredSchema = $rows->schema()->makeNullable(); - } else { - $this->inferredSchema = $this->inferredSchema->merge($rows->schema())->makeNullable(); - } - } - private function openWriter(DestinationStream $stream): Writer { $writer = new Writer( @@ -214,10 +218,8 @@ private function openWriter(DestinationStream $stream): Writer private function schema(): Schema { - return ( - $this->schema ?? $this->inferredSchema ?? throw new RuntimeException( - 'Schema has not been inferred yet. Load at least one batch of rows first.', - ) - ); + return ($this->schema ?? $this->inferredSchema ?? throw new RuntimeException( + 'Schema has not been inferred yet. Load at least one batch of rows first.', + ))->gracefulRemove(...$this->router->droppedNames()); } } diff --git a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Context/ParquetFilesContext.php b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Context/ParquetFilesContext.php index a8543356e9..d9b0152c08 100644 --- a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Context/ParquetFilesContext.php +++ b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Context/ParquetFilesContext.php @@ -6,14 +6,45 @@ use Flow\ETL\Rows; use Flow\Filesystem\Filesystem; +use Flow\Parquet\ParquetFile\Schema\Column; +use Flow\Parquet\Reader; +use function array_map; use function Flow\ETL\Adapter\Parquet\to_parquet; use function Flow\ETL\DSL\data_frame; use function Flow\ETL\DSL\from_rows; use function Flow\Filesystem\DSL\path; +use function iterator_to_array; final class ParquetFilesContext { + /** + * @return array + */ + public static function columnNames(Filesystem $filesystem, string $uri): array + { + return array_map( + static fn(Column $column): string => $column->name(), + (new Reader()) + ->readStream($filesystem->readFrom(path($uri))) + ->schema() + ->columns(), + ); + } + + /** + * @return list> + */ + public static function values(Filesystem $filesystem, string $uri): array + { + return iterator_to_array( + (new Reader()) + ->readStream($filesystem->readFrom(path($uri))) + ->values(), + false, + ); + } + /** * @param array $files - uri => the rows written to it */ diff --git a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetTest.php b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetTest.php index f523ef0e24..62611a9154 100644 --- a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetTest.php +++ b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetTest.php @@ -4,10 +4,16 @@ namespace Flow\ETL\Adapter\Parquet\Tests\Integration; +use DateTimeImmutable; +use Flow\ETL\Adapter\Parquet\Tests\Context\ParquetFilesContext; +use Flow\ETL\Tests\Context\LoaderEndingContext; +use Flow\ETL\Tests\Context\MemoryTelemetryContext; use Flow\ETL\Tests\Double\FakeExtractor; use Flow\ETL\Tests\Double\FakeRandomOrdersExtractor; use Flow\ETL\Tests\FlowTestCase; +use Flow\Filesystem\Exception\RuntimeException as FilesystemRuntimeException; use Flow\Filesystem\SizeUnits; +use Flow\Filesystem\Tests\Double\FailingCloseFilesystem; use Flow\Parquet\Engine\ArrowParquetEngine; use Flow\Parquet\Engine\PhpParquetEngine; use Flow\Parquet\Option; @@ -23,14 +29,19 @@ use function Flow\ETL\Adapter\Parquet\to_parquet; use function Flow\ETL\DSL\config; use function Flow\ETL\DSL\data_frame; +use function Flow\ETL\DSL\datetime_schema; use function Flow\ETL\DSL\from_array; use function Flow\ETL\DSL\from_rows; use function Flow\ETL\DSL\from_sequence_number; +use function Flow\ETL\DSL\int_schema; use function Flow\ETL\DSL\json_schema; use function Flow\ETL\DSL\list_schema; use function Flow\ETL\DSL\lit; use function Flow\ETL\DSL\map_schema; use function Flow\ETL\DSL\overwrite; +use function Flow\ETL\DSL\partition_by; +use function Flow\ETL\DSL\partition_types; +use function Flow\ETL\DSL\ref; use function Flow\ETL\DSL\row; use function Flow\ETL\DSL\rows; use function Flow\ETL\DSL\schema; @@ -40,6 +51,7 @@ use function Flow\ETL\DSL\to_transformation; use function Flow\Filesystem\DSL\memory_filesystem; use function Flow\Filesystem\DSL\path; +use function Flow\Types\DSL\type_datetime; use function Flow\Types\DSL\type_integer; use function Flow\Types\DSL\type_json; use function Flow\Types\DSL\type_list; @@ -47,6 +59,7 @@ use function Flow\Types\DSL\type_string; use function Flow\Types\DSL\type_structure; use function Flow\Types\DSL\type_uuid; +use function iterator_to_array; use function unlink; final class ParquetTest extends FlowTestCase @@ -363,4 +376,174 @@ public function test_transformation_loader_writes_all_batches_to_parquet(): void unlink($path); } } + + public function test_a_close_that_fails_during_closure_leaves_no_file(): void + { + $memory = memory_filesystem(); + $telemetry = new MemoryTelemetryContext(); + + try { + data_frame($telemetry->config) + ->read(from_array([['p' => 'a', 't' => 'x'], ['p' => 'b', 't' => 'y'], ['p' => 'c', 't' => 'z']])) + ->write(to_parquet( + path('memory://var/staged/file.parquet'), + filesystem: new FailingCloseFilesystem($memory, failingStreams: 2), + )->partitionBy(partition_by('p'))) + ->run(); + static::fail('the run was expected to throw'); + } catch (FilesystemRuntimeException $failure) { + static::assertSame('Closing "memory://var/staged/p=a/file.parquet" failed', $failure->getMessage()); + } + + static::assertSame([], iterator_to_array($memory->list(path('memory://var/staged/**/*')), false)); + static::assertSame([], $telemetry->logs->entriesContaining('failed to discard')); + } + + public function test_a_close_that_fails_while_discarding_a_failed_run_leaves_no_file(): void + { + $memory = memory_filesystem(); + + LoaderEndingContext::failedRun(to_parquet( + path('memory://var/failed/file.parquet'), + filesystem: new FailingCloseFilesystem($memory), + )->partitionBy(partition_by('id'))); + + static::assertSame([], iterator_to_array($memory->list(path('memory://var/failed/**/*')), false)); + } + + public function test_a_path_only_partition_column_is_left_out_of_a_declared_file_schema(): void + { + $memory = memory_filesystem(); + + data_frame() + ->read(from_array([ + ['id' => 'a', 'date' => new DateTimeImmutable('2026-09-01'), 'clicks' => 1], + ['id' => 'b', 'date' => new DateTimeImmutable('2026-09-02'), 'clicks' => 2], + ])) + ->write(to_parquet( + path('memory://var/declared/file.parquet'), + schema: schema(str_schema('id'), datetime_schema('date'), int_schema('clicks')), + filesystem: $memory, + )->partitionBy(partition_by('date'))) + ->run(); + + static::assertSame( + ['id', 'clicks'], + ParquetFilesContext::columnNames($memory, 'memory://var/declared/date=2026-09-01/file.parquet'), + ); + static::assertEquals( + [ + ['id' => 'a', 'clicks' => 1, 'date' => new DateTimeImmutable('2026-09-01 00:00:00 UTC')], + ['id' => 'b', 'clicks' => 2, 'date' => new DateTimeImmutable('2026-09-02 00:00:00 UTC')], + ], + data_frame() + ->read(from_parquet(path('memory://var/declared/**/*.parquet'), filesystem: $memory)->partitionTypes( + partition_types(date: type_datetime()), + )) + ->sortBy(ref('id')) + ->fetch() + ->toArray(), + ); + } + + public function test_a_path_only_partition_column_is_left_out_of_an_inferred_file_schema(): void + { + $memory = memory_filesystem(); + + data_frame() + ->read(from_array([ + ['id' => 'a', 'date' => new DateTimeImmutable('2026-09-01'), 'clicks' => 1], + ['id' => 'b', 'date' => new DateTimeImmutable('2026-09-02'), 'clicks' => 2], + ])) + ->write(to_parquet( + path('memory://var/inferred/file.parquet'), + filesystem: $memory, + )->partitionBy(partition_by('date'))) + ->run(); + + static::assertSame( + ['id', 'clicks'], + ParquetFilesContext::columnNames($memory, 'memory://var/inferred/date=2026-09-01/file.parquet'), + ); + } + + public function test_partitions_written_with_an_explicit_arrow_engine_each_get_their_own_file(): void + { + if (!extension_loaded('arrow')) { + static::markTestSkipped('arrow extension is not loaded'); + } + + $memory = memory_filesystem(); + + data_frame() + ->read(from_array([ + ['p' => 'a', 'v' => 1], + ['p' => 'b', 'v' => 2], + ['p' => 'a', 'v' => 3], + ['p' => 'c', 'v' => 4], + ])) + ->write(to_parquet( + path('memory://var/engine/file.parquet'), + engine: new ArrowParquetEngine(), + filesystem: $memory, + )->partitionBy(partition_by('p'))) + ->run(); + + static::assertSame( + [['v' => 1], ['v' => 3]], + ParquetFilesContext::values($memory, 'memory://var/engine/p=a/file.parquet'), + ); + static::assertSame([['v' => 2]], ParquetFilesContext::values($memory, 'memory://var/engine/p=b/file.parquet')); + static::assertSame([['v' => 4]], ParquetFilesContext::values($memory, 'memory://var/engine/p=c/file.parquet')); + } + + public function test_partitions_written_with_an_explicit_php_engine_each_get_their_own_file(): void + { + $memory = memory_filesystem(); + + data_frame() + ->read(from_array([ + ['p' => 'a', 'v' => 1], + ['p' => 'b', 'v' => 2], + ['p' => 'a', 'v' => 3], + ['p' => 'c', 'v' => 4], + ])) + ->write(to_parquet( + path('memory://var/engine/file.parquet'), + engine: new PhpParquetEngine(), + filesystem: $memory, + )->partitionBy(partition_by('p'))) + ->run(); + + static::assertSame( + [['v' => 1], ['v' => 3]], + ParquetFilesContext::values($memory, 'memory://var/engine/p=a/file.parquet'), + ); + static::assertSame([['v' => 2]], ParquetFilesContext::values($memory, 'memory://var/engine/p=b/file.parquet')); + static::assertSame([['v' => 4]], ParquetFilesContext::values($memory, 'memory://var/engine/p=c/file.parquet')); + } + + public function test_write_columns_keeps_the_partition_column_in_the_file(): void + { + $memory = memory_filesystem(); + + data_frame() + ->read(from_array([ + ['id' => 'a', 'date' => new DateTimeImmutable('2026-09-01'), 'clicks' => 1], + ['id' => 'b', 'date' => new DateTimeImmutable('2026-09-02'), 'clicks' => 2], + ])) + ->write( + to_parquet( + path('memory://var/write_columns/file.parquet'), + schema: schema(str_schema('id'), datetime_schema('date'), int_schema('clicks')), + filesystem: $memory, + )->partitionBy(partition_by('date')->writeColumns()), + ) + ->run(); + + static::assertSame( + ['id', 'date', 'clicks'], + ParquetFilesContext::columnNames($memory, 'memory://var/write_columns/date=2026-09-01/file.parquet'), + ); + } } diff --git a/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/Loader/XMLLoaderTest.php b/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/Loader/XMLLoaderTest.php index e26551f04c..8d358534eb 100644 --- a/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/Loader/XMLLoaderTest.php +++ b/src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Integration/Loader/XMLLoaderTest.php @@ -6,6 +6,8 @@ use Flow\ETL\Tests\Double\FakeExtractor; use Flow\ETL\Tests\FlowIntegrationTestCase; +use Flow\Filesystem\Exception\RuntimeException as FilesystemRuntimeException; +use Flow\Filesystem\Tests\Double\FailingCloseFilesystem; use function file_exists; use function file_get_contents; @@ -20,6 +22,9 @@ use function Flow\ETL\DSL\ref; use function Flow\ETL\DSL\select; use function Flow\ETL\DSL\to_transformation; +use function Flow\Filesystem\DSL\memory_filesystem; +use function Flow\Filesystem\DSL\path; +use function iterator_to_array; final class XMLLoaderTest extends FlowIntegrationTestCase { @@ -133,4 +138,24 @@ public function test_writing_xml_with_attributes(): void XML, $content); } + + public function test_a_close_that_fails_during_closure_leaves_no_file(): void + { + $memory = memory_filesystem(); + + try { + df() + ->read(from_array([['p' => 'a', 't' => 'x'], ['p' => 'b', 't' => 'y'], ['p' => 'c', 't' => 'z']])) + ->write(to_xml( + path('memory://var/staged/file.xml'), + filesystem: new FailingCloseFilesystem($memory, failingStreams: 2), + )->partitionBy(partition_by('p'))) + ->run(); + static::fail('the run was expected to throw'); + } catch (FilesystemRuntimeException $failure) { + static::assertSame('Closing "memory://var/staged/p=a/file.xml" failed', $failure->getMessage()); + } + + static::assertSame([], iterator_to_array($memory->list(path('memory://var/staged/**/*')), false)); + } } diff --git a/src/bridge/filesystem/async-aws/tests/Flow/Filesystem/Bridge/AsyncAWS/Tests/Integration/AsyncAWSS3FilesystemTest.php b/src/bridge/filesystem/async-aws/tests/Flow/Filesystem/Bridge/AsyncAWS/Tests/Integration/AsyncAWSS3FilesystemTest.php index 1c6ead5f9a..64cfe15e5d 100644 --- a/src/bridge/filesystem/async-aws/tests/Flow/Filesystem/Bridge/AsyncAWS/Tests/Integration/AsyncAWSS3FilesystemTest.php +++ b/src/bridge/filesystem/async-aws/tests/Flow/Filesystem/Bridge/AsyncAWS/Tests/Integration/AsyncAWSS3FilesystemTest.php @@ -5,6 +5,10 @@ namespace Flow\Filesystem\Bridge\AsyncAWS\Tests\Integration; use Flow\Filesystem\FileStatus; +use Flow\Filesystem\Path\Filter\OnlyFiles; +use Flow\Filesystem\Tests\Context\GlobMatrixContext; +use Generator; +use PHPUnit\Framework\Attributes\DataProvider; use function array_map; use function file_get_contents; @@ -12,10 +16,42 @@ use function Flow\Filesystem\DSL\path; use function fopen; use function iterator_to_array; +use function sort; use function str_repeat; +use function strlen; +use function substr; final class AsyncAWSS3FilesystemTest extends AsyncAWSS3TestCase { + public static function glob_matrix(): Generator + { + yield from GlobMatrixContext::patterns(); + } + + /** + * @param list $expected + */ + #[DataProvider('glob_matrix')] + public function test_list_agrees_with_the_shared_glob_matrix(string $pattern, array $expected): void + { + $fs = aws_s3_filesystem($this->bucket(), $this->s3Client()); + + foreach (GlobMatrixContext::files() as $file) { + $fs + ->writeTo(path('aws-s3://' . $file)) + ->append($file) + ->close(); + } + + $listed = array_map( + static fn(FileStatus $status): string => substr($status->path->uri(), strlen('aws-s3://')), + iterator_to_array($fs->list(path('aws-s3://' . $pattern), new OnlyFiles()), false), + ); + sort($listed); + + static::assertSame($expected, $listed); + } + public function test_appending_to_existing_5mb_blob(): void { $fs = aws_s3_filesystem($this->bucket(), $this->s3Client()); diff --git a/src/bridge/filesystem/azure/tests/Flow/Filesystem/Bridge/Azure/Tests/Integration/AzureBlobFilesystemTest.php b/src/bridge/filesystem/azure/tests/Flow/Filesystem/Bridge/Azure/Tests/Integration/AzureBlobFilesystemTest.php index 488cbfaffd..ebf4f01e3e 100644 --- a/src/bridge/filesystem/azure/tests/Flow/Filesystem/Bridge/Azure/Tests/Integration/AzureBlobFilesystemTest.php +++ b/src/bridge/filesystem/azure/tests/Flow/Filesystem/Bridge/Azure/Tests/Integration/AzureBlobFilesystemTest.php @@ -5,16 +5,55 @@ namespace Flow\Filesystem\Bridge\Azure\Tests\Integration; use Flow\Filesystem\Bridge\Azure\Options; +use Flow\Filesystem\FileStatus; +use Flow\Filesystem\Path\Filter\OnlyFiles; +use Flow\Filesystem\Tests\Context\GlobMatrixContext; +use Generator; +use PHPUnit\Framework\Attributes\DataProvider; +use function array_map; use function file_get_contents; use function Flow\Filesystem\Bridge\Azure\DSL\azure_filesystem; use function Flow\Filesystem\Bridge\Azure\DSL\azure_filesystem_options; use function Flow\Filesystem\DSL\path; use function fopen; +use function iterator_to_array; +use function sort; use function str_repeat; +use function strlen; +use function substr; final class AzureBlobFilesystemTest extends AzureBlobServiceTestCase { + public static function glob_matrix(): Generator + { + yield from GlobMatrixContext::patterns(); + } + + /** + * @param list $expected + */ + #[DataProvider('glob_matrix')] + public function test_list_agrees_with_the_shared_glob_matrix(string $pattern, array $expected): void + { + $fs = azure_filesystem($this->blobService('flow-php')); + + foreach (GlobMatrixContext::files() as $file) { + $fs + ->writeTo(path('azure-blob://' . $file)) + ->append($file) + ->close(); + } + + $listed = array_map( + static fn(FileStatus $status): string => substr($status->path->uri(), strlen('azure-blob://')), + iterator_to_array($fs->list(path('azure-blob://' . $pattern), new OnlyFiles()), false), + ); + sort($listed); + + static::assertSame($expected, $listed); + } + public function test_appending_to_existing_blob(): void { $fs = azure_filesystem($this->blobService('flow-php')); diff --git a/src/bridge/symfony/filesystem-bundle/tests/Flow/Bridge/Symfony/FilesystemBundle/Tests/Integration/Command/FilesystemCommandsIntegrationTest.php b/src/bridge/symfony/filesystem-bundle/tests/Flow/Bridge/Symfony/FilesystemBundle/Tests/Integration/Command/FilesystemCommandsIntegrationTest.php index 8ff2fc3cfb..1c2d460861 100644 --- a/src/bridge/symfony/filesystem-bundle/tests/Flow/Bridge/Symfony/FilesystemBundle/Tests/Integration/Command/FilesystemCommandsIntegrationTest.php +++ b/src/bridge/symfony/filesystem-bundle/tests/Flow/Bridge/Symfony/FilesystemBundle/Tests/Integration/Command/FilesystemCommandsIntegrationTest.php @@ -119,6 +119,20 @@ public function test_ls_default_emits_size_and_modified_columns(): void static::assertStringContainsString('5 B', $display); } + public function test_ls_recursive_lists_top_level_files_on_a_memory_filesystem(): void + { + $resolver = $this->bootWithMultiFstab(); + $this->seed($resolver->resolve(null), 'memory://tree/top.txt', 'top'); + $this->seed($resolver->resolve(null), 'memory://tree/nested/deep.txt', 'deep'); + + $tester = new CommandTester(new LsCommand($resolver)); + $exit = $tester->execute(['path' => 'memory://tree', '--recursive' => true]); + + static::assertSame(Command::SUCCESS, $exit); + static::assertStringContainsString('top.txt', $tester->getDisplay()); + static::assertStringContainsString('deep.txt', $tester->getDisplay()); + } + public function test_ls_does_not_prompt_when_limit_exactly_equals_page_size(): void { $resolver = $this->bootWithMultiFstab(); diff --git a/src/core/etl/composer.json b/src/core/etl/composer.json index 15c83a2446..8ad3b01d4f 100644 --- a/src/core/etl/composer.json +++ b/src/core/etl/composer.json @@ -19,7 +19,6 @@ "flow-php/array-dot": "self.version", "flow-php/filesystem": "self.version", "psr/simple-cache": "^1.0 || ^2.0 || ^3.0", - "webmozart/glob": "^3.0 || ^4.0", "symfony/string": "^6.4 || ^7.4 || ^8.0" }, "suggest": { diff --git a/src/core/etl/src/Flow/ETL/Filesystem/FilesSink.php b/src/core/etl/src/Flow/ETL/Filesystem/FilesSink.php index e3a5c21119..8090e628af 100644 --- a/src/core/etl/src/Flow/ETL/Filesystem/FilesSink.php +++ b/src/core/etl/src/Flow/ETL/Filesystem/FilesSink.php @@ -11,6 +11,7 @@ use Flow\Filesystem\Path; use Flow\Filesystem\Stream\VoidStream; use Generator; +use Throwable; use function array_key_exists; use function count; @@ -45,12 +46,17 @@ public function abandon(): void $created = $this->created; $this->streams = []; $this->created = []; + $failure = null; // every handle closes first: a format writer flushes its footer on close, and the file has to be gone // after that, not before foreach ($streams as $stream) { if ($stream->isOpen()) { - $stream->close(); + try { + $stream->close(); + } catch (Throwable $closeFailure) { + $failure ??= $closeFailure; + } } } @@ -59,47 +65,49 @@ public function abandon(): void $this->filesystem->rm($path); } } + + if ($failure !== null) { + throw $failure; + } } public function publish(): void { - $streams = $this->streams; - $this->streams = []; - $this->created = []; - - foreach ($streams as $stream) { + // a stream leaves the registry only once it is published, so abandon() after a failure part way through + // still removes every file that never made it + foreach ($this->streams as $uri => $stream) { if ($stream->isOpen()) { $stream->close(); } - if ($this->saveMode !== SaveMode::Overwrite) { - continue; - } - - if ($stream->path()->partitions()->count() || [] !== $this->destination->partitionPlaceholders()) { - $writtenFiles = path( - $stream->path()->parentDirectory()->uri() - . '/' - . str_replace(self::FLOW_TMP_FILE_PREFIX, '', $stream->path()->filename()) - . '*.' - // @mago-ignore analysis:possibly-false-operand - . $stream->path()->extension(), - $stream->path()->options(), - ); - - foreach ($this->filesystem->list($writtenFiles) as $stale) { - if (str_contains($stale->path->path(), self::FLOW_TMP_FILE_PREFIX)) { - continue; + if ($this->saveMode === SaveMode::Overwrite) { + if ($stream->path()->partitions()->count() || [] !== $this->destination->partitionPlaceholders()) { + $writtenFiles = path( + $stream->path()->parentDirectory()->uri() + . '/' + . str_replace(self::FLOW_TMP_FILE_PREFIX, '', $stream->path()->filename()) + . '*.' + // @mago-ignore analysis:possibly-false-operand + . $stream->path()->extension(), + $stream->path()->options(), + ); + + foreach ($this->filesystem->list($writtenFiles) as $stale) { + if (str_contains($stale->path->path(), self::FLOW_TMP_FILE_PREFIX)) { + continue; + } + + $this->filesystem->rm($stale->path); } - - $this->filesystem->rm($stale->path); } + + $this->filesystem->mv($stream->path(), path( + str_replace(self::FLOW_TMP_FILE_PREFIX, '', $stream->path()->uri()), + $stream->path()->options(), + )); } - $this->filesystem->mv($stream->path(), path( - str_replace(self::FLOW_TMP_FILE_PREFIX, '', $stream->path()->uri()), - $stream->path()->options(), - )); + unset($this->streams[$uri], $this->created[$uri]); } } diff --git a/src/core/etl/src/Flow/ETL/Loader/Discardable.php b/src/core/etl/src/Flow/ETL/Loader/Discardable.php index 5c06519bab..6562d766ae 100644 --- a/src/core/etl/src/Flow/ETL/Loader/Discardable.php +++ b/src/core/etl/src/Flow/ETL/Loader/Discardable.php @@ -8,7 +8,7 @@ /** * Loaders implementing this adapter will be notified by the pipeline when a run ends without reaching its last set of - * Rows - it threw, or the caller walked away from the generator. + * Rows - it threw, or the caller walked away from the generator - and when their own closure() threw. */ interface Discardable { diff --git a/src/core/etl/src/Flow/ETL/Pipeline/Segment.php b/src/core/etl/src/Flow/ETL/Pipeline/Segment.php index bd37130bef..786b8fc57b 100644 --- a/src/core/etl/src/Flow/ETL/Pipeline/Segment.php +++ b/src/core/etl/src/Flow/ETL/Pipeline/Segment.php @@ -252,6 +252,25 @@ private function endLoaders(array $loaders, FlowContext $context, bool $complete // one sink failing to end must not strand the others if ($completed) { $ending[] = $failure; + + // a closure() that threw published at most part of its output; the rest is abandoned as on + // a failed run + foreach ((new LoaderTree())->flatten($loader) as $node) { + if (!$node instanceof Discardable) { + continue; + } + + try { + $node->discard($context); + } catch (Throwable $discardFailure) { + $context + ->telemetry() + ->logger() + ->error('Loader failed to discard after its closure failed.', [ + 'exception' => $discardFailure, + ]); + } + } } else { $context ->telemetry() diff --git a/src/core/etl/src/Flow/Floe/FloeLoader.php b/src/core/etl/src/Flow/Floe/FloeLoader.php index 3fef89bbe0..b194150987 100644 --- a/src/core/etl/src/Flow/Floe/FloeLoader.php +++ b/src/core/etl/src/Flow/Floe/FloeLoader.php @@ -97,10 +97,12 @@ public function closure(FlowContext $context): void public function discard(FlowContext $context): void { - $this->closeWriters(); - - $this->files?->abandon(); - $this->files = null; + try { + $this->closeWriters(); + } finally { + $this->files?->abandon(); + $this->files = null; + } } public function destination(): Path @@ -137,9 +139,20 @@ public function load(Rows $rows, FlowContext $context): void private function closeWriters(): void { + $failure = null; + foreach ($this->writers as $uri => $writer) { unset($this->writers[$uri]); - $writer->close(); + + try { + $writer->close(); + } catch (Throwable $closeFailure) { + $failure ??= $closeFailure; + } + } + + if ($failure !== null) { + throw $failure; } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Context/LoaderEndingContext.php b/src/core/etl/tests/Flow/ETL/Tests/Context/LoaderEndingContext.php new file mode 100644 index 0000000000..39014d5253 --- /dev/null +++ b/src/core/etl/tests/Flow/ETL/Tests/Context/LoaderEndingContext.php @@ -0,0 +1,45 @@ +read(from_array([['id' => 1, 'v' => 'a'], ['id' => 2, 'v' => 'b']])) + ->batchSize(1) + ->with(new ThrowWhenRowMatches('id', 2, new RuntimeException('boom'))) + ->write($loader) + ->run(); + } catch (Throwable) { + // the run is expected to fail; what matters is which ending the sink was given + } + } + + public static function thrownByRun(Loader $loader, ?Config $config = null): ?Throwable + { + try { + data_frame($config) + ->read(from_array([['id' => 1]])) + ->write($loader) + ->run(); + } catch (Throwable $failure) { + return $failure; + } + + return null; + } +} diff --git a/src/core/etl/tests/Flow/ETL/Tests/Double/ClosureThrowingLoader.php b/src/core/etl/tests/Flow/ETL/Tests/Double/ClosureThrowingLoader.php index adca63e515..bdc2062022 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Double/ClosureThrowingLoader.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Double/ClosureThrowingLoader.php @@ -7,15 +7,19 @@ use Flow\ETL\FlowContext; use Flow\ETL\Loader; use Flow\ETL\Loader\Closure; +use Flow\ETL\Loader\Discardable; use Flow\ETL\Rows; use Throwable; -final class ClosureThrowingLoader implements Closure, Loader +final class ClosureThrowingLoader implements Closure, Discardable, Loader { + public int $discarded = 0; + public int $loadsCount = 0; public function __construct( private readonly Throwable $throwable, + private readonly ?Throwable $discardFailure = null, ) {} public function closure(FlowContext $context): void @@ -23,6 +27,15 @@ public function closure(FlowContext $context): void throw $this->throwable; } + public function discard(FlowContext $context): void + { + $this->discarded++; + + if ($this->discardFailure !== null) { + throw $this->discardFailure; + } + } + public function load(Rows $rows, FlowContext $context): void { $this->loadsCount++; diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/Filesystem/FilesSink/FilesSinkTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/Filesystem/FilesSink/FilesSinkTest.php index 7620227d04..99042d3479 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Integration/Filesystem/FilesSink/FilesSinkTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Integration/Filesystem/FilesSink/FilesSinkTest.php @@ -6,12 +6,17 @@ use Flow\ETL\Filesystem\FilesSink; use Flow\ETL\Filesystem\SaveMode; +use Flow\Filesystem\Exception\RuntimeException as FilesystemRuntimeException; use Flow\Filesystem\FileListing; use Flow\Filesystem\Path\Filter\KeepAll; +use Flow\Filesystem\Tests\Double\FailingCloseFilesystem; +use Generator; use Override; +use PHPUnit\Framework\Attributes\DataProvider; use function file_get_contents; use function Flow\ETL\DSL\exception_if_exists; +use function Flow\Filesystem\DSL\partition; use function Flow\Filesystem\DSL\path; use function Flow\Filesystem\DSL\stdout_filesystem; use function iterator_to_array; @@ -25,6 +30,66 @@ protected function tearDown(): void $this->cleanFiles(); } + public static function save_modes(): Generator + { + yield 'overwrite' => [SaveMode::Overwrite]; + yield 'exception if exists' => [SaveMode::ExceptionIfExists]; + yield 'append' => [SaveMode::Append]; + } + + #[DataProvider('save_modes')] + public function test_abandon_after_a_publish_that_failed_part_way_removes_only_what_was_not_published(SaveMode $saveMode): void + { + $this->setupFiles([__FUNCTION__ => []]); + + $files = new FilesSink( + new FailingCloseFilesystem($this->fs, healthyStreams: 1), + $this->getPath(__FUNCTION__ . '/file.txt'), + $saveMode, + ); + $files->writeTo([partition('p', 'a')])->append('a'); + $b = $files->writeTo([partition('p', 'b')]); + $b->append('b'); + + try { + $files->publish(); + static::fail('publish() was expected to throw'); + } catch (FilesystemRuntimeException $failure) { + static::assertSame('Closing "' . $b->path()->uri() . '" failed', $failure->getMessage()); + } + + $files->abandon(); + + static::assertSame('a', file_get_contents($this->getPath(__FUNCTION__ . '/p=a/file.txt')->path())); + static::assertFileDoesNotExist($this->getPath(__FUNCTION__ . '/p=b/file.txt')->path()); + static::assertFileDoesNotExist($b->path()->path()); + } + + public function test_abandon_removes_every_file_even_when_a_stream_fails_to_close(): void + { + $this->setupFiles([__FUNCTION__ => []]); + + $files = new FilesSink( + new FailingCloseFilesystem($this->fs, failingStreams: 2), + $this->getPath(__FUNCTION__ . '/file.txt'), + SaveMode::Overwrite, + ); + $a = $files->writeTo([partition('p', 'a')]); + $a->append('a'); + $b = $files->writeTo([partition('p', 'b')]); + $b->append('b'); + + try { + $files->abandon(); + static::fail('abandon() was expected to throw'); + } catch (FilesystemRuntimeException $failure) { + static::assertSame('Closing "' . $a->path()->uri() . '" failed', $failure->getMessage()); + } + + static::assertFileDoesNotExist($a->path()->path()); + static::assertFileDoesNotExist($b->path()->path()); + } + public function test_abandon_leaves_the_destination_untouched(): void { $this->setupFiles([ @@ -133,8 +198,7 @@ public function test_touched_reports_a_file_this_run_has_written_to(): void { $this->setupFiles([__FUNCTION__ => []]); - $file = $this->getPath(__FUNCTION__ . '/file.txt'); - $files = $this->files($file); + $files = $this->files($this->getPath(__FUNCTION__ . '/file.txt')); static::assertFalse($files->touched()); @@ -151,8 +215,7 @@ public function test_open_streams_lists_only_streams_that_are_still_open(): void { $this->setupFiles([__FUNCTION__ => []]); - $file = $this->getPath(__FUNCTION__ . '/file.txt'); - $files = $this->files($file); + $files = $this->files($this->getPath(__FUNCTION__ . '/file.txt')); $stream = $files->writeTo(); static::assertSame([$stream], iterator_to_array($files->openStreams(), false)); diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Pipeline/DiscardableTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Pipeline/DiscardableTest.php index 7bf2d3531d..cdf9e598c5 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Pipeline/DiscardableTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Pipeline/DiscardableTest.php @@ -6,12 +6,12 @@ use Flow\ETL\DataFrame; use Flow\ETL\Exception\RuntimeException; -use Flow\ETL\Loader; +use Flow\ETL\Tests\Context\LoaderEndingContext; +use Flow\ETL\Tests\Context\MemoryTelemetryContext; use Flow\ETL\Tests\Double\CallbackTransformation; +use Flow\ETL\Tests\Double\ClosureThrowingLoader; use Flow\ETL\Tests\Double\RecordingSink; -use Flow\ETL\Tests\Double\ThrowWhenRowMatches; use Flow\ETL\Tests\FlowTestCase; -use Throwable; use function Flow\ETL\DSL\data_frame; use function Flow\ETL\DSL\from_array; @@ -57,17 +57,72 @@ public function test_a_run_that_throws_discards_the_sink_and_never_closes_it(): { $sink = new RecordingSink(); - $this->failedRun($sink); + LoaderEndingContext::failedRun($sink); static::assertSame(1, $sink->discarded); static::assertSame(0, $sink->closed); } + public function test_a_discard_that_throws_after_a_failed_closure_does_not_replace_the_closure_failure(): void + { + $telemetry = new MemoryTelemetryContext(); + $closureFailure = new RuntimeException('closure failed'); + $sink = new ClosureThrowingLoader($closureFailure, new RuntimeException('discard failed')); + + static::assertSame($closureFailure, LoaderEndingContext::thrownByRun($sink, $telemetry->config)); + static::assertSame(1, $sink->discarded); + static::assertCount( + 1, + $telemetry->logs->entriesContaining('Loader failed to discard after its closure failed.'), + ); + } + + public function test_a_discard_that_throws_after_a_failed_run_is_logged(): void + { + $telemetry = new MemoryTelemetryContext(); + $sink = new ClosureThrowingLoader( + new RuntimeException('closure failed'), + new RuntimeException('discard failed'), + ); + + LoaderEndingContext::failedRun($sink, $telemetry->config); + + static::assertSame(1, $sink->discarded); + static::assertCount(1, $telemetry->logs->entriesContaining('Loader failed to end after a failed run.')); + } + + public function test_a_sink_wrapped_in_a_retrying_loader_is_discarded_when_its_closure_throws(): void + { + $closureFailure = new RuntimeException('closure failed'); + $sink = new ClosureThrowingLoader($closureFailure); + + static::assertSame($closureFailure, LoaderEndingContext::thrownByRun(write_with_retries($sink))); + static::assertSame(1, $sink->discarded); + } + + public function test_a_sink_whose_closure_throws_is_discarded_and_the_failure_rethrown(): void + { + $closureFailure = new RuntimeException('closure failed'); + $sink = new ClosureThrowingLoader($closureFailure); + + static::assertSame($closureFailure, LoaderEndingContext::thrownByRun($sink)); + static::assertSame(1, $sink->discarded); + } + + public function test_a_sink_wrapped_in_a_branching_loader_is_discarded_when_its_closure_throws(): void + { + $closureFailure = new RuntimeException('closure failed'); + $sink = new ClosureThrowingLoader($closureFailure); + + static::assertSame($closureFailure, LoaderEndingContext::thrownByRun(to_branch(ref('id')->isNotNull(), $sink))); + static::assertSame(1, $sink->discarded); + } + public function test_a_sink_wrapped_in_a_branching_loader_is_discarded(): void { $sink = new RecordingSink(); - $this->failedRun(to_branch(ref('id')->isNotNull(), $sink)); + LoaderEndingContext::failedRun(to_branch(ref('id')->isNotNull(), $sink)); static::assertSame(1, $sink->discarded); static::assertSame(0, $sink->closed); @@ -77,7 +132,7 @@ public function test_a_sink_wrapped_in_a_retry_loader_is_discarded(): void { $sink = new RecordingSink(); - $this->failedRun(write_with_retries($sink)); + LoaderEndingContext::failedRun(write_with_retries($sink)); static::assertSame(1, $sink->discarded); static::assertSame(0, $sink->closed); @@ -87,7 +142,7 @@ public function test_a_sink_wrapped_in_a_transformation_loader_is_discarded(): v { $sink = new RecordingSink(); - $this->failedRun(to_transformation( + LoaderEndingContext::failedRun(to_transformation( new CallbackTransformation(static fn(DataFrame $df): DataFrame => $df), $sink, )); @@ -100,22 +155,8 @@ public function test_a_sink_wrapped_twice_is_discarded_once(): void { $sink = new RecordingSink(); - $this->failedRun(write_with_retries(to_branch(ref('id')->isNotNull(), $sink))); + LoaderEndingContext::failedRun(write_with_retries(to_branch(ref('id')->isNotNull(), $sink))); static::assertSame(1, $sink->discarded); } - - protected function failedRun(Loader $loader): void - { - try { - data_frame() - ->read(from_array([['id' => 1], ['id' => 2]])) - ->batchSize(1) - ->with(new ThrowWhenRowMatches('id', 2, new RuntimeException('boom'))) - ->write($loader) - ->run(); - } catch (Throwable) { - // the run is expected to fail; what matters is which ending the sink was given - } - } } diff --git a/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeDataFrameTest.php b/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeDataFrameTest.php index 6d7a03341b..4e81f57ab2 100644 --- a/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeDataFrameTest.php +++ b/src/core/etl/tests/Flow/Floe/Tests/Integration/FloeDataFrameTest.php @@ -6,7 +6,10 @@ use DateTimeZone; use Flow\ETL\Row\PhpRowHydrator; +use Flow\ETL\Tests\Context\LoaderEndingContext; use Flow\ETL\Tests\FlowIntegrationTestCase; +use Flow\Filesystem\Exception\RuntimeException as FilesystemRuntimeException; +use Flow\Filesystem\Tests\Double\FailingCloseFilesystem; use Flow\Floe\FloeEngine; use Flow\Floe\NativeFloeEncoder; use Flow\Types\Value\Json; @@ -28,12 +31,15 @@ use function Flow\ETL\DSL\select; use function Flow\ETL\DSL\time_zone_schema; use function Flow\ETL\DSL\to_transformation; +use function Flow\Filesystem\DSL\memory_filesystem; +use function Flow\Filesystem\DSL\path; use function Flow\Floe\DSL\from_floe; use function Flow\Floe\DSL\to_floe; use function Flow\Types\DSL\type_instance_of; use function Flow\Types\DSL\type_json; use function Flow\Types\DSL\type_list; use function Flow\Types\DSL\type_time_zone; +use function iterator_to_array; final class FloeDataFrameTest extends FlowIntegrationTestCase { @@ -285,4 +291,36 @@ public function test_round_trip(): void static::assertSame(2, $result->count()); static::assertSame(['id', 'name'], $result->first()->names()); } + + public function test_a_close_that_fails_during_closure_leaves_no_file(): void + { + $memory = memory_filesystem(); + + try { + data_frame() + ->read(from_array([['p' => 'a', 't' => 'x'], ['p' => 'b', 't' => 'y'], ['p' => 'c', 't' => 'z']])) + ->write(to_floe( + path('memory://var/staged/file.floe'), + filesystem: new FailingCloseFilesystem($memory, failingStreams: 2), + )->partitionBy(partition_by('p'))) + ->run(); + static::fail('the run was expected to throw'); + } catch (FilesystemRuntimeException $failure) { + static::assertSame('Closing "memory://var/staged/p=a/file.floe" failed', $failure->getMessage()); + } + + static::assertSame([], iterator_to_array($memory->list(path('memory://var/staged/**/*')), false)); + } + + public function test_a_close_that_fails_while_discarding_a_failed_run_leaves_no_file(): void + { + $memory = memory_filesystem(); + + LoaderEndingContext::failedRun(to_floe( + path('memory://var/failed/file.floe'), + filesystem: new FailingCloseFilesystem($memory), + )->partitionBy(partition_by('id'))); + + static::assertSame([], iterator_to_array($memory->list(path('memory://var/failed/**/*')), false)); + } } diff --git a/src/lib/filesystem/composer.json b/src/lib/filesystem/composer.json index 3f56f8fd62..4a3bbde8fc 100644 --- a/src/lib/filesystem/composer.json +++ b/src/lib/filesystem/composer.json @@ -17,8 +17,7 @@ "php": "~8.3.0 || ~8.4.0 || ~8.5.0", "flow-php/telemetry": "self.version", "flow-php/types": "self.version", - "symfony/polyfill-mbstring": "^1.33", - "webmozart/glob": "^3.0 || ^4.0" + "symfony/polyfill-mbstring": "^1.33" }, "config": { "optimize-autoloader": true, diff --git a/src/lib/filesystem/src/Flow/Filesystem/Local/GlobWalker.php b/src/lib/filesystem/src/Flow/Filesystem/Local/GlobWalker.php new file mode 100644 index 0000000000..a04ca4ddca --- /dev/null +++ b/src/lib/filesystem/src/Flow/Filesystem/Local/GlobWalker.php @@ -0,0 +1,106 @@ + every file and directory the pattern matches, sorted + */ + public function walk(Path $pattern): array + { + $segments = explode('/', preg_replace('#/+#', '/', $pattern->path()) ?? $pattern->path()); + $static = []; + + while ($segments !== [] && strpbrk($segments[0], '*?[{') === false) { + $static[] = array_shift($segments); + } + + $base = implode('/', $static) ?: '/'; + + if ($segments === [] || !is_dir($base) || !is_readable($base)) { + return []; + } + + // ** repeated matches exactly what one ** matches, and every extra one would walk the tree again + $collapsed = []; + + foreach ($segments as $segment) { + if ($segment !== '**' || ($collapsed[count($collapsed) - 1] ?? null) !== '**') { + $collapsed[] = $segment; + } + } + + $matchers = array_map(static fn(string $segment): GlobPattern => new GlobPattern($segment), $collapsed); + $last = count($collapsed) - 1; + $found = []; + $queue = [[$base, 0]]; + + while ($queue !== []) { + [$directory, $index] = array_pop($queue); + $spanning = $collapsed[$index] === '**'; + // a whole-segment ** also matches zero directories, so the segment after it is tried in the same pass + $explicit = $spanning ? ($index < $last ? $index + 1 : null) : $index; + + foreach (scandir($directory) ?: [] as $name) { + if ($name === '.' || $name === '..') { + continue; + } + + $path = rtrim($directory, '/') . '/' . $name; + + if ($spanning) { + if ($index === $last) { + $found[$path] = true; + } + + // ** never follows a symlinked directory - a link back up would never end + if (is_dir($path) && !is_link($path) && is_readable($path)) { + $queue[] = [$path, $index]; + } + } + + if ($explicit === null || !$matchers[$explicit]->matches($name)) { + continue; + } + + if ($explicit === $last) { + $found[$path] = true; + } elseif (is_dir($path) && is_readable($path)) { + $queue[] = [$path, $explicit + 1]; + } + } + } + + $paths = array_values(array_filter( + array_keys($found), + (new GlobPattern(rtrim($base, '/') . '/' . implode('/', $collapsed)))->matches(...), + )); + sort($paths, SORT_STRING); + + return $paths; + } +} diff --git a/src/lib/filesystem/src/Flow/Filesystem/Local/NativeLocalFilesystem.php b/src/lib/filesystem/src/Flow/Filesystem/Local/NativeLocalFilesystem.php index 233255ad51..9e3a1f55fd 100644 --- a/src/lib/filesystem/src/Flow/Filesystem/Local/NativeLocalFilesystem.php +++ b/src/lib/filesystem/src/Flow/Filesystem/Local/NativeLocalFilesystem.php @@ -5,7 +5,6 @@ namespace Flow\Filesystem\Local; use DateTimeImmutable; -use EmptyIterator; use Flow\Filesystem\DestinationStream; use Flow\Filesystem\Exception\InvalidArgumentException; use Flow\Filesystem\Exception\InvalidSchemeException; @@ -20,39 +19,26 @@ use Flow\Filesystem\Stream\NativeLocalDestinationStream; use Flow\Filesystem\Stream\NativeLocalSourceStream; use Generator; -use Iterator; -use RecursiveDirectoryIterator; -use RecursiveIteratorIterator; -use Webmozart\Glob\Glob; -use Webmozart\Glob\Iterator\GlobFilterIterator; -use Webmozart\Glob\Iterator\GlobIterator; +use function array_reverse; use function file_exists; use function filemtime; use function filesize; use function Flow\Filesystem\DSL\path; use function Flow\Filesystem\DSL\path_real; -use function Flow\Types\DSL\type_string; use function in_array; use function is_dir; use function is_file; use function is_link; use function mkdir; -use function preg_replace; use function rename; use function rmdir; use function scandir; -use function sort; use function sprintf; use function str_ends_with; -use function str_replace; use function sys_get_temp_dir; use function unlink; -/** - * This implementation is based on the native PHP filesystem functions documented here: https://www.php.net/manual/en/book.filesystem.php - * Additionally, in order to support glob pattern `\/**\/` for matching zero or more directories it's using webmozart/glob library. - */ final readonly class NativeLocalFilesystem implements Filesystem { public function __construct( @@ -104,15 +90,7 @@ public function list(Path $path, Filter $pathFilter = new OnlyFiles()): Generato return; } - $filePaths = []; - - foreach (new GlobIterator($path->glob()) as $filePath) { - $filePaths[] = type_string()->assert($filePath); - } - - sort($filePaths, SORT_STRING); - - foreach ($filePaths as $filePath) { + foreach ((new GlobWalker())->walk($path) as $filePath) { $status = self::statFor(path_real($filePath, $path->options()), $filePath); if ($pathFilter->accept($status)) { @@ -190,9 +168,7 @@ public function rm(Path $path): bool $deletedCount = 0; - foreach ($this->matchChildFirst($path->glob()) as $filePath) { - $filePath = type_string()->assert($filePath); - + foreach (array_reverse((new GlobWalker())->walk($path)) as $filePath) { if (is_dir($filePath)) { $this->rmdir($filePath); } else { @@ -219,15 +195,9 @@ public function status(Path $path): ?FileStatus return self::statFor($path, $path->path()); } - foreach (new GlobIterator($path->glob()) as $filePath) { - $filePath = type_string()->assert($filePath); + $filePath = (new GlobWalker())->walk($path)[0] ?? null; - if (file_exists($filePath)) { - return self::statFor(path($filePath, $path->options()), $filePath); - } - } - - return null; + return $filePath === null ? null : self::statFor(path($filePath, $path->options()), $filePath); } public function supports(Path $path): bool @@ -261,41 +231,6 @@ public function writeTo(Path $path): DestinationStream return NativeLocalDestinationStream::openBlank($path); } - /** - * Lazy iterator over glob matches in CHILD_FIRST order so callers can safely delete each match - * without confusing webmozart/glob's internal RecursiveIteratorIterator (which descends with SELF_FIRST). - */ - /** - * @return \Iterator - */ - private function matchChildFirst(string $glob): Iterator - { - $glob = self::canonicalizePath($glob); - $basePath = Glob::getBasePath($glob); - - if (!is_dir($basePath)) { - return new EmptyIterator(); - } - - $recursive = new RecursiveIteratorIterator( - new RecursiveDirectoryIterator( - $basePath, - RecursiveDirectoryIterator::CURRENT_AS_PATHNAME | RecursiveDirectoryIterator::SKIP_DOTS, - ), - RecursiveIteratorIterator::CHILD_FIRST, - ); - - return new GlobFilterIterator( - $glob, - (static function () use ($recursive) { - foreach ($recursive as $path) { - yield self::canonicalizePath(type_string()->assert($path)); - } - })(), - GlobFilterIterator::FILTER_VALUE, - ); - } - private function rmdir(string $dirPath): void { if (is_link($dirPath)) { @@ -341,11 +276,6 @@ private function rmdir(string $dirPath): void rmdir($dirPath); } - private static function canonicalizePath(string $path): string - { - return type_string()->cast(preg_replace('#/+#', '/', str_replace('\\', '/', $path))); - } - private static function statFor(Path $path, string $absolutePath): FileStatus { $isFile = is_file($absolutePath); diff --git a/src/lib/filesystem/src/Flow/Filesystem/Path/GlobPattern.php b/src/lib/filesystem/src/Flow/Filesystem/Path/GlobPattern.php new file mode 100644 index 0000000000..096f002274 --- /dev/null +++ b/src/lib/filesystem/src/Flow/Filesystem/Path/GlobPattern.php @@ -0,0 +1,123 @@ + '[^/' . $class . ']', + $class === '' => '(?!)', + default => '(?!/)[' . $class . ']', + }; + $i = $close; + + continue; + } + } + + $regex .= preg_quote($char, '~'); + } + + $this->byteRegex = '~^' . $regex . '$~'; + $this->regex = $unicode ? $this->byteRegex . 'u' : $this->byteRegex; + } + + public function matches(string $path): bool + { + $matched = preg_match($this->regex, $path); + + // a path that is not valid UTF-8 can only be matched byte by byte + return ($matched === false ? preg_match($this->byteRegex, $path) : $matched) === 1; + } +} diff --git a/src/lib/filesystem/src/Flow/Filesystem/Path/UnixPath.php b/src/lib/filesystem/src/Flow/Filesystem/Path/UnixPath.php index 137326f350..a408c7ee94 100644 --- a/src/lib/filesystem/src/Flow/Filesystem/Path/UnixPath.php +++ b/src/lib/filesystem/src/Flow/Filesystem/Path/UnixPath.php @@ -52,6 +52,8 @@ private const string PLACEHOLDER_SENTINEL = "\x01"; + private ?GlobPattern $glob; + private Options $options; private string $path; @@ -76,6 +78,8 @@ public function __construct(string $uri, array|Options $options = []) } $this->path = $this->normalizePath($this->resolveHomePath($path)); + // listings match every entry against one pattern, so it is compiled once + $this->glob = $this->isPattern() ? new GlobPattern($this->path) : null; } /** @@ -330,7 +334,7 @@ public function matches(self $path): bool return false; } - return $this->fnmatch($this->path, $path->path); + return $this->glob?->matches($path->path) ?? false; } public function options(): Options @@ -529,27 +533,6 @@ public function withOptions(Options $options): self return new self($this->uri(), $options); } - private function fnmatch(string $pattern, string $filename, int $flags = 0): bool - { - if ($flags & 4) { - if ($filename[0] === '.' && $pattern[0] !== '.') { - return false; - } - } - - $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); - } - private function isAbsolutePath(string $path): bool { return str_starts_with($path, '/'); diff --git a/src/lib/filesystem/src/Flow/Filesystem/Path/WindowsPath.php b/src/lib/filesystem/src/Flow/Filesystem/Path/WindowsPath.php index 14fc1e6ffc..7e603e5aee 100644 --- a/src/lib/filesystem/src/Flow/Filesystem/Path/WindowsPath.php +++ b/src/lib/filesystem/src/Flow/Filesystem/Path/WindowsPath.php @@ -49,6 +49,8 @@ private const string PLACEHOLDER_SENTINEL = "\x01"; + private ?GlobPattern $glob; + private Options $options; private string $path; @@ -73,6 +75,8 @@ public function __construct(string $uri, array|Options $options = []) } $this->path = $this->normalizePath($this->resolveHomePath($path)); + // listings match every entry against one pattern, so it is compiled once + $this->glob = $this->isPattern() ? new GlobPattern($this->path) : null; } /** @@ -330,7 +334,7 @@ public function matches(self $path): bool return false; } - return $this->fnmatch($this->path, $path->path); + return $this->glob?->matches($path->path) ?? false; } public function options(): Options @@ -561,27 +565,6 @@ public function withOptions(Options $options): self return new self($this->uri(), $options); } - private function fnmatch(string $pattern, string $filename, int $flags = 0): bool - { - if ($flags & 4) { - if ($filename[0] === '.' && $pattern[0] !== '.') { - return false; - } - } - - $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); - } - private function isAbsolutePath(string $path): bool { return ( diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Context/GlobMatrixContext.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Context/GlobMatrixContext.php new file mode 100644 index 0000000000..53aa0f4453 --- /dev/null +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Context/GlobMatrixContext.php @@ -0,0 +1,58 @@ + + */ + public static function files(): array + { + return [ + 'data/.dir/x.parquet', + 'data/.hidden.parquet', + 'data/date=2026-09-01/one.parquet', + 'data/flat.parquet', + 'data/id=1/date=2026-09-01/two.parquet', + ]; + } + + /** + * @return Generator}> + */ + public static function patterns(): Generator + { + $all = self::files(); + + yield 'star stays in one directory' => ['data/*.parquet', ['data/.hidden.parquet', 'data/flat.parquet']]; + yield 'double star inside a name is a star' => [ + 'data/**.parquet', + ['data/.hidden.parquet', 'data/flat.parquet'], + ]; + yield 'double star segment spans zero or more directories' => ['data/**/*.parquet', $all]; + yield 'trailing double star is everything below' => ['data/**', $all]; + yield 'one directory level' => [ + 'data/*/*.parquet', + ['data/.dir/x.parquet', 'data/date=2026-09-01/one.parquet'], + ]; + yield 'negated class' => [ + 'data/**/[!f]*.parquet', + [ + 'data/.dir/x.parquet', + 'data/.hidden.parquet', + 'data/date=2026-09-01/one.parquet', + 'data/id=1/date=2026-09-01/two.parquet', + ], + ]; + yield 'caret is a literal in a class' => ['data/[^f]*.parquet', ['data/flat.parquet']]; + yield 'range' => ['data/[a-g]*.parquet', ['data/flat.parquet']]; + yield 'question mark never crosses a directory' => ['data/date=2026-09-01?one.parquet', []]; + yield 'partition placeholder' => ['data/date={date}/one.parquet', ['data/date=2026-09-01/one.parquet']]; + yield 'double star glued to a name stays in one directory' => ['data/x**/*.parquet', []]; + } +} diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FailingCloseDestinationStream.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FailingCloseDestinationStream.php new file mode 100644 index 0000000000..b5bac275ff --- /dev/null +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FailingCloseDestinationStream.php @@ -0,0 +1,53 @@ +inner->append($data); + + return $this; + } + + public function close(): void + { + // the handle is released first - a real failed close leaves nothing to retry + $this->closed = true; + $this->inner->close(); + + throw new RuntimeException(sprintf('Closing "%s" failed', $this->inner->path()->uri())); + } + + public function fromResource($resource): self + { + $this->inner->fromResource($resource); + + return $this; + } + + public function isOpen(): bool + { + return !$this->closed && $this->inner->isOpen(); + } + + public function path(): Path + { + return $this->inner->path(); + } +} diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FailingCloseFilesystem.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FailingCloseFilesystem.php new file mode 100644 index 0000000000..5654d649c2 --- /dev/null +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FailingCloseFilesystem.php @@ -0,0 +1,86 @@ +inner->appendTo($path); + $index = $this->opened++; + + return $index >= $this->healthyStreams && $index < ($this->healthyStreams + $this->failingStreams) + ? new FailingCloseDestinationStream($stream) + : $stream; + } + + public function getSystemTmpDir(): Path + { + return $this->inner->getSystemTmpDir(); + } + + public function list(Path $path, Filter $pathFilter = new KeepAll()): Generator + { + yield from $this->inner->list($path, $pathFilter); + } + + public function mount(): Mount + { + return $this->inner->mount(); + } + + public function mv(Path $from, Path $to): bool + { + return $this->inner->mv($from, $to); + } + + public function readFrom(Path $path): SourceStream + { + return $this->inner->readFrom($path); + } + + public function rm(Path $path): bool + { + return $this->inner->rm($path); + } + + public function status(Path $path): ?FileStatus + { + return $this->inner->status($path); + } + + public function supports(Path $path): bool + { + return $this->inner->supports($path); + } + + public function writeTo(Path $path): DestinationStream + { + $stream = $this->inner->writeTo($path); + $index = $this->opened++; + + return $index >= $this->healthyStreams && $index < ($this->healthyStreams + $this->failingStreams) + ? new FailingCloseDestinationStream($stream) + : $stream; + } +} diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FakeNativeLocalFilesystem.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FakeNativeLocalFilesystem.php index f3103c62ba..ee5c147569 100644 --- a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FakeNativeLocalFilesystem.php +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Double/FakeNativeLocalFilesystem.php @@ -10,6 +10,7 @@ use Flow\Filesystem\Exception\RuntimeException; use Flow\Filesystem\FileStatus; use Flow\Filesystem\Filesystem; +use Flow\Filesystem\Local\GlobWalker; use Flow\Filesystem\Mount; use Flow\Filesystem\Path; use Flow\Filesystem\Path\Filter; @@ -18,12 +19,11 @@ use Flow\Filesystem\Stream\NativeLocalDestinationStream; use Flow\Filesystem\Stream\NativeLocalSourceStream; use Generator; -use Webmozart\Glob\Iterator\GlobIterator; +use function array_reverse; use function file_exists; use function Flow\Filesystem\DSL\path; use function Flow\Filesystem\DSL\path_real; -use function Flow\Types\DSL\type_string; use function in_array; use function is_dir; use function is_file; @@ -90,8 +90,7 @@ public function list(Path $path, Filter $pathFilter = new OnlyFiles()): Generato return; } - foreach (new GlobIterator($path->path()) as $filePath) { - $filePath = type_string()->assert($filePath); + foreach ((new GlobWalker())->walk($path) as $filePath) { $status = new FileStatus(path_real($filePath, $path->options()), is_file($filePath)); if ($pathFilter->accept($status)) { @@ -169,9 +168,7 @@ public function rm(Path $path): bool $deletedCount = 0; - foreach (new GlobIterator($path->path()) as $filePath) { - $filePath = type_string()->assert($filePath); - + foreach (array_reverse((new GlobWalker())->walk($path)) as $filePath) { if (is_dir($filePath)) { $this->rmdir($filePath); } else { @@ -194,15 +191,9 @@ public function status(Path $path): ?FileStatus return new FileStatus($path, is_file($path->path())); } - foreach (new GlobIterator($path->path()) as $filePath) { - $filePath = type_string()->assert($filePath); - - if (file_exists($filePath)) { - return new FileStatus(path($filePath, $path->options()), true); - } - } + $filePath = (new GlobWalker())->walk($path)[0] ?? null; - return null; + return $filePath === null ? null : new FileStatus(path($filePath, $path->options()), true); } public function supports(Path $path): bool diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/GlobListingAgreementTest.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/GlobListingAgreementTest.php new file mode 100644 index 0000000000..79252c4798 --- /dev/null +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/GlobListingAgreementTest.php @@ -0,0 +1,89 @@ +writeTo(path(__DIR__ . '/var/glob_agreement/' . $file)) + ->append($file) + ->close(); + } + } + + protected function tearDown(): void + { + native_local_filesystem()->rm(path(__DIR__ . '/var/glob_agreement')); + } + + /** + * @param list $expected + */ + #[DataProvider('patterns')] + public function test_memory_filesystem_lists_the_matrix(string $pattern, array $expected): void + { + $memory = memory_filesystem(); + + foreach (GlobMatrixContext::files() as $file) { + $memory + ->writeTo(path('memory://' . $file)) + ->append($file) + ->close(); + } + + $listed = array_map( + static fn(FileStatus $status): string => substr($status->path->uri(), strlen('memory://')), + iterator_to_array($memory->list(path('memory://' . $pattern), new OnlyFiles()), false), + ); + sort($listed); + + static::assertSame($expected, $listed); + } + + /** + * @param list $expected + */ + #[DataProvider('patterns')] + public function test_native_local_filesystem_lists_the_matrix(string $pattern, array $expected): void + { + $listed = array_map( + static fn(FileStatus $status): string => substr( + $status->path->path(), + strlen(__DIR__ . '/var/glob_agreement/'), + ), + iterator_to_array( + native_local_filesystem()->list(path(__DIR__ . '/var/glob_agreement/' . $pattern), new OnlyFiles()), + false, + ), + ); + sort($listed); + + static::assertSame($expected, $listed); + } +} diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/Local/GlobWalkerTest.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/Local/GlobWalkerTest.php new file mode 100644 index 0000000000..ea43fbb254 --- /dev/null +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Integration/Local/GlobWalkerTest.php @@ -0,0 +1,107 @@ +writeTo(path(__DIR__ . '/var/glob_walker/data/a.csv')) + ->append('a') + ->close(); + native_local_filesystem() + ->writeTo(path(__DIR__ . '/var/glob_walker/real/sub/x.csv')) + ->append('x') + ->close(); + } + + protected function tearDown(): void + { + native_local_filesystem()->rm(path(__DIR__ . '/var/glob_walker')); + } + + public function test_a_missing_base_directory_yields_nothing(): void + { + static::assertSame([], (new GlobWalker())->walk(path(__DIR__ . '/var/glob_walker/missing/*.csv'))); + } + + public function test_directories_matching_the_last_segment_are_returned(): void + { + static::assertSame( + [__DIR__ . '/var/glob_walker/data/a.csv', __DIR__ . '/var/glob_walker/data/sub'], + (new GlobWalker())->walk(path(__DIR__ . '/var/glob_walker/data/*')), + ); + } + + public function test_double_star_does_not_descend_into_a_symlinked_directory(): void + { + symlink(__DIR__ . '/var/glob_walker/real', __DIR__ . '/var/glob_walker/data/linked'); + symlink(__DIR__ . '/var/glob_walker/data', __DIR__ . '/var/glob_walker/data/loop'); + + static::assertSame([], (new GlobWalker())->walk(path(__DIR__ . '/var/glob_walker/data/**/x.csv'))); + } + + public function test_explicit_segments_follow_a_symlinked_directory(): void + { + symlink(__DIR__ . '/var/glob_walker/real', __DIR__ . '/var/glob_walker/data/linked'); + + static::assertSame( + [__DIR__ . '/var/glob_walker/data/linked/sub/x.csv'], + (new GlobWalker())->walk(path(__DIR__ . '/var/glob_walker/data/*/sub/x.csv')), + ); + } + + public function test_paths_are_sorted(): void + { + native_local_filesystem() + ->writeTo(path(__DIR__ . '/var/glob_walker/sorted/0.csv')) + ->append('0') + ->close(); + native_local_filesystem() + ->writeTo(path(__DIR__ . '/var/glob_walker/sorted/a/1.csv')) + ->append('1') + ->close(); + native_local_filesystem() + ->writeTo(path(__DIR__ . '/var/glob_walker/sorted/b/1.csv')) + ->append('1') + ->close(); + + static::assertSame( + [ + __DIR__ . '/var/glob_walker/sorted/0.csv', + __DIR__ . '/var/glob_walker/sorted/a/1.csv', + __DIR__ . '/var/glob_walker/sorted/b/1.csv', + ], + (new GlobWalker())->walk(path(__DIR__ . '/var/glob_walker/sorted/**/*.csv')), + ); + } + + public function test_consecutive_double_stars_match_like_one(): void + { + static::assertSame( + [__DIR__ . '/var/glob_walker/real/sub/x.csv'], + (new GlobWalker())->walk(path(__DIR__ . '/var/glob_walker/real/**/**/*.csv')), + ); + } + + public function test_redundant_slashes_are_collapsed(): void + { + static::assertSame( + [__DIR__ . '/var/glob_walker/real/sub/x.csv'], + (new GlobWalker())->walk(path(__DIR__ . '/var/glob_walker/real/*//x.csv')), + ); + } +} diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/GlobPatternTest.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/GlobPatternTest.php new file mode 100644 index 0000000000..be08ab5535 --- /dev/null +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/Path/GlobPatternTest.php @@ -0,0 +1,77 @@ +matches($path)); + } +} 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 d5a0039a5f..74c86fa814 100644 --- a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PathTest.php +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PathTest.php @@ -61,6 +61,10 @@ public static function paths_pattern_matching(): Generator yield ['/nested/folder/[a]*/file.csv', '/nested/folder/ab/file.csv', true]; yield ['/nested/folder/**/file.csv', '/nested/folder/any/nested/file.csv', true]; yield ['/nested/folder/**/fil?.csv', '/nested/folder/any/nested/file.csv', true]; + yield ['/nested/folder/**/file.csv', '/nested/folder/file.csv', true]; + yield ['/nested/folder/**.csv', '/nested/folder/any/file.csv', false]; + yield ['/nested/folder/[!a]*.csv', '/nested/folder/b.csv', true]; + yield ['/nested/folder/[a-c].csv', '/nested/folder/b.csv', true]; } /** diff --git a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PathTestCase.php b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PathTestCase.php index 04501081c0..a29439520a 100644 --- a/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PathTestCase.php +++ b/src/lib/filesystem/tests/Flow/Filesystem/Tests/Unit/PathTestCase.php @@ -56,6 +56,10 @@ public static function paths_pattern_matching(): Generator yield ['/nested/folder/[a]*/file.csv', '/nested/folder/ab/file.csv', true]; yield ['/nested/folder/**/file.csv', '/nested/folder/any/nested/file.csv', true]; yield ['/nested/folder/**/fil?.csv', '/nested/folder/any/nested/file.csv', true]; + yield ['/nested/folder/**/file.csv', '/nested/folder/file.csv', true]; + yield ['/nested/folder/**.csv', '/nested/folder/any/file.csv', false]; + yield ['/nested/folder/[!a]*.csv', '/nested/folder/b.csv', true]; + yield ['/nested/folder/[a-c].csv', '/nested/folder/b.csv', true]; } /** diff --git a/src/lib/parquet/src/Flow/Parquet/Engine/AdaptiveParquetEngine.php b/src/lib/parquet/src/Flow/Parquet/Engine/AdaptiveParquetEngine.php index f3cf11ffed..5ddf84a895 100644 --- a/src/lib/parquet/src/Flow/Parquet/Engine/AdaptiveParquetEngine.php +++ b/src/lib/parquet/src/Flow/Parquet/Engine/AdaptiveParquetEngine.php @@ -11,6 +11,7 @@ use Flow\Parquet\ParquetEngine; use Flow\Parquet\ParquetFile\Compressions; use Flow\Parquet\ParquetFile\Schema; +use Flow\Parquet\ParquetFileWriter; use Generator; use function extension_loaded; @@ -26,18 +27,13 @@ public function __construct(ByteOrder $byteOrder = ByteOrder::LITTLE_ENDIAN, Opt : new PhpParquetEngine($byteOrder, $options); } - public function closeWrite(): void - { - $this->delegate->closeWrite(); - } - public function openForWrite( DestinationStream $stream, Schema $schema, Compressions $compression, Options $options, - ): void { - $this->delegate->openForWrite($stream, $schema, $compression, $options); + ): ParquetFileWriter { + return $this->delegate->openForWrite($stream, $schema, $compression, $options); } public function readValues( @@ -50,16 +46,6 @@ public function readValues( return $this->delegate->readValues($stream, $schema, $columns, $limit, $offset); } - public function writeBatch(iterable $rows): void - { - $this->delegate->writeBatch($rows); - } - - public function writeRow(array $row): void - { - $this->delegate->writeRow($row); - } - public function writeRows( DestinationStream $stream, Schema $schema, diff --git a/src/lib/parquet/src/Flow/Parquet/Engine/ArrowParquetEngine.php b/src/lib/parquet/src/Flow/Parquet/Engine/ArrowParquetEngine.php index b304f4563f..535ae2701f 100644 --- a/src/lib/parquet/src/Flow/Parquet/Engine/ArrowParquetEngine.php +++ b/src/lib/parquet/src/Flow/Parquet/Engine/ArrowParquetEngine.php @@ -18,6 +18,7 @@ use Flow\Parquet\ParquetEngine; use Flow\Parquet\ParquetFile\Compressions; use Flow\Parquet\ParquetFile\Schema; +use Flow\Parquet\ParquetFileWriter; use Generator; use function array_column; @@ -27,20 +28,6 @@ final class ArrowParquetEngine implements ParquetEngine { - private ?Writer $arrowWriter = null; - - /** - * @var array> - */ - private array $batch = []; - - private int $batchSize = 0; - - /** - * @var array - */ - private array $colNames = []; - public function __construct( private readonly Options $options = new Options(), ) { @@ -63,40 +50,31 @@ public static function mapCompression(Compressions $compression): string }; } - public function closeWrite(): void - { - if ($this->arrowWriter === null) { - throw new RuntimeException('Writer is not open'); - } - - if ($this->batchSize > 0) { - $this->arrowWriter->writeBatch($this->batch); - } - - $this->arrowWriter->close(); - $this->arrowWriter = null; - $this->batch = []; - $this->batchSize = 0; - $this->colNames = []; - } - public function openForWrite( DestinationStream $stream, Schema $schema, Compressions $compression, Options $options, - ): void { - $adapter = new DestinationStreamAdapter($stream); + ): ParquetFileWriter { $extensionSchema = SchemaConverter::toExtension($schema); - $compressionStr = self::mapCompression($compression); - $extensionOptions = OptionsConverter::toExtension($options); - $this->arrowWriter = new Writer($adapter, $extensionSchema, $compressionStr, $extensionOptions); + /** @var list $columnNames */ + $columnNames = array_column($extensionSchema, 'name'); + // the arrow Writer takes the stream by reference, so it has to be a variable + $adapter = new DestinationStreamAdapter($stream); - /** @var array $colNames */ - $colNames = array_column($extensionSchema, 'name'); - $this->colNames = $colNames; - $this->resetBatch(); + return new ArrowParquetFileWriter( + new Writer( + $adapter, + $extensionSchema, + self::mapCompression($compression), + OptionsConverter::toExtension($options), + ), + $stream, + $columnNames, + // write batching is an engine option, not a per-file one + $this->options->getInt(Option::ARROW_WRITE_BATCH_SIZE), + ); } /** @@ -154,31 +132,6 @@ public function readValues( } } - public function writeBatch(iterable $rows): void - { - foreach ($rows as $row) { - $this->writeRow($row); - } - } - - public function writeRow(array $row): void - { - if ($this->arrowWriter === null) { - throw new RuntimeException('Writer is not open'); - } - - foreach ($this->colNames as $name) { - $this->batch[$name][] = $row[$name] ?? null; - } - - $this->batchSize++; - - if ($this->batchSize >= $this->options->getInt(Option::ARROW_WRITE_BATCH_SIZE)) { - $this->arrowWriter->writeBatch($this->batch); - $this->resetBatch(); - } - } - public function writeRows( DestinationStream $stream, Schema $schema, @@ -186,23 +139,12 @@ public function writeRows( Options $options, iterable $rows, ): void { - $this->openForWrite($stream, $schema, $compression, $options); + $file = $this->openForWrite($stream, $schema, $compression, $options); try { - $this->writeBatch($rows); + $file->writeBatch($rows); } finally { - $this->closeWrite(); + $file->close(); } } - - private function resetBatch(): void - { - $this->batch = []; - - foreach ($this->colNames as $name) { - $this->batch[$name] = []; - } - - $this->batchSize = 0; - } } diff --git a/src/lib/parquet/src/Flow/Parquet/Engine/ArrowParquetFileWriter.php b/src/lib/parquet/src/Flow/Parquet/Engine/ArrowParquetFileWriter.php new file mode 100644 index 0000000000..c3e3fb35dc --- /dev/null +++ b/src/lib/parquet/src/Flow/Parquet/Engine/ArrowParquetFileWriter.php @@ -0,0 +1,77 @@ +> + */ + private array $batch; + + private int $batchSize = 0; + + private ?Writer $arrowWriter; + + /** + * @param list $columnNames + */ + public function __construct( + Writer $arrowWriter, + private readonly DestinationStream $stream, + private readonly array $columnNames, + private readonly int $batchSizeLimit, + ) { + $this->arrowWriter = $arrowWriter; + $this->batch = array_fill_keys($this->columnNames, []); + } + + public function close(): void + { + $writer = $this->arrowWriter ?? throw new RuntimeException('Writer is not open'); + + try { + if ($this->batchSize > 0) { + $writer->writeBatch($this->batch); + } + + $writer->close(); + $this->stream->close(); + } finally { + $this->arrowWriter = null; + } + } + + public function writeBatch(iterable $rows): void + { + foreach ($rows as $row) { + $this->writeRow($row); + } + } + + public function writeRow(array $row): void + { + $writer = $this->arrowWriter ?? throw new RuntimeException('Writer is not open'); + + foreach ($this->columnNames as $name) { + $this->batch[$name][] = $row[$name] ?? null; + } + + $this->batchSize++; + + if ($this->batchSize >= $this->batchSizeLimit) { + $writer->writeBatch($this->batch); + $this->batch = array_fill_keys($this->columnNames, []); + $this->batchSize = 0; + } + } +} diff --git a/src/lib/parquet/src/Flow/Parquet/Engine/PhpParquetEngine.php b/src/lib/parquet/src/Flow/Parquet/Engine/PhpParquetEngine.php index bb8dc94568..9d0d2eb58b 100644 --- a/src/lib/parquet/src/Flow/Parquet/Engine/PhpParquetEngine.php +++ b/src/lib/parquet/src/Flow/Parquet/Engine/PhpParquetEngine.php @@ -4,18 +4,13 @@ namespace Flow\Parquet\Engine; -use Composer\InstalledVersions; use Flow\Filesystem\DestinationStream; use Flow\Filesystem\SourceStream; use Flow\Parquet\Binary\ByteOrder; use Flow\Parquet\Dremel\ColumnData\ReadFlatColumnValues; use Flow\Parquet\Dremel\DremelAssembler; -use Flow\Parquet\Dremel\DremelShredder; use Flow\Parquet\Dremel\ReadColumnData; -use Flow\Parquet\Dremel\Validator\ColumnDataValidator; -use Flow\Parquet\Dremel\Validator\DisabledValidator; use Flow\Parquet\Exception\InvalidArgumentException; -use Flow\Parquet\Exception\RuntimeException; use Flow\Parquet\Option; use Flow\Parquet\Options; use Flow\Parquet\ParquetEngine; @@ -23,118 +18,45 @@ use Flow\Parquet\ParquetFile\Compressions; use Flow\Parquet\ParquetFile\Data\DataConverter; use Flow\Parquet\ParquetFile\Metadata; -use Flow\Parquet\ParquetFile\RowGroups; use Flow\Parquet\ParquetFile\Schema; use Flow\Parquet\ParquetFile\Schema\Column; use Flow\Parquet\ParquetFile\Schema\FlatColumn; use Flow\Parquet\ParquetFile\Schema\NestedColumn; +use Flow\Parquet\ParquetFileWriter; use Flow\Parquet\Reader\ColumnChunkReader; use Flow\Parquet\Reader\PageReader; use Flow\Parquet\Thrift\CompactProtocol; use Flow\Parquet\Thrift\MemoryBuffer; -use Flow\Parquet\Thrift\PhpFileStream; use Flow\Parquet\ThriftModel\FileMetaData; -use Flow\Parquet\Writer\RowGroupBuilder; use Generator; use MultipleIterator; use function array_map; use function array_push; use function count; -use function fclose; -use function fopen; use function is_array; use function iterator_to_array; -use function pack; -use function stream_get_contents; -use function strlen; use function unpack; final class PhpParquetEngine implements ParquetEngine { - private int $fileOffset = 0; - - private ?Metadata $metadata = null; - - private ?RowGroupBuilder $rowGroupBuilder = null; - - private ?DestinationStream $writeStream = null; - public function __construct( private readonly ByteOrder $byteOrder = ByteOrder::LITTLE_ENDIAN, private readonly Options $options = new Options(), ) {} - public function closeWrite(): void - { - if ($this->writeStream === null) { - throw new RuntimeException('Writer is not open'); - } - - if (!$this->activeRowGroupBuilder()->isEmpty()) { - $rowGroupContainer = $this->activeRowGroupBuilder()->flush($this->fileOffset); - $this->activeStream()->append($rowGroupContainer->binaryBuffer); - $this->activeMetadata()->rowGroups()->add($rowGroupContainer->rowGroup); - $this->fileOffset += strlen($rowGroupContainer->binaryBuffer); - } - - $this->rowGroupBuilder = null; - - $metadataHandle = fopen('php://temp/maxmemory:' . (5 * 1024 * 1024), 'rb+'); - - if ($metadataHandle === false) { - throw new RuntimeException('Cannot open temporary stream'); - } - - $this - ->activeMetadata() - ->toThrift() - ->write(new CompactProtocol(new PhpFileStream($metadataHandle))); - $metadataBytes = stream_get_contents($metadataHandle, offset: 0); - - if ($metadataBytes === false) { - throw new RuntimeException('Cannot read metadata from temporary stream'); - } - - $this->activeStream()->append($metadataBytes); - fclose($metadataHandle); - - $this->activeStream()->append(pack('l', strlen($metadataBytes))); - $this->activeStream()->append(ParquetFile::PARQUET_MAGIC_NUMBER); - - $this->activeStream()->close(); - - $this->writeStream = null; - $this->metadata = null; - $this->fileOffset = 0; - } - public function openForWrite( DestinationStream $stream, Schema $schema, Compressions $compression, Options $options, - ): void { - $this->writeStream = $stream; - $this->activeStream()->append(ParquetFile::PARQUET_MAGIC_NUMBER); - $this->fileOffset = strlen(ParquetFile::PARQUET_MAGIC_NUMBER); - - $this->metadata = new Metadata( - $schema, - new RowGroups([]), - 0, - $options->getInt(Option::WRITER_VERSION), - 'flow-php parquet version ' . InstalledVersions::getRootPackage()['pretty_version'], - ); - - $dataConverter = DataConverter::initialize($options); - $validator = $options->getBool(Option::VALIDATE_DATA) ? new ColumnDataValidator() : new DisabledValidator(); - - $this->rowGroupBuilder = new RowGroupBuilder( + ): ParquetFileWriter { + return new PhpParquetFileWriter( + $stream, $schema, $compression, $options, - new DremelShredder($validator, $dataConverter), + $this->options->getInt(Option::ROW_GROUP_SIZE_CHECK_INTERVAL), ); } @@ -217,35 +139,6 @@ public function readValues( } } - public function writeBatch(iterable $rows): void - { - if (is_array($rows)) { - $this->activeRowGroupBuilder()->addRows($rows); - - return; - } - - foreach ($rows as $row) { - $this->writeRow($row); - } - } - - public function writeRow(array $row): void - { - $this->activeRowGroupBuilder()->addRow($row); - $interval = $this->options->getInt(Option::ROW_GROUP_SIZE_CHECK_INTERVAL); - - if ( - ($this->activeRowGroupBuilder()->rowsCount() % $interval) === 0 - && $this->activeRowGroupBuilder()->isFull() - ) { - $rowGroupContainer = $this->activeRowGroupBuilder()->flush($this->fileOffset); - $this->activeStream()->append($rowGroupContainer->binaryBuffer); - $this->activeMetadata()->rowGroups()->add($rowGroupContainer->rowGroup); - $this->fileOffset += strlen($rowGroupContainer->binaryBuffer); - } - } - public function writeRows( DestinationStream $stream, Schema $schema, @@ -253,92 +146,20 @@ public function writeRows( Options $options, iterable $rows, ): void { - $stream->append(ParquetFile::PARQUET_MAGIC_NUMBER); - $fileOffset = strlen(ParquetFile::PARQUET_MAGIC_NUMBER); - - $metadata = new Metadata( - $schema, - new RowGroups([]), - 0, - $options->getInt(Option::WRITER_VERSION), - 'flow-php parquet version ' . InstalledVersions::getRootPackage()['pretty_version'], - ); - - $dataConverter = DataConverter::initialize($options); - $validator = $options->getBool(Option::VALIDATE_DATA) ? new ColumnDataValidator() : new DisabledValidator(); - - $rowGroupBuilder = new RowGroupBuilder( + // writeRows() reads the row group check interval from its own $options, openForWrite() from the engine's + $file = new PhpParquetFileWriter( + $stream, $schema, $compression, $options, - new DremelShredder($validator, $dataConverter), + $options->getInt(Option::ROW_GROUP_SIZE_CHECK_INTERVAL), ); foreach ($rows as $row) { - $rowGroupBuilder->addRow($row); - $interval = $options->getInt(Option::ROW_GROUP_SIZE_CHECK_INTERVAL); - - if (($rowGroupBuilder->rowsCount() % $interval) === 0 && $rowGroupBuilder->isFull()) { - $rowGroupContainer = $rowGroupBuilder->flush($fileOffset); - $stream->append($rowGroupContainer->binaryBuffer); - $metadata->rowGroups()->add($rowGroupContainer->rowGroup); - $fileOffset += strlen($rowGroupContainer->binaryBuffer); - } - } - - if (!$rowGroupBuilder->isEmpty()) { - $rowGroupContainer = $rowGroupBuilder->flush($fileOffset); - $stream->append($rowGroupContainer->binaryBuffer); - $metadata->rowGroups()->add($rowGroupContainer->rowGroup); - } - - $metadataHandle = fopen('php://temp/maxmemory:' . (5 * 1024 * 1024), 'rb+'); - - if ($metadataHandle === false) { - throw new RuntimeException('Cannot open temporary stream'); - } - - $metadata->toThrift()->write(new CompactProtocol(new PhpFileStream($metadataHandle))); - $metadataBytes = stream_get_contents($metadataHandle, offset: 0); - - if ($metadataBytes === false) { - throw new RuntimeException('Cannot read metadata from temporary stream'); - } - - $stream->append($metadataBytes); - fclose($metadataHandle); - - $stream->append(pack('l', strlen($metadataBytes))); - $stream->append(ParquetFile::PARQUET_MAGIC_NUMBER); - - $stream->close(); - } - - private function activeMetadata(): Metadata - { - if ($this->metadata === null) { - throw new RuntimeException('Writer is not open'); - } - - return $this->metadata; - } - - private function activeRowGroupBuilder(): RowGroupBuilder - { - if ($this->rowGroupBuilder === null) { - throw new RuntimeException('Writer is not open'); - } - - return $this->rowGroupBuilder; - } - - private function activeStream(): DestinationStream - { - if ($this->writeStream === null) { - throw new RuntimeException('Writer is not open'); + $file->writeRow($row); } - return $this->writeStream; + $file->close(); } private function readColumn( diff --git a/src/lib/parquet/src/Flow/Parquet/Engine/PhpParquetFileWriter.php b/src/lib/parquet/src/Flow/Parquet/Engine/PhpParquetFileWriter.php new file mode 100644 index 0000000000..88ecc82f02 --- /dev/null +++ b/src/lib/parquet/src/Flow/Parquet/Engine/PhpParquetFileWriter.php @@ -0,0 +1,149 @@ +append(ParquetFile::PARQUET_MAGIC_NUMBER); + $this->stream = $stream; + $this->fileOffset = strlen(ParquetFile::PARQUET_MAGIC_NUMBER); + $this->metadata = new Metadata( + $schema, + new RowGroups([]), + 0, + $options->getInt(Option::WRITER_VERSION), + 'flow-php parquet version ' . InstalledVersions::getRootPackage()['pretty_version'], + ); + $this->rowGroupBuilder = new RowGroupBuilder( + $schema, + $compression, + $options, + new DremelShredder( + $options->getBool(Option::VALIDATE_DATA) ? new ColumnDataValidator() : new DisabledValidator(), + DataConverter::initialize($options), + ), + ); + } + + public function close(): void + { + $stream = $this->stream ?? throw new RuntimeException('Writer is not open'); + + try { + if (!$this->rowGroupBuilder->isEmpty()) { + $this->flushRowGroup($stream); + } + + $metadataHandle = fopen('php://temp/maxmemory:' . (5 * 1024 * 1024), 'rb+'); + + if ($metadataHandle === false) { + throw new RuntimeException('Cannot open temporary stream'); + } + + $this->metadata->toThrift()->write(new CompactProtocol(new PhpFileStream($metadataHandle))); + $metadataBytes = stream_get_contents($metadataHandle, offset: 0); + + if ($metadataBytes === false) { + throw new RuntimeException('Cannot read metadata from temporary stream'); + } + + $stream->append($metadataBytes); + fclose($metadataHandle); + + $stream->append(pack('l', strlen($metadataBytes))); + $stream->append(ParquetFile::PARQUET_MAGIC_NUMBER); + $stream->close(); + } finally { + $this->stream = null; + } + } + + public function writeBatch(iterable $rows): void + { + $stream = $this->stream ?? throw new RuntimeException('Writer is not open'); + + if (!is_array($rows)) { + foreach ($rows as $row) { + $this->writeRow($row); + } + + return; + } + + /** @var int<1, max> $interval */ + $interval = $this->rowGroupSizeCheckInterval; + + foreach (array_chunk($rows, $interval) as $chunk) { + $this->rowGroupBuilder->addRows($chunk); + + if ($this->rowGroupBuilder->isFull()) { + $this->flushRowGroup($stream); + } + } + } + + public function writeRow(array $row): void + { + $stream = $this->stream ?? throw new RuntimeException('Writer is not open'); + $this->rowGroupBuilder->addRow($row); + + if ( + ($this->rowGroupBuilder->rowsCount() % $this->rowGroupSizeCheckInterval) === 0 + && $this->rowGroupBuilder->isFull() + ) { + $this->flushRowGroup($stream); + } + } + + private function flushRowGroup(DestinationStream $stream): void + { + $rowGroupContainer = $this->rowGroupBuilder->flush($this->fileOffset); + $stream->append($rowGroupContainer->binaryBuffer); + $this->metadata->rowGroups()->add($rowGroupContainer->rowGroup); + $this->fileOffset += strlen($rowGroupContainer->binaryBuffer); + } +} diff --git a/src/lib/parquet/src/Flow/Parquet/ParquetEngine.php b/src/lib/parquet/src/Flow/Parquet/ParquetEngine.php index 2a3e67ede1..a33557ed29 100644 --- a/src/lib/parquet/src/Flow/Parquet/ParquetEngine.php +++ b/src/lib/parquet/src/Flow/Parquet/ParquetEngine.php @@ -12,14 +12,12 @@ interface ParquetEngine { - public function closeWrite(): void; - public function openForWrite( DestinationStream $stream, Schema $schema, Compressions $compression, Options $options, - ): void; + ): ParquetFileWriter; /** * @param array $columns @@ -34,16 +32,6 @@ public function readValues( ?int $offset = null, ): Generator; - /** - * @param iterable> $rows - */ - public function writeBatch(iterable $rows): void; - - /** - * @param array $row - */ - public function writeRow(array $row): void; - /** * @param iterable> $rows */ diff --git a/src/lib/parquet/src/Flow/Parquet/ParquetFileWriter.php b/src/lib/parquet/src/Flow/Parquet/ParquetFileWriter.php new file mode 100644 index 0000000000..dfd7fb2f1c --- /dev/null +++ b/src/lib/parquet/src/Flow/Parquet/ParquetFileWriter.php @@ -0,0 +1,24 @@ +> $rows + */ + public function writeBatch(iterable $rows): void; + + /** + * @param array $row + */ + public function writeRow(array $row): void; +} diff --git a/src/lib/parquet/src/Flow/Parquet/Writer.php b/src/lib/parquet/src/Flow/Parquet/Writer.php index 86e801f10e..298f853750 100644 --- a/src/lib/parquet/src/Flow/Parquet/Writer.php +++ b/src/lib/parquet/src/Flow/Parquet/Writer.php @@ -19,7 +19,7 @@ final class Writer { - private bool $isOpen = false; + private ?ParquetFileWriter $file = null; public function __construct( private readonly Compressions $compression = Compressions::SNAPPY, @@ -62,17 +62,18 @@ public function __destruct() public function close(): void { - if (!$this->isOpen()) { - throw new RuntimeException('Writer is not open'); - } + $file = $this->file ?? throw new RuntimeException('Writer is not open'); - $this->engine->closeWrite(); - $this->isOpen = false; + try { + $file->close(); + } finally { + $this->file = null; + } } public function isOpen(): bool { - return $this->isOpen; + return $this->file !== null; } public function open(string $path, Schema $schema): void @@ -85,19 +86,17 @@ public function open(string $path, Schema $schema): void throw new InvalidArgumentException("File {$path} already exists"); } - $this->engine->openForWrite( + $this->file = $this->engine->openForWrite( NativeLocalDestinationStream::openBlank(path($path)), $schema, $this->compression, $this->options, ); - $this->isOpen = true; } public function openForStream(DestinationStream $stream, Schema $schema): void { - $this->engine->openForWrite($stream, $schema, $this->compression, $this->options); - $this->isOpen = true; + $this->file = $this->engine->openForWrite($stream, $schema, $this->compression, $this->options); } /** @@ -123,7 +122,7 @@ public function write(string $path, Schema $schema, iterable $rows): void */ public function writeBatch(iterable $rows): void { - $this->engine->writeBatch($rows); + ($this->file ?? throw new RuntimeException('Writer is not open'))->writeBatch($rows); } /** @@ -131,7 +130,7 @@ public function writeBatch(iterable $rows): void */ public function writeRow(array $row): void { - $this->engine->writeRow($row); + ($this->file ?? throw new RuntimeException('Writer is not open'))->writeRow($row); } /** diff --git a/src/lib/parquet/src/Flow/Parquet/Writer/RowGroupBuilder.php b/src/lib/parquet/src/Flow/Parquet/Writer/RowGroupBuilder.php index bdba33cb2c..d2d36c6911 100644 --- a/src/lib/parquet/src/Flow/Parquet/Writer/RowGroupBuilder.php +++ b/src/lib/parquet/src/Flow/Parquet/Writer/RowGroupBuilder.php @@ -58,6 +58,9 @@ public function addRows(array $rows): void /** @var int<1, max> $interval */ $interval = $this->options->getInt(Option::PAGE_SIZE_CHECK_INTERVAL); + // rows still buffered by addRow() came first - parquet identifies a row by its position + $this->flushBuffer(); + foreach (array_chunk($rows, $interval) as $chunk) { $flatColumnsData = $this->shredder->shred($this->schema, $chunk); diff --git a/src/lib/parquet/tests/Flow/Parquet/Tests/Integration/IO/WriterTest.php b/src/lib/parquet/tests/Flow/Parquet/Tests/Integration/IO/WriterTest.php index 1ce1ae4160..c79e2bbb8a 100644 --- a/src/lib/parquet/tests/Flow/Parquet/Tests/Integration/IO/WriterTest.php +++ b/src/lib/parquet/tests/Flow/Parquet/Tests/Integration/IO/WriterTest.php @@ -6,7 +6,9 @@ use Composer\InstalledVersions; use Faker\Factory; +use Flow\Filesystem\Exception\RuntimeException as FilesystemRuntimeException; use Flow\Filesystem\Stream\NativeLocalDestinationStream; +use Flow\Filesystem\Tests\Double\FailingCloseDestinationStream; use Flow\Parquet\Consts; use Flow\Parquet\Engine\PhpParquetEngine; use Flow\Parquet\Option; @@ -26,6 +28,7 @@ use function array_map; use function Flow\ETL\DSL\generate_random_int; +use function Flow\Filesystem\DSL\memory_filesystem; use function Flow\Filesystem\DSL\path; use function fopen; use function iterator_to_array; @@ -33,6 +36,31 @@ class WriterTest extends ParquetIntegrationTestCase { + #[DataProvider('engine_provider')] + public function test_a_close_that_throws_leaves_the_writer_closed(ParquetEngine $engine): void + { + $writer = new Writer(engine: $engine); + $writer->openForStream( + new FailingCloseDestinationStream(memory_filesystem()->writeTo(path('memory://file.parquet'))), + Schema::with(FlatColumn::int32('id')), + ); + $writer->writeRow(['id' => 1]); + + try { + $writer->close(); + static::fail('close() was expected to throw'); + } catch (FilesystemRuntimeException $failure) { + static::assertSame('Closing "memory://file.parquet" failed', $failure->getMessage()); + } + + static::assertFalse($writer->isOpen()); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('Writer is not open'); + + $writer->close(); + } + #[DataProvider('engine_provider')] public function test_closing_not_open_writer(ParquetEngine $engine): void { @@ -79,6 +107,42 @@ public function test_opening_already_open_writer(ParquetEngine $engine): void $writer->open($path, $schema); } + #[DataProvider('engine_provider')] + public function test_writers_sharing_one_engine_write_independent_files(ParquetEngine $engine): void + { + $memory = memory_filesystem(); + $schema = Schema::with(FlatColumn::int32('id')); + $first = new Writer(engine: $engine); + $second = new Writer(engine: $engine); + + $first->openForStream($memory->writeTo(path('memory://first.parquet')), $schema); + $second->openForStream($memory->writeTo(path('memory://second.parquet')), $schema); + $first->writeRow(['id' => 1]); + $second->writeRow(['id' => 2]); + $first->writeRow(['id' => 3]); + $first->close(); + $second->close(); + + static::assertSame( + [['id' => 1], ['id' => 3]], + iterator_to_array( + (new Reader(engine: $engine)) + ->readStream($memory->readFrom(path('memory://first.parquet'))) + ->values(), + false, + ), + ); + static::assertSame( + [['id' => 2]], + iterator_to_array( + (new Reader(engine: $engine)) + ->readStream($memory->readFrom(path('memory://second.parquet'))) + ->values(), + false, + ), + ); + } + #[DataProvider('engine_provider')] public function test_writing_all_column_types(ParquetEngine $engine): void { diff --git a/src/lib/parquet/tests/Flow/Parquet/Tests/Mother/ParquetFileWriterMother.php b/src/lib/parquet/tests/Flow/Parquet/Tests/Mother/ParquetFileWriterMother.php new file mode 100644 index 0000000000..17cd13dfda --- /dev/null +++ b/src/lib/parquet/tests/Flow/Parquet/Tests/Mother/ParquetFileWriterMother.php @@ -0,0 +1,29 @@ +openForWrite( + $stream, + Schema::with(FlatColumn::int32('id')), + Compressions::UNCOMPRESSED, + $options, + ); + } +} diff --git a/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/ArrowParquetFileWriterTest.php b/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/ArrowParquetFileWriterTest.php new file mode 100644 index 0000000000..75621106cb --- /dev/null +++ b/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/ArrowParquetFileWriterTest.php @@ -0,0 +1,140 @@ +writeTo(path('memory://file.parquet'))), + ); + + try { + $file->close(); + static::fail('close() was expected to throw'); + } catch (FilesystemRuntimeException $failure) { + static::assertSame('Closing "memory://file.parquet" failed', $failure->getMessage()); + } + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('Writer is not open'); + + $file->close(); + } + + public function test_closing_twice_throws(): void + { + $file = ParquetFileWriterMother::open( + new ArrowParquetEngine(), + memory_filesystem()->writeTo(path('memory://file.parquet')), + ); + $file->close(); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('Writer is not open'); + + $file->close(); + } + + public function test_closing_writes_a_file_the_reader_reads_back(): void + { + $memory = memory_filesystem(); + $file = ParquetFileWriterMother::open( + new ArrowParquetEngine(), + $memory->writeTo(path('memory://file.parquet')), + ); + + $file->writeRow(['id' => 1]); + $file->writeBatch([['id' => 2]]); + $file->close(); + + static::assertSame( + [['id' => 1], ['id' => 2]], + iterator_to_array( + (new Reader(engine: new PhpParquetEngine())) + ->readStream($memory->readFrom(path('memory://file.parquet'))) + ->values(), + false, + ), + ); + } + + public function test_writing_a_batch_after_close_throws(): void + { + $file = ParquetFileWriterMother::open( + new ArrowParquetEngine(), + memory_filesystem()->writeTo(path('memory://file.parquet')), + ); + $file->close(); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('Writer is not open'); + + $file->writeBatch([['id' => 1]]); + } + + public function test_writing_a_row_after_close_throws(): void + { + $file = ParquetFileWriterMother::open( + new ArrowParquetEngine(), + memory_filesystem()->writeTo(path('memory://file.parquet')), + ); + $file->close(); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('Writer is not open'); + + $file->writeRow(['id' => 1]); + } + + public function test_rows_beyond_the_batch_size_are_written_once(): void + { + $memory = memory_filesystem(); + $file = ParquetFileWriterMother::open( + new ArrowParquetEngine(Options::default()->set(Option::ARROW_WRITE_BATCH_SIZE, 2)), + $memory->writeTo(path('memory://file.parquet')), + ); + + $file->writeRow(['id' => 1]); + $file->writeRow(['id' => 2]); + $file->writeRow(['id' => 3]); + $file->close(); + + static::assertSame( + [['id' => 1], ['id' => 2], ['id' => 3]], + iterator_to_array( + (new Reader(engine: new PhpParquetEngine())) + ->readStream($memory->readFrom(path('memory://file.parquet'))) + ->values(), + false, + ), + ); + } +} diff --git a/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/PhpParquetEngineTest.php b/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/PhpParquetEngineTest.php index cd7576505c..b09201027c 100644 --- a/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/PhpParquetEngineTest.php +++ b/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/PhpParquetEngineTest.php @@ -5,38 +5,25 @@ namespace Flow\Parquet\Tests\Unit\Engine; use Flow\Parquet\Engine\PhpParquetEngine; -use Flow\Parquet\Exception\RuntimeException; +use Flow\Parquet\Engine\PhpParquetFileWriter; +use Flow\Parquet\Tests\Mother\ParquetFileWriterMother; use PHPUnit\Framework\TestCase; +use function Flow\Filesystem\DSL\memory_filesystem; +use function Flow\Filesystem\DSL\path; + final class PhpParquetEngineTest extends TestCase { - public function test_close_write_throws_when_writer_not_open(): void - { - $engine = new PhpParquetEngine(); - - $this->expectException(RuntimeException::class); - $this->expectExceptionMessage('Writer is not open'); - - $engine->closeWrite(); - } - - public function test_write_batch_throws_when_writer_not_open(): void - { - $engine = new PhpParquetEngine(); - - $this->expectException(RuntimeException::class); - $this->expectExceptionMessage('Writer is not open'); - - $engine->writeBatch([['col' => 'value']]); - } - - public function test_write_row_throws_when_writer_not_open(): void + public function test_open_for_write_returns_a_new_writer_for_every_call(): void { $engine = new PhpParquetEngine(); + $memory = memory_filesystem(); - $this->expectException(RuntimeException::class); - $this->expectExceptionMessage('Writer is not open'); + $first = ParquetFileWriterMother::open($engine, $memory->writeTo(path('memory://first.parquet'))); + $second = ParquetFileWriterMother::open($engine, $memory->writeTo(path('memory://second.parquet'))); - $engine->writeRow(['col' => 'value']); + static::assertInstanceOf(PhpParquetFileWriter::class, $first); + static::assertInstanceOf(PhpParquetFileWriter::class, $second); + static::assertNotSame($first, $second); } } diff --git a/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/PhpParquetFileWriterTest.php b/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/PhpParquetFileWriterTest.php new file mode 100644 index 0000000000..5e3d51b623 --- /dev/null +++ b/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Engine/PhpParquetFileWriterTest.php @@ -0,0 +1,143 @@ +writeTo(path('memory://file.parquet'))), + ); + + try { + $file->close(); + static::fail('close() was expected to throw'); + } catch (FilesystemRuntimeException $failure) { + static::assertSame('Closing "memory://file.parquet" failed', $failure->getMessage()); + } + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('Writer is not open'); + + $file->close(); + } + + public function test_closing_twice_throws(): void + { + $file = ParquetFileWriterMother::open( + new PhpParquetEngine(), + memory_filesystem()->writeTo(path('memory://file.parquet')), + ); + $file->close(); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('Writer is not open'); + + $file->close(); + } + + public function test_closing_writes_a_file_the_reader_reads_back(): void + { + $memory = memory_filesystem(); + $file = ParquetFileWriterMother::open(new PhpParquetEngine(), $memory->writeTo(path('memory://file.parquet'))); + + $file->writeRow(['id' => 1]); + $file->writeBatch([['id' => 2]]); + $file->close(); + + static::assertSame( + [['id' => 1], ['id' => 2]], + iterator_to_array( + (new Reader(engine: new PhpParquetEngine())) + ->readStream($memory->readFrom(path('memory://file.parquet'))) + ->values(), + false, + ), + ); + } + + public function test_writing_a_batch_after_close_throws(): void + { + $file = ParquetFileWriterMother::open( + new PhpParquetEngine(), + memory_filesystem()->writeTo(path('memory://file.parquet')), + ); + $file->close(); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('Writer is not open'); + + $file->writeBatch([['id' => 1]]); + } + + public function test_writing_a_row_after_close_throws(): void + { + $file = ParquetFileWriterMother::open( + new PhpParquetEngine(), + memory_filesystem()->writeTo(path('memory://file.parquet')), + ); + $file->close(); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('Writer is not open'); + + $file->writeRow(['id' => 1]); + } + + public function test_an_array_batch_is_split_into_row_groups(): void + { + $memory = memory_filesystem(); + $file = ParquetFileWriterMother::open( + new PhpParquetEngine(options: Options::default()->set(Option::ROW_GROUP_SIZE_CHECK_INTERVAL, 1)), + $memory->writeTo(path('memory://file.parquet')), + Options::default()->set(Option::ROW_GROUP_SIZE_BYTES, 1)->set(Option::PAGE_SIZE_CHECK_INTERVAL, 1), + ); + + $file->writeBatch([['id' => 1], ['id' => 2], ['id' => 3]]); + $file->close(); + + $parquet = (new Reader(engine: new PhpParquetEngine()))->readStream($memory->readFrom(path( + 'memory://file.parquet', + ))); + static::assertSame([['id' => 1], ['id' => 2], ['id' => 3]], iterator_to_array($parquet->values(), false)); + static::assertCount(3, $parquet->metadata()->rowGroups()->all()); + } + + public function test_a_non_array_batch_is_split_into_row_groups(): void + { + $memory = memory_filesystem(); + $file = ParquetFileWriterMother::open( + new PhpParquetEngine(options: Options::default()->set(Option::ROW_GROUP_SIZE_CHECK_INTERVAL, 1)), + $memory->writeTo(path('memory://file.parquet')), + Options::default()->set(Option::ROW_GROUP_SIZE_BYTES, 1)->set(Option::PAGE_SIZE_CHECK_INTERVAL, 1), + ); + + $file->writeBatch(new ArrayIterator([['id' => 1], ['id' => 2], ['id' => 3]])); + $file->close(); + + $parquet = (new Reader(engine: new PhpParquetEngine()))->readStream($memory->readFrom(path( + 'memory://file.parquet', + ))); + static::assertSame([['id' => 1], ['id' => 2], ['id' => 3]], iterator_to_array($parquet->values(), false)); + static::assertCount(3, $parquet->metadata()->rowGroups()->all()); + } +} diff --git a/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Writer/RowGroupBuilderTest.php b/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Writer/RowGroupBuilderTest.php index 4db85be454..9eeb7e09df 100644 --- a/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Writer/RowGroupBuilderTest.php +++ b/src/lib/parquet/tests/Flow/Parquet/Tests/Unit/Writer/RowGroupBuilderTest.php @@ -404,6 +404,22 @@ public function test_rows_count_increases_with_each_row(): void } } + public function test_rows_added_one_by_one_stay_before_a_later_batch(): void + { + $schema = Schema::with(FlatColumn::int32('id')); + $options = new Options(); + $compression = Compressions::UNCOMPRESSED; + $shredder = new DremelShredder(new ColumnDataValidator(), DataConverter::initialize(Options::default())); + $mixed = new RowGroupBuilder($schema, $compression, $options, $shredder); + $batched = new RowGroupBuilder($schema, $compression, $options, $shredder); + + $mixed->addRow(['id' => 1]); + $mixed->addRows([['id' => 2]]); + $batched->addRows([['id' => 1], ['id' => 2]]); + + static::assertSame($batched->flush(0)->binaryBuffer, $mixed->flush(0)->binaryBuffer); + } + public function test_rows_count_initially_zero(): void { $schema = Schema::with(FlatColumn::int32('id'));