From 15b6b2287ec963d92a25b4415be78c1dd74944c5 Mon Sep 17 00:00:00 2001 From: Hans Peter Nielsen Date: Fri, 4 Sep 2026 13:32:52 +0200 Subject: [PATCH 1/3] perf: JIRA-17851 stream export rows instead of holding them in memory filterQuery() built every row of an export into one array and returned it, and normalizeOptionalItems() then made a second full pass over that array to fill in the optional columns. Memory grew with the size of the export, and exports around 100,000 rows were exhausting a 2 GB process limit even after the generator had been made to stream. Rows are now written to a buffer as they are built. The buffer lives in memory up to 8 MB and spills to a temp file beyond that, and it tracks the fixed and optional column names as rows arrive. Reading it back fills each row out to the final column list, so the optional-column behaviour is unchanged: optional columns follow the fixed ones in the order first seen, an optional value takes precedence over a fixed column of the same name, and a row without a value gets an empty string. The CSV writer consumes the rows one at a time and takes the header from the column list, so nothing holds the whole export at once. Measured on 50,000 rows: memory growth stays under 24 MB where it was previously proportional to the row count, and a three-chunk export no longer grows between chunks. BREAKING CHANGE: filterQuery() returns a RowStream rather than an array, and ProcessorData accepts any iterable of rows. ProcessorData gains rows(), getColumns() and getCount(); getData() still returns an array but now loads every row to do so, and should not be used for large exports. getAllOptionalKeys() and normalizeOptionalItems() are removed; their behaviour lives in RowStream. Co-Authored-By: Claude Opus 5 --- phpstan-baseline.neon | 5 - src/Generator/CsvDriver.php | 32 +---- src/Processor/EloquentProcessor.php | 83 ++--------- src/Processor/ProcessorData.php | 47 ++++++- src/Processor/RowStream.php | 130 ++++++++++++++++++ tests/Feature/Generator/CsvDriverTest.php | 10 ++ tests/Feature/Processor/ProcessorDataTest.php | 30 ++++ tests/Feature/Processor/RowStreamTest.php | 69 ++++++++++ 8 files changed, 299 insertions(+), 107 deletions(-) create mode 100644 src/Processor/RowStream.php create mode 100644 tests/Feature/Processor/RowStreamTest.php diff --git a/phpstan-baseline.neon b/phpstan-baseline.neon index a1ab11f..18f930b 100644 --- a/phpstan-baseline.neon +++ b/phpstan-baseline.neon @@ -120,11 +120,6 @@ parameters: count: 1 path: src/Processor/EloquentProcessor.php - - - message: '#^Parameter \#2 \$value of method Illuminate\\Support\\Collection\<\(int\|string\),string\>\:\:put\(\) expects string, array given\.$#' - identifier: argument.type - count: 1 - path: src/Processor/EloquentProcessor.php - message: '#^PHP files should declare strict types\.$#' diff --git a/src/Generator/CsvDriver.php b/src/Generator/CsvDriver.php index 4a85a4b..c420f04 100644 --- a/src/Generator/CsvDriver.php +++ b/src/Generator/CsvDriver.php @@ -69,7 +69,7 @@ public function saveToStorage( path: $filepath, size: Storage::size($filepath), url: Storage::url($filepath), - count: count($processorData->getData()), + count: $processorData->getCount(), mimeType: 'text/csv', ); } @@ -97,12 +97,7 @@ public function toStream(ProcessorData $processorData) */ public function writeCsv(ProcessorData $processorData, $stream): int { - if (! is_resource($stream)) { - throw new ValueError('The stream must be a resource.'); - } - - $rows = $processorData->getData(); - $columns = $this->columns($rows); + $columns = $processorData->getColumns(); if ($columns === []) { return 0; @@ -112,7 +107,7 @@ public function writeCsv(ProcessorData $processorData, $stream): int $written = 0; - foreach ($rows as $row) { + foreach ($processorData->rows() as $row) { $fields = []; foreach ($columns as $column) { @@ -126,27 +121,6 @@ public function writeCsv(ProcessorData $processorData, $stream): int return $written; } - /** - * The header names every column any row has, in the order first seen, so a - * row whose keys differ cannot line up against the wrong header. - * - * @param array> $rows - * - * @return array - */ - private function columns(array $rows): array - { - $columns = []; - - foreach ($rows as $row) { - foreach (array_keys($row) as $key) { - $columns[$key] = true; - } - } - - return array_keys($columns); - } - /** * @param resource $stream * @param array $fields diff --git a/src/Processor/EloquentProcessor.php b/src/Processor/EloquentProcessor.php index df68a64..6c8c109 100644 --- a/src/Processor/EloquentProcessor.php +++ b/src/Processor/EloquentProcessor.php @@ -87,16 +87,15 @@ protected function columns(): Collection }); } - protected function filterQuery(Builder $query): array + protected function filterQuery(Builder $query): RowStream { $columns = $this->columns(); $allowedColumns = $columns->keys(); - $items = collect(); + $rows = new RowStream(); - $query->chunk(1000, function ($data) use ($columns, $allowedColumns, $items) { - // Combine columns plus additional and optional fields - $filtered = $data->map(function ($item) use ($columns, $allowedColumns) { + $query->chunk(1000, function ($data) use ($columns, $allowedColumns, $rows) { + foreach ($data as $item) { $additional = $this->additional($item); $optional = $this->optional($item); $additionalKeys = array_keys($additional); @@ -106,13 +105,13 @@ protected function filterQuery(Builder $query): array $attributes = $allowedColumns->isEmpty() ? [] : $this->serialiseAttributes($item); // Merge fieldsets - $item = collect(array_merge($attributes, $additional)); + $row = collect(array_merge($attributes, $additional)); // Only let through the desired columns - $item = $item->only($allowedColumns->concat($additionalKeys)); + $row = $row->only($allowedColumns->concat($additionalKeys)); // Ensure some values are properly casted - $item = $item->map(function ($value) { + $row = $row->map(function ($value) { if (is_bool($value)) { $value = (int) $value; } @@ -120,75 +119,15 @@ protected function filterQuery(Builder $query): array return strval($value); }); - // add optional into the item as array so we can later - // normalize the columns - $item->put('optional', $optional); - // update keys based on desired key names provided in the processor - $item = $item->keyBy(function ($value, $key) use ($columns) { + $row = $row->keyBy(function ($value, $key) use ($columns) { return $columns->get($key, $key); })->all(); - // we return an array - return $item; - }); - - $items->push($filtered); - }); - - $formattedItems = $this->normalizeOptionalItems($items->flatten(1)->toArray()); - - return $formattedItems; - } - - protected function getAllOptionalKeys(array $items): array - { - $optionalKeys = []; - - foreach ($items as $item) { - $optional = $item['optional']; - - if (count($optional)) { - foreach ($item['optional'] as $optional) { - $key = key($optional); - if (! in_array($key, $optionalKeys)) { - $optionalKeys[] = $key; - } - } - } - } - - return $optionalKeys; - } - - protected function normalizeOptionalItems(array $items): array - { - // get optional keys - $optionalKeys = $this->getAllOptionalKeys($items); - - // go through each item - foreach ($items as $key => $item) { - $itemOptionals = call_user_func_array('array_merge', $items[$key]['optional']); - - // for each item we wanna add each optional key - // and also populate value if we got one - foreach ($optionalKeys as $optionalArrKey => $optionalKey) { - $value = ''; - - // check if we have value for this item and populate it - if (isset($itemOptionals[$optionalKey])) { - $value = $itemOptionals[$optionalKey]; - } - - // add the optional key and it's value - $items[$key][$optionalKey] = $value; + $rows->push($row, $optional); } + }); - // unset our "temp" optional array as now we have them - // in the correct format - unset($items[$key]['optional']); - } - - return $items; + return $rows; } } diff --git a/src/Processor/ProcessorData.php b/src/Processor/ProcessorData.php index afe66db..4d6449e 100644 --- a/src/Processor/ProcessorData.php +++ b/src/Processor/ProcessorData.php @@ -4,17 +4,62 @@ class ProcessorData { + /** + * @param iterable> $data + */ public function __construct( - private readonly array $data, + private readonly iterable $data, private readonly string $type, ) { } + /** + * Every row, loaded into memory. Prefer rows() for anything large. + * + * @return array> + */ public function getData(): array + { + return is_array($this->data) ? $this->data : iterator_to_array($this->data, false); + } + + /** + * The rows, one at a time. + * + * @return iterable> + */ + public function rows(): iterable { return $this->data; } + public function getCount(): int + { + return is_countable($this->data) ? count($this->data) : count($this->getData()); + } + + /** + * Every column any row has, in the order first seen. + * + * @return array + */ + public function getColumns(): array + { + if ($this->data instanceof RowStream) { + return $this->data->columns(); + } + + $columns = []; + + foreach ($this->data as $row) { + foreach (array_keys($row) as $key) { + $columns[$key] = true; + } + } + + return array_keys($columns); + } + public function getType(): string { return $this->type; diff --git a/src/Processor/RowStream.php b/src/Processor/RowStream.php new file mode 100644 index 0000000..a597370 --- /dev/null +++ b/src/Processor/RowStream.php @@ -0,0 +1,130 @@ +> + */ +final class RowStream implements IteratorAggregate, Countable +{ + /** Rows stay in memory up to this size and go to a temp file beyond it. */ + private const BUFFER_BYTES = 8 * 1024 * 1024; + + /** @var resource */ + private $buffer; + + /** @var array the fixed columns, in the order first seen */ + private array $columns = []; + + /** @var array the optional columns, in the order first seen */ + private array $optionalColumns = []; + + private int $count = 0; + + public function __construct() + { + $buffer = fopen('php://temp/maxmemory:' . self::BUFFER_BYTES, 'w+b'); + + if ($buffer === false) { + throw new RuntimeException('Unable to open a buffer to hold the export rows.'); + } + + $this->buffer = $buffer; + } + + public function __destruct() + { + if (is_resource($this->buffer)) { + fclose($this->buffer); + } + } + + /** + * @param array $row the row's fixed columns + * @param array> $optional one single-entry array per optional value + */ + public function push(array $row, array $optional): void + { + foreach (array_keys($row) as $key) { + $this->columns[$key] = true; + } + + foreach ($optional as $entry) { + foreach (array_keys($entry) as $key) { + $this->optionalColumns[$key] = true; + } + } + + fwrite($this->buffer, json_encode( + [$row, $optional], + JSON_THROW_ON_ERROR|JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES + ) . "\n"); + + $this->count++; + } + + /** + * The fixed columns first, then any optional column that is not already one of them. + * + * @return array + */ + public function columns(): array + { + return array_merge( + array_keys($this->columns), + array_keys(array_diff_key($this->optionalColumns, $this->columns)), + ); + } + + public function count(): int + { + return $this->count; + } + + /** + * Every row, filled out to the final column list. Can be iterated more than once. + * + * @return Generator> + */ + public function getIterator(): Generator + { + $columns = $this->columns(); + + rewind($this->buffer); + + while (($line = fgets($this->buffer)) !== false) { + /** @var array{0: array, 1: array>} $decoded */ + $decoded = json_decode($line, true, 512, JSON_THROW_ON_ERROR); + [$row, $optional] = $decoded; + + $optionals = $optional === [] ? [] : array_merge(...$optional); + + $filled = []; + + foreach ($columns as $column) { + // An optional column wins over a fixed one of the same name, and a + // row without a value for it gets an empty string. + if (isset($this->optionalColumns[$column])) { + $filled[$column] = isset($optionals[$column]) ? $optionals[$column] : ''; + } else { + $filled[$column] = $row[$column] ?? ''; + } + } + + yield $filled; + } + } +} diff --git a/tests/Feature/Generator/CsvDriverTest.php b/tests/Feature/Generator/CsvDriverTest.php index 117c2ae..306bbe0 100644 --- a/tests/Feature/Generator/CsvDriverTest.php +++ b/tests/Feature/Generator/CsvDriverTest.php @@ -123,3 +123,13 @@ it('returns an empty string when there is nothing to export', function () { expect((new CsvDriver())->exportToCsv(new ProcessorData([], 'empty')))->toBe(''); }); + +it('writes a row stream without loading it, with the header from its columns', function () { + $rows = new \Worksome\DataExport\Processor\RowStream(); + $rows->push(['id' => '1', 'name' => 'One'], [['UK' => 'applies']]); + $rows->push(['id' => '2', 'name' => 'Two'], []); + + $csv = (new CsvDriver())->exportToCsv(new ProcessorData($rows, 'people')); + + expect($csv)->toBe("id,name,UK\r\n1,One,applies\r\n2,Two,\r\n"); +}); diff --git a/tests/Feature/Processor/ProcessorDataTest.php b/tests/Feature/Processor/ProcessorDataTest.php index 4027f88..51e3814 100644 --- a/tests/Feature/Processor/ProcessorDataTest.php +++ b/tests/Feature/Processor/ProcessorDataTest.php @@ -15,3 +15,33 @@ expect($processorData->getType())->toBe('randomstuff'); }); + +it('streams rows and reports the columns and count for a row stream', function () { + $rows = new \Worksome\DataExport\Processor\RowStream(); + $rows->push(['id' => '1'], [['UK' => 'applies']]); + $rows->push(['id' => '2'], []); + + $processorData = new ProcessorData($rows, 'compliance'); + + expect($processorData->getColumns())->toBe(['id', 'UK']) + ->and($processorData->getCount())->toBe(2) + ->and(iterator_to_array($processorData->rows(), false))->toBe([ + ['id' => '1', 'UK' => 'applies'], + ['id' => '2', 'UK' => ''], + ]) + ->and($processorData->getData())->toBe([ + ['id' => '1', 'UK' => 'applies'], + ['id' => '2', 'UK' => ''], + ]); +}); + +it('derives the columns and count from a plain array of rows', function () { + $processorData = new ProcessorData([ + ['a' => '1', 'b' => '2'], + ['b' => '3', 'c' => '4'], + ], 'shapes'); + + expect($processorData->getColumns())->toBe(['a', 'b', 'c']) + ->and($processorData->getCount())->toBe(2) + ->and($processorData->rows())->toBe($processorData->getData()); +}); diff --git a/tests/Feature/Processor/RowStreamTest.php b/tests/Feature/Processor/RowStreamTest.php new file mode 100644 index 0000000..0342054 --- /dev/null +++ b/tests/Feature/Processor/RowStreamTest.php @@ -0,0 +1,69 @@ +push(['id' => '1', 'name' => 'One'], [['UK' => 'applies']]); + $rows->push(['id' => '2', 'name' => 'Two'], [['US' => 'none'], ['UK' => 'exempt']]); + $rows->push(['id' => '3', 'name' => 'Three'], []); + + expect($rows->columns())->toBe(['id', 'name', 'UK', 'US']) + ->and(count($rows))->toBe(3) + ->and(iterator_to_array($rows, false))->toBe([ + ['id' => '1', 'name' => 'One', 'UK' => 'applies', 'US' => ''], + ['id' => '2', 'name' => 'Two', 'UK' => 'exempt', 'US' => 'none'], + ['id' => '3', 'name' => 'Three', 'UK' => '', 'US' => ''], + ]); +}); + +it('lets an optional value take precedence over a fixed column of the same name', function () { + $rows = new RowStream(); + $rows->push(['id' => '1', 'status' => 'fixed'], [['status' => 'optional']]); + $rows->push(['id' => '2', 'status' => 'fixed'], []); + + // The column keeps its fixed position, and a row without the optional value is blank + expect($rows->columns())->toBe(['id', 'status']) + ->and(iterator_to_array($rows, false))->toBe([ + ['id' => '1', 'status' => 'optional'], + ['id' => '2', 'status' => ''], + ]); +}); + +it('can be read more than once', function () { + $rows = new RowStream(); + $rows->push(['a' => '1'], []); + + expect(iterator_to_array($rows, false))->toBe([['a' => '1']]) + ->and(iterator_to_array($rows, false))->toBe([['a' => '1']]); +}); + +it('yields nothing and no columns when empty', function () { + $rows = new RowStream(); + + expect($rows->columns())->toBe([]) + ->and(count($rows))->toBe(0) + ->and(iterator_to_array($rows, false))->toBe([]); +}); + +it('does not hold the rows in memory', function () { + $rows = new RowStream(); + $row = array_fill_keys(array_map(fn ($i) => "column_$i", range(1, 20)), str_repeat('x', 40)); + + $before = memory_get_usage(true); + for ($i = 0; $i < 50000; $i++) { + $rows->push($row, [['extra' => 'y']]); + } + $count = 0; + foreach ($rows as $filled) { + $count++; + } + $grown = (memory_get_peak_usage(true) - $before) / 1048576; + + // 50k rows of ~1 KB each would be ~50 MB if kept in memory. The buffer spills + // to disk past 8 MB, so growth stays a small constant however many rows there are. + expect($count)->toBe(50000) + ->and($grown)->toBeLessThan(24); +}); From a81bce83ff38b0c5b1b10f823d723d721d21bf76 Mon Sep 17 00:00:00 2001 From: Hans Peter Nielsen Date: Fri, 4 Sep 2026 14:05:14 +0200 Subject: [PATCH 2/3] fix: JIRA-17851 harden the export row buffer after review ProcessorData now takes an array or a RowStream rather than any iterable. A one-shot iterable would be exhausted by the column lookup before the rows were read, and nothing passes one anyway. The buffer stores each value as the string the file will contain and uses a length-framed serialize() spool rather than JSON lines. JSON turned a Stringable into an array and refused bytes that are not valid UTF-8; both now round-trip exactly, and a value that cannot become a string fails at once with a clear message. Writes are checked against the frame length and the row count only moves after a complete write, and reads verify each frame so a truncated buffer fails loudly rather than yielding a partial row. The CSV writer checks the stream is a resource again before doing any work, which the streaming rewrite had dropped. The memory test resets the process peak before its baseline so the bound measures only itself. Co-Authored-By: Claude Opus 5 --- src/Generator/CsvDriver.php | 4 ++ src/Processor/ProcessorData.php | 9 ++- src/Processor/RowStream.php | 84 ++++++++++++++++++----- tests/Feature/Generator/CsvDriverTest.php | 5 ++ tests/Feature/Processor/RowStreamTest.php | 36 ++++++++++ 5 files changed, 117 insertions(+), 21 deletions(-) diff --git a/src/Generator/CsvDriver.php b/src/Generator/CsvDriver.php index c420f04..443cc0c 100644 --- a/src/Generator/CsvDriver.php +++ b/src/Generator/CsvDriver.php @@ -97,6 +97,10 @@ public function toStream(ProcessorData $processorData) */ public function writeCsv(ProcessorData $processorData, $stream): int { + if (! is_resource($stream)) { + throw new ValueError('The stream must be a resource.'); + } + $columns = $processorData->getColumns(); if ($columns === []) { diff --git a/src/Processor/ProcessorData.php b/src/Processor/ProcessorData.php index 4d6449e..bbc6173 100644 --- a/src/Processor/ProcessorData.php +++ b/src/Processor/ProcessorData.php @@ -5,10 +5,13 @@ class ProcessorData { /** - * @param iterable> $data + * Rows either already built as an array, or held in a stream that can be read + * more than once. A one-shot iterable would be exhausted by the first reader. + * + * @param array>|RowStream $data */ public function __construct( - private readonly iterable $data, + private readonly array|RowStream $data, private readonly string $type, ) { } @@ -35,7 +38,7 @@ public function rows(): iterable public function getCount(): int { - return is_countable($this->data) ? count($this->data) : count($this->getData()); + return count($this->data); } /** diff --git a/src/Processor/RowStream.php b/src/Processor/RowStream.php index a597370..d9733c8 100644 --- a/src/Processor/RowStream.php +++ b/src/Processor/RowStream.php @@ -8,6 +8,7 @@ use Generator; use IteratorAggregate; use RuntimeException; +use Stringable; /** * The rows of an export, kept in a buffer that spills to disk rather than in memory. @@ -16,7 +17,7 @@ * known - an optional column only exists once some row has a value for it - * so reading the rows back fills each one out to the final column list. * - * @implements IteratorAggregate> + * @implements IteratorAggregate> */ final class RowStream implements IteratorAggregate, Countable { @@ -53,25 +54,38 @@ public function __destruct() } /** + * Values are stored as the strings the file will contain, so nothing is lost + * or reinterpreted on the way through the buffer. + * * @param array $row the row's fixed columns * @param array> $optional one single-entry array per optional value */ public function push(array $row, array $optional): void { - foreach (array_keys($row) as $key) { + $fixed = []; + + foreach ($row as $key => $value) { $this->columns[$key] = true; + $fixed[$key] = $this->stringify($value); } + // A later optional value for the same column wins, as array_merge would have it. + $merged = []; + foreach ($optional as $entry) { - foreach (array_keys($entry) as $key) { + foreach ($entry as $key => $value) { $this->optionalColumns[$key] = true; + $merged[$key] = $this->stringify($value); } } - fwrite($this->buffer, json_encode( - [$row, $optional], - JSON_THROW_ON_ERROR|JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES - ) . "\n"); + // Length-framed so a value may contain anything, including newlines. + $payload = serialize([$fixed, $merged]); + $frame = strlen($payload) . "\n" . $payload; + + if (fwrite($this->buffer, $frame) !== strlen($frame)) { + throw new RuntimeException('Unable to write an export row to the buffer.'); + } $this->count++; } @@ -97,7 +111,7 @@ public function count(): int /** * Every row, filled out to the final column list. Can be iterated more than once. * - * @return Generator> + * @return Generator> */ public function getIterator(): Generator { @@ -105,26 +119,60 @@ public function getIterator(): Generator rewind($this->buffer); - while (($line = fgets($this->buffer)) !== false) { - /** @var array{0: array, 1: array>} $decoded */ - $decoded = json_decode($line, true, 512, JSON_THROW_ON_ERROR); - [$row, $optional] = $decoded; + while (($header = fgets($this->buffer)) !== false) { + $payload = $this->read((int) $header); - $optionals = $optional === [] ? [] : array_merge(...$optional); + /** @var array{0: array, 1: array} $decoded */ + $decoded = unserialize($payload, ['allowed_classes' => false]); + [$fixed, $optionals] = $decoded; $filled = []; foreach ($columns as $column) { // An optional column wins over a fixed one of the same name, and a // row without a value for it gets an empty string. - if (isset($this->optionalColumns[$column])) { - $filled[$column] = isset($optionals[$column]) ? $optionals[$column] : ''; - } else { - $filled[$column] = $row[$column] ?? ''; - } + $filled[$column] = isset($this->optionalColumns[$column]) + ? ($optionals[$column] ?? '') + : ($fixed[$column] ?? ''); } yield $filled; } } + + private function read(int $length): string + { + $payload = ''; + + while (strlen($payload) < $length) { + $chunk = fread($this->buffer, $length - strlen($payload)); + + if ($chunk === false || $chunk === '') { + throw new RuntimeException('The export row buffer is truncated.'); + } + + $payload .= $chunk; + } + + return $payload; + } + + private function stringify(mixed $value): string + { + if ($value === null) { + return ''; + } + + if (is_string($value)) { + return $value; + } + + if (is_scalar($value) || $value instanceof Stringable) { + return (string) $value; + } + + throw new RuntimeException( + sprintf('An export value must be a string or stringable, %s given.', get_debug_type($value)) + ); + } } diff --git a/tests/Feature/Generator/CsvDriverTest.php b/tests/Feature/Generator/CsvDriverTest.php index 306bbe0..c5b4610 100644 --- a/tests/Feature/Generator/CsvDriverTest.php +++ b/tests/Feature/Generator/CsvDriverTest.php @@ -133,3 +133,8 @@ expect($csv)->toBe("id,name,UK\r\n1,One,applies\r\n2,Two,\r\n"); }); + +it('rejects a stream that is not a resource, even when there is nothing to write', function () { + expect(fn () => (new CsvDriver())->writeCsv(new ProcessorData([], 'empty'), 'not a stream')) + ->toThrow(\ValueError::class); +}); diff --git a/tests/Feature/Processor/RowStreamTest.php b/tests/Feature/Processor/RowStreamTest.php index 0342054..dd3ff92 100644 --- a/tests/Feature/Processor/RowStreamTest.php +++ b/tests/Feature/Processor/RowStreamTest.php @@ -48,10 +48,46 @@ ->and(iterator_to_array($rows, false))->toBe([]); }); +it('keeps values byte for byte, including newlines and bytes that are not valid UTF-8', function () { + $rows = new RowStream(); + $rows->push(['note' => "line one\nline two"], [['raw' => "bad \xB1\xFF bytes"]]); + + expect(iterator_to_array($rows, false))->toBe([ + ['note' => "line one\nline two", 'raw' => "bad \xB1\xFF bytes"], + ]); +}); + +it('stores values as the strings the file will contain', function () { + $stringable = new class() implements \Stringable { + public function __toString(): string + { + return 'from __toString'; + } + }; + + $rows = new RowStream(); + $rows->push( + ['id' => 7, 'flag' => true, 'off' => false, 'none' => null], + [['label' => $stringable], ['rate' => 12.5]] + ); + + expect(iterator_to_array($rows, false))->toBe([ + ['id' => '7', 'flag' => '1', 'off' => '', 'none' => '', 'label' => 'from __toString', 'rate' => '12.5'], + ]); +}); + +it('refuses a value it cannot turn into a string', function () { + $rows = new RowStream(); + + expect(fn () => $rows->push(['id' => ['not', 'a', 'string']], [])) + ->toThrow(\RuntimeException::class); +}); + it('does not hold the rows in memory', function () { $rows = new RowStream(); $row = array_fill_keys(array_map(fn ($i) => "column_$i", range(1, 20)), str_repeat('x', 40)); + memory_reset_peak_usage(); $before = memory_get_usage(true); for ($i = 0; $i < 50000; $i++) { $rows->push($row, [['extra' => 'y']]); From cee7fd15cf9486b5ebf25d90c32ab4595bcf8b65 Mon Sep 17 00:00:00 2001 From: Hans Peter Nielsen Date: Fri, 4 Sep 2026 14:54:17 +0200 Subject: [PATCH 3/3] fix: JIRA-17851 restore the legacy header order and drop getData() Second review round on the row stream. The header put every fixed column before the optional ones. The old normalisation appended the optional keys to every row and the writer took the header as the union of row keys in first-seen order, so a fixed column that first appeared in a later row used to come after the optional columns. The stream now remembers how many fixed columns the first row had and reproduces that order, with a test for the case. ProcessorData::getData() is removed on the maintainer's request. It was the one way left to pull a whole export back into memory, the package tests were its only callers, and this is a major release. Co-Authored-By: Claude Fable 5.1 --- src/Processor/ProcessorData.php | 10 --------- src/Processor/RowStream.php | 21 ++++++++++++++----- .../Processor/EloquentProcessorTest.php | 12 +++++------ tests/Feature/Processor/ProcessorDataTest.php | 11 +++++----- tests/Feature/Processor/RowStreamTest.php | 13 ++++++++++++ 5 files changed, 40 insertions(+), 27 deletions(-) diff --git a/src/Processor/ProcessorData.php b/src/Processor/ProcessorData.php index bbc6173..656694c 100644 --- a/src/Processor/ProcessorData.php +++ b/src/Processor/ProcessorData.php @@ -16,16 +16,6 @@ public function __construct( ) { } - /** - * Every row, loaded into memory. Prefer rows() for anything large. - * - * @return array> - */ - public function getData(): array - { - return is_array($this->data) ? $this->data : iterator_to_array($this->data, false); - } - /** * The rows, one at a time. * diff --git a/src/Processor/RowStream.php b/src/Processor/RowStream.php index d9733c8..cd8263a 100644 --- a/src/Processor/RowStream.php +++ b/src/Processor/RowStream.php @@ -33,6 +33,9 @@ final class RowStream implements IteratorAggregate, Countable /** @var array the optional columns, in the order first seen */ private array $optionalColumns = []; + /** How many of the fixed columns the first row had. */ + private int $firstRowColumnCount = 0; + private int $count = 0; public function __construct() @@ -69,6 +72,10 @@ public function push(array $row, array $optional): void $fixed[$key] = $this->stringify($value); } + if ($this->count === 0) { + $this->firstRowColumnCount = count($this->columns); + } + // A later optional value for the same column wins, as array_merge would have it. $merged = []; @@ -91,16 +98,20 @@ public function push(array $row, array $optional): void } /** - * The fixed columns first, then any optional column that is not already one of them. + * The first row's columns, then the optional columns, then any column that only + * appears in a later row. A column keeps the first position it is given. * * @return array */ public function columns(): array { - return array_merge( - array_keys($this->columns), - array_keys(array_diff_key($this->optionalColumns, $this->columns)), - ); + $fixed = array_keys($this->columns); + + return array_values(array_unique(array_merge( + array_slice($fixed, 0, $this->firstRowColumnCount), + array_keys($this->optionalColumns), + array_slice($fixed, $this->firstRowColumnCount), + ))); } public function count(): int diff --git a/tests/Feature/Processor/EloquentProcessorTest.php b/tests/Feature/Processor/EloquentProcessorTest.php index f6a7f63..9829eea 100644 --- a/tests/Feature/Processor/EloquentProcessorTest.php +++ b/tests/Feature/Processor/EloquentProcessorTest.php @@ -19,7 +19,7 @@ $processedData = $processor->process($export); - $data = $processedData->getData(); + $data = iterator_to_array($processedData->rows(), false); expect($data)->toBeEmpty(); }); @@ -41,7 +41,7 @@ $processedData = $processor->process($export); - $data = $processedData->getData(); + $data = iterator_to_array($processedData->rows(), false); expect($data)->not->toBeEmpty(); expect($data)->toHaveCount(2); @@ -71,7 +71,7 @@ $export = new Export(); $processor = new FakeProcessorWithOptionalDriver(); $processedData = $processor->process($export); - $data = $processedData->getData(); + $data = iterator_to_array($processedData->rows(), false); foreach ($data as $item) { $this->assertArrayHasKey('Compliance UK', $item); @@ -103,7 +103,7 @@ $queries[] = $query->sql; }); - $data = (new FakeProcessorWithRelationDriver())->process(new Export())->getData(); + $data = iterator_to_array((new FakeProcessorWithRelationDriver())->process(new Export())->rows(), false); // One query for the users, one for the eagerly loaded teams. Serialising a // team would query its members on top of that. @@ -123,7 +123,7 @@ $queries[] = $query->sql; }); - $data = (new FakeProcessorWithoutColumnsDriver())->process(new Export())->getData(); + $data = iterator_to_array((new FakeProcessorWithoutColumnsDriver())->process(new Export())->rows(), false); // Only the chunk query over teams. Reading a team's attributes would run its // appended accessor, which queries the team's members. @@ -141,7 +141,7 @@ 'email' => 'one@example.com', ]); - $data = (new FakeProcessorWithCustomToArrayDriver())->process(new Export())->getData(); + $data = iterator_to_array((new FakeProcessorWithCustomToArrayDriver())->process(new Export())->rows(), false); // The model redacts the address when it serialises itself, so the export // must not fall back to the raw attribute. diff --git a/tests/Feature/Processor/ProcessorDataTest.php b/tests/Feature/Processor/ProcessorDataTest.php index 51e3814..9a89b29 100644 --- a/tests/Feature/Processor/ProcessorDataTest.php +++ b/tests/Feature/Processor/ProcessorDataTest.php @@ -9,7 +9,7 @@ 'foo' => 'Bar', ], 'randomstuff'); - expect($processorData->getData())->toBe([ + expect($processorData->rows())->toBe([ 'foo' => 'Bar', ]); @@ -28,10 +28,6 @@ ->and(iterator_to_array($processorData->rows(), false))->toBe([ ['id' => '1', 'UK' => 'applies'], ['id' => '2', 'UK' => ''], - ]) - ->and($processorData->getData())->toBe([ - ['id' => '1', 'UK' => 'applies'], - ['id' => '2', 'UK' => ''], ]); }); @@ -43,5 +39,8 @@ expect($processorData->getColumns())->toBe(['a', 'b', 'c']) ->and($processorData->getCount())->toBe(2) - ->and($processorData->rows())->toBe($processorData->getData()); + ->and($processorData->rows())->toBe([ + ['a' => '1', 'b' => '2'], + ['b' => '3', 'c' => '4'], + ]); }); diff --git a/tests/Feature/Processor/RowStreamTest.php b/tests/Feature/Processor/RowStreamTest.php index dd3ff92..a445636 100644 --- a/tests/Feature/Processor/RowStreamTest.php +++ b/tests/Feature/Processor/RowStreamTest.php @@ -19,6 +19,19 @@ ]); }); +it('places a column that only appears in a later row after the optional columns', function () { + $rows = new RowStream(); + $rows->push(['id' => '1'], [['opt' => 'a']]); + $rows->push(['id' => '2', 'later' => 'b'], []); + + // The first row's columns, then the optional ones, then the newcomer + expect($rows->columns())->toBe(['id', 'opt', 'later']) + ->and(iterator_to_array($rows, false))->toBe([ + ['id' => '1', 'opt' => 'a', 'later' => ''], + ['id' => '2', 'opt' => '', 'later' => 'b'], + ]); +}); + it('lets an optional value take precedence over a fixed column of the same name', function () { $rows = new RowStream(); $rows->push(['id' => '1', 'status' => 'fixed'], [['status' => 'optional']]);