Fixed batch size optimization for deeply nested pipelines

norberttech · norberttech · commit 3ed245d368b1 · 2025-05-08T22:35:03.000+02:00
diff --git a/src/adapter/etl-adapter-xml/src/Flow/ETL/Adapter/XML/RowsNormalizer/EntryNormalizer.php b/src/adapter/etl-adapter-xml/src/Flow/ETL/Adapter/XML/RowsNormalizer/EntryNormalizer.php
@@ -10,7 +10,18 @@
 use Flow\ETL\PHP\Type\Logical\{ListType, MapType, StructureType};
 use Flow\ETL\PHP\Type\Type;
 use Flow\ETL\Row\Entry;
-use Flow\ETL\Row\Entry\{BooleanEntry, DateTimeEntry, EnumEntry, FloatEntry, IntegerEntry, JsonEntry, ListEntry, MapEntry, StringEntry, StructureEntry, UuidEntry};
+use Flow\ETL\Row\Entry\{BooleanEntry,
+    DateTimeEntry,
+    EnumEntry,
+    FloatEntry,
+    IntegerEntry,
+    JsonEntry,
+    ListEntry,
+    MapEntry,
+    StringEntry,
+    StructureEntry,
+    UuidEntry,
+    XMLEntry};
 
 final readonly class EntryNormalizer
 {
@@ -50,6 +61,7 @@ public function normalize(Entry $entry) : XMLNode|XMLAttribute
             EnumEntry::class => XMLNode::flatNode($entry->name(), $entry->toString()),
             JsonEntry::class => XMLNode::flatNode($entry->name(), $entry->toString()),
             UuidEntry::class => XMLNode::flatNode($entry->name(), $entry->toString()),
+            XMLEntry::class => XMLNode::flatNode($entry->name(), $entry->toString()),
             default => throw new InvalidArgumentException("Given entry type can't be converted to node, given entry type: " . $entry::class),
         };
     }
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/BatchingPipeline.php b/src/core/etl/src/Flow/ETL/Pipeline/BatchingPipeline.php
@@ -8,8 +8,10 @@
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\{Extractor, FlowContext, Loader, Pipeline, Transformer};
 
