feat(flow-php/etl): corrupted records are refused or routed, never repaired silently

documentation/components/adapters/csv.md

@@ -36,11 +36,17 @@ $rows = data_frame()
 ### Header and missing cells
 
 ```
-header row  [' id ', '', 'name']   →  columns ['id', 'e01', 'name']   (trimmed, a blank cell named e + its position)
-record      ['1', 'a']             →  ['id' => 1, 'e01' => 'a', 'name' => null]   (a cell the record lacks is null)
-record      ['1', '', 'x']         →  'e01' => null   (a present '' is null; `withEmptyToNull(false)` keeps it '')
+header row  [' id ', '', 'name']   ->  columns ['id', 'e01', 'name']   (trimmed, a blank cell named e + its position)
+record      ['1', 'a']             ->  corrupted record (2 cells, the header has 3); `withMissingColumnsAsNull(true)` makes 'name' null
+record      ['1', 'a', 'x', '']    ->  corrupted record (a non-empty cell past the header), an empty one is dropped
+record      ['1', '', 'x']         ->  'e01' => null   (a present '' is null; `withEmptyToNull(false)` keeps it '')
+empty line                         ->  skipped
 ```
 
+A corrupted record throws `CorruptedRecordException`, or goes to `->withCorruptedRecords($loader)`, see
+[Corrupted Records](/documentation/components/core/corrupted-records.md). A column of `->withSchema()` missing from the
+header throws `SchemaValidationException`.
+
 ## Loader
 
 ```php

documentation/components/adapters/excel.md

@@ -71,11 +71,16 @@ starts at row 5.
 ### Header and missing cells
 
 ```
-header row  [' id ', '', 'name']   →  columns ['id', 'e01', 'name']   (trimmed, a blank cell named e + its position)
-record      ['1', 'a']             →  ['id' => '1', 'e01' => 'a', 'name' => null]   (a cell the record lacks is null)
-record      ['1', '', 'x']         →  'e01' => null   (a present '' is null; `withConvertEmptyToNull(false)` keeps it '')
+header row  [' id ', '', 'name']   ->  columns ['id', 'e01', 'name']   (trimmed, a blank cell named e + its position)
+record      ['1', 'a']             ->  ['id' => '1', 'e01' => 'a', 'name' => null]; `withMissingColumnsAsNull(false)` makes it a corrupted record
+record      ['1', 'a', 'x', '']    ->  corrupted record (a non-empty cell past the header), an empty one is dropped
+record      ['1', '', 'x']         ->  'e01' => null   (a present '' is null; `withConvertEmptyToNull(false)` keeps it '')
 ```
 
+A corrupted record throws `CorruptedRecordException`, or goes to `->withCorruptedRecords($loader)`, see
+[Corrupted Records](/documentation/components/core/corrupted-records.md). A column of `->withSchema()` missing from the
+header throws `SchemaValidationException`.
+
 A header cell that is not text (a date, a number) is named like a blank one.
 
 ### Schema

documentation/components/adapters/google-sheet.md

@@ -32,11 +32,16 @@ data_frame()
 ### Header and missing cells
 
 ```
-header row  [' id ', '', 'name']   →  columns ['id', 'e01', 'name']   (trimmed, a blank cell named e + its position)
-record      ['1', 'a']             →  ['id' => '1', 'e01' => 'a', 'name' => null]   (a cell the record lacks is null)
-record      ['1', '', 'x']         →  'e01' => null   (a present '' is null; `withEmptyToNull(false)` keeps it '')
+header row  [' id ', '', 'name']   ->  columns ['id', 'e01', 'name']   (trimmed, a blank cell named e + its position)
+record      ['1', 'a']             ->  corrupted record (2 cells, the header has 3); `withMissingColumnsAsNull(true)` makes 'name' null
+record      ['1', 'a', 'x', '']    ->  corrupted record (a non-empty cell past the header), an empty one is dropped
+record      ['1', '', 'x']         ->  'e01' => null   (a present '' is null; `withEmptyToNull(false)` keeps it '')
 ```
 
+A corrupted record throws `CorruptedRecordException`, or goes to `->withCorruptedRecords($loader)`, see
+[Corrupted Records](/documentation/components/core/corrupted-records.md). A column of `->withSchema()` missing from the
+header throws `SchemaValidationException`.
+
 Blank rows before the header are skipped, also under `withHeader(false)`.
 
 ## Needed parameters

documentation/components/adapters/json.md

@@ -54,6 +54,10 @@ holding a top-level array are read natively, as strict JSON: malformed input thr
 `RuntimeException('Malformed JSON in "<uri>" at line N: ...')` (`at element N` in a document). A pointer, or a document
 that is not an array, is read by the PHP reader.
 
+A value that does not cast to the type `->withSchema()` declares throws `CorruptedRecordException`, or goes to
+`->withCorruptedRecords($loader)`, see [Corrupted Records](/documentation/components/core/corrupted-records.md). Under an
+inferred schema, a key first seen after the inference sample throws `InferredSchemaException`.
+
 ## Loader
 
 ```php

