Enabled more integration tests with pyarrow

.github/workflows/integration-parquet.yml

@@ -46,7 +46,7 @@ jobs:
           python -m venv venv
           source venv/bin/activate
           pip install --upgrade pip
-          pip install pyarrow==6 pyspark==3
+          pip install pyarrow==8 pyspark==3
           python main.py
           # test against spark
           python main_spark.py

arrow-parquet-integration-testing/main.py

@@ -49,11 +49,8 @@ def _expected(file: str):
     pyarrow.date64(),
     pyarrow.time32("s"),
     pyarrow.timestamp("s"),
+    # the issue here is the second, not the tz
     pyarrow.timestamp("s", tz="UTC"),
-    pyarrow.duration("s"),
-    pyarrow.duration("ms"),
-    pyarrow.duration("us"),
-    pyarrow.duration("ns"),
 ]
 
 

arrow-parquet-integration-testing/src/main.rs

@@ -11,8 +11,8 @@ use arrow2::{
         json_integration::read,
         json_integration::ArrowJson,
         parquet::write::{
-            CompressionOptions as ParquetCompression, Encoding, FileWriter, RowGroupIterator,
-            Version as ParquetVersion, WriteOptions,
+            transverse, CompressionOptions as ParquetCompression, Encoding, FileWriter,
+            RowGroupIterator, Version as ParquetVersion, WriteOptions,
         },
     },
 };
@@ -174,16 +174,18 @@ fn main() -> Result<()> {
     let encodings = schema
         .fields
         .iter()
-        .map(|x| match x.data_type() {
-            DataType::Dictionary(..) => vec![Encoding::RleDictionary],
-            DataType::Utf8 | DataType::LargeUtf8 => {
-                vec![if args.encoding_utf8 == EncodingScheme::Delta {
-                    Encoding::DeltaLengthByteArray
-                } else {
-                    Encoding::Plain
-                }]
-            }
-            _ => vec![Encoding::Plain],
+        .map(|f| {
+            transverse(&f.data_type, |dt| match dt {
+                DataType::Dictionary(..) => Encoding::RleDictionary,
+                DataType::Utf8 | DataType::LargeUtf8 => {
+                    if args.encoding_utf8 == EncodingScheme::Delta {
+                        Encoding::DeltaLengthByteArray
+                    } else {
+                        Encoding::Plain
+                    }
+                }
+                _ => Encoding::Plain,
+            })
         })
         .collect();
 

src/io/parquet/write/pages.rs

@@ -172,7 +172,7 @@ fn to_leafs_recursive<'a>(array: &'a dyn Array, leafs: &mut Vec<&'a dyn Array>)
         }
         Null | Boolean | Primitive(_) | Binary | FixedSizeBinary | LargeBinary | Utf8
         | LargeUtf8 | Dictionary(_) => leafs.push(array),
-        _ => todo!(),
+        other => todo!("Writing {:?} to parquet not yet implemented", other),
     }
 }