ARROW-7842: [Rust] [Parquet] implement array_reader for list type columns

rust/arrow/src/ipc/convert.rs

@@ -511,10 +511,11 @@ pub(crate) fn get_fb_field_type<'a: 'b, 'b>(
         }
         List(ref list_type) => {
             let inner_types = get_fb_field_type(list_type, &mut fbb);
+            let field_name = fbb.create_string("list"); // field schema requires name to be not None
             let child = ipc::Field::create(
                 &mut fbb,
                 &ipc::FieldArgs {
-                    name: None,
+                    name: Some(field_name),
                     nullable: false,
                     type_type: inner_types.0,
                     type_: Some(inner_types.1),

rust/arrow/src/record_batch.rs

@@ -214,6 +214,25 @@ mod tests {
             .len(4)
             .add_buffer(Buffer::from([42, 28, 19, 31].to_byte_slice()))
             .build();
+
+        // Construct a value array
+        let value_data = ArrayData::builder(DataType::Int32)
+            .len(9)
+            .add_buffer(Buffer::from(&[0, 1, 2, 3, 4, 5, 6, 7, 8].to_byte_slice()))
+            .build();
+
+        // Construct a buffer for value offsets, for the nested array:
+        //  [[0, 1, 2], [3, 4, 5], [6, 7], [8]]
+        let value_offsets = Buffer::from(&[0, 3, 6, 8, 9].to_byte_slice());
+
+        // Construct a list array from the above two
+        let list_data_type = DataType::List(Box::new(DataType::Int32));
+        let list_data = ArrayData::builder(list_data_type.clone())
+            .len(4)
+            .add_buffer(value_offsets.clone())
+            .add_child_data(value_data.clone())
+            .build();
+
         let struct_array = StructArray::from(vec![
             (
                 Field::new("b", DataType::Boolean, false),
@@ -224,16 +243,111 @@ mod tests {
                 Field::new("c", DataType::Int32, false),
                 Arc::new(Int32Array::from(vec![42, 28, 19, 31])),
             ),
+            (
+                Field::new("d", DataType::List(Box::new(DataType::Int32)), false),
+                Arc::new(ListArray::from(list_data.clone())),
+            ),
         ]);
 
         let batch = RecordBatch::from(&struct_array);
-        assert_eq!(2, batch.num_columns());
+        assert_eq!(3, batch.num_columns());
         assert_eq!(4, batch.num_rows());
         assert_eq!(
             struct_array.data_type(),
             &DataType::Struct(batch.schema().fields().to_vec())
         );
         assert_eq!(batch.column(0).data(), boolean_data);
         assert_eq!(batch.column(1).data(), int_data);
+        assert_eq!(batch.column(2).data(), list_data);
+    }
+
+    #[test]
+    fn create_record_batch_with_list_column() {
+        let schema = Schema::new(vec![Field::new(
+            "a",
+            DataType::List(Box::new(DataType::Int32)),
+            false,
+        )]);
+
+        // Construct a value array
+        let value_data = ArrayData::builder(DataType::Int32)
+            .len(8)
+            .add_buffer(Buffer::from(&[0, 1, 2, 3, 4, 5, 6, 7].to_byte_slice()))
+            .build();
+
+        // Construct a buffer for value offsets, for the nested array:
+        //  [[0, 1, 2], [3, 4, 5], [6, 7]]
+        let value_offsets = Buffer::from(&[0, 3, 6, 8].to_byte_slice());
+
+        // Construct a list array from the above two
+        let list_data_type = DataType::List(Box::new(DataType::Int32));
+        let list_data = ArrayData::builder(list_data_type.clone())
+            .len(3)
+            .add_buffer(value_offsets.clone())
+            .add_child_data(value_data.clone())
+            .build();
+        let a = ListArray::from(list_data);
+
+        let record_batch =
+            RecordBatch::try_new(Arc::new(schema), vec![Arc::new(a)]).unwrap();
+
+        assert_eq!(3, record_batch.num_rows());
+        assert_eq!(1, record_batch.num_columns());
+        assert_eq!(
+            &DataType::List(Box::new(DataType::Int32)),
+            record_batch.schema().field(0).data_type()
+        );
+        assert_eq!(3, record_batch.column(0).data().len());
+    }
+
+    #[test]
+    fn create_record_batch_with_list_column_nulls() {
+        let schema = Schema::new(vec![Field::new(
+            "a",
+            DataType::List(Box::new(DataType::Int32)),
+            false,
+        )]);
+
+        let values_builder = PrimitiveBuilder::<Int32Type>::new(10);
+        let mut builder = ListBuilder::new(values_builder);
+
+        builder.values().append_null().unwrap();
+        builder.values().append_null().unwrap();
+        builder.append(true).unwrap();
+        builder.append(false).unwrap();
+        builder.append(true).unwrap();
+
+        // [[null, null], null, []]
+        let list_array = builder.finish();
+
+        let record_batch =
+            RecordBatch::try_new(Arc::new(schema), vec![Arc::new(list_array)]).unwrap();
+
+        assert_eq!(3, record_batch.num_rows());
+        assert_eq!(1, record_batch.num_columns());
+        assert_eq!(
+            &DataType::List(Box::new(DataType::Int32)),
+            record_batch.schema().field(0).data_type()
+        );
+        assert_eq!(3, record_batch.column(0).data().len());
+
+        assert_eq!(false, record_batch.column(0).is_null(0));
+        assert_eq!(true, record_batch.column(0).is_null(1));
+        assert_eq!(false, record_batch.column(0).is_null(2));
+
+        let col_as_list_array = record_batch
+            .column(0)
+            .as_any()
+            .downcast_ref::<ListArray>()
+            .unwrap();
+
+        assert_eq!(2, col_as_list_array.value(0).len());
+        assert_eq!(0, col_as_list_array.value(2).len());
+
+        let sublist_0_val = col_as_list_array.value(0);
+        let sublist_0 = sublist_0_val.as_any().downcast_ref::<Int32Array>().unwrap();
+
+        assert_eq!(true, sublist_0.is_null(0));
+        assert_eq!(true, sublist_0.is_null(1));
     }
 }

rust/datafusion/src/execution/physical_plan/expressions.rs

@@ -39,9 +39,9 @@ use arrow::compute;
 use arrow::compute::kernels::arithmetic::{add, divide, multiply, subtract};
 use arrow::compute::kernels::boolean::{and, or};
 use arrow::compute::kernels::cast::cast;
-use arrow::compute::kernels::comparison::{eq, gt, gt_eq, lt, lt_eq, neq};
 use arrow::compute::kernels::comparison::{
-    eq_utf8, gt_eq_utf8, gt_utf8, like_utf8, lt_eq_utf8, lt_utf8, neq_utf8, nlike_utf8,
+    eq, eq_utf8, gt, gt_eq, gt_eq_utf8, gt_utf8, like_utf8, lt, lt_eq, lt_eq_utf8,
+    lt_utf8, neq, neq_utf8, nlike_utf8,
 };
 use arrow::datatypes::{DataType, Schema, TimeUnit};
 use arrow::record_batch::RecordBatch;

rust/datafusion/src/logicalplan.rs

@@ -828,8 +828,8 @@ mod tests {
         .build()?;
 
         let expected = "Projection: #id\
-        \n  Selection: #state Eq Utf8(\"CO\")\
-        \n    TableScan: employee.csv projection=Some([0, 3])";
+                        \n  Selection: #state Eq Utf8(\"CO\")\
+                        \n    TableScan: employee.csv projection=Some([0, 3])";
 
         assert_eq!(expected, format!("{:?}", plan));
 
@@ -852,9 +852,10 @@ mod tests {
         .project(vec![col("state"), col("total_salary")])?
         .build()?;
 
-        let expected = "Projection: #state, #total_salary\
-        \n  Aggregate: groupBy=[[#state]], aggr=[[SUM(#salary) AS total_salary]]\
-        \n    TableScan: employee.csv projection=Some([3, 4])";
+        let expected =
+            "Projection: #state, #total_salary\
+             \n  Aggregate: groupBy=[[#state]], aggr=[[SUM(#salary) AS total_salary]]\
+             \n    TableScan: employee.csv projection=Some([3, 4])";
 
         assert_eq!(expected, format!("{:?}", plan));
 

rust/datafusion/src/optimizer/projection_push_down.rs

@@ -274,7 +274,7 @@ mod tests {
             .build()?;
 
         let expected = "Aggregate: groupBy=[[]], aggr=[[MAX(#0)]]\
-        \n  TableScan: test projection=Some([1])";
+                        \n  TableScan: test projection=Some([1])";
 
         assert_optimized_plan_eq(&plan, expected);
 
@@ -290,7 +290,7 @@ mod tests {
             .build()?;
 
         let expected = "Aggregate: groupBy=[[#1]], aggr=[[MAX(#0)]]\
-        \n  TableScan: test projection=Some([1, 2])";
+                        \n  TableScan: test projection=Some([1, 2])";
 
         assert_optimized_plan_eq(&plan, expected);
 
@@ -307,8 +307,8 @@ mod tests {
             .build()?;
 
         let expected = "Aggregate: groupBy=[[]], aggr=[[MAX(#0)]]\
-        \n  Selection: #1\
-        \n    TableScan: test projection=Some([1, 2])";
+                        \n  Selection: #1\
+                        \n    TableScan: test projection=Some([1, 2])";
 
         assert_optimized_plan_eq(&plan, expected);
 
@@ -327,7 +327,7 @@ mod tests {
             .build()?;
 
         let expected = "Projection: CAST(#0 AS Float64)\
-        \n  TableScan: test projection=Some([2])";
+                        \n  TableScan: test projection=Some([2])";
 
         assert_optimized_plan_eq(&projection, expected);
 
@@ -347,7 +347,7 @@ mod tests {
         assert_fields_eq(&plan, vec!["a", "b"]);
 
         let expected = "Projection: #0, #1\
-        \n  TableScan: test projection=Some([0, 1])";
+                        \n  TableScan: test projection=Some([0, 1])";
 
         assert_optimized_plan_eq(&plan, expected);
 
@@ -368,8 +368,8 @@ mod tests {
         assert_fields_eq(&plan, vec!["c", "a"]);
 
         let expected = "Limit: UInt32(5)\
-        \n  Projection: #1, #0\
-        \n    TableScan: test projection=Some([0, 2])";
+                        \n  Projection: #1, #0\
+                        \n    TableScan: test projection=Some([0, 2])";
 
         assert_optimized_plan_eq(&plan, expected);
 

rust/datafusion/src/sql/planner.rs

@@ -579,7 +579,7 @@ mod tests {
         quick_test(
             "SELECT * from person",
             "Projection: #0, #1, #2, #3, #4, #5, #6\
-            \n  TableScan: person projection=None",
+             \n  TableScan: person projection=None",
         );
     }
 

rust/datafusion/src/utils.rs

@@ -74,6 +74,22 @@ macro_rules! make_string {
     }};
 }
 
+macro_rules! make_string_from_list {
+    ($column: ident, $row: ident) => {{
+        let list = $column
+            .as_any()
+            .downcast_ref::<array::ListArray>()
+            .ok_or(ExecutionError::ExecutionError(format!(
+                "Repl error: could not convert list column to list array."
+            )))?
+            .value($row);
+        let string_values = (0..list.len())
+            .map(|i| array_value_to_string(list.clone(), i))
+            .collect::<Result<Vec<String>>>()?;
+        Ok(format!("[{}]", string_values.join(", ")))
+    }};
+}
+
 /// Get the value at the given row in an array as a string
 pub fn array_value_to_string(column: array::ArrayRef, row: usize) -> Result<String> {
     match column.data_type() {
@@ -120,6 +136,7 @@ pub fn array_value_to_string(column: array::ArrayRef, row: usize) -> Result<Stri
         DataType::Time64(unit) if *unit == TimeUnit::Nanosecond => {
             make_string!(array::Time64NanosecondArray, column, row)
         }
+        DataType::List(_) => make_string_from_list!(column, row),
         _ => Err(ExecutionError::ExecutionError(format!(
             "Unsupported {:?} type for repl.",
             column.data_type()