documentation/components/core/core.md

@@ -343,6 +343,7 @@ For detailed information about specific DataFrame operations, see the following
 - **[Schema](/documentation/components/core/schema.md)** - Schema management and validation
 - **[Constraints](/documentation/components/core/constraints.md)** - Data integrity constraints and business rules
 - **[Error Handling](/documentation/components/core/error-handling.md)** - Error management strategies
+- **[Corrupted Records](/documentation/components/core/corrupted-records.md)** - Routing records that cannot become rows
 
 ### Observability
 - **[Telemetry](/documentation/components/core/telemetry.md)** - Distributed tracing, metrics, and logging integration

documentation/components/core/corrupted-records.md

@@ -0,0 +1,96 @@
+# Corrupted Records
+
+[DOC_LINK:/documentation/components/core/core.md]
+
+A corrupted record is a source record that cannot become a row without losing or inventing data: a CSV line or a
+sheet row whose cells do not match the header, or a value that does not cast to the declared type. A read throws
+`CorruptedRecordException` at the first one. Route them to any loader instead:
+
+```php
+<?php
+
+use function Flow\ETL\Adapter\CSV\from_csv;
+use function Flow\ETL\Adapter\JSON\{from_json_lines, to_json, to_json_lines};
+use function Flow\ETL\DSL\{count, data_frame, float_schema, int_schema, lit, not, ref, schema, to_branch, to_output};
+
+$valid = ref('amount')->greaterThan(lit(0));
+
+// read, route corrupted records, split valid and invalid rows
+data_frame()
+    ->read(from_csv('orders.csv')
+        ->withSchema(schema(int_schema('id'), float_schema('amount')))
+        ->withCorruptedRecords(to_json_lines('rejects/broken.jsonl')))
+    ->write(to_branch($valid, to_json('clean.json')))
+    ->write(to_branch(not($valid), to_json('rejects/invalid.json')))
+    ->run();
+
+// report
+data_frame()
+    ->read(from_json_lines('rejects/broken.jsonl'))
+    ->groupBy(ref('reason'))
+    ->aggregate(count(ref('offset')))
+    ->write(to_output())
+    ->run();
+
+// repair: a record with an extra field keeps its first two
+data_frame()
+    ->read(from_json_lines('rejects/broken.jsonl'))
+    ->filter(ref('fields')->size()->greaterThan(lit(2)))
+    ->select('offset', 'fields')
+    ->withColumn('id', ref('fields')->arrayGet('0')->cast('int'))
+    ->withColumn('amount', ref('fields')->arrayGet('1')->cast('float'))
+    ->drop('fields')
+    ->write(to_output())
+    ->run();
+```
+
+`orders.csv`:
+
+```
+id,amount
+1,10.5
+2,abc
+3,-4
+4,7,extra
+5,1.5
+```
+
+Output:
+
+```
++----------------------+--------------+
+|               reason | offset_count |
++----------------------+--------------+
+| 3 fields, the header |            1 |
+| column "amount": cou |            1 |
++----------------------+--------------+
+2 rows
++--------+----+----------+
+| offset | id |   amount |
++--------+----+----------+
+|     28 |  4 | 7.000000 |
++--------+----+----------+
+1 rows
+```
+
+## The routed row
+
+| Column   | Type            | Value                                                                    |
+|----------|-----------------|--------------------------------------------------------------------------|
+| `uri`    | `?string`       | the file, null for `from_array()` and sequences                          |
+| `offset` | `int`           | where the record starts, see below                                       |
+| `line`   | `?int`          | if the read starts at the first byte of the file, the line of the record |
+| `reason` | `string`        | `3 fields, the header has 2`, `column "amount": could not convert ...`   |
+| `record` | `string`        | the record as the source holds it, else its cells or values as JSON      |
+| `fields` | `?list<string>` | the cells a CSV, Excel or Google Sheet reader split, null for JSON       |
+
+| Reader                    | `offset`                     |
+|---------------------------|------------------------------|
+| CSV, JSON lines           | byte offset of the record    |
+| Excel, Google Sheet       | 1-based sheet row number     |
+| JSON array                | 1-based element number       |
+| `from_array()`, sequences | 1-based entry number         |
+
+A run that routed records logs how many and where. A corrupted record is not an extraction error. Without
+`withCorruptedRecords()` it fails the run, whatever the `ErrorHandler`. Route it to skip it. `limit()` and `offset()`
+count rows only.

