Cleanups after refactoring

Author: norberttech

examples/topics/aggregations/group_by/code.php

@@ -19,6 +19,7 @@
         ['id' => 9, 'group' => 'A'],
         ['id' => 10, 'group' => 'B'],
     ]))
-    ->groupBy(ref('group'))
+    ->groupBy(ref('group')) // GroupedDataFrame
+    ->toDF() // DataFrame
     ->write(to_stream(__DIR__ . '/output.txt', truncate: false))
     ->run();

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

@@ -109,7 +109,7 @@ public function aggregate(AggregatingFunction ...$aggregations) : self
         $groupBy = new GroupBy();
         $groupBy->aggregate(...$aggregations);
 
-        $this->pipeline = new LinkedPipeline(new GroupByPipeline($groupBy, $this->pipeline), new SynchronousPipeline());
+        $this->pipeline = new LinkedPipeline(new GroupByPipeline($groupBy, $this->pipeline));
 
         return $this;
     }
@@ -135,7 +135,7 @@ public function autoCast() : self
      */
     public function batchSize(int $size) : self
     {
-        $this->pipeline = new LinkedPipeline(new BatchingPipeline($this->pipeline, $size), new SynchronousPipeline());
+        $this->pipeline = new LinkedPipeline(new BatchingPipeline($this->pipeline, $size));
 
         return $this;
     }
@@ -159,7 +159,7 @@ public function cache(?string $id = null, ?int $cacheBatchSize = null) : self
         }
 
         $this->batchSize($cacheBatchSize ?? $this->context->config->cacheBatchSize());
-        $this->pipeline = new LinkedPipeline(new CachingPipeline($this->pipeline, $id), new SynchronousPipeline());
+        $this->pipeline = new LinkedPipeline(new CachingPipeline($this->pipeline, $id));
 
         return $this;
     }
@@ -172,7 +172,7 @@ public function cache(?string $id = null, ?int $cacheBatchSize = null) : self
      */
     public function collect() : self
     {
-        $this->pipeline = new LinkedPipeline(new CollectingPipeline($this->pipeline), new SynchronousPipeline());
+        $this->pipeline = new LinkedPipeline(new CollectingPipeline($this->pipeline));
 
         return $this;
     }
@@ -459,7 +459,7 @@ public function join(self $dataFrame, Expression $on, string|Join $type = Join::
             $type = Join::from($type);
         }
 
-        $this->pipeline = new LinkedPipeline(new HashJoinPipeline($this->pipeline, $dataFrame, $on, $type), new SynchronousPipeline());
+        $this->pipeline = new LinkedPipeline(new HashJoinPipeline($this->pipeline, $dataFrame, $on, $type));
 
         return $this;
     }
@@ -556,7 +556,7 @@ public function partitionBy(string|Reference $entry, string|Reference ...$entrie
     {
         \array_unshift($entries, $entry);
 
-        $this->pipeline = new LinkedPipeline(new PartitioningPipeline($this->pipeline, References::init(...$entries)->all()), new SynchronousPipeline());
+        $this->pipeline = new LinkedPipeline(new PartitioningPipeline($this->pipeline, References::init(...$entries)->all()));
 
         return $this;
     }
@@ -852,7 +852,7 @@ public function withEntry(string $entryName, ScalarFunction|WindowFunction $ref)
     {
         if ($ref instanceof WindowFunction) {
             if (\count($ref->window()->partitions())) {
-                $this->pipeline = new LinkedPipeline(new PartitioningPipeline($this->pipeline, $ref->window()->partitions(), $ref->window()->order()), new SynchronousPipeline());
+                $this->pipeline = new LinkedPipeline(new PartitioningPipeline($this->pipeline, $ref->window()->partitions(), $ref->window()->order()));
             } else {
                 $this->collect();
 

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

@@ -32,7 +32,7 @@ public function toDataFrame() : DataFrame
     {
         return $this->df->rebuild(function (Pipeline $pipeline, FlowContext $context) : DataFrame {
             return new DataFrame(
-                new Pipeline\LinkedPipeline(new Pipeline\GroupByPipeline($this->groupBy, $pipeline), new Pipeline\SynchronousPipeline()),
+                new Pipeline\LinkedPipeline(new Pipeline\GroupByPipeline($this->groupBy, $pipeline)),
                 $context
             );
         });

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

@@ -8,7 +8,6 @@
 use Flow\ETL\Exception\{DuplicatedEntriesException, JoinException};
 use Flow\ETL\Hash\NativePHPHash;
 use Flow\ETL\Join\{Expression, Join};
-use Flow\ETL\Loader\Closure;
 use Flow\ETL\Pipeline\HashJoin\HashTable;
 use Flow\ETL\Row\Entry;
 use Flow\ETL\{DataFrame, Extractor, FlowContext, Loader, Pipeline, Row, Rows, Transformer};
@@ -29,23 +28,14 @@ public function __construct(
 
     public function add(Loader|Transformer $pipe) : Pipeline
     {
-        $this->left->pipes()->add($pipe);
+        $this->left->add($pipe);
 
         return $this;
     }
 
-    public function closure(FlowContext $context) : void
-    {
-        foreach ($this->left->pipes()->all() as $pipe) {
-            if ($pipe instanceof Loader && $pipe instanceof Closure) {
-                $pipe->closure($context);
-            }
-        }
-    }
-
     public function has(string $transformerClass) : bool
     {
-        return $this->left->pipes()->has($transformerClass);
+        return $this->left->has($transformerClass);
     }
 
     public function pipes() : Pipes

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

@@ -75,7 +75,7 @@ public function optimize(Loader|Transformer $element, Pipeline $pipeline) : Pipe
             return $pipeline;
         }
 
-        $pipeline = new Pipeline\LinkedPipeline(new BatchingPipeline($pipeline, $this->batchSize), new Pipeline\SynchronousPipeline());
+        $pipeline = new Pipeline\LinkedPipeline(new BatchingPipeline($pipeline, $this->batchSize));
         $pipeline->add($element);
 
         return $pipeline;