Bug parquet reader memory consumption (#1757)

* Split FlatColumnData to Read and Write

* Use Read/Write FlatColumnData in reader/writer

* Updated ReadFlatColumnValues constructor

* Make ParquetFile yield rows instead of buffering them in array

* Refactored REad/WriteFlatcolumnValues iterator method

* Yield pages from column chunk reader instead of buffering them

* Optimized Dremel Assembler null level processing

* Make assemblyFlat to yield rows instead of buffering them

* Fix yielding one page at time from ColumnChunkReader

* Optimized microseconds to date time conversio in parquet

* Optimize DremelAssembler assemblyList method

* Optimize DremelAssemble assemblyList assemblyMap assemblyStructure methods

* Optimize using stack for parquet flat columns

* Optimize BinaryReady to yield values

* Make PlainValueUnpacker return generator

* Fix bug in write column values and read even deeply nested columns read page by page

* Cleaned up ParquetFile

* Move to flat number of rows per page vs estimations basde on rows in memory

* Smal refactoring of ReadFlatColumnValues

* Fixed skipping rows in Read/Write flat column values

* Simplified read flat column values

* Make ReadFlatColumnValues take values as generator

* Updated dsl definitions

* CS Fixes

* Regenerate data for parquet extractor benchmark

* Added missing file

* Added a failing test related to data page sizes

* Temporarly save nested column children in one page

* Restored data page size option

* PoC of new Optimized Row Group Writer

* Fixed bug when writing empty pages

* Fix not equal rows distribution across pges

* Fixed bug related to inverting booleans

* More performance optimizations

* Regenerated parquet fixtures in order to fix extractors benchmarks

* Removed legacy row group builder

* Reorganized parquet library namespaces

* Covered column Page Builders with unit tests

* Fixed namespace for ChunkColumnBuilder implementtaions

* Covered RowGroupBuilder and dependencies with unit tests

* Added Benchmarks for Parquet Library

* Updated github workflow with benchmarks

* Added snappy extension to parquet benchmarks

Author: norberttech

.github/workflows/test-benchmark.yml

@@ -43,7 +43,9 @@ jobs:
           tools: composer:v2
           php-version: "${{ matrix.php-version }}"
           ini-values: memory_limit=-1
-          extensions: :psr, bcmath, dom, hash, json, mbstring, xml, xmlwriter, xmlreader, zlib
+          extensions: :psr, bcmath, dom, hash, json, mbstring, xml, xmlwriter, xmlreader, zlib, snappy-https://github.com/kjdev/php-ext-snappy@0.2.1
+        env:
+          SNAPPY_CONFIGURE_PREFIX_OPTS: "CXXFLAGS=-std=c++11"
 
       - name: "Get Composer Cache Directory"
         id: composer-cache
@@ -107,6 +109,14 @@ jobs:
             echo '  '
             echo '</details>'
             echo '  '
+            echo '<details><summary>Parquet Library</summary>'
+            echo '  '
+            echo '```shell'
+            composer test:benchmark:parquet-library -- --ref=1.x --progress=none
+            echo '```'
+            echo '  '
+            echo '</details>'
+            echo '  '
           } >> "./var/phpbench/summary.txt"
 
       - uses: actions/upload-artifact@v4

composer.json

@@ -361,7 +361,8 @@
             "@test:benchmark:building_blocks",
             "@test:benchmark:extractor",
             "@test:benchmark:loader",
-            "@test:benchmark:transformer"
+            "@test:benchmark:transformer",
+            "@test:benchmark:parquet-library"
         ],
         "test:website": [
             "composer test --working-dir=./web/landing"
@@ -381,6 +382,9 @@
         "test:benchmark:transformer": [
             "tools/phpbench/vendor/bin/phpbench run --report=flow-report --group=transformer"
         ],
+        "test:benchmark:parquet-library": [
+            "tools/phpbench/vendor/bin/phpbench run --report=flow-report --group=parquet-library"
+        ],
         "test:mutation": [
             "tools/infection/vendor/bin/infection --threads=max"
         ],

phpbench.json

@@ -22,7 +22,8 @@
     "src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Benchmark/",
     "src/adapter/etl-adapter-text/tests/Flow/ETL/Adapter/Text/Tests/Benchmark/",
     "src/adapter/etl-adapter-xml/tests/Flow/ETL/Adapter/XML/Tests/Benchmark/",
-    "src/core/etl/tests/Flow/ETL/Tests/Benchmark/"
+    "src/core/etl/tests/Flow/ETL/Tests/Benchmark/",
+    "src/lib/parquet/tests/Flow/Parquet/Tests/Benchmark/"
   ],
   "runner.php_config": { "memory_limit": "1G" },
   "runner.iterations": 3,

phpstan.neon

