fix: extractors no longer mutate user-provided Schem

documentation/adrs.md

@@ -38,6 +38,7 @@ consistency in decision-making across the project.
 - [2025-01-07: Static Analysis Baseline](/documentation/adrs/static-analysis-baseline.md)
 - [2025-01-09: Extension Points](/documentation/adrs/extension-points.md)
 - [2026-03-02: Variadic Arguments Pattern for Required Parameters](/documentation/adrs/variadic-arguments-pattern.md)
+- [2026-07-27: Schema Immutability](/documentation/adrs/schema-immutability.md)
 
 ### [Proposed AD](https://github.com/flow-php/flow/pulls?q=is%3Apr+is%3Aopen+label%3AAD+)
 

documentation/adrs/schema-immutability.md

@@ -0,0 +1,116 @@
+# Schema Immutability
+
+[TOC]
+
+Proposed by: @norberttech
+Date: 2026-07-27
+
+## Context
+---
+
+`Schema` was mutable — `add()`, `remove()`, `rename()`, `merge()` and every other mutator rewrote
+`$this->definitions` and returned `$this`. `Definition::addMetadata()` and `Definition::setMetadata()` did the same
+with `$this->metadata`.
+
+File extractors extend the schema they were given so the hydrator can cast auto-added columns
+(`_input_file_uri`, partition columns, sheet metadata): `Hydrator::cast()` iterates schema definitions and drops
+undeclared row keys, so a column that should materialize in rows must be present in the schema. An extractor
+storing the caller's `Schema` therefore wrote every internal extension into the caller's object, producing three
+observable defects (#2536 regression):
+
+1. **Caller schema pollution** — a `Schema` the user holds for other purposes gains non-nullable columns it never
+   declared.
+2. **Extractor-lifetime pollution** — columns added during one `extract()` run persist into subsequent runs.
+3. **Cross-stream pollution** — partition columns of one stream leak into the next, and
+   `Hydrator::cast(fillMissing: true)` injects `null` into a non-nullable definition.
+
+The first fix cloned: `withSchema()` stored `clone $schema`, and extractors cloned again per run and per stream.
+That fix was incomplete. `clone` is shallow and `Schema`'s only state is `array<string, Definition>`, so a cloned
+`Schema` **shares its `Definition` instances**. `Schema::addMetadata()` / `setMetadata()` reached into a shared
+`Definition` and mutated it in place, so metadata writes aliased through every copy — including the caller's.
+
+The same mutable contract left latent aliasing traps elsewhere: `Rows::schema()` seeded its merge loop with row 0's
+*memoized* `Schema` and corrupted it, `FloeStreamWriter` retained a caller-owned `Schema` for the lifetime of a
+write session, and `merge()`'s fast paths returned `$this` or the argument.
+
+## Decision
+---
+
+**`Schema` and its whole state chain are immutable. Every mutator returns a new instance; nothing is ever written
+in place.**
+
+- `Schema` — a `final readonly class`. All 19 mutators return `new self(...)`; `setDefinitions()` is the
+  constructor's validation helper and is called from the constructor only.
+- `Definition` (19 implementations) — each a `final readonly class`. `addMetadata()` and `setMetadata()` return a
+  per-class `new self(...)`, matching the idiom `makeNullable()` and `rename()` already used.
+- `Metadata` — already a `final readonly class`.
+
+Immutability is declared at the class level, not per property: a `readonly class` cannot gain a writable property
+later, so the guarantee survives future edits instead of depending on whoever adds property number 20 remembering
+the rule. It is compiler-enforced, not convention. Extractors and the DSL hold caller-provided `Schema` instances
+directly: there is nothing to clone because there is nothing to mutate. Sharing an instance — `merge()`'s fast
+paths, a retained base `Definition` in the hydrator, `Rows`' memoized schema — is safe by construction.
+
+DSL `from_*()` functions stay pure delegation.
+
+### Out of scope
+
+`EntryReference` remains mutable — `as()`, `asc()` and `desc()` write `$alias` / `$sort` on `$this`. It is shared
+with the entire expression DSL, so making it immutable is a separate project and is not attempted here.
+
+### Breaking change
+
+Calling a mutator and discarding the result is now a **silent no-op**. There is no `#[\NoDiscard]`; the change is
+communicated through [upgrading.md](/documentation/upgrading.md).
+
+```php
+$schema->add(str_schema('x'));          // before: mutates $schema. now: no-op.
+$schema = $schema->add(str_schema('x')); // correct
+```
+
+## Pros & Cons
+---
+
+**Advantages:**
+
+- **The bug class is gone**, not patched — aliasing is impossible because there is no writable state to alias.
+- **Compiler-enforced**: a `readonly` violation is a fatal error, not a convention a new extractor can forget.
+- **Sharing becomes free**: no defensive clones in extractors, `PhpRowHydrator`, or the native hydrator.
+- Fixes the `Rows::schema()` and `FloeStreamWriter` aliasing traps without touching either.
+
+**Disadvantages:**
+
+- Breaking change for downstream code that discards a mutator's return value, and it breaks **silently**.
+- A long mutator chain allocates one `Schema` per link. Schemas are small and built once per pipeline, not per
+  row, so this does not show up in profiles.
+
+## Alternatives Considered
+---
+
+### 1. Clone at reception, clone before extending
+
+Every `withSchema()` stores `clone $schema`; extractors clone again per run and per stream.
+
+**Rejected because:** the clone is shallow, so `Definition` instances stay shared and metadata mutations alias
+through every copy anyway. It is also convention rather than a compiler-enforced rule — a new extractor can forget
+to clone — and it leaves unobservable dead clones in extractors that never extend the schema.
+
+### 2. Deep `Schema::__clone()` + `Definition::__clone()`
+
+Give `Schema` and every `Definition` a `__clone()` that copies the definition array and its objects.
+
+**Rejected because:** it patches the symptom while keeping the mutable contract, so every future aliasing trap
+(`Rows::schema()`, `FloeStreamWriter`, `merge()`'s fast paths) still has to be found and cloned around by hand. It
+also makes every clone more expensive without removing the need to remember to clone.
+
+### 3. Materialize auto-added columns post-hydration via `Row::add()` (Floe style)
+
+**Rejected because:** the extended schema *is* the hydrator's instruction set — `Hydrator::cast()` drops undeclared
+row keys, and the `findDefinition()` guard lets a user-declared partition column keep its user-defined type.
+Post-hydration adds would bypass both.
+
+## Links and References
+---
+
+- [PR #2536](https://github.com/flow-php/flow/pull/2536) - encoder/hydrator row contract that introduced the
+  schema extension in extractors

documentation/upgrading.md

@@ -478,6 +478,28 @@ Custom aggregators must implement `references()` - return the references the agg
 | `->groupBy(...)->aggregate(...)->pivot(ref('x'))` | `->groupBy(...)->pivot(ref('x'))->aggregate(...)` |
 | `DataFrame::pivot()`                              | removed; `GroupedDataFrame::pivot()` only         |
 
+### 32) `flow-php/etl-adapter-csv`, `-excel`, `-json`,
+`-xml` - explicit schema no longer projects partition columns away
+
+| Before                                                           | After                                                |
+|------------------------------------------------------------------|------------------------------------------------------|
+| partition columns undeclared in the schema are dropped from rows | force-added to rows as non-nullable `string` columns |
+
+Declare the partition column in the schema to control its type.
+
+### 33) `flow-php/etl` - `Schema` and `Schema\Definition` mutators return a new instance
+
+| Before                                  | After                                                |
+|-----------------------------------------|------------------------------------------------------|
+| `$schema->add(str_schema('x'));`        | `$schema = $schema->add(str_schema('x'));`           |
+| `$schema->addMetadata('id', 'k', 'v');` | `$schema = $schema->addMetadata('id', 'k', 'v');`    |
+| `$definition->setMetadata($metadata);`  | `$definition = $definition->setMetadata($metadata);` |
+
+Assign the result of every `Schema` mutator - `add`, `addAfter`, `addBefore`, `addMetadata`, `gracefulRemove`,
+`insertAt`, `keep`, `makeNullable`, `merge`, `moveAfter`, `moveBefore`, `moveTo`, `prepend`, `remove`, `rename`,
+`reorder`, `replace`, `setMetadata`, `sort` - and of `Definition::addMetadata()` / `Definition::setMetadata()`.
+Discarding it is a silent no-op.
+
 ---
 
 ## Upgrading from 0.40.x to 0.41.x
@@ -2233,7 +2255,7 @@ After:
     ->run();
 ```
 
-### 4) ConfigBuilder::putInputIntoRows () output is now prefixed with _  (underscore)
+### 4) ConfigBuilder::putInputIntoRows () output is now prefixed with _   (underscore)
 
 In order to avoid collisions with datasets columns, additional columns created after using putInputIntoRows ()
 would now be prefixed with `_` (underscore) symbol.

src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Fixtures/cross_stream/date=2026-01-01/data.csv

@@ -0,0 +1,2 @@
+id,value
+1,a

src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Fixtures/cross_stream/nopart/data.csv

@@ -0,0 +1,2 @@
+id,value
+2,b

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

@@ -18,8 +18,12 @@
 use function Flow\ETL\DSL\config;
 use function Flow\ETL\DSL\df;
 use function Flow\ETL\DSL\flow_context;
+use function Flow\ETL\DSL\int_schema;
 use function Flow\ETL\DSL\ref;
+use function Flow\ETL\DSL\schema;
+use function Flow\ETL\DSL\schema_metadata;
 use function Flow\ETL\DSL\schema_to_ascii;
+use function Flow\ETL\DSL\str_schema;
 use function Flow\Filesystem\DSL\path_real;
 use function iterator_to_array;
 
@@ -145,6 +149,33 @@ public function test_bom_removal_utf8(): void
         );
     }
 
+    public function test_extract_does_not_mutate_metadata_of_user_provided_schema(): void
+    {
+        $schema = schema(int_schema('id', metadata: schema_metadata(['primary_key' => true])), str_schema('value'));
+
+        $before = $schema->normalize();
+
+        $extractor = from_csv(__DIR__ . '/../Fixtures/cross_stream/*/data.csv', schema: $schema);
+
+        df(Config::builder()->putInputIntoRows())->read($extractor)->run();
+        df(Config::builder()->putInputIntoRows())->read($extractor)->run();
+
+        static::assertSame($before, $schema->normalize());
+    }
+
+    public function test_extract_does_not_mutate_user_provided_schema(): void
+    {
+        $schema = schema(int_schema('id'), str_schema('value'));
+
+        $extractor = from_csv(__DIR__ . '/../Fixtures/cross_stream/*/data.csv', schema: $schema);
+
+        df(Config::builder()->putInputIntoRows())->read($extractor)->run();
+        df(Config::builder()->putInputIntoRows())->read($extractor)->run();
+
+        static::assertNull($schema->findDefinition('date'));
+        static::assertNull($schema->findDefinition('_input_file_uri'));
+    }
+
     public function test_extracting_csv_empty_columns_as_empty_strings(): void
     {
         $extractor = from_csv(
@@ -438,6 +469,23 @@ public function test_loading_data_from_all_partitions(): void
         );
     }
 
+    public function test_partition_columns_are_not_leaking_between_streams(): void
+    {
+        static::assertSame(
+            [
+                ['id' => 1, 'value' => 'a', 'date' => '2026-01-01'],
+                ['id' => 2, 'value' => 'b'],
+            ],
+            df()
+                ->read(from_csv(__DIR__ . '/../Fixtures/cross_stream/*/data.csv', schema: schema(
+                    int_schema('id'),
+                    str_schema('value'),
+                )))
+                ->fetch()
+                ->toArray(),
+        );
+    }
+
     public function test_signal_stop(): void
     {
         $extractor = from_csv(path_real(__DIR__ . '/../Fixtures/orders_flow.csv'));

src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelExtractorTest.php

@@ -5,6 +5,7 @@
 namespace Flow\ETL\Adapter\Excel\Tests\Integration;
 
 use Flow\ETL\Adapter\Excel\ExcelReader;
+use Flow\ETL\Config;
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Extractor\Signal;
 use Flow\ETL\Rows;
@@ -250,6 +251,19 @@ public function test_is_valid_excel_sheet_name_function(): void
         static::assertTrue($result[4]['is_valid']);
     }
 
+    public function test_extract_does_not_mutate_user_provided_schema(): void
+    {
+        $schema = schema(string_schema('group'), int_schema('id'), string_schema('value'));
+
+        $extractor = from_excel(__DIR__ . '/../Fixtures/cross_stream/*/*.xlsx')->withSchema($schema);
+
+        df(Config::builder()->putInputIntoRows())->read($extractor)->run();
+        df(Config::builder()->putInputIntoRows())->read($extractor)->run();
+
+        static::assertNull($schema->findDefinition('date'));
+        static::assertNull($schema->findDefinition('_input_file_uri'));
+    }
+
     public function test_loading_data_from_all_partitions(): void
     {
         df()->read(from_excel(__DIR__ . '/../Fixtures/partitioned/group=*/*.xlsx'))->run(function (Rows $rows): void {
@@ -260,6 +274,26 @@ public function test_loading_data_from_all_partitions(): void
         });
     }
 
+    public function test_partition_columns_are_not_leaking_between_streams(): void
+    {
+        static::assertSame(
+            [
+                ['group' => '1', 'id' => 1, 'value' => 'a', 'date' => '2026-01-01'],
+                ['group' => '1', 'id' => 2, 'value' => 'b', 'date' => '2026-01-01'],
+                ['group' => '2', 'id' => 5, 'value' => 'e'],
+                ['group' => '2', 'id' => 6, 'value' => 'f'],
+            ],
+            df()
+                ->read(from_excel(__DIR__ . '/../Fixtures/cross_stream/*/*.xlsx')->withSchema(schema(
+                    string_schema('group'),
+                    int_schema('id'),
+                    string_schema('value'),
+                )))
+                ->fetch()
+                ->toArray(),
+        );
+    }
+
     public function test_signal_stop(): void
     {
         $generator = from_excel(path_real(__DIR__ . '/../Fixtures/fixture.xlsx'))->extract(flow_context(config()));

src/adapter/etl-adapter-google-sheet/tests/Flow/ETL/Adapter/GoogleSheet/Tests/Unit/GoogleSheetExtractorTest.php

@@ -15,12 +15,66 @@
 use function Flow\ETL\Adapter\GoogleSheet\from_google_sheet_columns;
 use function Flow\ETL\DSL\flow_context;
 use function Flow\ETL\DSL\row;
+use function Flow\ETL\DSL\schema;
 use function Flow\ETL\DSL\str_entry;
+use function Flow\ETL\DSL\str_schema;
 use function Flow\ETL\DSL\string_entry;
 use function iterator_to_array;
 
 final class GoogleSheetExtractorTest extends FlowTestCase
 {
+    public function test_extract_does_not_mutate_user_provided_schema(): void
+    {
+        $sheetName = 'sheet';
+
+        $gridProperties = new Sheets\GridProperties();
+        $gridProperties->setRowCount(100);
+
+        $properties = new Sheets\SheetProperties();
+        $properties->title = $sheetName;
+        $properties->setGridProperties($gridProperties);
+
+        $sheet = new Sheets\Sheet();
+        $sheet->setProperties($properties);
+
+        $spreadsheet = $this->createMock(Sheets\Spreadsheet::class);
+        $spreadsheet->expects(self::exactly(2))->method('getSheets')->willReturn([$sheet]);
+
+        $resource = $this->createMock(Sheets\Resource\Spreadsheets::class);
+        $resource
+            ->expects(self::exactly(2))
+            ->method('get')
+            ->with('spread-id', ['ranges' => [], 'includeGridData' => false])
+            ->willReturn($spreadsheet);
+
+        $service = new Sheets();
+        $service->spreadsheets = $resource;
+
+        $valueRange = new ValueRange();
+        $valueRange->setValues([['header'], ['row1']]);
+
+        $response = new Sheets\BatchGetValuesResponse();
+        $response->setValueRanges([$valueRange]);
+
+        $spreadsheetsValues = $this->createMock(SpreadsheetsValues::class);
+        $spreadsheetsValues->expects(self::exactly(2))->method('batchGet')->willReturn($response);
+
+        $service->spreadsheets_values = $spreadsheetsValues;
+
+        $schema = schema(str_schema('header'));
+
+        $extractor = from_google_sheet_columns($service, 'spread-id', $sheetName, 'A', 'B')
+            ->withHeader(true)
+            ->withRowsPerPage(2)
+            ->withSchema($schema);
+
+        iterator_to_array($extractor->extract(flow_context((new ConfigBuilder())->putInputIntoRows()->build())));
+        iterator_to_array($extractor->extract(flow_context((new ConfigBuilder())->putInputIntoRows()->build())));
+
+        static::assertNull($schema->findDefinition('_spread_sheet_id'));
+        static::assertNull($schema->findDefinition('_sheet_name'));
+    }
+
     public function test_its_fails_if_sheet_not_found(): void
     {
         $spreadsheet = $this->createMock(Sheets\Spreadsheet::class);

src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Fixtures/cross_stream/date=2026-01-01/data.json

@@ -0,0 +1 @@
+[{"id":1,"value":"a"}]

src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Fixtures/cross_stream/date=2026-01-01/data.jsonl

@@ -0,0 +1 @@
+{"id":1,"value":"a"}

src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Fixtures/cross_stream/nopart/data.json

@@ -0,0 +1 @@
+[{"id":2,"value":"b"}]

src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Fixtures/cross_stream/nopart/data.jsonl

@@ -0,0 +1 @@
+{"id":2,"value":"b"}