-final readonly class BatchingPipeline implements Pipeline
+final readonly class BatchingPipeline implements OverridingPipeline, Pipeline
 {
+    use RecursivePipelineIterator;
+
     /**
      * @param Pipeline $pipeline
      * @param int<1, max> $size
@@ -35,6 +37,11 @@ public function has(string $transformerClass) : bool
         return $this->pipeline->has($transformerClass);
     }
 
+    public function pipelines() : array
+    {
+        return $this->allPipelines($this->pipeline);
+    }
+
     public function pipes() : Pipes
     {
         return $this->pipeline->pipes();
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/CachingPipeline.php b/src/core/etl/src/Flow/ETL/Pipeline/CachingPipeline.php
@@ -6,8 +6,10 @@
 
 use Flow\ETL\{Cache\CacheIndex, Extractor, FlowContext, Loader, Pipeline, Transformer};
 
-final readonly class CachingPipeline implements Pipeline
+final readonly class CachingPipeline implements OverridingPipeline, Pipeline
 {
+    use RecursivePipelineIterator;
+
     public function __construct(private Pipeline $pipeline, private ?string $id = null)
     {
     }
@@ -24,6 +26,11 @@ public function has(string $transformerClass) : bool
         return $this->pipeline->has($transformerClass);
     }
 
+    public function pipelines() : array
+    {
+        return $this->allPipelines($this->pipeline);
+    }
+
     public function pipes() : Pipes
     {
         return $this->pipeline->pipes();
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/CollectingPipeline.php b/src/core/etl/src/Flow/ETL/Pipeline/CollectingPipeline.php
@@ -9,8 +9,10 @@
 /**
  * @internal
  */
-final readonly class CollectingPipeline implements Pipeline
+final readonly class CollectingPipeline implements OverridingPipeline, Pipeline
 {
+    use RecursivePipelineIterator;
+
     public function __construct(private Pipeline $pipeline)
     {
     }
@@ -27,6 +29,11 @@ public function has(string $transformerClass) : bool
         return $this->pipeline->has($transformerClass);
     }
 
+    public function pipelines() : array
+    {
+        return $this->allPipelines($this->pipeline);
+    }
+
     public function pipes() : Pipes
     {
         return $this->pipeline->pipes();
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/ConstrainedPipeline.php b/src/core/etl/src/Flow/ETL/Pipeline/ConstrainedPipeline.php
@@ -7,8 +7,10 @@
 use Flow\ETL\{Constraint, Extractor, FlowContext, Loader, Pipeline, Transformer};
 use Flow\ETL\Exception\{ConstraintViolationException, InvalidArgumentException};
 
-final class ConstrainedPipeline implements Pipeline
+final class ConstrainedPipeline implements OverridingPipeline, Pipeline
 {
+    use RecursivePipelineIterator;
+
     private int $rowIndex = 0;
 
     /**
@@ -38,6 +40,14 @@ public function has(string $transformerClass) : bool
         return $this->pipeline->has($transformerClass);
     }
 
+    /**
+     * @return array<Pipeline>
+     */
+    public function pipelines() : array
+    {
+        return $this->allPipelines($this->pipeline);
+    }
+
     public function pipes() : Pipes
     {
         return $this->pipeline->pipes();
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/GroupByPipeline.php b/src/core/etl/src/Flow/ETL/Pipeline/GroupByPipeline.php
@@ -6,8 +6,10 @@
 
 use Flow\ETL\{Extractor, FlowContext, GroupBy, Loader, Pipeline, Transformer};
 
-final readonly class GroupByPipeline implements Pipeline
+final readonly class GroupByPipeline implements OverridingPipeline, Pipeline
 {
+    use RecursivePipelineIterator;
+
     public function __construct(public GroupBy $groupBy, private Pipeline $pipeline)
     {
     }
@@ -24,6 +26,11 @@ public function has(string $transformerClass) : bool
         return $this->pipeline->has($transformerClass);
     }
 
+    public function pipelines() : array
+    {
+        return $this->allPipelines($this->pipeline);
+    }
+
     public function pipes() : Pipes
     {
         return $this->pipeline->pipes();
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/HashJoinPipeline.php b/src/core/etl/src/Flow/ETL/Pipeline/HashJoinPipeline.php
@@ -12,8 +12,10 @@
 use Flow\ETL\Pipeline\HashJoin\HashTable;
 use Flow\ETL\Row\Entry;
 
-final readonly class HashJoinPipeline implements Pipeline
+final readonly class HashJoinPipeline implements OverridingPipeline, Pipeline
 {
+    use RecursivePipelineIterator;
+
     private Extractor $extractor;
 
     public function __construct(
@@ -38,6 +40,11 @@ public function has(string $transformerClass) : bool
         return $this->left->has($transformerClass);
     }
 
+    public function pipelines() : array
+    {
+        return $this->allPipelines($this->left);
+    }
+
     public function pipes() : Pipes
     {
         return $this->left->pipes();
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/LinkedPipeline.php b/src/core/etl/src/Flow/ETL/Pipeline/LinkedPipeline.php
@@ -16,6 +16,8 @@
  */
 final readonly class LinkedPipeline implements OverridingPipeline, Pipeline
 {
+    use RecursivePipelineIterator;
+
     private Pipeline $nextPipeline;
 
     public function __construct(
@@ -41,18 +43,7 @@ public function has(string $transformerClass) : bool
      */
     public function pipelines() : array
     {
-        $pipelines = [];
-
-        if ($this->pipeline instanceof OverridingPipeline) {
-            $pipelines = $this->pipeline->pipelines();
-        }
-
-        $pipelines[] = $this->pipeline;
-
-        //        if ($this->nextPipeline instanceof OverridingPipeline) {
-        //            $pipelines = \array_merge($pipelines, $this->nextPipeline->pipelines());
-        //        }
-
+        $pipelines = $this->allPipelines($this->pipeline);
         $pipelines[] = $this->nextPipeline;
 
         return $pipelines;
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/Optimizer/BatchSizeOptimization.php b/src/core/etl/src/Flow/ETL/Pipeline/Optimizer/BatchSizeOptimization.php
@@ -33,16 +33,19 @@ final class BatchSizeOptimization implements Optimization
      * @var array<class-string<Loader>>
      */
     private array $supportedLoaders = [
-        \Flow\ETL\Adapter\Doctrine\DbalLoader::class,
-        \Flow\ETL\Adapter\Elasticsearch\ElasticsearchPHP\ElasticsearchLoader::class,
-        \Flow\ETL\Adapter\Meilisearch\MeilisearchPHP\MeilisearchLoader::class,
+        'Flow\ETL\Adapter\Doctrine\DbalLoader',
+        'Flow\ETL\Adapter\Elasticsearch\ElasticsearchPHP\ElasticsearchLoader',
+        'Flow\ETL\Adapter\Meilisearch\MeilisearchPHP\MeilisearchLoader',
     ];
 
     /**
      * @param int<1, max> $batchSize
      */
-    public function __construct(private readonly int $batchSize = 1000)
+    public function __construct(private readonly int $batchSize = 1000, ?array $supportedLoaders = null)
     {
+        if ($supportedLoaders !== null) {
+            $this->supportedLoaders = $supportedLoaders;
+        }
     }
 
     public function isFor(Loader|Transformer $element, Pipeline $pipeline) : bool
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/PartitioningPipeline.php b/src/core/etl/src/Flow/ETL/Pipeline/PartitioningPipeline.php
@@ -18,7 +18,7 @@
 use Flow\ETL\Row\Reference;
 use Flow\Filesystem\Partition;
 
-final readonly class PartitioningPipeline implements Pipeline
+final readonly class PartitioningPipeline implements OverridingPipeline, Pipeline
 {
     private Algorithm $hashAlgorithm;
 
@@ -52,6 +52,30 @@ public function has(string $transformerClass) : bool
         return $this->pipeline->has($transformerClass);
     }
 
+    /**
+     * @return array<Pipeline>
+     */
+    public function pipelines() : array
+    {
+        $pipelines = [];
+
+        $recursivelyAddPipelines = function (Pipeline $pipeline, array &$pipelines) use (&$recursivelyAddPipelines) : void {
+            if ($pipeline instanceof OverridingPipeline) {
+                $pipelines[] = $pipeline;
+
+                foreach ($pipeline->pipelines() as $p) {
+                    $recursivelyAddPipelines($p, $pipelines);
+                }
+            } else {
+                $pipelines[] = $pipeline;
+            }
+        };
+
+        $pipelines[] = $this->pipeline;
+
+        return $pipelines;
+    }
+
     public function pipes() : Pipes
     {
         return $this->pipeline->pipes();
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/RecursivePipelineIterator.php b/src/core/etl/src/Flow/ETL/Pipeline/RecursivePipelineIterator.php
@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Pipeline;
+
+use Flow\ETL\Pipeline;
+
+trait RecursivePipelineIterator
+{
+    /**
+     * @return array<Pipeline>
+     */
+    private function allPipelines(Pipeline $pipeline) : array
+    {
+        $pipelines = [];
+
+        if ($pipeline instanceof OverridingPipeline) {
+            $pipelines[] = $pipeline;
+
+            foreach ($pipeline->pipelines() as $nextPipeline) {
+                $pipelines = [...$pipelines, ...$this->allPipelines($nextPipeline)];
+            }
+        } else {
+            $pipelines[] = $pipeline;
+        }
+
+        return $pipelines;
+    }
+}
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/SortingPipeline.php b/src/core/etl/src/Flow/ETL/Pipeline/SortingPipeline.php
@@ -11,8 +11,10 @@
 use Flow\ETL\Row\References;
 use Flow\ETL\Sort\{ExternalSort, MemorySort};
 
-final readonly class SortingPipeline implements Pipeline
+final readonly class SortingPipeline implements OverridingPipeline, Pipeline
 {
+    use RecursivePipelineIterator;
+
     public function __construct(private Pipeline $pipeline, private References $refs)
     {
     }
@@ -29,6 +31,11 @@ public function has(string $transformerClass) : bool
         return $this->pipeline->has($transformerClass);
     }
 
+    public function pipelines() : array
+    {
+        return $this->allPipelines($this->pipeline);
+    }
+
     public function pipes() : Pipes
     {
         return $this->pipeline->pipes();
diff --git a/src/core/etl/src/Flow/ETL/Pipeline/VoidPipeline.php b/src/core/etl/src/Flow/ETL/Pipeline/VoidPipeline.php
@@ -6,8 +6,10 @@
 
 use Flow\ETL\{Extractor, FlowContext, Loader, Pipeline, Rows, Transformer};
 
-final readonly class VoidPipeline implements Pipeline
+final readonly class VoidPipeline implements OverridingPipeline, Pipeline
 {
+    use RecursivePipelineIterator;
+
     public function __construct(private Pipeline $pipeline)
     {
     }
@@ -22,6 +24,11 @@ public function has(string $transformerClass) : bool
         return $this->pipeline->has($transformerClass);
     }
 
+    public function pipelines() : array
+    {
+        return $this->allPipelines($this->pipeline);
+    }
+
     public function pipes() : Pipes
     {
         return $this->pipeline->pipes();
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Double/FakeExtractor.php b/src/core/etl/tests/Flow/ETL/Tests/Double/FakeExtractor.php
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Double/SpyLoader.php b/src/core/etl/tests/Flow/ETL/Tests/Double/SpyLoader.php
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/BatchSizeOptimizationTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/BatchSizeOptimizationTest.php
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Pipeline/LinkedPipelineTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Pipeline/LinkedPipelineTest.php

Original file line number	Diff line number	Diff line change
`@@ -6,8 +6,10 @@`
`6`	`6`
`7`	`7`	`use Flow\ETL\{Cache\CacheIndex, Extractor, FlowContext, Loader, Pipeline, Transformer};`
`8`	`8`
`9`		`-final readonly class CachingPipeline implements Pipeline`
	`9`	`+final readonly class CachingPipeline implements OverridingPipeline, Pipeline`
`10`	`10`	`{`
	`11`	`+ use RecursivePipelineIterator;`
	`12`	`+`
`11`	`13`	`public function __construct(private Pipeline $pipeline, private ?string $id = null)`
`12`	`14`	`{`
`13`	`15`	`}`
`@@ -24,6 +26,11 @@ public function has(string $transformerClass) : bool`
`24`	`26`	`return $this->pipeline->has($transformerClass);`
`25`	`27`	`}`
`26`	`28`
	`29`	`+ public function pipelines() : array`
	`30`	`+ {`
	`31`	`+ return $this->allPipelines($this->pipeline);`
	`32`	`+ }`
	`33`	`+`
`27`	`34`	`public function pipes() : Pipes`
`28`	`35`	`{`
`29`	`36`	`return $this->pipeline->pipes();`
Original file line number	Diff line number	Diff line change
`@@ -9,8 +9,10 @@`
`9`	`9`	`/**`
`10`	`10`	`* @internal`
`11`	`11`	`*/`
`12`		`-final readonly class CollectingPipeline implements Pipeline`
	`12`	`+final readonly class CollectingPipeline implements OverridingPipeline, Pipeline`
`13`	`13`	`{`
	`14`	`+ use RecursivePipelineIterator;`
	`15`	`+`
`14`	`16`	`public function __construct(private Pipeline $pipeline)`
`15`	`17`	`{`
`16`	`18`	`}`
`@@ -27,6 +29,11 @@ public function has(string $transformerClass) : bool`
`27`	`29`	`return $this->pipeline->has($transformerClass);`
`28`	`30`	`}`
`29`	`31`
	`32`	`+ public function pipelines() : array`
	`33`	`+ {`
	`34`	`+ return $this->allPipelines($this->pipeline);`
	`35`	`+ }`
	`36`	`+`
`30`	`37`	`public function pipes() : Pipes`
`31`	`38`	`{`
`32`	`39`	`return $this->pipeline->pipes();`
Original file line number	Diff line number	Diff line change
`@@ -6,8 +6,10 @@`
`6`	`6`
`7`	`7`	`use Flow\ETL\{Extractor, FlowContext, GroupBy, Loader, Pipeline, Transformer};`
`8`	`8`
`9`		`-final readonly class GroupByPipeline implements Pipeline`
	`9`	`+final readonly class GroupByPipeline implements OverridingPipeline, Pipeline`
`10`	`10`	`{`
	`11`	`+ use RecursivePipelineIterator;`
	`12`	`+`
`11`	`13`	`public function __construct(public GroupBy $groupBy, private Pipeline $pipeline)`
`12`	`14`	`{`
`13`	`15`	`}`
`@@ -24,6 +26,11 @@ public function has(string $transformerClass) : bool`
`24`	`26`	`return $this->pipeline->has($transformerClass);`
`25`	`27`	`}`
`26`	`28`
	`29`	`+ public function pipelines() : array`
	`30`	`+ {`
	`31`	`+ return $this->allPipelines($this->pipeline);`
	`32`	`+ }`
	`33`	`+`
`27`	`34`	`public function pipes() : Pipes`
`28`	`35`	`{`
`29`	`36`	`return $this->pipeline->pipes();`