From d504b3993871ca85e45570b7c71301e5293e1cff Mon Sep 17 00:00:00 2001 From: Joseph Bielawski Date: Sun, 12 Jul 2026 10:30:07 +0200 Subject: [PATCH] Move out receiving partitions in extractors out of loops --- .../src/Flow/ETL/Adapter/CSV/CSVExtractor.php | 2 +- .../src/Flow/ETL/Adapter/Excel/ExcelExtractor.php | 9 +++------ .../ETL/Adapter/JSON/JSONMachine/JsonExtractor.php | 8 ++------ .../JSON/JSONMachine/JsonLinesExtractor.php | 8 ++------ .../Flow/ETL/Adapter/Parquet/ParquetExtractor.php | 14 ++++++++------ .../src/Flow/ETL/Adapter/Text/TextExtractor.php | 3 ++- .../Flow/ETL/Adapter/XML/XMLParserExtractor.php | 8 ++------ 7 files changed, 20 insertions(+), 32 deletions(-) diff --git a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php index 6e0dabf377..adf00e8df2 100644 --- a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php +++ b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php @@ -73,11 +73,11 @@ public function extract(FlowContext $context): Generator $enclosure = $this->enclosure ?? $option->enclosure; $escape = $this->escape ?? $option->escape; $uri = $stream->path()->uri(); + $partitions = $stream->path()->partitions(); $headers = []; $headersCount = 0; $streamUri = $shouldPutInputIntoRows ? $uri : null; - $partitions = $stream->path()->partitions(); $csvLineReader = new CSVLineReader($enclosure, $this->charactersReadInLine, $this->removeBOM); $rowNormalizer = new CSVRowNormalizer($this->emptyToNull); diff --git a/src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/ExcelExtractor.php b/src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/ExcelExtractor.php index a93d1cb49f..895c9212b9 100644 --- a/src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/ExcelExtractor.php +++ b/src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/ExcelExtractor.php @@ -74,13 +74,10 @@ public function extract(FlowContext $context): Generator $offset = $this->offset ?? 1; foreach ($context->streams()->list($this->path, $this->filter()) as $stream) { + $partitions = $stream->path()->partitions(); + foreach ($this->extractRows($stream, $headers, $offset) as $row) { - $signal = yield array_to_rows( - $row, - $context->entryFactory(), - $stream->path()->partitions(), - schema: $this->schema, - ); + $signal = yield array_to_rows($row, $context->entryFactory(), $partitions, $this->schema); $this->incrementReturnedRows(); diff --git a/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonExtractor.php b/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonExtractor.php index 68c27d3c90..7ed3a8e211 100644 --- a/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonExtractor.php +++ b/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonExtractor.php @@ -47,6 +47,7 @@ public function extract(FlowContext $context): Generator foreach ($context->streams()->list($this->path, $this->filter()) as $stream) { $uri = $stream->path()->uri(); + $partitions = $stream->path()->partitions(); /** * @var array $rowData @@ -66,12 +67,7 @@ public function extract(FlowContext $context): Generator continue; } - $signal = yield array_to_rows( - [$row], - $context->entryFactory(), - $stream->path()->partitions(), - $this->schema, - ); + $signal = yield array_to_rows([$row], $context->entryFactory(), $partitions, $this->schema); $this->incrementReturnedRows(); diff --git a/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonLinesExtractor.php b/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonLinesExtractor.php index 75f6d6d98b..e9aad6f87c 100644 --- a/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonLinesExtractor.php +++ b/src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JSONMachine/JsonLinesExtractor.php @@ -61,6 +61,7 @@ public function extract(FlowContext $context): Generator foreach ($context->streams()->list($this->path, $this->filter()) as $stream) { $uri = $stream->path()->uri(); + $partitions = $stream->path()->partitions(); foreach ($stream->readLines() as $jsonLine) { /** @@ -81,12 +82,7 @@ public function extract(FlowContext $context): Generator continue; } - $signal = yield array_to_rows( - [$row], - $context->entryFactory(), - $stream->path()->partitions(), - $this->schema, - ); + $signal = yield array_to_rows([$row], $context->entryFactory(), $partitions, $this->schema); $this->incrementReturnedRows(); diff --git a/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetExtractor.php b/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetExtractor.php index de226e83d5..8b75a22a69 100644 --- a/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetExtractor.php +++ b/src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetExtractor.php @@ -69,12 +69,6 @@ public function extract(FlowContext $context): Generator foreach ($this->readers($context) as $fileData) { $fileRows = $fileData['file']->metadata()->rowsNumber(); - $flowSchema = $this->schemaConverter->toFlow($fileData['file']->schema()); - $uri = $fileData['stream']->path()->uri(); - - if (count($this->columns)) { - $flowSchema = $flowSchema->keep(...$this->columns); - } if ($fileOffset > $fileRows) { $fileData['stream']->close(); @@ -83,6 +77,14 @@ public function extract(FlowContext $context): Generator continue; } + $flowSchema = $this->schemaConverter->toFlow($fileData['file']->schema()); + + if (count($this->columns)) { + $flowSchema = $flowSchema->keep(...$this->columns); + } + + $uri = $fileData['stream']->path()->uri(); + foreach ($fileData['file']->values($this->columns, $this->limit(), $fileOffset) as $row) { $entries = []; diff --git a/src/adapter/etl-adapter-text/src/Flow/ETL/Adapter/Text/TextExtractor.php b/src/adapter/etl-adapter-text/src/Flow/ETL/Adapter/Text/TextExtractor.php index 919f200677..88aa2822be 100644 --- a/src/adapter/etl-adapter-text/src/Flow/ETL/Adapter/Text/TextExtractor.php +++ b/src/adapter/etl-adapter-text/src/Flow/ETL/Adapter/Text/TextExtractor.php @@ -38,6 +38,7 @@ public function extract(FlowContext $context): Generator foreach ($context->streams()->list($this->path, $this->filter()) as $stream) { $uri = $stream->path()->uri(); + $partitions = $stream->path()->partitions(); foreach ($stream->readLines() as $rowData) { if ($shouldPutInputIntoRows) { @@ -46,7 +47,7 @@ public function extract(FlowContext $context): Generator $row = [['text' => rtrim($rowData)]]; } - $signal = yield array_to_rows($row, $context->entryFactory(), $stream->path()->partitions()); + $signal = yield array_to_rows($row, $context->entryFactory(), $partitions); $this->incrementReturnedRows(); diff --git a/src/adapter/etl-adapter-xml/src/Flow/ETL/Adapter/XML/XMLParserExtractor.php b/src/adapter/etl-adapter-xml/src/Flow/ETL/Adapter/XML/XMLParserExtractor.php index ebade62727..653d09bad4 100644 --- a/src/adapter/etl-adapter-xml/src/Flow/ETL/Adapter/XML/XMLParserExtractor.php +++ b/src/adapter/etl-adapter-xml/src/Flow/ETL/Adapter/XML/XMLParserExtractor.php @@ -114,6 +114,7 @@ public function extract(FlowContext $context): Generator foreach ($context->streams()->list($this->path, $this->filter()) as $stream) { $uri = $stream->path()->uri(); + $partitions = $stream->path()->partitions(); foreach ($stream->iterate($this->bufferSize) as $chunk) { if (!xml_parse($this->parser(), $chunk)) { @@ -130,12 +131,7 @@ public function extract(FlowContext $context): Generator $rowData['_input_file_uri'] = $uri; } - $signal = yield array_to_rows( - $rowData, - $context->entryFactory(), - $stream->path()->partitions(), - $this->schema, - ); + $signal = yield array_to_rows($rowData, $context->entryFactory(), $partitions, $this->schema); $this->incrementReturnedRows();