Chore/thrift performance (#1867)

* Reorganized Thrift related namespaces

* Thrift building blocks cleanup

* Simplified Thrift Protocol and Transport

* Improved debug code precision

* Optimize MemoryBuffer and Compact Protocol

* Optimized Thrift PhpFileStream and BufferedTransports

* Fixed cleanining page containers after flush

* Fixed issue with non cleaning page coumn chunk buidlers properly after flushing

* Covered Thrift objects with unit tests

* Added parquet schema conversion to DSL

Author: norberttech

.php-cs-fixer.php

@@ -32,7 +32,7 @@
         __DIR__ . '/tools/rector/src',
     ])
     ->exclude([
-        'Flow/Parquet/Thrift',
+        'Flow/Parquet/ThriftModel',
         'Flow/CLI/Tests/Integration',
         'Flow/ETL/Tests/Unit/Loader',
         'Flow/ETL/Tests/Unit/Exception'

composer.json

@@ -456,8 +456,8 @@
             "./tools/phpdocumentor/vendor/bin/phpdoc --config=./phpdoc/bridge.symfony.http-foundation.xml"
         ],
         "build:parquet:thrift": [
-            "grep -q 'namespace php Flow.Parquet.Thrift' src/lib/parquet/src/Flow/Parquet/Resources/Thrift/parquet.thrift || { echo \"Flow php namespace not found in thrift definition!\"; exit 1; }\n",
-            "rm src/lib/parquet/src/Flow/Parquet/Thrift/*.php",
+            "grep -q 'namespace php Flow.Parquet.ThriftModel' src/lib/parquet/src/Flow/Parquet/Resources/Thrift/parquet.thrift || { echo \"Flow php namespace not found in thrift definition!\"; exit 1; }\n",
+            "rm src/lib/parquet/src/Flow/Parquet/ThriftModel/*.php",
             "thrift --gen php --out src/lib/parquet/src src/lib/parquet/src/Flow/Parquet/Resources/Thrift/parquet.thrift",
             "@cs:php:fix"
         ],

phpstan.neon

@@ -65,8 +65,8 @@ parameters:
         - src/adapter/etl-adapter-elasticsearch/src/Flow/ETL/Adapter/Elasticsearch/ElasticsearchPHP/SearchParams.php
         - src/adapter/etl-adapter-elasticsearch/src/Flow/ETL/Adapter/Elasticsearch/ElasticsearchPHP/PointInTime.php
         - src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/Sheet/SheetsManager.php
-        - src/lib/parquet/src/Flow/Parquet/ThriftStream/*
         - src/lib/parquet/src/Flow/Parquet/Thrift/*
+        - src/lib/parquet/src/Flow/Parquet/ThriftModel/*
         - src/lib/parquet/src/Flow/Parquet/BinaryReader/*
         - src/lib/parquet/src/Flow/Parquet/Dremel/ColumnData/DefinitionConverter.php
 

rector.src.php

@@ -17,7 +17,7 @@
     ])
     ->withSkip([
         StringClassNameToClassConstantRector::class,
-        __DIR__ . '/src/lib/parquet/src/Flow/Parquet/Thrift/*',
+        __DIR__ . '/src/lib/parquet/src/Flow/Parquet/ThriftModel/*',
     ])
     ->withCache(__DIR__ . '/var/rector/src')
     ->withImportNames(importShortClasses: false, removeUnusedImports: true)

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

@@ -93,3 +93,15 @@ function empty_generator() : \Generator
 {
     yield from [];
 }
+
+#[DocumentationDSL(module: Module::PARQUET, type: DSLType::HELPER)]
+function schema_to_parquet(Schema $schema) : \Flow\Parquet\ParquetFile\Schema
+{
+    return (new SchemaConverter())->toParquet($schema);
+}
+
+#[DocumentationDSL(module: Module::PARQUET, type: DSLType::HELPER)]
+function schema_from_parquet(\Flow\Parquet\ParquetFile\Schema $schema) : Schema
+{
+    return (new SchemaConverter())->toFlow($schema);
+}

src/core/etl/tests/Flow/ETL/Tests/Double/FakeStaticOrdersExtractor.php

@@ -4,7 +4,15 @@
 
 namespace Flow\ETL\Tests\Double;
 
-use function Flow\ETL\DSL\{array_to_rows, datetime_schema, float_schema, list_schema, schema, string_schema, structure_schema, uuid_schema};
+use function Flow\ETL\DSL\{array_to_rows,
+    datetime_schema,
+    float_schema,
+    integer_schema,
+    list_schema,
+    schema,
+    string_schema,
+    structure_schema,
+    uuid_schema};
 use function Flow\Types\DSL\{type_float, type_integer, type_list, type_string, type_structure};
 use Flow\ETL\{Extractor, FlowContext, Schema};
 
@@ -17,6 +25,7 @@ public function __construct(private int $count = 1_000)
     public static function schema() : Schema
     {
         return schema(
+            integer_schema('index'),
             uuid_schema('order_id'),
             datetime_schema('created_at'),
             datetime_schema('updated_at', true),
@@ -65,6 +74,7 @@ public function rawData() : \Generator
 
         for ($i = 0; $i < $this->count; $i++) {
             yield [
+                'index' => $i,
                 'order_id' => '254d61c5-22c8-4407-83a2-76f1cab53af2',
                 'created_at' => new \DateTimeImmutable('2025-01-01 12:00:00'),
                 'updated_at' => \random_int(0, 1) === 1 ? new \DateTimeImmutable('2025-01-01 12:10:00') : null,

src/lib/parquet/src/Flow/Parquet/ParquetFile.php

@@ -11,15 +11,15 @@
     ParquetFile\Metadata,
     ParquetFile\Page\ColumnPageHeader,
     ParquetFile\Schema,
-    Reader\PageReader};
+    Reader\PageReader,
+    Thrift\CompactProtocol,
+    Thrift\MemoryBuffer};
 use Flow\Parquet\Exception\{InvalidArgumentException, RuntimeException};
 use Flow\Parquet\ParquetFile\Data\DataConverter;
 use Flow\Parquet\ParquetFile\Schema\{Column, FlatColumn};
 use Flow\Parquet\ParquetFile\Schema\NestedColumn;
 use Flow\Parquet\Reader\{ColumnChunkReader, ColumnChunkViewer};
-use Flow\Parquet\Thrift\FileMetaData;
-use Thrift\Protocol\TCompactProtocol;
-use Thrift\Transport\TMemoryBuffer;
+use Flow\Parquet\ThriftModel\FileMetaData;
 
 final class ParquetFile
 {
@@ -64,8 +64,8 @@ public function metadata() : Metadata
 
         $thriftMetadata = new FileMetaData();
         $thriftMetadata->read(
-            new TCompactProtocol(
-                new TMemoryBuffer($metadata)
+            new CompactProtocol(
+                new MemoryBuffer($metadata)
             )
         );
 

src/lib/parquet/src/Flow/Parquet/ParquetFile/Metadata.php

@@ -6,7 +6,7 @@
 
 use Flow\Parquet\Options;
 use Flow\Parquet\ParquetFile\RowGroup\ColumnChunk;
-use Flow\Parquet\Thrift\FileMetaData;
+use Flow\Parquet\ThriftModel\FileMetaData;
 
 final readonly class Metadata
 {

src/lib/parquet/src/Flow/Parquet/ParquetFile/Page/Header/DataPageHeader.php

@@ -16,7 +16,7 @@ public function __construct(
     ) {
     }
 
-    public static function fromThrift(\Flow\Parquet\Thrift\DataPageHeader $thrift) : self
+    public static function fromThrift(\Flow\Parquet\ThriftModel\DataPageHeader $thrift) : self
     {
         return new self(
             Encodings::from($thrift->encoding),
@@ -41,9 +41,9 @@ public function repetitionLevelEncoding() : Encodings
         return $this->repetitionLevelEncoding;
     }
 
-    public function toThrift() : \Flow\Parquet\Thrift\DataPageHeader
+    public function toThrift() : \Flow\Parquet\ThriftModel\DataPageHeader
     {
-        return new \Flow\Parquet\Thrift\DataPageHeader([
+        return new \Flow\Parquet\ThriftModel\DataPageHeader([
             'num_values' => $this->valuesCount,
             'encoding' => $this->encoding->value,
             'definition_level_encoding' => $this->definitionLevelEncoding->value,

src/lib/parquet/src/Flow/Parquet/ParquetFile/Page/Header/DataPageHeaderV2.php

@@ -22,7 +22,7 @@ public function __construct(
     ) {
     }
 
-    public static function fromThrift(\Flow\Parquet\Thrift\DataPageHeaderV2 $thrift, Options $options) : self
+    public static function fromThrift(\Flow\Parquet\ThriftModel\DataPageHeaderV2 $thrift, Options $options) : self
     {
         return new self(
             $thrift->num_values,
@@ -61,9 +61,9 @@ public function statistics(Options $options) : ?StatisticsReader
         return new StatisticsReader($this->statistics, $options);
     }
 
-    public function toThrift() : \Flow\Parquet\Thrift\DataPageHeaderV2
+    public function toThrift() : \Flow\Parquet\ThriftModel\DataPageHeaderV2
     {
-        return new \Flow\Parquet\Thrift\DataPageHeaderV2([
+        return new \Flow\Parquet\ThriftModel\DataPageHeaderV2([
             'num_values' => $this->valuesCount,
             'num_nulls' => $this->nullsCount,
             'num_rows' => $this->rowsCount,