documentation/components/core/error-handling.md

@@ -11,6 +11,8 @@ By default an exception thrown while reading, transforming or loading stops the
 | `onTransformation(TransformationError): TransformationAction` | `cause`, `transformer`, `rows` | `skipBatch` - drop the batch, continue with the next one                           |
 | `onLoading(LoadingError): LoadingAction`                      | `cause`, `loader`, `rows`      | `skipLoader` - the failing loader misses this batch, the remaining steps still run |
 
+A corrupted record never reaches the `ErrorHandler`, see [Corrupted Records](/documentation/components/core/corrupted-records.md).
+
 Built-in handlers:
 
 | Handler                           | Extraction  | Transformation | Loading      |

documentation/upgrading.md

@@ -1347,7 +1347,7 @@ wrappers are gone, the class that builds the sink or source picks the lane
 | `Flow\ETL\Adapter\JSON\{JsonOpenSource, PhpJsonOpenSource, RustJsonOpenSource}`                                                                  | `Flow\ETL\Adapter\JSON\IO\Extractor\{JsonOpenSource, PhpJsonOpenSource, RustJsonOpenSource}`                                                                |
 | `Flow\ETL\Adapter\JSON\JSONMachine\{JsonArrayElements, JsonExtractor, JsonFileBatches, JsonFileReader}`                                          | `Flow\ETL\Adapter\JSON\IO\Extractor\{JsonArrayElements, JsonExtractor, JsonFileBatches, JsonFileReader}`                                                    |
 | `Flow\ETL\Adapter\JSON\JSONMachine\JsonFileSample`                                                                                               | `Flow\ETL\Adapter\JSON\IO\Inference\JsonFileSample`                                                                                                         |
-| `Flow\ETL\Adapter\JSON\JSONMachine\JsonReadBytes`                                                                                                | `Flow\ETL\Adapter\JSON\IO\Extractor\JsonReadBytes`                                                                                                          |
+| `Flow\ETL\Adapter\JSON\JSONMachine\JsonReadBytes`                                                                                                | `Flow\ETL\Adapter\JSON\IO\Extractor\JsonReadProgress`                                                                                                       |
 | `Flow\ETL\Adapter\JSON\JsonSchema\{JsonSchemaMetadata, ResolvedReference}`                                                                       | `Flow\ETL\Adapter\JSON\Values\Schema\JsonSchema\{JsonSchemaMetadata, ResolvedReference}`                                                                    |
 | `Flow\ETL\Adapter\JSON\JsonSchema\ReferenceResolver`                                                                                             | `Flow\ETL\Adapter\JSON\IO\Schema\JsonSchema\ReferenceResolver`                                                                                              |
 | `Flow\ETL\Adapter\JSON\JsonSchema\Exception\{CircularReferenceException, UnresolvableReferenceException, UnsupportedKeywordException}`           | `Flow\ETL\Adapter\JSON\Foundation\Exception\{CircularReferenceException, UnresolvableReferenceException, UnsupportedKeywordException}`                      |
@@ -1889,6 +1889,31 @@ by Spark's hidden-path rule.
 |------------------------------------------------------|-----------------------------------------------------------------------|
 | the first filled block put an empty blob at the path | nothing is visible at the path until `close()` commits the block list |
 
