Read performance

documentation/contributing/rust.md

@@ -62,6 +62,8 @@ src/extension/flow-php-ext/
 │   ├── format.rs           # Floe binary format primitives
 │   ├── hydrate.rs          # Row hydration against a schema
 │   ├── cast.rs             # Value casting
+│   ├── json_check.rs       # JSON validation shared by casting and CSV inference
+│   ├── csv/                # CSV tokenizer, reader and schema-inference fold
 │   ├── plan.rs             # Per-column plan resolved once per schema
 │   ├── ctx.rs              # Shared module context
 │   ├── values.rs           # Zval <-> PHP value helpers

documentation/upgrading.md

@@ -221,6 +221,23 @@ after it shift by one. `Stage::physical` is new - see the core documentation.
 | `new Explain\TreeLayout($details, declarations: true)`  | `new Explain\Outline(declarations: true)`, the layout takes no arguments          |
 | `new Explain\BoxLayout($details)`                       | `new Explain\BoxLayout()`                                                         |
 
+### 22) `flow-php/etl-adapter-csv` - an escaped or bare enclosure no longer merges a record with the next line
+
+| Input (3 lines)              | Before                                 | After                                             |
+|------------------------------|----------------------------------------|---------------------------------------------------|
+| `a,b` / `"x\"y",1` / `"p",2` | 1 row: `{"a":"x\\\"y","b":"1\n\"p\""}` | 2 rows: `{"a":"x\\\"y","b":1}`, `{"a":"p","b":2}` |
+| `a,b` / `x"y,1` / `"p",2`    | 1 row: `{"a":"x\"y","b":"1\n\"p\""}`   | 2 rows: `{"a":"x\"y","b":1}`, `{"a":"p","b":2}`   |
+
+### 23) `flow-php/etl-adapter-csv` - `CSVOpenSource` is an interface, `CSVLineReader` takes the separator and escape
+
+| Before                                                                     | After                                                                                          |
+|----------------------------------------------------------------------------|------------------------------------------------------------------------------------------------|
+| `new CSVLineReader($enclosure, $charactersReadInLine, $removeBOM)`         | `new CSVLineReader($enclosure, $separator, $escape, $charactersReadInLine, $removeBOM)`        |
+| `new CSVOpenSource($stream, $dialect, $encoder, $lineReader)`              | `new PhpCSVOpenSource($stream, $encoder, $lineReader)`; `CSVOpenSource` is its interface       |
+| `$open->stream`, `$open->dialect`, `$open->encoder`, `$open->lineReader`   | removed                                                                                        |
+| `CSVFileReader::samples()` yields `Generator`s                             | yields `CSVFileSample` (`IteratorAggregate`); `$unit->getIterator()` for the generator         |
+| `CSVFileReader::sample($source)`                                           | `new CSVFileSample($opener, $source)`                                                          |
+
 ---
 
 ## Upgrading from 0.43.x to 0.44.x

src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVEnclosureScan.php

@@ -0,0 +1,81 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\CSV;
+
+use function str_replace;
+use function strcspn;
+use function strlen;
+use function strpos;
+use function strspn;
+
+/**
+ * CSVRecordBoundary's rules as a byte walk, for the buffers PCRE gives up on.
+ */
+final readonly class CSVEnclosureScan
+{
+    /**
+     * isspace() in the C locale, which is what fgetcsv skips before an opening enclosure.
+     */
+    public const string BLANKS = " \t\n\v\f\r";
+
+    private string $blanks;
+
+    private string $specials;
+
+    public function __construct(
+        private string $separator,
+        private string $enclosure,
+        string $escape,
+    ) {
+        $this->blanks = str_replace($separator, '', self::BLANKS);
+        $this->specials = $escape === $enclosure ? $enclosure : $enclosure . $escape;
+    }
+
+    public function endsOutsideAnEnclosure(string $buffer): bool
+    {
+        $length = strlen($buffer);
+        $position = 0;
+
+        while (true) {
+            $fieldStart = $position + strspn($buffer, $this->blanks, $position);
+
+            if ($fieldStart < $length && $buffer[$fieldStart] === $this->enclosure) {
+                $position = $fieldStart + 1;
+
+                while (true) {
+                    $position += strcspn($buffer, $this->specials, $position);
+
+                    if ($position >= $length) {
+                        return false;
+                    }
+
+                    if ($buffer[$position] !== $this->enclosure) {
+                        $position += 2;
+
+                        continue;
+                    }
+
+                    if (($position + 1) < $length && $buffer[$position + 1] === $this->enclosure) {
+                        $position += 2;
+
+                        continue;
+                    }
+
+                    $position++;
+
+                    break;
+                }
+            }
+
+            $nextSeparator = strpos($buffer, $this->separator, $position);
+
+            if ($nextSeparator === false) {
+                return true;
+            }
+
+            $position = $nextSeparator + 1;
+        }
+    }
+}

