Pipeline Optimization - LimitOptimizer (#730)

Added Pipeline Optimizer with LimitOptimization

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

@@ -6,6 +6,7 @@
 
 use Flow\ETL\Filesystem\FilesystemStreams;
 use Flow\ETL\Pipeline\Execution\Processors;
+use Flow\ETL\Pipeline\Optimizer;
 use Flow\ETL\Row\EntryFactory;
 use Flow\Serializer\Serializer;
 
@@ -26,6 +27,7 @@ public function __construct(
         private readonly Serializer $serializer,
         private readonly FilesystemStreams $filesystemStreams,
         private readonly Processors $processors,
+        private readonly Optimizer $optimizer,
         private readonly bool $putInputIntoRows,
         private readonly EntryFactory $entryFactory
     ) {
@@ -66,6 +68,11 @@ public function id() : string
         return $this->id;
     }
 
+    public function optimizer() : Optimizer
+    {
+        return $this->optimizer;
+    }
+
     public function processors() : Processors
     {
         return $this->processors;

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

@@ -11,6 +11,7 @@
 use Flow\ETL\Monitoring\Memory\Unit;
 use Flow\ETL\Pipeline\Execution\Processor\FilesystemProcessor;
 use Flow\ETL\Pipeline\Execution\Processors;
+use Flow\ETL\Pipeline\Optimizer;
 use Flow\ETL\Row\Factory\NativeEntryFactory;
 use Flow\Serializer\CompressingSerializer;
 use Flow\Serializer\Serializer;
@@ -68,6 +69,9 @@ public function build() : Config
             new Processors(
                 new FilesystemProcessor()
             ),
+            new Optimizer(
+                new Optimizer\LimitOptimization()
+            ),
             $this->putInputIntoRows,
             new NativeEntryFactory()
         );

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

@@ -398,7 +398,7 @@ public function joinEach(DataFrameFactory $factory, Expression $on, string|Join
      */
     public function limit(int $limit) : self
     {
-        $this->pipeline->add(new LimitTransformer($limit));
+        $this->pipeline = $this->context->config->optimizer()->optimize(new LimitTransformer($limit), $this->pipeline);
 
         return $this;
     }
@@ -647,7 +647,7 @@ public function sortBy(EntryReference ...$entries) : self
             ->run();
 
         $this->pipeline = $this->pipeline->cleanCopy();
-        $this->pipeline->source($this->context->config->externalSort()->sortBy(...$entries));
+        $this->pipeline->setSource($this->context->config->externalSort()->sortBy(...$entries));
 
         return $this;
     }

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

@@ -24,6 +24,11 @@ public function countRow() : void
         $this->yieldedRows++;
     }
 
+    public function isLimited() : bool
+    {
+        return $this->limit !== null;
+    }
+
     public function limit() : ?int
     {
         return $this->limit;

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

@@ -10,4 +10,6 @@
 interface LimitableExtractor
 {
     public function changeLimit(int $limit) : void;
+
+    public function isLimited() : bool;
 }

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

@@ -28,15 +28,15 @@ public static function setUp(ConfigBuilder|Config $config) : self
     public function extract(Extractor $extractor) : DataFrame
     {
         return new DataFrame(
-            (new SynchronousPipeline())->source($extractor),
+            (new SynchronousPipeline())->setSource($extractor),
             $this->config
         );
     }
 
     public function process(Rows ...$rows) : DataFrame
     {
         return new DataFrame(
-            (new SynchronousPipeline())->source(new ProcessExtractor(...$rows)),
+            (new SynchronousPipeline())->setSource(new ProcessExtractor(...$rows)),
             $this->config
         );
     }

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

@@ -4,6 +4,8 @@
 
 namespace Flow\ETL;
 
+use Flow\ETL\Pipeline\Pipes;
+
 /**
  * @internal
  */
@@ -22,10 +24,14 @@ public function has(string $transformerClass) : bool;
 
     public function isAsync() : bool;
 
+    public function pipes() : Pipes;
+
     /**
      * @return \Generator<Rows>
      */
     public function process(FlowContext $context) : \Generator;
 
-    public function source(Extractor $extractor) : self;
+    public function setSource(Extractor $extractor) : self;
+
+    public function source() : Extractor;
 }

src/core/etl/src/Flow/ETL/Pipeline/BatchingPipeline.php

@@ -61,9 +61,14 @@ public function isAsync() : bool
         return $this->pipeline->isAsync();
     }
 
+    public function pipes() : Pipes
+    {
+        return $this->pipeline->pipes()->merge($this->nextPipeline->pipes());
+    }
+
     public function process(FlowContext $context) : \Generator
     {
-        $this->nextPipeline->source(
+        $this->nextPipeline->setSource(
             From::chunks_from(
                 From::pipeline($this->pipeline),
                 $this->size
@@ -73,10 +78,15 @@ public function process(FlowContext $context) : \Generator
         return $this->nextPipeline->process($context);
     }
 
-    public function source(Extractor $extractor) : self
+    public function setSource(Extractor $extractor) : self
     {
-        $this->pipeline->source($extractor);
+        $this->pipeline->setSource($extractor);
 
         return $this;
     }
+
+    public function source() : Extractor
+    {
+        return $this->pipeline->source();
+    }
 }

src/core/etl/src/Flow/ETL/Pipeline/CachingPipeline.php

@@ -43,6 +43,11 @@ public function isAsync() : bool
         return false;
     }
 
+    public function pipes() : Pipes
+    {
+        return $this->pipeline->pipes();
+    }
+
     public function process(FlowContext $context) : \Generator
     {
         $context->config->cache()->clear($id = $this->id ?: $context->config->id());
@@ -53,10 +58,15 @@ public function process(FlowContext $context) : \Generator
         }
     }
 
