Repository: jorgecarleitao/arrow2 (rust)

Issue:
Arrow Flight deserialization of batch containing DataType::Struct leads to OutOfSpec
I am using `arrow2-convert` in combination with `arrow2::io::flight` to go from Rust types -> Arrow -> Flight -> Arrow -> Rust types.

I am trying to `arrow2::io::flight::serialize_batch` and `arrow2::io::flight::deserialize_batch` data using the following `DataType`:

```rust
DataType::Struct(vec![
    Field::new(
        "target",
        DataType::Struct(vec![
            Field::new("name", DataType::Utf8, false),
            Field::new("value", DataType::Float64, false)
        ]),
        false
    ),
    Field::new(
        "assignments",
        DataType::List(Box::new(Field::new(
            "item",
            DataType::Struct(vec![
                Field::new("name", DataType::Utf8, false),
                Field::new("value", DataType::Float64, false)
            ]),
            false
        ))),
        false
    )
]);
``` 
After serializing data of the above type and trying to deserialize it, the following error arises: `OutOfSpec("The slots of the array times the physical size must be smaller or equal to the length of the IPC buffer. However, this array reports 2 slots, which, for physical type \"f64\", corresponds to 16 bytes, which is larger than the buffer length 0")`.

Misuse of the API or a bug in either the serialization or deserialization for the `Schema` using the fields specified in this `DataType`?

A repository containing an example demonstrating the above behaviour can be found here: https://github.com/nielsmeima/convert-flight-test/blob/master/src/main.rs

Pull request description:
Raise error when invalid fields are passed to flight
Related to #1092

Patch:
diff --git a/src/io/flight/mod.rs b/src/io/flight/mod.rs
index 4b848e3adf..b439ced2d9 100644
--- a/src/io/flight/mod.rs
+++ b/src/io/flight/mod.rs
@@ -14,16 +14,22 @@ use crate::{
     io::ipc::write::common::{encode_chunk, DictionaryTracker, EncodedData, WriteOptions},
 };
 
-use super::ipc::write::default_ipc_fields;
+pub use super::ipc::write::default_ipc_fields;
 use super::ipc::{IpcField, IpcSchema};
 
 /// Serializes [`Chunk`] to a vector of [`FlightData`] representing the serialized dictionaries
 /// and a [`FlightData`] representing the batch.
