fix: write and read back partitioned datasets

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",

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
+<?php
+
+use function Flow\Filesystem\DSL\native_local_filesystem;
+use function Flow\Filesystem\DSL\path;
+
+// data/flat.parquet, data/date=2026-09-01/one.parquet, data/id=1/date=2026-09-01/two.parquet
+native_local_filesystem()->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

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()

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

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

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

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

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