+### 126) `flow-php/etl` - extractors yield corrupted records, the run routes or refuses them
+
+| Before                                                     | After                                                                                                                                                                                                                           |
+|------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `Extractor::extract()`, `Reader::extract()` yielded `Rows` | they may also yield a `CorruptedRecordException`, one per record that cannot become a row; a direct caller routes or refuses it (`instanceof Rows`)                                                                             |
+| a reader refused a corrupted record itself                 | the run routes it to `withCorruptedRecords($loader)`, loaded, closed and discarded like the run's other loaders; without a loader it fails the run, whatever the `ErrorHandler` - a corrupted record is not an extraction error |
+| `ExtractionError::$extractor` was an `Extractor`           | `Extractor\|Reader`: a task's source is a `Reader`                                                                                                                                                                              |
+| `StepRunner::steps()`, `run()`: `?Extractor $source`       | `Extractor\|Reader\|null $source`; a source implementing `CorruptedRecordsRouter` routes its corrupted records                                                                                                                  |
+
+### 127) `flow-php/etl`, `flow-php/etl-adapter-csv`, `-json`, `-excel`, `-google-sheet` - a record that cannot become a row throws
+
+| Before                                                                                                                | After                                                                                                                                                                                    |
+|-----------------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| a CSV line or a sheet row with fewer cells than the header was padded with nulls                                      | `CorruptedRecordException`; `->withMissingColumnsAsNull(true)` pads it (CSV, Google Sheet), Excel pads by default (`->withMissingColumnsAsNull(false)` refuses)                          |
+| a CSV line or an Excel row with more cells than the header lost them                                                  | `CorruptedRecordException` for a non-empty extra cell; an empty one is dropped                                                                                                           |
+| Google Sheet `->withDropExtraColumns(bool)`                                                                           | removed: a non-empty extra cell is a corrupted record; `->withMissingColumnsAsNull(bool)` pads a short row                                                                               |
+| an empty CSV line was a row of nulls                                                                                  | skipped                                                                                                                                                                                  |
+| a value that does not cast to the type `->withSchema()` declares threw `SchemaMismatchException`                      | `CorruptedRecordException` (route it with `->withCorruptedRecords($loader)`)                                                                                                             |
+| a value or a JSON key a file shows only after the inference sample: `SchemaMismatchException`, or the key was dropped | `InferredSchemaException::outsideTheSample()`, naming the file, offset, line and sample size; `from_array()` and sequences throw `InferredSchemaException`                               |
+| a `->withSchema()` column missing from a CSV, Excel or Google Sheet header was null on every row                      | `SchemaValidationException`                                                                                                                                                              |
+| `new RecordDecoder($withHeader, $emptyToNull, $dropExtraColumns, $headers)`, `decode(list $batch): list`              | `new RecordDecoder($withHeader, $emptyToNull, $missingColumnsAsNull, $headers)`, `decode(array $cells): null\|array\|string` (null: header or empty record, string: why it is corrupted) |
+| `CSVDecoder::decode(list<string> $lines)`                                                                             | `CSVDecoder::decode(CSVDecoder::cells(string $line))`, one record                                                                                                                        |
+| `CSVOpenSource::batches($schema, $batchSize, $backend)`                                                               | `+ ?int $sampleSize = null`; `JsonOpenSource::batches()` also `+ ?Schema $known = null`; both yield `CorruptedRecordException\|Rows`                                                     |
+| `to_date_time(null)` threw `InvalidArgumentException`                                                                 | returns null; a null format or time zone still throws                                                                                                                                    |
+
 ## Upgrading from 0.45.x to 0.46.x
 
 ### 1) `flow-php/etl-adapter-postgresql`, `flow-php/etl-adapter-doctrine` - keyset reads reject NULL keys up front

src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/IO/Extractor/CSVDecoder.php

@@ -6,7 +6,6 @@
 
 use Flow\ETL\IO\Extractor\Grid\RecordDecoder;
 
-use function array_map;
 use function str_getcsv;
 
 final class CSVDecoder
@@ -23,24 +22,29 @@ public function __construct(
         private string $escape = '\\',
         bool $emptyToNull = true,
         ?array $headers = null,
+        bool $missingColumnsAsNull = false,
     ) {
-        $this->records = new RecordDecoder($withHeader, $emptyToNull, headers: $headers);
+        $this->records = new RecordDecoder($withHeader, $emptyToNull, $missingColumnsAsNull, $headers);
     }
 
     /**
-     * @param list<string> $batch
+     * @return list<?string>
+     */
+    public function cells(string $line): array
+    {
+        /** @var list<?string> */
+        return str_getcsv($line, $this->separator, $this->enclosure, $this->escape);
+    }
+
+    /**
+     * @param list<?string> $cells
      *
-     * @return list<array<string, ?string>>
+     * @return null|array<string, ?string>|string see RecordDecoder::decode()
      */
-    public function decode(array $batch): array
+    public function decode(array $cells): array|string|null
     {
-        /** @var list<array<string, ?string>> */
-        return $this->records->decode(array_map(fn(string $line): array => str_getcsv(
-            $line,
-            $this->separator,
-            $this->enclosure,
-            $this->escape,
-        ), $batch));
+        /** @var null|array<string, ?string>|string */
+        return $this->records->decode($cells);
     }
 
     /**