+/// # Errors
+/// This function errors iff `fields` is not consistent with `columns`
 pub fn serialize_batch(
     columns: &Chunk<Box<dyn Array>>,
     fields: &[IpcField],
     options: &WriteOptions,
-) -> (Vec<FlightData>, FlightData) {
+) -> Result<(Vec<FlightData>, FlightData)> {
+    if fields.len() != columns.arrays().len() {
+        return Err(Error::InvalidArgumentError("The argument `fields` must be consistent with the columns' schema. Use e.g. &arrow2::io::flight::default_ipc_fields(&schema.fields)".to_string()));
+    }
+
     let mut dictionary_tracker = DictionaryTracker {
         dictionaries: Default::default(),
         cannot_replace: false,
@@ -36,7 +42,7 @@ pub fn serialize_batch(
     let flight_dictionaries = encoded_dictionaries.into_iter().map(Into::into).collect();
     let flight_batch = encoded_batch.into();
 
-    (flight_dictionaries, flight_batch)
+    Ok((flight_dictionaries, flight_batch))
 }
 
 impl From<EncodedData> for FlightData {
diff --git a/src/io/parquet/read/deserialize/mod.rs b/src/io/parquet/read/deserialize/mod.rs
index 433d807baf..d4a7de6ee8 100644
--- a/src/io/parquet/read/deserialize/mod.rs
+++ b/src/io/parquet/read/deserialize/mod.rs
@@ -188,14 +188,29 @@ where
         }
         UInt32 => {
             init.push(InitNested::Primitive(field.is_nullable));
-            types.pop();
-            primitive::iter_to_arrays_nested(
-                columns.pop().unwrap(),
-                init,
-                field.data_type().clone(),
-                chunk_size,
-                |x: i32| x as u32,
-            )
+            let type_ = types.pop().unwrap();
+            match type_.physical_type {
+                PhysicalType::Int32 => primitive::iter_to_arrays_nested(
+                    columns.pop().unwrap(),
+                    init,
+                    field.data_type().clone(),
+                    chunk_size,
+                    |x: i32| x as u32,
+                ),
+                // some implementations of parquet write arrow's u32 into i64.
+                PhysicalType::Int64 => primitive::iter_to_arrays_nested(
+                    columns.pop().unwrap(),
+                    init,
+                    field.data_type().clone(),
+                    chunk_size,
+                    |x: i64| x as u32,
+                ),
+                other => {
+                    return Err(Error::nyi(format!(
+                        "Deserializing UInt32 from {other:?}'s parquet"
+                    )))
+                }
+            }
         }
         UInt64 => {
             init.push(InitNested::Primitive(field.is_nullable));
diff --git a/src/io/parquet/read/deserialize/simple.rs b/src/io/parquet/read/deserialize/simple.rs
index 6d3026c18b..8a8ef653da 100644
--- a/src/io/parquet/read/deserialize/simple.rs
+++ b/src/io/parquet/read/deserialize/simple.rs
@@ -82,12 +82,27 @@ pub fn page_iter_to_arrays<'a, I: 'a + DataPages>(
             chunk_size,
             |x: i32| x as u16,
         ))),
-        UInt32 => dyn_iter(iden(primitive::Iter::new(
-            pages,
-            data_type,
-            chunk_size,
-            |x: i32| x as u32,
-        ))),
+        UInt32 => match physical_type {
+            PhysicalType::Int32 => dyn_iter(iden(primitive::Iter::new(
+                pages,
+                data_type,
+                chunk_size,
+                |x: i32| x as u32,
+            ))),
+            // some implementations of parquet write arrow's u32 into i64.
+            PhysicalType::Int64 => dyn_iter(iden(primitive::Iter::new(
+                pages,
+                data_type,
+                chunk_size,
+                |x: i64| x as u32,
+            ))),
+            other => {
+                return Err(Error::NotYetImplemented(format!(
+                    "Reading uin32 from {:?}-encoded parquet still not implemented",
+                    other
+                )))
+            }
+        },
         Int8 => dyn_iter(iden(primitive::Iter::new(
             pages,
             data_type,
diff --git a/src/io/parquet/read/statistics/mod.rs b/src/io/parquet/read/statistics/mod.rs
index 4f3eaeff7b..540dac4df9 100644
--- a/src/io/parquet/read/statistics/mod.rs
+++ b/src/io/parquet/read/statistics/mod.rs
@@ -410,7 +410,17 @@ fn push(
         }
         UInt8 => primitive::push(from, min, max, |x: i32| Ok(x as u8)),
         UInt16 => primitive::push(from, min, max, |x: i32| Ok(x as u16)),
-        UInt32 => primitive::push(from, min, max, |x: i32| Ok(x as u32)),
+        UInt32 => match physical_type {
+            // some implementations of parquet write arrow's u32 into i64.
+            ParquetPhysicalType::Int64 => primitive::push(from, min, max, |x: i64| Ok(x as u32)),
+            ParquetPhysicalType::Int32 => primitive::push(from, min, max, |x: i32| Ok(x as u32)),
+            other => {
+                return Err(Error::NotYetImplemented(format!(
+                    "Can't decode UInt32 type from parquet type {:?}",
+                    other
+                )))
+            }
+        },
         Int32 => primitive::push(from, min, max, |x: i32| Ok(x as i32)),
         Int64 | Date64 | Time64(_) | Duration(_) => {
             primitive::push(from, min, max, |x: i64| Ok(x as i64))


Test patch:
diff --git a/integration-testing/src/flight_client_scenarios/integration_test.rs b/integration-testing/src/flight_client_scenarios/integration_test.rs
index 4d028292f4..19a4b1a4ef 100644
--- a/integration-testing/src/flight_client_scenarios/integration_test.rs
+++ b/integration-testing/src/flight_client_scenarios/integration_test.rs
@@ -147,7 +147,7 @@ async fn send_batch(
     fields: &[IpcField],
     options: &write::WriteOptions,
 ) -> Result {
-    let (dictionary_flight_data, mut batch_flight_data) = serialize_batch(batch, fields, options);
+    let (dictionary_flight_data, mut batch_flight_data) = serialize_batch(batch, fields, options)?;
 
     upload_tx
         .send_all(&mut stream::iter(dictionary_flight_data).map(Ok))
diff --git a/integration-testing/src/flight_server_scenarios/integration_test.rs b/integration-testing/src/flight_server_scenarios/integration_test.rs
index 73b4cd9dbe..07453f5c72 100644
--- a/integration-testing/src/flight_server_scenarios/integration_test.rs
+++ b/integration-testing/src/flight_server_scenarios/integration_test.rs
@@ -122,7 +122,7 @@ impl FlightService for FlightServiceImpl {
             .enumerate()
             .flat_map(|(counter, batch)| {
                 let (dictionary_flight_data, mut batch_flight_data) =
-                    serialize_batch(batch, &flight.ipc_schema.fields, &options);
+                    serialize_batch(batch, &flight.ipc_schema.fields, &options).unwrap();
 
                 // Only the record batch's FlightData gets app_metadata
                 let metadata = counter.to_string().into_bytes();
diff --git a/tests/it/io/parquet/read.rs b/tests/it/io/parquet/read.rs
index 4e426ef5ba..8ab783dbab 100644
--- a/tests/it/io/parquet/read.rs
+++ b/tests/it/io/parquet/read.rs
@@ -130,11 +130,15 @@ fn v1_timestamp_ms_nullable() -> Result<()> {
 }
 
 #[test]
-#[ignore] // pyarrow issue; see https://issues.apache.org/jira/browse/ARROW-12201
 fn v1_u32_nullable() -> Result<()> {
     test_pyarrow_integration("uint32", 1, "basic", false, false, None)
 }
 
+#[test]
+fn v2_u32_nullable() -> Result<()> {
+    test_pyarrow_integration("uint32", 2, "basic", false, false, None)
+}
+
 #[test]
 fn v2_int64_nullable() -> Result<()> {
     test_pyarrow_integration("int64", 2, "basic", false, false, None)