Improve performance for Rows: dropRight, filter, map, `partit…

…ionBy` & `sortBy`

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

@@ -134,13 +134,7 @@ public function drop(int $size) : self
 
     public function dropRight(int $size) : self
     {
-        $rows = $this->rows;
-
-        for ($i = 0; $i < $size; $i++) {
-            \array_pop($rows);
-        }
-
-        return new self(...$rows);
+        return new self(...\array_slice($this->rows, 0, -$size));
     }
 
     /**
@@ -177,15 +171,7 @@ public function entries() : array
      */
     public function filter(callable $callable) : self
     {
-        $results = [];
-
-        foreach ($this->rows as $row) {
-            if ($callable($row)) {
-                $results[] = $row;
-            }
-        }
-
-        return new self(...$results);
+        return new self(...\array_filter($this->rows, $callable));
     }
 
     public function find(callable $callable) : self
@@ -226,13 +212,7 @@ public function first() : Row
      */
     public function flatMap(callable $callable) : self
     {
-        $rows = [];
-
-        foreach ($this->rows as $row) {
-            $rows[] = $callable($row);
-        }
-
-        return new self(...\array_merge(...$rows));
+        return new self(...\array_merge(...\array_map($callable, $this->rows)));
     }
 
     /**
@@ -432,13 +412,7 @@ public function joinRight(self $right, Expression $expression) : self
      */
     public function map(callable $callable) : self
     {
-        $rows = [];
-
-        foreach ($this->rows as $row) {
-            $rows[] = $callable($row);
-        }
-
-        return new self(...$rows);
+        return new self(...\array_map($callable, $this->rows));
     }
 
     public function merge(self ...$rows) : self
@@ -556,7 +530,7 @@ public function partitionBy(string|Reference $entry, string|Reference ...$entrie
          * @var array<string, mixed> $partitionsData
          */
         foreach ($cartesianProduct($partitions) as $partitionsData) {
-            $rows = $this->filter(function (Row $row) use ($partitionsData) : bool {
+            $rows = \array_filter($this->rows, function (Row $row) use ($partitionsData) : bool {
                 /**
                  * @var mixed $value
                  */
@@ -569,8 +543,8 @@ public function partitionBy(string|Reference $entry, string|Reference ...$entrie
                 return true;
             });
 
-            if ($rows->count()) {
-                $partitionedRows[] = new PartitionedRows($rows, ...Partition::fromArray($partitionsData));
+            if ($rows) {
+                $partitionedRows[] = new PartitionedRows(new self(...$rows), ...Partition::fromArray($partitionsData));
             }
         }
 
@@ -672,11 +646,9 @@ public function sortAscending(string|EntryReference $ref) : self
      */
     public function sortBy(EntryReference ...$refs) : self
     {
-        $sortBy = References::init(...$refs)->reverse();
-
         $rows = $this;
 
-        foreach ($sortBy->all() as $ref) {
+        foreach (\array_reverse($refs) as $ref) {
             $rows = $ref->sort() === SortOrder::ASC ? $rows->sortAscending($ref) : $rows->sortDescending($ref);
         }