chore: upgrade arrow-format (#14675)

upgrade arrow-format

Author: ariesdevil

Cargo.lock

@@ -253,7 +253,7 @@ rpath = false
 
 [patch.crates-io]
 # If there are dependencies that need patching, they can be listed below.
-arrow-format = { git = "https://github.com/everpcpc/arrow-format", rev = "588d371" }
+arrow-format = { git = "https://github.com/everpcpc/arrow-format", rev = "ad8f2dd" }
 parquet2 = { git = "https://github.com/jorgecarleitao/parquet2", rev = "b0e6545" }
 metrics = { git = "https://github.com/datafuse-extras/metrics.git", rev = "fc2ecd1" }
 icelake = { git = "https://github.com/icelake-io/icelake", rev = "f06cdf3" }

Cargo.toml

@@ -364,6 +364,7 @@ fn get_data_type(
         Struct(_) => deserialize_struct(field)?,
         Union(union_) => deserialize_union(union_, field)?,
         Map(map) => deserialize_map(map, field)?,
+        _ => unimplemented!(),
     })
 }
 

src/common/arrow/src/arrow/io/ipc/read/schema.rs

@@ -247,6 +247,33 @@ fn serialize_compression(
     }
 }
 
+fn set_variadic_buffer_counts(_counts: &mut Vec<i64>, array: &dyn Array) {
+    match array.data_type() {
+        DataType::Struct(_) => {
+            let array = array.as_any().downcast_ref::<StructArray>().unwrap();
+            for array in array.values() {
+                set_variadic_buffer_counts(_counts, array.as_ref())
+            }
+        }
+        DataType::LargeList(_) => {
+            let array = array.as_any().downcast_ref::<ListArray<i64>>().unwrap();
+            set_variadic_buffer_counts(_counts, array.values().as_ref())
+        }
+        DataType::FixedSizeList(_, _) => {
+            let array = array.as_any().downcast_ref::<FixedSizeListArray>().unwrap();
+            set_variadic_buffer_counts(_counts, array.values().as_ref())
+        }
+        DataType::Dictionary(_, _, _) => {
+            let array = array
+                .as_any()
+                .downcast_ref::<DictionaryArray<u32>>()
+                .unwrap();
+            set_variadic_buffer_counts(_counts, array.values().as_ref())
+        }
+        _ => (),
+    }
+}
+
 /// Write [`Chunk`] into two sets of bytes, one for the header (ipc::Schema::Message) and the
 /// other for the batch's data
 fn chunk_to_bytes_amortized(
@@ -260,7 +287,10 @@ fn chunk_to_bytes_amortized(
     arrow_data.clear();
 
     let mut offset = 0;
+    let mut variadic_buffer_counts = vec![];
+
     for array in chunk.arrays() {
+        set_variadic_buffer_counts(&mut variadic_buffer_counts, array.as_ref());
         write(
             array.as_ref(),
             &mut buffers,
@@ -272,6 +302,12 @@ fn chunk_to_bytes_amortized(
         )
     }
 
+    let variadic_buffer_counts = if variadic_buffer_counts.is_empty() {
+        None
+    } else {
+        Some(variadic_buffer_counts)
+    };
+
     let compression = serialize_compression(options.compression);
 
     let message = arrow_format::ipc::Message {
@@ -282,6 +318,7 @@ fn chunk_to_bytes_amortized(
                 nodes: Some(nodes),
                 buffers: Some(buffers),
                 compression,
+                variadic_buffer_counts,
             },
         ))),
         body_length: arrow_data.len() as i64,
@@ -306,6 +343,15 @@ fn dictionary_batch_to_bytes<K: DictionaryKey>(
     let mut buffers: Vec<arrow_format::ipc::Buffer> = vec![];
     let mut arrow_data: Vec<u8> = vec![];
 
+    let mut variadic_buffer_counts = vec![];
+    set_variadic_buffer_counts(&mut variadic_buffer_counts, array.values().as_ref());
+
+    let variadic_buffer_counts = if variadic_buffer_counts.is_empty() {
+        None
+    } else {
+        Some(variadic_buffer_counts)
+    };
+
     let length = write_dictionary(
         array,
         &mut buffers,
@@ -329,6 +375,7 @@ fn dictionary_batch_to_bytes<K: DictionaryKey>(
                     nodes: Some(nodes),
                     buffers: Some(buffers),
                     compression,
+                    variadic_buffer_counts,
                 })),
                 is_delta: false,
             },