@@ -66,7 +66,7 @@ parameters:
         - src/lib/parquet/src/Flow/Parquet/ThriftStream/*
         - src/lib/parquet/src/Flow/Parquet/Thrift/*
         - src/lib/parquet/src/Flow/Parquet/BinaryReader/*
-        - src/lib/parquet/src/Flow/Parquet/ParquetFile/RowGroupBuilder/ColumnData/DefinitionConverter.php
+        - src/lib/parquet/src/Flow/Parquet/Dremel/ColumnData/DefinitionConverter.php
 
     tmpDir: var/phpstan/cache
 

rector.tests.php

@@ -57,7 +57,7 @@
 use Rector\Set\ValueObject\LevelSetList;
 use Rector\Transform\Rector\StaticCall\StaticCallToFuncCallRector;
 use \Rector\Transform\ValueObject\StaticCallToFuncCall;
-use Rector\Renaming\Rector\FuncCall\RenameFunctionRector;
+use Rector\PHPUnit\AnnotationsToAttributes\Rector\ClassMethod\DataProviderAnnotationToAttributeRector;
 
 return RectorConfig::configure()
     ->withPaths([
@@ -71,6 +71,9 @@
     ->withSets([
         LevelSetList::UP_TO_PHP_82
     ])
+    ->withRules([
+        DataProviderAnnotationToAttributeRector::class,
+    ])
     ->withConfiguredRule(
         StaticCallToFuncCallRector::class,
         [

src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/RowsNormalizer.php

@@ -27,7 +27,9 @@ public function normalize(Rows $rows, Schema $schema) : array
             $columns = [];
 
             foreach ($row->entries() as $entry) {
-                if ($schema->get($entry->ref())->isNullable() && $entry->value() === null) {
+                $definition = $schema->get($entry->ref());
+
+                if ($definition->isNullable() && $entry->value() === null) {
                     $columns[$entry->name()] = null;
 
                     continue;

src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/functions.php

@@ -4,7 +4,11 @@
 
 namespace Flow\ETL\Adapter\Parquet;
 
-use Flow\ETL\{Attribute\DocumentationDSL, Attribute\DocumentationExample, Attribute\Module, Attribute\Type as DSLType};
+use Flow\ETL\{Attribute\DocumentationDSL,
+    Attribute\DocumentationExample,
+    Attribute\Module,
+    Attribute\Type as DSLType
+};
 use Flow\ETL\Schema;
 use Flow\Filesystem\Path;
 use Flow\Parquet\{ByteOrder, Options};
@@ -68,3 +72,24 @@ function to_parquet(
 
     return $loader;
 }
+
+/**
+ * @template T
+ *
+ * @param array<T> $data
+ *
+ * @return \Generator<T>
+ */
+#[DocumentationDSL(module: Module::PARQUET, type: DSLType::HELPER)]
+function array_to_generator(array $data) : \Generator
+{
+    foreach ($data as $row) {
+        yield $row;
+    }
+}
+
+#[DocumentationDSL(module: Module::PARQUET, type: DSLType::HELPER)]
+function empty_generator() : \Generator
+{
+    yield from [];
+}

src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Benchmark/Fixtures/orders_10k.parquet

@@ -5,11 +5,13 @@
 namespace Flow\ETL\Adapter\Parquet\Tests\Integration;
 
 use function Flow\ETL\Adapter\Parquet\{from_parquet, to_parquet};
-use function Flow\ETL\DSL\{config, from_array, json_schema, schema, str_schema};
+use function Flow\ETL\DSL\{config, from_array, json_schema, overwrite, schema, str_schema};
 use function Flow\ETL\DSL\data_frame;
 use function Flow\Filesystem\DSL\{path};
 use Flow\ETL\Tests\Double\FakeExtractor;
-use Flow\ETL\{Tests\FlowTestCase};
+use Flow\ETL\{Tests\Double\FakeRandomOrdersExtractor, Tests\FlowTestCase};
+use Flow\Filesystem\SizeUnits;
+use Flow\Parquet\{Option, Options};
 use Ramsey\Uuid\Uuid;
 
 final class ParquetTest extends FlowTestCase
@@ -33,6 +35,32 @@ public function test_writing_and_reading_into_parquet() : void
         );
     }
 
+    public function test_writing_and_reading_parquet_orders() : void
+    {
+        $path = path(__DIR__ . '/var/orders.snappy.parquet');
+        $config = config();
+        data_frame($config)
+            ->read(new FakeRandomOrdersExtractor(1000))
+            ->mode(overwrite())
+            ->write(
+                to_parquet($path)
+                    ->withOptions(
+                        Options::default()
+                            ->set(Option::ROW_GROUP_SIZE_CHECK_INTERVAL, 500)
+                            ->set(Option::ROW_GROUP_SIZE_BYTES, SizeUnits::MiB_SIZE)
+                            ->set(Option::PAGE_MAXIMUM_ROWS_COUNT, 10)
+                    )
+            )
+            ->run();
+
+        self::assertEquals(
+            1000,
+            (data_frame($config))
+                ->read(from_parquet($path))
+                ->count()
+        );
+    }
+
     public function test_writing_with_provided_schema() : void
     {
         $path = path('memory://var/file_schema.snappy.parquet');