Added support for NullArray at FFI. (#598)

arrow-pyarrow-integration-testing/tests/test_sql.py

@@ -39,6 +39,17 @@ def tearDown(self):
         # No leak of C++ memory
         self.assertEqual(self.old_allocated_cpp, pyarrow.total_allocated_bytes())
 
+    # see https://issues.apache.org/jira/browse/ARROW-14680
+    def _test_null(self):
+        """
+        Python -> Rust -> Python
+        """
+        a = pyarrow.array([None], type=pyarrow.null())
+        b = arrow_pyarrow_integration_testing.round_trip_array(a)
+        b.validate(full=True)
+        assert a.to_pylist() == b.to_pylist()
+        assert a.type == b.type
+
     def test_primitive(self):
         a = pyarrow.array([0, None, 2, 3, 4])
         b = arrow_pyarrow_integration_testing.round_trip_array(a)

src/array/null.rs

@@ -1,6 +1,10 @@
 use crate::{bitmap::Bitmap, datatypes::DataType};
 
-use super::{ffi::ToFfi, Array};
+use crate::{
+    array::{Array, FromFfi, ToFfi},
+    error::Result,
+    ffi,
+};
 
 /// The concrete [`Array`] of [`DataType::Null`].
 #[derive(Debug, Clone)]
@@ -89,3 +93,10 @@ unsafe impl ToFfi for NullArray {
         self.clone()
     }
 }
+
+impl<A: ffi::ArrowArrayRef> FromFfi<A> for NullArray {
+    unsafe fn try_from_ffi(array: A) -> Result<Self> {
+        let data_type = array.field().data_type().clone();
+        Ok(Self::from_data(data_type, array.array().len()))
+    }
+}

src/ffi/array.rs

@@ -2,17 +2,17 @@
 
 use super::ffi::ArrowArrayRef;
 use crate::array::{BooleanArray, FromFfi};
-use crate::error::{ArrowError, Result};
+use crate::error::Result;
 use crate::{array::*, datatypes::PhysicalType};
 
 /// Reads a valid `ffi` interface into a `Box<dyn Array>`
 /// # Errors
 /// If and only if:
-/// * the data type is not supported
 /// * the interface is not valid (e.g. a null pointer)
 pub unsafe fn try_from<A: ArrowArrayRef>(array: A) -> Result<Box<dyn Array>> {
     use PhysicalType::*;
     Ok(match array.field().data_type().to_physical_type() {
+        Null => Box::new(NullArray::try_from_ffi(array)?),
         Boolean => Box::new(BooleanArray::try_from_ffi(array)?),
         Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
             Box::new(PrimitiveArray::<$T>::try_from_ffi(array)?)
@@ -33,11 +33,5 @@ pub unsafe fn try_from<A: ArrowArrayRef>(array: A) -> Result<Box<dyn Array>> {
         }
         Union => Box::new(UnionArray::try_from_ffi(array)?),
         Map => Box::new(MapArray::try_from_ffi(array)?),
-        data_type => {
-            return Err(ArrowError::NotYetImplemented(format!(
-                "Importing PhysicalType \"{:?}\" is not yet supported.",
-                data_type
-            )))
-        }
     })
 }

src/io/parquet/read/mod.rs

@@ -27,7 +27,7 @@ pub use parquet2::{
 };
 
 use crate::{
-    array::{Array, DictionaryKey, PrimitiveArray},
+    array::{Array, DictionaryKey, NullArray, PrimitiveArray},
     datatypes::{DataType, IntervalUnit, TimeUnit},
     error::{ArrowError, Result},
     io::parquet::read::nested_utils::create_list,
@@ -234,6 +234,10 @@ pub fn page_iter_to_array<I: FallibleStreamingIterator<Item = DataPage, Error =
 ) -> Result<Box<dyn Array>> {
     use DataType::*;
     match data_type.to_logical_type() {
+        Null => Ok(Box::new(NullArray::from_data(
+            data_type,
+            metadata.num_values() as usize,
+        ))),
         // INT32
         UInt8 => primitive::iter_to_array(iter, metadata, data_type, |x: i32| x as u8),
         UInt16 => primitive::iter_to_array(iter, metadata, data_type, |x: i32| x as u16),
@@ -346,6 +350,10 @@ pub async fn page_stream_to_array<I: Stream<Item = std::result::Result<DataPage,
 ) -> Result<Box<dyn Array>> {
     use DataType::*;
     match data_type.to_logical_type() {
+        Null => Ok(Box::new(NullArray::from_data(
+            data_type,
+            metadata.num_values() as usize,
+        ))),
         // INT32
         UInt8 => primitive::stream_to_array(pages, metadata, data_type, |x: i32| x as u8).await,
         UInt16 => primitive::stream_to_array(pages, metadata, data_type, |x: i32| x as u16).await,