-    public function source(Extractor $extractor) : Pipeline
+    public function setSource(Extractor $extractor) : Pipeline
     {
-        $this->pipeline->source($extractor);
+        $this->pipeline->setSource($extractor);
 
         return $this;
     }
+
+    public function source() : Extractor
+    {
+        return $this->pipeline->source();
+    }
 }

src/core/etl/src/Flow/ETL/Pipeline/CollectingPipeline.php

@@ -57,19 +57,29 @@ public function isAsync() : bool
         return $this->pipeline->isAsync();
     }
 
+    public function pipes() : Pipes
+    {
+        return $this->pipeline->pipes()->merge($this->nextPipeline->pipes());
+    }
+
     public function process(FlowContext $context) : \Generator
     {
-        $this->nextPipeline->source(From::rows(
+        $this->nextPipeline->setSource(From::rows(
             (new Rows())->merge(...\iterator_to_array($this->pipeline->process($context)))
         ));
 
         return $this->nextPipeline->process($context);
     }
 
-    public function source(Extractor $extractor) : self
+    public function setSource(Extractor $extractor) : self
     {
-        $this->pipeline->source($extractor);
+        $this->pipeline->setSource($extractor);
 
         return $this;
     }
+
+    public function source() : Extractor
+    {
+        return $this->pipeline->source();
+    }
 }

src/core/etl/src/Flow/ETL/Pipeline/Execution/LogicalPlan.php → src/core/etl/src/Flow/ETL/Pipeline/Execution/ExecutionPlan.php

@@ -8,7 +8,7 @@
 use Flow\ETL\Pipeline\Execution\Plan\FilesystemOperations;
 use Flow\ETL\Pipeline\Pipes;
 
-final class LogicalPlan
+final class ExecutionPlan
 {
     public function __construct(public readonly Extractor $extractor, public readonly Pipes $pipes)
     {

src/core/etl/src/Flow/ETL/Pipeline/Execution/Processor/FilesystemProcessor.php

@@ -8,12 +8,12 @@
 use Flow\ETL\Filesystem\SaveMode;
 use Flow\ETL\FlowContext;
 use Flow\ETL\Loader\FileLoader;
-use Flow\ETL\Pipeline\Execution\LogicalPlan;
+use Flow\ETL\Pipeline\Execution\ExecutionPlan;
 use Flow\ETL\Pipeline\Pipes;
 
 final class FilesystemProcessor implements Processor
 {
-    public function process(LogicalPlan $plan, FlowContext $context) : LogicalPlan
+    public function process(ExecutionPlan $plan, FlowContext $context) : ExecutionPlan
     {
         $operations = $plan->filesystemOperations();
 
@@ -76,6 +76,6 @@ public function process(LogicalPlan $plan, FlowContext $context) : LogicalPlan
             }
         }
 
-        return new LogicalPlan($plan->extractor, $newPipes);
+        return new ExecutionPlan($plan->extractor, $newPipes);
     }
 }

src/core/etl/src/Flow/ETL/Pipeline/Execution/Processor/Processor.php

@@ -5,9 +5,9 @@
 namespace Flow\ETL\Pipeline\Execution\Processor;
 
 use Flow\ETL\FlowContext;
-use Flow\ETL\Pipeline\Execution\LogicalPlan;
+use Flow\ETL\Pipeline\Execution\ExecutionPlan;
 
 interface Processor
 {
-    public function process(LogicalPlan $plan, FlowContext $context) : LogicalPlan;
+    public function process(ExecutionPlan $plan, FlowContext $context) : ExecutionPlan;
 }

src/core/etl/src/Flow/ETL/Pipeline/Execution/Processors.php

@@ -19,7 +19,7 @@ public function __construct(Processor ...$processors)
         $this->processors = $processors;
     }
 
-    public function process(LogicalPlan $plan, FlowContext $context) : LogicalPlan
+    public function process(ExecutionPlan $plan, FlowContext $context) : ExecutionPlan
     {
         $processedPlan = $plan;
 

src/core/etl/src/Flow/ETL/Pipeline/GroupByPipeline.php

@@ -52,23 +52,33 @@ public function isAsync() : bool
         return $this->pipeline->isAsync();
     }
 
+    public function pipes() : Pipes
+    {
+        return $this->pipeline->pipes()->merge($this->nextPipeline->pipes());
+    }
+
     public function process(FlowContext $context) : \Generator
     {
         foreach ($this->pipeline->process($context) as $nextRows) {
             $this->groupBy->group($nextRows);
         }
 
-        $this->nextPipeline->source(new Extractor\ProcessExtractor($this->groupBy->result($context)));
+        $this->nextPipeline->setSource(new Extractor\ProcessExtractor($this->groupBy->result($context)));
 
         foreach ($this->nextPipeline->process($context) as $nextRows) {
             yield $nextRows;
         }
     }
 
-    public function source(Extractor $extractor) : self
+    public function setSource(Extractor $extractor) : self
     {
-        $this->pipeline->source($extractor);
+        $this->pipeline->setSource($extractor);
 
         return $this;
     }
+
+    public function source() : Extractor
+    {
+        return $this->pipeline->source();
+    }
 }