src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVEncoder.php

@@ -89,10 +89,8 @@ public function decode(array $batch): array
         $maps = [];
 
         foreach ($batch as $line) {
-            $fields = array_values(array_map(
-                static fn(mixed $field): ?string => is_string($field) ? $field : null,
-                str_getcsv($line, $this->separator, $this->enclosure, $this->escape),
-            ));
+            /** @var list<null|string> $fields */
+            $fields = str_getcsv($line, $this->separator, $this->enclosure, $this->escape);
 
             if ($this->headers === null) {
                 if ($this->withHeader) {

src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVFileReader.php

@@ -79,28 +79,14 @@ public function header(): CSVHeader
     }
 
     /**
-     * @return Generator<int, RawRowValues>
-     */
-    public function sample(SourceFile $source): Generator
-    {
-        $open = $this->opener->open($source);
-
-        try {
-            yield from $open->records();
-        } finally {
-            $open->close();
-        }
-    }
-
-    /**
-     * $rowBudget is deliberately unused: sample() is lazy and SchemaInferrer stops advancing it.
+     * $rowBudget is unused: SchemaInferrer hands each unit its remaining budget through sniffColumnTypes().
      *
-     * @return Generator<int, Generator<int, RawRowValues>>
+     * @return Generator<int, CSVFileSample>
      */
     public function samples(int $rowBudget): iterable
     {
         foreach ($this->sources as $source) {
-            yield $this->sample($source);
+            yield new CSVFileSample($this->opener, $source);
         }
     }
 }

src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVFileSample.php

@@ -0,0 +1,56 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\CSV;
+
+use Flow\ETL\Extractor\SourceFile;
+use Flow\ETL\Row\RawRowValues;
+use Flow\ETL\Schema\Inference\ColumnTypes;
+use Flow\ETL\Schema\Inference\SchemaInference;
+use Flow\ETL\Schema\Inference\SniffsColumnTypes;
+use Flow\Types\Type\TypeNarrower;
+use Generator;
+use IteratorAggregate;
+
+/**
+ * @implements IteratorAggregate<int, RawRowValues>
+ */
+final readonly class CSVFileSample implements IteratorAggregate, SniffsColumnTypes
+{
+    public function __construct(
+        private CSVSourceOpener $opener,
+        private SourceFile $source,
+    ) {}
+
+    /**
+     * The source is opened on the first advance; abandoning the generator closes it.
+     *
+     * @return Generator<int, RawRowValues>
+     */
+    public function getIterator(): Generator
+    {
+        $open = $this->opener->open($this->source);
+
+        try {
+            yield from $open->records();
+        } finally {
+            $open->close();
+        }
+    }
+
+    public function sniffColumnTypes(
+        array $names,
+        int $rowBudget,
+        SchemaInference $inference,
+        TypeNarrower $typer,
+    ): ColumnTypes {
+        $open = $this->opener->open($this->source);
+
+        try {
+            return $open->sniff($names, $rowBudget, $inference, $typer);
+        } finally {
+            $open->close();
+        }
+    }
+}

src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVLineReader.php

@@ -11,18 +11,23 @@
 use function str_contains;
 use function str_starts_with;
 use function substr;
-use function substr_count;
 
 final readonly class CSVLineReader
 {
+    private CSVRecordBoundary $boundary;
+
     /**
      * @param null|int<1, max> $charactersReadInLine
      */
     public function __construct(
         private string $enclosure,
+        string $separator = ',',
+        string $escape = '\\',
         private ?int $charactersReadInLine = null,
         private bool $removeBOM = true,
-    ) {}
+    ) {
+        $this->boundary = new CSVRecordBoundary($enclosure, $separator, $escape);
+    }
 
     /**
      * @return \Generator<int, string>
@@ -42,7 +47,7 @@ public function readLines(SourceStream $stream): Generator
                 $lineNumber++;
                 $buffer = '';
             } else {
-                if ($this->isCompleteCSVRecord($buffer)) {
+                if ($this->boundary->isComplete($buffer)) {
                     yield $this->removeBOM && $lineNumber === 0
                         ? $this->removeBOMFromLine(rtrim($buffer, "\r\n"))
                         : rtrim($buffer, "\r\n");
@@ -61,19 +66,6 @@ public function readLines(SourceStream $stream): Generator
         }
     }
 
-    /**
-     * Check if the current buffer contains a complete CSV record
-     * by counting enclosures and ensuring they are properly paired.
-     */
-    private function isCompleteCSVRecord(string $buffer): bool
-    {
-        if (!str_contains($buffer, $this->enclosure)) {
-            return true;
-        }
-
-        return (substr_count($buffer, $this->enclosure) % 2) === 0;
-    }
-
     /**
      * Remove Byte Order Mark (BOM) from the beginning of a line if present.
      */

src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVOpenSource.php

@@ -5,51 +5,34 @@
 namespace Flow\ETL\Adapter\CSV;
 
 use Flow\ETL\Row\RawRowValues;
-use Flow\Filesystem\SourceStream;
+use Flow\ETL\Schema\Inference\ColumnTypes;
+use Flow\ETL\Schema\Inference\SchemaInference;
+use Flow\Types\Type\TypeNarrower;
 use Generator;
 
-final readonly class CSVOpenSource
+interface CSVOpenSource
 {
-    public function __construct(
-        public SourceStream $stream,
-        public CSVDialect $dialect,
-        public CSVEncoder $encoder,
-        public CSVLineReader $lineReader,
-    ) {}
-
-    public function close(): void
-    {
-        $this->stream->close();
-    }
+    public function close(): void;
 
     /**
      * This instance is consumed afterwards.
      *
      * @return list<string>
      */
-    public function columns(): array
-    {
-        foreach ($this->lineReader->readLines($this->stream) as $line) {
-            $this->encoder->decode([$line]);
-
-            break;
-        }
-
-        return $this->encoder->headers() ?? [];
-    }
+    public function columns(): array;
 
     /**
-     * CSVLineReader::readLines() already joins a quoted multi-line record, so never re-split or re-join here.
      * This instance is consumed afterwards.
      *
      * @return Generator<int, RawRowValues>
      */
-    public function records(): Generator
-    {
-        foreach ($this->lineReader->readLines($this->stream) as $line) {
-            foreach ($this->encoder->decode([$line]) as $values) {
-                yield $values;
-            }
-        }
-    }
+    public function records(): Generator;
+
+    /**
+     * SchemaInferrer::sniff() over records(). This instance is consumed afterwards.
+     *
+     * @param list<string> $names
+     * @param int<0, max>|-1 $rowBudget
+     */
+    public function sniff(array $names, int $rowBudget, SchemaInference $inference, TypeNarrower $typer): ColumnTypes;
 }

src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVRecordBoundary.php

@@ -0,0 +1,56 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\CSV;
+
+use function preg_match;
+use function preg_quote;
+use function str_contains;
+use function str_replace;
+
+/**
+ * A record is complete when the buffer ends outside an enclosure. An enclosure only OPENS at a field start - buffer
+ * start or just after the separator, blanks allowed - which is what fgetcsv does and what counting enclosures cannot
+ * express. Possessive quantifiers throughout: the pattern must stay linear on a multi-megabyte buffer.
+ */
+final readonly class CSVRecordBoundary
+{
+    private string $pattern;
+
+    private CSVEnclosureScan $scan;
+
+    public function __construct(
+        private string $enclosure,
+        string $separator = ',',
+        string $escape = '\\',
+    ) {
+        $quotedSeparator = preg_quote($separator, '/');
+        $quotedEnclosure = preg_quote($enclosure, '/');
+        $quotedEscape = preg_quote($escape, '/');
+
+        $blanks = '[' . preg_quote(str_replace($separator, '', CSVEnclosureScan::BLANKS), '/') . ']*+';
+        $escapedByte = $escape === '' || $escape === $enclosure ? '' : $quotedEscape . '.|';
+        $plainBytes = '[^' . $quotedEnclosure . ($escapedByte === '' ? '' : $quotedEscape) . ']++';
+        $enclosedBytes = '(?:' . $plainBytes . '|' . $escapedByte . $quotedEnclosure . $quotedEnclosure . ')*+';
+        $enclosedField = $quotedEnclosure . $enclosedBytes . $quotedEnclosure . '[^' . $quotedSeparator . ']*+';
+        $unenclosedField = '(?!' . $quotedEnclosure . ')[^' . $quotedSeparator . ']*+';
+        $fieldEnd = '(?:' . $quotedSeparator . '|\z)';
+
+        $this->pattern =
+            '/\A(?:' . $blanks . '(?:' . $enclosedField . '|' . $unenclosedField . ')' . $fieldEnd . ')*+\z/s';
+        $this->scan = new CSVEnclosureScan($separator, $enclosure, $escape);
+    }
+
+    public function isComplete(string $buffer): bool
+    {
+        if (!str_contains($buffer, $this->enclosure)) {
+            return true;
+        }
+
+        $matched = preg_match($this->pattern, $buffer);
+
+        // PCRE gives up at about 142k fields in one record; reading that as "incomplete" would glue every following line
+        return $matched === false ? $this->scan->endsOutsideAnEnclosure($buffer) : $matched === 1;
+    }
+}