Improve performance of sorting operation by reducing number of writes to cache (#1036)

* Improve performance of sorting operation by reducing number of writes to cache

* CS Fixes

Author: norberttech

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

@@ -40,28 +40,26 @@ public function has(string $id) : bool
     public function read(string $id) : \Generator
     {
         foreach ($this->index($id) as $entry) {
-            if ($this->cache->has($entry)) {
-                /**
-                 * @var Rows $rows
-                 */
-                $rows = $this->serializer->unserialize((string) $this->cache->get($entry), Rows::class);
+            $serializedRows = $this->cache->get($entry);
 
-                yield $rows;
+            if ($serializedRows === null) {
+                continue;
             }
+
+            /**
+             * @var Rows $rows
+             */
+            $rows = $this->serializer->unserialize((string) $serializedRows, Rows::class);
+
+            yield $rows;
         }
     }
 
     private function addToIndex(string $indexId, string $id) : void
     {
-        if (!$this->cache->has($indexId)) {
-            $this->cache->set($indexId, [$id], $this->ttl);
-
-            return;
-        }
-
-        /** @var array<string> $index */
+        /** @var null|array<string> $index */
         $index = $this->cache->get($indexId);
-        $this->cache->set($indexId, \array_merge($index, [$id]), $this->ttl);
+        $this->cache->set($indexId, \array_merge($index ?? [], [$id]), $this->ttl);
     }
 
     /**
@@ -71,13 +69,9 @@ private function addToIndex(string $indexId, string $id) : void
      */
     private function index(string $indexId) : array
     {
-        if (!$this->cache->has($indexId)) {
-            return [];
-        }
-
-        /** @var array<string> $index */
+        /** @var null|array<string> $index */
         $index = $this->cache->get($indexId);
 
-        return $index;
+        return $index ?? [];
     }
 }

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

@@ -36,13 +36,8 @@ public function sortBy(Reference ...$refs) : Extractor
         /** @var int $i */
         foreach ($this->cache->read($this->id) as $i => $rows) {
             $maxRowsSize = \max($maxRowsSize, $rows->count());
-            /** @var Rows $singleRowRows */
             $partId = $this->id . '_tmp_' . $i;
-
-            foreach ($rows->sortBy(...$refs)->chunks(1) as $singleRowRows) {
-                $this->cache->add($partId, $singleRowRows);
-            }
-
+            $this->cache->add($partId, $rows->sortBy(...$refs));
             $cachedPartsArray[$partId] = $this->cache->read($partId);
         }
 

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

@@ -33,7 +33,12 @@ public function createHeap(Reference ...$refs) : RowsMinHeap
 
         foreach ($this->generators as $cacheId => $generator) {
             if ($generator->valid()) {
-                $heap->insert(CachedRow::fromRows($generator->current(), $cacheId));
+                /** @var Rows $rows */
+                $rows = $generator->current();
+
+                foreach ($rows->chunks(1) as $chunk) {
+                    $heap->insert(CachedRow::fromRows($chunk, $cacheId));
+                }
                 $generator->next();
             }
         }

src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/SortTest.php

@@ -31,7 +31,7 @@ public function test_etl_sort_at_disk_in_memory() : void
         $cache = \array_diff(\scandir($this->cacheDir), ['..', '.']);
 
         self::assertEmpty($cache);
-        self::assertSame(2506, $cacheSpy->writes());
+        self::assertSame(9, $cacheSpy->writes());
         self::assertSame(5, $cacheSpy->clears());
     }