Fixed writing nullable parquet values that are marked in the schema as optional (#1025)

* Fixed writing nullable parquet values that are marked in the schema as optional

* Fixed failing validation test

* Fixed reading empty chunks without snappy extension

Author: norberttech

composer.lock

@@ -37,6 +37,10 @@ public function decodeBitPacked(BinaryReader $reader, int $bitWidth, int $varInt
         $readBytes = $reader->readBytes(\min($remainingByteCount, $totalByteCount));
         $actualByteCount = $readBytes->count();
 
+        if ($actualByteCount === 0) {
+            return;
+        }
+
         $bitMask = (1 << $bitWidth) - 1;
         $byteIndex = 0;
         $currentByte = $readBytes[$byteIndex];

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

@@ -79,7 +79,7 @@ public function readDictionary(FlatColumn $column, PageHeader $pageHeader, Compr
                 (new Codec($this->options))
                     ->decompress(
                         /** @phpstan-ignore-next-line */
-                        \fread($stream, $pageHeader->compressedPageSize()),
+                        $pageHeader->compressedPageSize() === 0 ? '' : \fread($stream, $pageHeader->compressedPageSize()),
                         $codec
                     ),
                 $column,

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

@@ -5,7 +5,6 @@
 namespace Flow\Parquet\ParquetFile;
 
 use Flow\Parquet\Data\DataConverter;
-use Flow\Parquet\Exception\InvalidArgumentException;
 use Flow\Parquet\ParquetFile\RowGroupBuilder\Validator\{ColumnDataValidator, DisabledValidator};
 use Flow\Parquet\ParquetFile\RowGroupBuilder\{ColumnChunkBuilder, Flattener, PageSizeCalculator, RowGroupContainer, RowGroupStatistics};
 use Flow\Parquet\{Option, Options};
@@ -49,9 +48,6 @@ public function addRow(array $row) : void
         $flatRow = [];
 
         foreach ($this->schema->columns() as $column) {
-            if (!\array_key_exists($column->name(), $row)) {
-                throw new InvalidArgumentException(\sprintf("Column '%s' not found in row", $column->name()));
-            }
             $flatRow[] = $this->flattener->flattenColumn($column, $row);
         }
 

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

@@ -18,7 +18,7 @@ public function flattenColumn(Column $column, array $row) : array
         if (!\array_key_exists($column->name(), $row)) {
             $this->validator->validate($column, null);
 
-            return [];
+            return [$column->name() => null];
         }
 
         /**

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

@@ -20,6 +20,8 @@ public function isMap() : bool;
 
     public function isMapElement() : bool;
 
+    public function isRequired() : bool;
+
     public function isStruct() : bool;
 
     public function isStructElement() : bool;

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

@@ -243,6 +243,11 @@ public function isMapElement() : bool
         return false;
     }
 
+    public function isRequired() : bool
+    {
+        return $this->repetition !== Repetition::OPTIONAL;
+    }
+
     public function isStruct() : bool
     {
         return false;

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

@@ -264,6 +264,11 @@ public function isMapElement() : bool
         return false;
     }
 
+    public function isRequired() : bool
+    {
+        return $this->repetition !== Repetition::OPTIONAL;
+    }
+
     public function isStruct() : bool
     {
         if ($this->isMap()) {