Merge pull request #9171 from sundy-li/binary-arrow

feat(query): bump arrow2 crate

Author: mergify[bot]

Cargo.lock

@@ -34,12 +34,13 @@ simd = ["arrow/simd"]
 # Workspace dependencies
 
 # Crates.io dependencies
-arrow = { package = "arrow2", git = "https://github.com/jorgecarleitao/arrow2", rev = "95e117d", default-features = false, features = [
+arrow = { package = "arrow2", git = "https://github.com/jorgecarleitao/arrow2", rev = "1da33ac", default-features = false, features = [
     "io_parquet",
     "io_parquet_compression",
 ] }
+
 arrow-format = { version = "0.8.0", features = ["flight-data", "flight-service", "ipc"] }
 futures = "0.3.24"
-parquet2 = { version = "0.16.3", default_features = false }
+parquet2 = { version = "0.17.0", default_features = false }
 
 [dev-dependencies]

src/common/arrow/Cargo.toml

@@ -40,6 +40,7 @@ pub fn serialize_data_blocks_with_compression(
         write_statistics: false,
         compression,
         version: Version::V2,
+        data_pagesize_limit: None,
     };
     let batches = blocks
         .into_iter()
@@ -87,16 +88,12 @@ pub fn serialize_data_blocks(
     serialize_data_blocks_with_compression(blocks, schema, buf, CompressionOptions::Lz4Raw)
 }
 
-fn col_encoding(_data_type: &ArrowDataType) -> Encoding {
-    // Although encoding does work, parquet2 has not implemented decoding of DeltaLengthByteArray yet, we fallback to Plain
-    // From parquet2: Decoding "DeltaLengthByteArray"-encoded required V2 pages is not yet implemented for Binary.
-    //
-    // match data_type {
-    //    ArrowDataType::Binary
-    //    | ArrowDataType::LargeBinary
-    //    | ArrowDataType::Utf8
-    //    | ArrowDataType::LargeUtf8 => Encoding::DeltaLengthByteArray,
-    //    _ => Encoding::Plain,
-    //}
-    Encoding::Plain
+fn col_encoding(data_type: &ArrowDataType) -> Encoding {
+    match data_type {
+        ArrowDataType::Binary
+        | ArrowDataType::LargeBinary
+        | ArrowDataType::Utf8
+        | ArrowDataType::LargeUtf8 => Encoding::DeltaLengthByteArray,
+        _ => Encoding::Plain,
+    }
 }

src/query/datablocks/src/serialize.rs

@@ -430,6 +430,9 @@ impl BlockReader {
                 Err(ErrorCode::StorageOther(err_msg))
             }
             Compression::Lz4Raw => Ok(ParquetCompression::Lz4Raw),
+            Compression::Snappy => Ok(ParquetCompression::Snappy),
+            Compression::Zstd => Ok(ParquetCompression::Zstd),
+            Compression::Gzip => Ok(ParquetCompression::Gzip),
         }
     }
 

src/query/storages/fuse/fuse/src/io/read/block_reader.rs

@@ -85,6 +85,9 @@ where Self: Sized
 pub enum Compression {
     Lz4,
     Lz4Raw,
+    Snappy,
+    Zstd,
+    Gzip,
 }
 
 impl Compression {