perf(flow-php/etl): linear sort accumulation, working memory-to-exter…

src/core/etl/src/Flow/ETL/DataFrame.php

@@ -361,12 +361,8 @@ public function fetch(?int $limit = null): Rows
             $this->limit($limit);
         }
 
-        $rows = new Rows();
-
         try {
-            foreach ($this->pipeline->process($this->context) as $nextRows) {
-                $rows = $rows->merge($nextRows);
-            }
+            $rows = Rows::mergeAll($this->pipeline->process($this->context));
             $this->context->telemetry()->dataFrameCompleted($this->context);
         } catch (Throwable $e) {
             $this->context->telemetry()->dataFrameFailed($this->context, $e);

src/core/etl/src/Flow/ETL/Exception/OutOfMemoryException.php

@@ -4,4 +4,18 @@
 
 namespace Flow\ETL\Exception;
 
-final class OutOfMemoryException extends RuntimeException {}
+use Flow\ETL\Rows;
+use Throwable;
+
+final class OutOfMemoryException extends RuntimeException
+{
+    /**
+     * @param null|Rows $collectedRows
+     */
+    public function __construct(
+        public readonly ?Rows $collectedRows = null,
+        ?Throwable $previous = null,
+    ) {
+        parent::__construct('Memory limit exceeded', 0, $previous);
+    }
+}

src/core/etl/src/Flow/ETL/Extractor/CollectingExtractor.php

@@ -17,13 +17,7 @@ public function __construct(
 
     public function extract(FlowContext $context): Generator
     {
-        $collectedRows = new Rows();
-
-        foreach ($this->extractor->extract($context) as $rows) {
-            $collectedRows = $collectedRows->merge($rows);
-        }
-
-        yield $collectedRows;
+        yield Rows::mergeAll($this->extractor->extract($context));
     }
 
     public function extractors(): array

src/core/etl/src/Flow/ETL/Processor/CollectingProcessor.php

@@ -20,14 +20,13 @@
  */
 final readonly class CollectingProcessor implements Processor
 {
+    /**
+     * @param \Generator<Rows> $rows
+     *
+     * @return \Generator<Rows>
+     */
     public function process(Generator $rows, FlowContext $context): Generator
     {
-        $collected = new Rows();
-
-        foreach ($rows as $batch) {
-            $collected = $collected->merge($batch);
-        }
-
-        yield $collected;
+        yield Rows::mergeAll($rows);
     }
 }

src/core/etl/src/Flow/ETL/Processor/SortingProcessor.php

@@ -5,13 +5,15 @@
 namespace Flow\ETL\Processor;
 
 use Flow\ETL\Dataset\Memory\Unit;
+use Flow\ETL\Exception\OutOfMemoryException;
 use Flow\ETL\FlowContext;
 use Flow\ETL\Processor;
 use Flow\ETL\Row\References;
 use Flow\ETL\Rows;
 use Flow\ETL\Sort\ExternalSort;
 use Flow\ETL\Sort\ExternalSort\BucketsCache\FilesystemBucketsCache;
 use Flow\ETL\Sort\MemorySort;
+use Flow\ETL\Sort\SortAlgorithms;
 use Generator;
 
 /**
@@ -35,14 +37,28 @@ public function process(Generator $rows, FlowContext $context): Generator
             $context->config->sort->algorithm->useMemory()
             && $context->config->sort->memoryLimit->isGreaterThan($minMemoryForMemorySort)
         ) {
-            yield from (new MemorySort($context->config->sort->memoryLimit))->sortGenerator(
-                $rows,
-                $context,
-                $this->refs,
-            );
-        } else {
-            yield from $this->externalSort($rows, $context);
+            try {
+                yield from (new MemorySort($context->config->sort->memoryLimit))->sortGenerator(
+                    $rows,
+                    $context,
+                    $this->refs,
+                );
+
+                return;
+            } catch (OutOfMemoryException $exception) {
+                if ($context->config->sort->algorithm !== SortAlgorithms::MEMORY_FALLBACK_EXTERNAL_SORT) {
+                    throw $exception;
+                }
+
+                $rows->next();
+
+                yield from $this->externalSort(self::resume($exception->collectedRows, $rows), $context);
+
+                return;
+            }
         }
+
+        yield from $this->externalSort($rows, $context);
     }
 
     /**
@@ -59,6 +75,21 @@ private function externalSort(Generator $rows, FlowContext $context): Generator
                 batchSize: $context->config->cache->externalSortBatchSize,
             ),
             $context->config->cache->externalSortBucketsCount,
+            $context->config->cache->externalSortBatchSize,
         ))->sortGenerator($rows, $context, $this->refs);
     }
+
+    /**
+     * @param \Generator<Rows> $rows
+     *
+     * @return \Generator<Rows>
+     */
+    private static function resume(?Rows $collectedRows, Generator $rows): Generator
+    {
+        if ($collectedRows !== null && !$collectedRows->empty()) {
+            yield $collectedRows;
+        }
+
+        yield from $rows;
+    }
 }