Repository: jorgecarleitao/arrow2
Pull request #296 (2021-08-18 08:15:47)

Generalized ZipIterator to accept a `BitmapIter`
# Rational

It allows to re-use the `ZipValidity` on `MutableArrays`.

# backwards incompatible

`bitmap::utils::zip_validity` and `bitmap::utils::ZipValidity::new` now expect a `BitmapIter` in their arguments.





Patch:
diff --git a/src/array/binary/iterator.rs b/src/array/binary/iterator.rs
index d0b116f17f..7d2679798a 100644
--- a/src/array/binary/iterator.rs
+++ b/src/array/binary/iterator.rs
@@ -52,7 +52,10 @@ impl<'a, O: Offset> IntoIterator for &'a BinaryArray<O> {
 impl<'a, O: Offset> BinaryArray<O> {
     /// Returns an iterator of `Option<&[u8]>`
     pub fn iter(&'a self) -> ZipValidity<'a, &'a [u8], BinaryValueIter<'a, O>> {
-        zip_validity(BinaryValueIter::new(self), &self.validity)
+        zip_validity(
+            BinaryValueIter::new(self),
+            self.validity.as_ref().map(|x| x.iter()),
+        )
     }
 
     /// Returns an iterator of `&[u8]`
diff --git a/src/array/boolean/iterator.rs b/src/array/boolean/iterator.rs
index 8cc334b6d4..9b2d3821da 100644
--- a/src/array/boolean/iterator.rs
+++ b/src/array/boolean/iterator.rs
@@ -16,7 +16,10 @@ impl<'a> BooleanArray {
     /// constructs a new iterator
     #[inline]
     pub fn iter(&'a self) -> ZipValidity<'a, bool, BitmapIter<'a>> {
-        zip_validity(self.values().iter(), &self.validity)
+        zip_validity(
+            self.values().iter(),
+            self.validity.as_ref().map(|x| x.iter()),
+        )
     }
 
     /// Returns an iterator of `bool`
diff --git a/src/array/dictionary/iterator.rs b/src/array/dictionary/iterator.rs
index 12530cf96f..ae7d6951dd 100644
--- a/src/array/dictionary/iterator.rs
+++ b/src/array/dictionary/iterator.rs
@@ -70,7 +70,10 @@ impl<'a, K: DictionaryKey> IntoIterator for &'a DictionaryArray<K> {
 impl<'a, K: DictionaryKey> DictionaryArray<K> {
     /// Returns an iterator of `Option<Box<dyn Array>>`
     pub fn iter(&'a self) -> ZipIter<'a, K> {
-        zip_validity(DictionaryValuesIter::new(self), self.keys.validity())
+        zip_validity(
+            DictionaryValuesIter::new(self),
+            self.keys.validity().as_ref().map(|x| x.iter()),
+        )
     }
 
     /// Returns an iterator of `Box<dyn Array>`
diff --git a/src/array/fixed_size_binary/iterator.rs b/src/array/fixed_size_binary/iterator.rs
index 91781f0011..b24117fea7 100644
--- a/src/array/fixed_size_binary/iterator.rs
+++ b/src/array/fixed_size_binary/iterator.rs
@@ -50,6 +50,9 @@ impl<'a> IntoIterator for &'a FixedSizeBinaryArray {
 impl<'a> FixedSizeBinaryArray {
     /// constructs a new iterator
     pub fn iter(&'a self) -> ZipValidity<'a, &'a [u8], FixedSizeBinaryValuesIter<'a>> {
-        zip_validity(FixedSizeBinaryValuesIter::new(self), &self.validity)
+        zip_validity(
+            FixedSizeBinaryValuesIter::new(self),
+            self.validity.as_ref().map(|x| x.iter()),
+        )
     }
 }
diff --git a/src/array/fixed_size_list/iterator.rs b/src/array/fixed_size_list/iterator.rs
index 883144dbd0..031f3f15f4 100644
--- a/src/array/fixed_size_list/iterator.rs
+++ b/src/array/fixed_size_list/iterator.rs
@@ -26,7 +26,10 @@ impl<'a> IntoIterator for &'a FixedSizeListArray {
 impl<'a> FixedSizeListArray {
     /// Returns an iterator of `Option<Box<dyn Array>>`
     pub fn iter(&'a self) -> ZipIter<'a> {
-        zip_validity(ListValuesIter::new(self), &self.validity)
+        zip_validity(
+            ListValuesIter::new(self),
+            self.validity.as_ref().map(|x| x.iter()),
+        )
     }
 
     /// Returns an iterator of `Box<dyn Array>`
diff --git a/src/array/list/iterator.rs b/src/array/list/iterator.rs
index b6c184fb97..7b55ff0313 100644
--- a/src/array/list/iterator.rs
+++ b/src/array/list/iterator.rs
@@ -76,7 +76,10 @@ impl<'a, O: Offset> IntoIterator for &'a ListArray<O> {
 impl<'a, O: Offset> ListArray<O> {
     /// Returns an iterator of `Option<Box<dyn Array>>`
     pub fn iter(&'a self) -> ZipIter<'a, O> {
-        zip_validity(ListValuesIter::new(self), &self.validity)
+        zip_validity(
+            ListValuesIter::new(self),
+            self.validity.as_ref().map(|x| x.iter()),
+        )
     }
 
     /// Returns an iterator of `Box<dyn Array>`
diff --git a/src/array/primitive/iterator.rs b/src/array/primitive/iterator.rs
index 281fd99ec5..2d878baa59 100644
--- a/src/array/primitive/iterator.rs
+++ b/src/array/primitive/iterator.rs
@@ -19,6 +19,9 @@ impl<'a, T: NativeType> PrimitiveArray<T> {
     /// constructs a new iterator
     #[inline]
     pub fn iter(&'a self) -> ZipValidity<'a, &'a T, std::slice::Iter<'a, T>> {
-        zip_validity(self.values().iter(), &self.validity)
+        zip_validity(
+            self.values().iter(),
+            self.validity.as_ref().map(|x| x.iter()),
+        )
     }
 }
diff --git a/src/array/utf8/iterator.rs b/src/array/utf8/iterator.rs
index 5fe1b285f3..fd6f5d01df 100644
--- a/src/array/utf8/iterator.rs
+++ b/src/array/utf8/iterator.rs
@@ -68,7 +68,10 @@ impl<'a, O: Offset> IntoIterator for &'a Utf8Array<O> {
 impl<'a, O: Offset> Utf8Array<O> {
     /// Returns an iterator of `Option<&str>`
     pub fn iter(&'a self) -> ZipValidity<'a, &'a str, Utf8ValuesIter<'a, O>> {
-        zip_validity(Utf8ValuesIter::new(self), &self.validity)
+        zip_validity(
+            Utf8ValuesIter::new(self),
+            self.validity.as_ref().map(|x| x.iter()),
+        )
     }
 
     /// Returns an iterator of `&str`
diff --git a/src/bitmap/utils/zip_validity.rs b/src/bitmap/utils/zip_validity.rs
index 615d749ce2..b23983aff1 100644
--- a/src/bitmap/utils/zip_validity.rs
+++ b/src/bitmap/utils/zip_validity.rs
@@ -1,4 +1,4 @@
-use crate::{bitmap::Bitmap, trusted_len::TrustedLen};
+use crate::trusted_len::TrustedLen;
 
 use super::BitmapIter;
 
@@ -21,17 +21,14 @@ impl<'a, T, I: Iterator<Item = T> + Clone> Clone for ZipValidity<'a, T, I> {
 }
 
 impl<'a, T, I: Iterator<Item = T>> ZipValidity<'a, T, I> {
-    #[inline]
-    pub fn new(values: I, validity: &'a Option<Bitmap>) -> Self {
-        let validity_iter = validity
-            .as_ref()
-            .map(|x| x.iter())
-            .unwrap_or_else(|| BitmapIter::new(&[], 0, 0));
+    pub fn new(values: I, validity: Option<BitmapIter<'a>>) -> Self {
+        let has_validity = validity.as_ref().is_some();
+        let validity_iter = validity.unwrap_or_else(|| BitmapIter::new(&[], 0, 0));
 
         Self {
             values,
             validity_iter,
-            has_validity: validity.is_some(),
+            has_validity,
         }
     }
 }
@@ -75,10 +72,9 @@ impl<'a, T, I: Iterator<Item = T>> std::iter::ExactSizeIterator for ZipValidity<
 unsafe impl<T, I: TrustedLen<Item = T>> TrustedLen for ZipValidity<'_, T, I> {}
 
 /// Returns an iterator adapter that returns Option<T> according to an optional validity.
-#[inline]
 pub fn zip_validity<T, I: Iterator<Item = T>>(
     values: I,
-    validity: &Option<Bitmap>,
+    validity: Option<BitmapIter>,
 ) -> ZipValidity<T, I> {
-    ZipValidity::<T, I>::new(values, validity)
+    ZipValidity::new(values, validity)
 }
diff --git a/src/compute/cast/mod.rs b/src/compute/cast/mod.rs
index 9d09e6bc75..eca849ea48 100644
--- a/src/compute/cast/mod.rs
+++ b/src/compute/cast/mod.rs
@@ -116,18 +116,22 @@ pub fn can_cast_types(from_type: &DataType, to_type: &DataType) -> bool {
         (_, Dictionary(_, value_type)) => can_cast_types(from_type, value_type),
 
         (_, Boolean) => is_numeric(from_type),
-        (Boolean, _) => is_numeric(to_type) || to_type == &Utf8,
+        (Boolean, _) => is_numeric(to_type) || to_type == &Utf8 || to_type == &LargeUtf8,
 
         (Utf8, Date32) => true,
         (Utf8, Date64) => true,
         (Utf8, Timestamp(TimeUnit::Nanosecond, None)) => true,
         (Utf8, LargeUtf8) => true,
         (Utf8, _) => is_numeric(to_type),
+        (LargeUtf8, Date32) => true,
+        (LargeUtf8, Date64) => true,
         (LargeUtf8, Timestamp(TimeUnit::Nanosecond, None)) => true,
         (LargeUtf8, Utf8) => true,
         (LargeUtf8, _) => is_numeric(to_type),
         (_, Utf8) => is_numeric(from_type) || from_type == &Binary,
         (_, LargeUtf8) => is_numeric(from_type) || from_type == &Binary,
+        (Binary, LargeBinary) => true,
+        (LargeBinary, Binary) => true,
 
         // start numeric casts
         (UInt8, UInt16) => true,
@@ -356,6 +360,7 @@ fn cast_with_options(
 
     let as_options = options.with_wrapped(true);
     match (from_type, to_type) {
+        (Null, Int32) => Ok(new_null_array(to_type.clone(), array.len())),
         (Struct(_), _) => Err(ArrowError::NotYetImplemented(
             "Cannot cast from struct to other types".to_string(),
         )),
@@ -375,7 +380,7 @@ fn cast_with_options(
             to_type,
         ))
         .map(|x| Box::new(x) as Box<dyn Array>),
-        (List(lhs), LargeList(rhs)) if lhs == rhs => Ok(cast_large_to_list(
+        (LargeList(lhs), List(rhs)) if lhs == rhs => Ok(cast_large_to_list(
             array.as_any().downcast_ref().unwrap(),
             to_type,
         ))
@@ -388,8 +393,7 @@ fn cast_with_options(
             let offsets =
                 unsafe { Buffer::from_trusted_len_iter_unchecked(0..=array.len() as i32) };
 
-            let data_type = ListArray::<i32>::default_datatype(to.data_type().clone());
-            let list_array = ListArray::<i32>::from_data(data_type, offsets, values, None);
+            let list_array = ListArray::<i32>::from_data(to_type.clone(), offsets, values, None);
 
             Ok(Box::new(list_array))
         }
@@ -750,646 +754,3 @@ fn cast_to_dictionary<K: DictionaryKey>(
         ))),
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::types::NativeType;
-
-    #[test]
-    fn i32_to_f64() {
-        let array = Int32Array::from_slice(&[5, 6, 7, 8, 9]);
-        let b = cast(&array, &DataType::Float64).unwrap();
-        let c = b.as_any().downcast_ref::<Float64Array>().unwrap();
-        assert!((5.0 - c.value(0)).abs() < f64::EPSILON);
-        assert!((6.0 - c.value(1)).abs() < f64::EPSILON);
-        assert!((7.0 - c.value(2)).abs() < f64::EPSILON);
-        assert!((8.0 - c.value(3)).abs() < f64::EPSILON);
-        assert!((9.0 - c.value(4)).abs() < f64::EPSILON);
-    }
-
-    #[test]
-    fn i32_as_f64_no_overflow() {
-        let array = Int32Array::from_slice(&[5, 6, 7, 8, 9]);
-        let b = wrapping_cast(&array, &DataType::Float64).unwrap();
-        let c = b.as_any().downcast_ref::<Float64Array>().unwrap();
-        assert!((5.0 - c.value(0)).abs() < f64::EPSILON);
-        assert!((6.0 - c.value(1)).abs() < f64::EPSILON);
-        assert!((7.0 - c.value(2)).abs() < f64::EPSILON);
-        assert!((8.0 - c.value(3)).abs() < f64::EPSILON);
-        assert!((9.0 - c.value(4)).abs() < f64::EPSILON);
-    }
-
-    #[test]
-    fn u16_as_u8_overflow() {
-        let array = UInt16Array::from_slice(&[255, 256, 257, 258, 259]);
-        let b = wrapping_cast(&array, &DataType::UInt8).unwrap();
-        let c = b.as_any().downcast_ref::<UInt8Array>().unwrap();
-        let values = c.values().as_slice();
-
-        println!("{}", 255u8.wrapping_add(10));
-
-        assert_eq!(values, &[255, 0, 1, 2, 3])
-    }
-
-    #[test]
-    fn u16_as_u8_no_overflow() {
-        let array = UInt16Array::from_slice(&[1, 2, 3, 4, 5]);
-        let b = wrapping_cast(&array, &DataType::UInt8).unwrap();
-        let c = b.as_any().downcast_ref::<UInt8Array>().unwrap();
-        let values = c.values().as_slice();
-        assert_eq!(values, &[1, 2, 3, 4, 5])
-    }
-
-    #[test]
-    fn float_range_max() {
-        //floats to integers
-        let u: Option<u32> = num::cast(f32::MAX);
-        assert_eq!(u, None);
-        let u: Option<u64> = num::cast(f32::MAX);
-        assert_eq!(u, None);
-
-        let u: Option<i32> = num::cast(f32::MAX);
-        assert_eq!(u, None);
-        let u: Option<i64> = num::cast(f32::MAX);
-        assert_eq!(u, None);
-
-        let u: Option<u32> = num::cast(f64::MAX);
-        assert_eq!(u, None);
-        let u: Option<u64> = num::cast(f64::MAX);
-        assert_eq!(u, None);
-
-        let u: Option<i32> = num::cast(f64::MAX);
-        assert_eq!(u, None);
-        let u: Option<i64> = num::cast(f64::MAX);
-        assert_eq!(u, None);
-
-        //integers to floats
-        let u: Option<f32> = num::cast(u32::MAX);
-        assert!(u.is_some());
-        let u: Option<f64> = num::cast(u32::MAX);
-        assert!(u.is_some());
-
-        let u: Option<f32> = num::cast(i32::MAX);
-        assert!(u.is_some());
-        let u: Option<f64> = num::cast(i32::MAX);
-        assert!(u.is_some());
-
-        let u: Option<f64> = num::cast(i64::MAX);
-        assert!(u.is_some());
-        let u: Option<f64> = num::cast(u64::MAX);
-        assert!(u.is_some());
-
-        let u: Option<f64> = num::cast(f32::MAX);
-        assert!(u.is_some());
-    }
-
-    #[test]
-    fn float_range_min() {
-        //floats to integers
-        let u: Option<u32> = num::cast(f32::MIN);
-        assert_eq!(u, None);
-        let u: Option<u64> = num::cast(f32::MIN);
-        assert_eq!(u, None);
-
-        let u: Option<i32> = num::cast(f32::MIN);
-        assert_eq!(u, None);
-        let u: Option<i64> = num::cast(f32::MIN);
-        assert_eq!(u, None);
-
-        let u: Option<u32> = num::cast(f64::MIN);
-        assert_eq!(u, None);
-        let u: Option<u64> = num::cast(f64::MIN);
-        assert_eq!(u, None);
-
-        let u: Option<i32> = num::cast(f64::MIN);
-        assert_eq!(u, None);
-        let u: Option<i64> = num::cast(f64::MIN);
-        assert_eq!(u, None);
-
-        //integers to floats
-        let u: Option<f32> = num::cast(u32::MIN);
-        assert!(u.is_some());
-        let u: Option<f64> = num::cast(u32::MIN);
-        assert!(u.is_some());
-
-        let u: Option<f32> = num::cast(i32::MIN);
-        assert!(u.is_some());
-        let u: Option<f64> = num::cast(i32::MIN);
-        assert!(u.is_some());
-
-        let u: Option<f64> = num::cast(i64::MIN);
-        assert!(u.is_some());
-        let u: Option<f64> = num::cast(u64::MIN);
-        assert!(u.is_some());
-
-        let u: Option<f64> = num::cast(f32::MIN);
-        assert!(u.is_some());
-    }
-
-    #[test]
-    fn f32_as_u8_overflow() {
-        let array = Float32Array::from_slice(&[1.1, 5000.0]);
-        let b = cast(&array, &DataType::UInt8).unwrap();
-        let c = b.as_any().downcast_ref::<UInt8Array>().unwrap();
-        let expected = UInt8Array::from(&[Some(1), None]);
-        assert_eq!(c, &expected);
-
-        let b = cast_with_options(&array, &DataType::UInt8, CastOptions { wrapped: true }).unwrap();
-        let c = b.as_any().downcast_ref::<UInt8Array>().unwrap();
-        let expected = UInt8Array::from(&[Some(1), Some(255)]);
-        assert_eq!(c, &expected);
-    }
-
-    #[test]
-    fn i32_to_u8() {
-        let array = Int32Array::from_slice(&[-5, 6, -7, 8, 100000000]);
-        let b = cast(&array, &DataType::UInt8).unwrap();
-        let expected = UInt8Array::from(&[None, Some(6), None, Some(8), None]);
-        let c = b.as_any().downcast_ref::<UInt8Array>().unwrap();
-        assert_eq!(c, &expected);
-    }
-
-    #[test]
-    fn i32_to_u8_sliced() {
-        let array = Int32Array::from_slice(&[-5, 6, -7, 8, 100000000]);
-        let array = array.slice(2, 3);
-        let b = cast(&array, &DataType::UInt8).unwrap();
-        let expected = UInt8Array::from(&[None, Some(8), None]);
-        let c = b.as_any().downcast_ref::<UInt8Array>().unwrap();
-        assert_eq!(c, &expected);
-    }
-
-    #[test]
-    fn i32_to_i32() {
-        let array = Int32Array::from_slice(&[5, 6, 7, 8, 9]);
-        let b = cast(&array, &DataType::Int32).unwrap();
-        let c = b.as_any().downcast_ref::<Int32Array>().unwrap();
-
-        let expected = &[5, 6, 7, 8, 9];
-        let expected = Int32Array::from_slice(expected);
-        assert_eq!(c, &expected);
-    }
-
-    #[test]
-    fn i32_to_list_i32() {
-        let array = Int32Array::from_slice(&[5, 6, 7, 8, 9]);
-        let b = cast(
-            &array,
-            &DataType::List(Box::new(Field::new("item", DataType::Int32, true))),
-        )
-        .unwrap();
-
-        let arr = b.as_any().downcast_ref::<ListArray<i32>>().unwrap();
-        assert_eq!(&[0, 1, 2, 3, 4, 5], arr.offsets().as_slice());
-        let values = arr.values();
-        let c = values
-            .as_any()
-            .downcast_ref::<PrimitiveArray<i32>>()
-            .unwrap();
-
-        let expected = Int32Array::from_slice(&[5, 6, 7, 8, 9]);
-        assert_eq!(c, &expected);
-    }
-
-    #[test]
-    fn i32_to_list_i32_nullable() {
-        let input = [Some(5), None, Some(7), Some(8), Some(9)];
-
-        let array = Int32Array::from(input);
-        let b = cast(
-            &array,
-            &DataType::List(Box::new(Field::new("item", DataType::Int32, true))),
-        )
-        .unwrap();
-
-        let arr = b.as_any().downcast_ref::<ListArray<i32>>().unwrap();
-        assert_eq!(&[0, 1, 2, 3, 4, 5], arr.offsets().as_slice());
-        let values = arr.values();
-        let c = values.as_any().downcast_ref::<Int32Array>().unwrap();
-
-        let expected = &[Some(5), None, Some(7), Some(8), Some(9)];
-        let expected = Int32Array::from(expected);
-        assert_eq!(c, &expected);
-    }
-
-    #[test]
-    fn i32_to_list_f64_nullable_sliced() {
-        let input = [Some(5), None, Some(7), Some(8), None, Some(10)];
-
-        let array = Int32Array::from(input);
-
-        let array = array.slice(2, 4);
-        let b = cast(
-            &array,
-            &DataType::List(Box::new(Field::new("item", DataType::Float64, true))),
-        )
-        .unwrap();
-
-        let arr = b.as_any().downcast_ref::<ListArray<i32>>().unwrap();
-        assert_eq!(&[0, 1, 2, 3, 4], arr.offsets().as_slice());
-        let values = arr.values();
-        let c = values.as_any().downcast_ref::<Float64Array>().unwrap();
-
-        let expected = &[Some(7.0), Some(8.0), None, Some(10.0)];
-        let expected = Float64Array::from(expected);
-        assert_eq!(c, &expected);
-    }
-
-    #[test]
-    fn utf8_to_i32() {
-        let array = Utf8Array::<i32>::from_slice(&["5", "6", "seven", "8", "9.1"]);
-        let b = cast(&array, &DataType::Int32).unwrap();
-        let c = b.as_any().downcast_ref::<PrimitiveArray<i32>>().unwrap();
-
-        let expected = &[Some(5), Some(6), None, Some(8), None];
-        let expected = Int32Array::from(expected);
-        assert_eq!(c, &expected);
-    }
-
-    #[test]
-    fn bool_to_i32() {
-        let array = BooleanArray::from(vec![Some(true), Some(false), None]);
-        let b = cast(&array, &DataType::Int32).unwrap();
-        let c = b.as_any().downcast_ref::<Int32Array>().unwrap();
-
-        let expected = &[Some(1), Some(0), None];
-        let expected = Int32Array::from(expected);
-        assert_eq!(c, &expected);
-    }
-
-    #[test]
-    fn bool_to_f64() {
-        let array = BooleanArray::from(vec![Some(true), Some(false), None]);
-        let b = cast(&array, &DataType::Float64).unwrap();
-        let c = b.as_any().downcast_ref::<Float64Array>().unwrap();
-
-        let expected = &[Some(1.0), Some(0.0), None];
-        let expected = Float64Array::from(expected);
-        assert_eq!(c, &expected);
-    }
-
-    #[test]
-    #[should_panic(expected = "Casting from Int32 to Timestamp(Microsecond, None) not supported")]
-    fn int32_to_timestamp() {
-        let array = Int32Array::from(&[Some(2), Some(10), None]);
-        cast(&array, &DataType::Timestamp(TimeUnit::Microsecond, None)).unwrap();
-    }
-
-    #[test]
-    fn consistency() {
-        use crate::datatypes::DataType::*;
-        let datatypes = vec![
-            Null,
-            Boolean,
-            UInt8,
-            UInt16,
-            UInt32,
-            UInt64,
-            Int8,
-            Int16,
-            Int32,
-            Int64,
-            Float32,
-            Float64,
-            Timestamp(TimeUnit::Second, None),
-            Timestamp(TimeUnit::Millisecond, None),
-            Timestamp(TimeUnit::Microsecond, None),
-            Timestamp(TimeUnit::Nanosecond, None),
-            Time64(TimeUnit::Microsecond),
-            Time64(TimeUnit::Nanosecond),
-            Date32,
-            Time32(TimeUnit::Second),
-            Time32(TimeUnit::Millisecond),
-            Date64,
-            Utf8,
-            LargeUtf8,
-            Binary,
-            LargeBinary,
-            Duration(TimeUnit::Second),
-            Duration(TimeUnit::Millisecond),
-            Duration(TimeUnit::Microsecond),
-            Duration(TimeUnit::Nanosecond),
-            List(Box::new(Field::new("a", Utf8, true))),
-            LargeList(Box::new(Field::new("a", Utf8, true))),
-        ];
-        datatypes
-            .clone()
-            .into_iter()
-            .zip(datatypes.into_iter())
-            .for_each(|(d1, d2)| {
-                let array = new_null_array(d1.clone(), 10);
-                if can_cast_types(&d1, &d2) {
-                    let result = cast(array.as_ref(), &d2);
-                    if let Ok(result) = result {
-                        assert_eq!(result.data_type(), &d2);
-                    } else {
-                        panic!("Cast should have not failed")
-                    }
-                } else {
-                    assert!(
-                        cast_with_options(array.as_ref(), &d2, CastOptions::default()).is_err()
-                    );
-                }
-            });
-    }
-
-    fn test_primitive_to_primitive<I: NativeType, O: NativeType>(
-        lhs: &[I],
-        lhs_type: DataType,
-        expected: &[O],
-        expected_type: DataType,
-    ) {
-        let a = PrimitiveArray::<I>::from_slice(lhs).to(lhs_type);
-        let b = cast(&a, &expected_type).unwrap();
-        let b = b.as_any().downcast_ref::<PrimitiveArray<O>>().unwrap();
-        let expected = PrimitiveArray::<O>::from_slice(expected).to(expected_type);
-        assert_eq!(b, &expected);
-    }
-
-    #[test]
-    fn date32_to_date64() {
-        test_primitive_to_primitive(
-            &[10000i32, 17890],
-            DataType::Date32,
-            &[864000000000i64, 1545696000000],
-            DataType::Date64,
-        );
-    }
-
-    #[test]
-    fn date64_to_date32() {
-        test_primitive_to_primitive(
-            &[864000000005i64, 1545696000001],
-            DataType::Date64,
-            &[10000i32, 17890],
-            DataType::Date32,
-        );
-    }
-
-    #[test]
-    fn date32_to_int32() {
-        test_primitive_to_primitive(
-            &[10000i32, 17890],
-            DataType::Date32,
-            &[10000i32, 17890],
-            DataType::Int32,
-        );
-    }
-
-    #[test]
-    fn int32_to_date32() {
-        test_primitive_to_primitive(
-            &[10000i32, 17890],
-            DataType::Int32,
-            &[10000i32, 17890],
-            DataType::Date32,
-        );
-    }
-
-    #[test]
-    fn timestamp_to_date32() {
-        test_primitive_to_primitive(
-            &[864000000005i64, 1545696000001],
-            DataType::Timestamp(TimeUnit::Millisecond, Some(String::from("UTC"))),
-            &[10000i32, 17890],
-            DataType::Date32,
-        );
-    }
-
-    #[test]
-    fn timestamp_to_date64() {
-        test_primitive_to_primitive(
-            &[864000000005i64, 1545696000001],
-            DataType::Timestamp(TimeUnit::Millisecond, Some(String::from("UTC"))),
-            &[864000000005i64, 1545696000001i64],
-            DataType::Date64,
-        );
-    }
-
-    #[test]
-    fn timestamp_to_i64() {
-        test_primitive_to_primitive(
-            &[864000000005i64, 1545696000001],
-            DataType::Timestamp(TimeUnit::Millisecond, Some(String::from("UTC"))),
-            &[864000000005i64, 1545696000001i64],
-            DataType::Int64,
-        );
-    }
-
-    #[test]
-    fn timestamp_to_timestamp() {
-        test_primitive_to_primitive(
-            &[864000003005i64, 1545696002001],
-            DataType::Timestamp(TimeUnit::Millisecond, None),
-            &[864000003i64, 1545696002],
-            DataType::Timestamp(TimeUnit::Second, None),
-        );
-    }
-
-    #[test]
-    fn utf8_to_dict() {
-        let array = Utf8Array::<i32>::from(&[Some("one"), None, Some("three"), Some("one")]);
-
-        // Cast to a dictionary (same value type, Utf8)
-        let cast_type = DataType::Dictionary(Box::new(DataType::UInt8), Box::new(DataType::Utf8));
-        let result = cast(&array, &cast_type).expect("cast failed");
-
-        let mut expected = MutableDictionaryArray::<u8, MutableUtf8Array<i32>>::new();
-        expected
-            .try_extend([Some("one"), None, Some("three"), Some("one")])
-            .unwrap();
-        let expected: DictionaryArray<u8> = expected.into();
-        assert_eq!(expected, result.as_ref());
-    }
-
-    #[test]
-    fn dict_to_utf8() {
-        let mut array = MutableDictionaryArray::<u8, MutableUtf8Array<i32>>::new();
-        array
-            .try_extend([Some("one"), None, Some("three"), Some("one")])
-            .unwrap();
-        let array: DictionaryArray<u8> = array.into();
-
-        let result = cast(&array, &DataType::Utf8).expect("cast failed");
-
-        let expected = Utf8Array::<i32>::from(&[Some("one"), None, Some("three"), Some("one")]);
-
-        assert_eq!(expected, result.as_ref());
-    }
-
-    #[test]
-    fn i32_to_dict() {
-        let array = Int32Array::from(&[Some(1), None, Some(3), Some(1)]);
-
-        // Cast to a dictionary (same value type, Utf8)
-        let cast_type = DataType::Dictionary(Box::new(DataType::UInt8), Box::new(DataType::Int32));
-        let result = cast(&array, &cast_type).expect("cast failed");
-
-        let mut expected = MutableDictionaryArray::<u8, MutablePrimitiveArray<i32>>::new();
-        expected
-            .try_extend([Some(1), None, Some(3), Some(1)])
-            .unwrap();
-        let expected: DictionaryArray<u8> = expected.into();
-        assert_eq!(expected, result.as_ref());
-    }
-
-    #[test]
-    fn list_to_list() {
-        let data = vec![
-            Some(vec![Some(1i32), Some(2), Some(3)]),
-            None,
-            Some(vec![Some(4), None, Some(6)]),
-        ];
-
-        let expected_data = data
-            .iter()
-            .map(|x| x.as_ref().map(|x| x.iter().map(|x| x.map(|x| x as u16))));
-
-        let mut array = MutableListArray::<i32, MutablePrimitiveArray<i32>>::new();
-        array.try_extend(data.clone()).unwrap();
-        let array: ListArray<i32> = array.into();
-
-        let mut expected = MutableListArray::<i32, MutablePrimitiveArray<u16>>::new();
-        expected.try_extend(expected_data).unwrap();
-        let expected: ListArray<i32> = expected.into();
-
-        let result = cast(&array, expected.data_type()).unwrap();
-        assert_eq!(expected, result.as_ref());
-    }
-
-    /*
-    #[test]
-    fn dict_to_dict_bad_index_value_primitive() {
-        use DataType::*;
-        // test converting from an array that has indexes of a type
-        // that are out of bounds for a particular other kind of
-        // index.
-
-        let keys_builder = PrimitiveBuilder::<i32>::new(10);
-        let values_builder = PrimitiveBuilder::<i64>::new(10);
-        let mut builder = PrimitiveDictionaryBuilder::new(keys_builder, values_builder);
-
-        // add 200 distinct values (which can be stored by a
-        // dictionary indexed by int32, but not a dictionary indexed
-        // with int8)
-        for i in 0..200 {
-            builder.append(i).unwrap();
-        }
-        let array: ArrayRef = Arc::new(builder.finish());
-
-        let cast_type = Dictionary(Box::new(Int8), Box::new(Utf8));
-        let res = cast_with_options(&array, &cast_type);
-        assert, CastOptions::default())!(res.is_err());
-        let actual_error = format!("{:?}", res);
-        let expected_error = "Could not convert 72 dictionary indexes from Int32 to Int8";
-        assert!(
-            actual_error.contains(expected_error),
-            "did not find expected error '{}' in actual error '{}'",
-            actual_error,
-            expected_error
-        );
-    }
-
-    #[test]
-    fn dict_to_dict_bad_index_value_utf8() {
-        use DataType::*;
-        // Same test as dict_to_dict_bad_index_value but use
-        // string values (and encode the expected behavior here);
-
-        let keys_builder = PrimitiveBuilder::<i32>::new(10);
-        let values_builder = StringBuilder::new(10);
-        let mut builder = StringDictionaryBuilder::new(keys_builder, values_builder);
-
-        // add 200 distinct values (which can be stored by a
-        // dictionary indexed by int32, but not a dictionary indexed
-        // with int8)
-        for i in 0..200 {
-            let val = format!("val{}", i);
-            builder.append(&val).unwrap();
-        }
-        let array: ArrayRef = Arc::new(builder.finish());
-
-        let cast_type = Dictionary(Box::new(Int8), Box::new(Utf8));
-        let res = cast_with_options(&array, &cast_type);
-        assert, CastOptions::default())!(res.is_err());
-        let actual_error = format!("{:?}", res);
-        let expected_error = "Could not convert 72 dictionary indexes from Int32 to Int8";
-        assert!(
-            actual_error.contains(expected_error),
-            "did not find expected error '{}' in actual error '{}'",
-            actual_error,
-            expected_error
-        );
-    }
-
-    #[test]
-    fn utf8_to_date32() {
-        use chrono::NaiveDate;
-        let from_ymd = chrono::NaiveDate::from_ymd;
-        let since = chrono::NaiveDate::signed_duration_since;
-
-        let a = StringArray::from(vec![
-            "2000-01-01",          // valid date with leading 0s
-            "2000-2-2",            // valid date without leading 0s
-            "2000-00-00",          // invalid month and day
-            "2000-01-01T12:00:00", // date + time is invalid
-            "2000",                // just a year is invalid
-        ]);
-        let array = Arc::new(a) as ArrayRef;
-        let b = cast_with_options(&array, &DataType::Date32, CastOptions::default()).unwrap();
-        let c = b.as_any().downcast_ref::<Date32Array>().unwrap();
-
-        // test valid inputs
-        let date_value = since(NaiveDate::from_ymd(2000, 1, 1), from_ymd(1970, 1, 1))
-            .num_days() as i32;
-        assert_eq!(true, c.is_valid(0)); // "2000-01-01"
-        assert_eq!(date_value, c.value(0));
-
-        let date_value = since(NaiveDate::from_ymd(2000, 2, 2), from_ymd(1970, 1, 1))
-            .num_days() as i32;
-        assert_eq!(true, c.is_valid(1)); // "2000-2-2"
-        assert_eq!(date_value, c.value(1));
-
-        // test invalid inputs
-        assert_eq!(false, c.is_valid(2)); // "2000-00-00"
-        assert_eq!(false, c.is_valid(3)); // "2000-01-01T12:00:00"
-        assert_eq!(false, c.is_valid(4)); // "2000"
-    }
-
-    #[test]
-    fn utf8_to_date64() {
-        let a = StringArray::from(vec![
-            "2000-01-01T12:00:00", // date + time valid
-            "2020-12-15T12:34:56", // date + time valid
-            "2020-2-2T12:34:56",   // valid date time without leading 0s
-            "2000-00-00T12:00:00", // invalid month and day
-            "2000-01-01 12:00:00", // missing the 'T'
-            "2000-01-01",          // just a date is invalid
-        ]);
-        let array = Arc::new(a) as ArrayRef;
-        let b = cast_with_options(&array, &DataType::Date64, CastOptions::default()).unwrap();
-        let c = b.as_any().downcast_ref::<Date64Array>().unwrap();
-
-        // test valid inputs
-        assert_eq!(true, c.is_valid(0)); // "2000-01-01T12:00:00"
-        assert_eq!(946728000000, c.value(0));
-        assert_eq!(true, c.is_valid(1)); // "2020-12-15T12:34:56"
-        assert_eq!(1608035696000, c.value(1));
-        assert_eq!(true, c.is_valid(2)); // "2020-2-2T12:34:56"
-        assert_eq!(1580646896000, c.value(2));
-
-        // test invalid inputs
-        assert_eq!(false, c.is_valid(3)); // "2000-00-00T12:00:00"
-        assert_eq!(false, c.is_valid(4)); // "2000-01-01 12:00:00"
-        assert_eq!(false, c.is_valid(5)); // "2000-01-01"
-    }
-
-    fn make_union_array() -> UnionArray {
-        let mut builder = UnionBuilder::new_dense(7);
-        builder.append::<i32>("a", 1).unwrap();
-        builder.append::<i64>("b", 2).unwrap();
-        builder.build().unwrap()
-    }
-    */
-}


Test patch:
diff --git a/tests/it/bitmap/utils/zip_validity.rs b/tests/it/bitmap/utils/zip_validity.rs
index b49ec62381..d778fb001b 100644
--- a/tests/it/bitmap/utils/zip_validity.rs
+++ b/tests/it/bitmap/utils/zip_validity.rs
@@ -2,9 +2,10 @@ use arrow2::bitmap::{utils::zip_validity, Bitmap};
 
 #[test]
 fn basic() {
-    let a = Some(Bitmap::from([true, false]));
+    let a = Bitmap::from([true, false]);
+    let a = Some(a.iter());
     let values = vec![0, 1];
-    let zip = zip_validity(values.into_iter(), &a);
+    let zip = zip_validity(values.into_iter(), a);
 
     let a = zip.collect::<Vec<_>>();
     assert_eq!(a, vec![Some(0), None]);
@@ -12,11 +13,10 @@ fn basic() {
 
 #[test]
 fn complete() {
-    let a = Some(Bitmap::from([
-        true, false, true, false, true, false, true, false,
-    ]));
+    let a = Bitmap::from([true, false, true, false, true, false, true, false]);
+    let a = Some(a.iter());
     let values = vec![0, 1, 2, 3, 4, 5, 6, 7];
-    let zip = zip_validity(values.into_iter(), &a);
+    let zip = zip_validity(values.into_iter(), a);
 
     let a = zip.collect::<Vec<_>>();
     assert_eq!(
@@ -27,7 +27,8 @@ fn complete() {
 
 #[test]
 fn slices() {
-    let a = Some(Bitmap::from([true, false]));
+    let a = Bitmap::from([true, false]);
+    let a = Some(a.iter());
     let offsets = vec![0, 2, 3];
     let values = vec![1, 2, 3];
     let iter = offsets.windows(2).map(|x| {
@@ -35,7 +36,7 @@ fn slices() {
         let end = x[1];
         &values[start..end]
     });
-    let zip = zip_validity(iter, &a);
+    let zip = zip_validity(iter, a);
 
     let a = zip.collect::<Vec<_>>();
     assert_eq!(a, vec![Some([1, 2].as_ref()), None]);
@@ -43,11 +44,10 @@ fn slices() {
 
 #[test]
 fn byte() {
-    let a = Some(Bitmap::from([
-        true, false, true, false, false, true, true, false, true,
-    ]));
+    let a = Bitmap::from([true, false, true, false, false, true, true, false, true]);
+    let a = Some(a.iter());
     let values = vec![0, 1, 2, 3, 4, 5, 6, 7, 8];
-    let zip = zip_validity(values.into_iter(), &a);
+    let zip = zip_validity(values.into_iter(), a);
 
     let a = zip.collect::<Vec<_>>();
     assert_eq!(
@@ -68,10 +68,10 @@ fn byte() {
 
 #[test]
 fn offset() {
-    let a = Bitmap::from([true, false, true, false, false, true, true, false, true]);
-    let a = Some(a.slice(1, 8));
+    let a = Bitmap::from([true, false, true, false, false, true, true, false, true]).slice(1, 8);
+    let a = Some(a.iter());
     let values = vec![0, 1, 2, 3, 4, 5, 6, 7];
-    let zip = zip_validity(values.into_iter(), &a);
+    let zip = zip_validity(values.into_iter(), a);
 
     let a = zip.collect::<Vec<_>>();
     assert_eq!(
@@ -83,7 +83,7 @@ fn offset() {
 #[test]
 fn none() {
     let values = vec![0, 1, 2];
-    let zip = zip_validity(values.into_iter(), &None);
+    let zip = zip_validity(values.into_iter(), None);
 
     let a = zip.collect::<Vec<_>>();
     assert_eq!(a, vec![Some(0), Some(1), Some(2)]);
@@ -91,10 +91,10 @@ fn none() {
 
 #[test]
 fn rev() {
-    let a = Bitmap::from([true, false, true, false, false, true, true, false, true]);
-    let a = Some(a.slice(1, 8));
+    let a = Bitmap::from([true, false, true, false, false, true, true, false, true]).slice(1, 8);
+    let a = Some(a.iter());
     let values = vec![0, 1, 2, 3, 4, 5, 6, 7];
-    let zip = zip_validity(values.into_iter(), &a);
+    let zip = zip_validity(values.into_iter(), a);
 
     let result = zip.rev().collect::<Vec<_>>();
     let expected = vec![None, Some(1), None, None, Some(4), Some(5), None, Some(7)]
diff --git a/tests/it/compute/cast.rs b/tests/it/compute/cast.rs
new file mode 100644
index 0000000000..8abb8afc29
--- /dev/null
+++ b/tests/it/compute/cast.rs
@@ -0,0 +1,635 @@
+use arrow2::array::*;
+use arrow2::compute::cast::{can_cast_types, cast, wrapping_cast};
+use arrow2::datatypes::*;
+use arrow2::types::NativeType;
+
+#[test]
+fn i32_to_f64() {
+    let array = Int32Array::from_slice(&[5, 6, 7, 8, 9]);
+    let b = cast(&array, &DataType::Float64).unwrap();
+    let c = b.as_any().downcast_ref::<Float64Array>().unwrap();
+    assert!((5.0 - c.value(0)).abs() < f64::EPSILON);
+    assert!((6.0 - c.value(1)).abs() < f64::EPSILON);
+    assert!((7.0 - c.value(2)).abs() < f64::EPSILON);
+    assert!((8.0 - c.value(3)).abs() < f64::EPSILON);
+    assert!((9.0 - c.value(4)).abs() < f64::EPSILON);
+}
+
+#[test]
+fn i32_as_f64_no_overflow() {
+    let array = Int32Array::from_slice(&[5, 6, 7, 8, 9]);
+    let b = wrapping_cast(&array, &DataType::Float64).unwrap();
+    let c = b.as_any().downcast_ref::<Float64Array>().unwrap();
+    assert!((5.0 - c.value(0)).abs() < f64::EPSILON);
+    assert!((6.0 - c.value(1)).abs() < f64::EPSILON);
+    assert!((7.0 - c.value(2)).abs() < f64::EPSILON);
+    assert!((8.0 - c.value(3)).abs() < f64::EPSILON);
+    assert!((9.0 - c.value(4)).abs() < f64::EPSILON);
+}
+
+#[test]
+fn u16_as_u8_overflow() {
+    let array = UInt16Array::from_slice(&[255, 256, 257, 258, 259]);
+    let b = wrapping_cast(&array, &DataType::UInt8).unwrap();
+    let c = b.as_any().downcast_ref::<UInt8Array>().unwrap();
+    let values = c.values().as_slice();
+
+    println!("{}", 255u8.wrapping_add(10));
+
+    assert_eq!(values, &[255, 0, 1, 2, 3])
+}
+
+#[test]
+fn u16_as_u8_no_overflow() {
+    let array = UInt16Array::from_slice(&[1, 2, 3, 4, 5]);
+    let b = wrapping_cast(&array, &DataType::UInt8).unwrap();
+    let c = b.as_any().downcast_ref::<UInt8Array>().unwrap();
+    let values = c.values().as_slice();
+    assert_eq!(values, &[1, 2, 3, 4, 5])
+}
+
+#[test]
+fn float_range_max() {
+    //floats to integers
+    let u: Option<u32> = num::cast(f32::MAX);
+    assert_eq!(u, None);
+    let u: Option<u64> = num::cast(f32::MAX);
+    assert_eq!(u, None);
+
+    let u: Option<i32> = num::cast(f32::MAX);
+    assert_eq!(u, None);
+    let u: Option<i64> = num::cast(f32::MAX);
+    assert_eq!(u, None);
+
+    let u: Option<u32> = num::cast(f64::MAX);
+    assert_eq!(u, None);
+    let u: Option<u64> = num::cast(f64::MAX);
+    assert_eq!(u, None);
+
+    let u: Option<i32> = num::cast(f64::MAX);
+    assert_eq!(u, None);
+    let u: Option<i64> = num::cast(f64::MAX);
+    assert_eq!(u, None);
+
+    //integers to floats
+    let u: Option<f32> = num::cast(u32::MAX);
+    assert!(u.is_some());
+    let u: Option<f64> = num::cast(u32::MAX);
+    assert!(u.is_some());
+
+    let u: Option<f32> = num::cast(i32::MAX);
+    assert!(u.is_some());
+    let u: Option<f64> = num::cast(i32::MAX);
+    assert!(u.is_some());
+
+    let u: Option<f64> = num::cast(i64::MAX);
+    assert!(u.is_some());
+    let u: Option<f64> = num::cast(u64::MAX);
+    assert!(u.is_some());
+
+    let u: Option<f64> = num::cast(f32::MAX);
+    assert!(u.is_some());
+}
+
+#[test]
+fn float_range_min() {
+    //floats to integers
+    let u: Option<u32> = num::cast(f32::MIN);
+    assert_eq!(u, None);
+    let u: Option<u64> = num::cast(f32::MIN);
+    assert_eq!(u, None);
+
+    let u: Option<i32> = num::cast(f32::MIN);
+    assert_eq!(u, None);
+    let u: Option<i64> = num::cast(f32::MIN);
+    assert_eq!(u, None);
+
+    let u: Option<u32> = num::cast(f64::MIN);
+    assert_eq!(u, None);
+    let u: Option<u64> = num::cast(f64::MIN);
+    assert_eq!(u, None);
+
+    let u: Option<i32> = num::cast(f64::MIN);
+    assert_eq!(u, None);
+    let u: Option<i64> = num::cast(f64::MIN);
+    assert_eq!(u, None);
+
+    //integers to floats
+    let u: Option<f32> = num::cast(u32::MIN);
+    assert!(u.is_some());
+    let u: Option<f64> = num::cast(u32::MIN);
+    assert!(u.is_some());
+
+    let u: Option<f32> = num::cast(i32::MIN);
+    assert!(u.is_some());
+    let u: Option<f64> = num::cast(i32::MIN);
+    assert!(u.is_some());
+
+    let u: Option<f64> = num::cast(i64::MIN);
+    assert!(u.is_some());
+    let u: Option<f64> = num::cast(u64::MIN);
+    assert!(u.is_some());
+
+    let u: Option<f64> = num::cast(f32::MIN);
+    assert!(u.is_some());
+}
+
+#[test]
+fn f32_as_u8_overflow() {
+    let array = Float32Array::from_slice(&[1.1, 5000.0]);
+    let b = cast(&array, &DataType::UInt8).unwrap();
+    let expected = UInt8Array::from(&[Some(1), None]);
+    assert_eq!(expected, b.as_ref());
+
+    let b = wrapping_cast(&array, &DataType::UInt8).unwrap();
+    let expected = UInt8Array::from(&[Some(1), Some(255)]);
+    assert_eq!(expected, b.as_ref());
+}
+
+#[test]
+fn i32_to_u8() {
+    let array = Int32Array::from_slice(&[-5, 6, -7, 8, 100000000]);
+    let b = cast(&array, &DataType::UInt8).unwrap();
+    let expected = UInt8Array::from(&[None, Some(6), None, Some(8), None]);
+    let c = b.as_any().downcast_ref::<UInt8Array>().unwrap();
+    assert_eq!(c, &expected);
+}
+
+#[test]
+fn i32_to_u8_sliced() {
+    let array = Int32Array::from_slice(&[-5, 6, -7, 8, 100000000]);
+    let array = array.slice(2, 3);
+    let b = cast(&array, &DataType::UInt8).unwrap();
+    let expected = UInt8Array::from(&[None, Some(8), None]);
+    let c = b.as_any().downcast_ref::<UInt8Array>().unwrap();
+    assert_eq!(c, &expected);
+}
+
+#[test]
+fn i32_to_i32() {
+    let array = Int32Array::from_slice(&[5, 6, 7, 8, 9]);
+    let b = cast(&array, &DataType::Int32).unwrap();
+    let c = b.as_any().downcast_ref::<Int32Array>().unwrap();
+
+    let expected = &[5, 6, 7, 8, 9];
+    let expected = Int32Array::from_slice(expected);
+    assert_eq!(c, &expected);
+}
+
+#[test]
+fn i32_to_list_i32() {
+    let array = Int32Array::from_slice(&[5, 6, 7, 8, 9]);
+    let b = cast(
+        &array,
+        &DataType::List(Box::new(Field::new("item", DataType::Int32, true))),
+    )
+    .unwrap();
+
+    let arr = b.as_any().downcast_ref::<ListArray<i32>>().unwrap();
+    assert_eq!(&[0, 1, 2, 3, 4, 5], arr.offsets().as_slice());
+    let values = arr.values();
+    let c = values
+        .as_any()
+        .downcast_ref::<PrimitiveArray<i32>>()
+        .unwrap();
+
+    let expected = Int32Array::from_slice(&[5, 6, 7, 8, 9]);
+    assert_eq!(c, &expected);
+}
+
+#[test]
+fn i32_to_list_i32_nullable() {
+    let input = [Some(5), None, Some(7), Some(8), Some(9)];
+
+    let array = Int32Array::from(input);
+    let b = cast(
+        &array,
+        &DataType::List(Box::new(Field::new("item", DataType::Int32, true))),
+    )
+    .unwrap();
+
+    let arr = b.as_any().downcast_ref::<ListArray<i32>>().unwrap();
+    assert_eq!(&[0, 1, 2, 3, 4, 5], arr.offsets().as_slice());
+    let values = arr.values();
+    let c = values.as_any().downcast_ref::<Int32Array>().unwrap();
+
+    let expected = &[Some(5), None, Some(7), Some(8), Some(9)];
+    let expected = Int32Array::from(expected);
+    assert_eq!(c, &expected);
+}
+
+#[test]
+fn i32_to_list_f64_nullable_sliced() {
+    let input = [Some(5), None, Some(7), Some(8), None, Some(10)];
+
+    let array = Int32Array::from(input);
+
+    let array = array.slice(2, 4);
+    let b = cast(
+        &array,
+        &DataType::List(Box::new(Field::new("item", DataType::Float64, true))),
+    )
+    .unwrap();
+
+    let arr = b.as_any().downcast_ref::<ListArray<i32>>().unwrap();
+    assert_eq!(&[0, 1, 2, 3, 4], arr.offsets().as_slice());
+    let values = arr.values();
+    let c = values.as_any().downcast_ref::<Float64Array>().unwrap();
+
+    let expected = &[Some(7.0), Some(8.0), None, Some(10.0)];
+    let expected = Float64Array::from(expected);
+    assert_eq!(c, &expected);
+}
+
+#[test]
+fn utf8_to_i32() {
+    let array = Utf8Array::<i32>::from_slice(&["5", "6", "seven", "8", "9.1"]);
+    let b = cast(&array, &DataType::Int32).unwrap();
+    let c = b.as_any().downcast_ref::<PrimitiveArray<i32>>().unwrap();
+
+    let expected = &[Some(5), Some(6), None, Some(8), None];
+    let expected = Int32Array::from(expected);
+    assert_eq!(c, &expected);
+}
+
+#[test]
+fn bool_to_i32() {
+    let array = BooleanArray::from(vec![Some(true), Some(false), None]);
+    let b = cast(&array, &DataType::Int32).unwrap();
+    let c = b.as_any().downcast_ref::<Int32Array>().unwrap();
+
+    let expected = &[Some(1), Some(0), None];
+    let expected = Int32Array::from(expected);
+    assert_eq!(c, &expected);
+}
+
+#[test]
+fn bool_to_f64() {
+    let array = BooleanArray::from(vec![Some(true), Some(false), None]);
+    let b = cast(&array, &DataType::Float64).unwrap();
+    let c = b.as_any().downcast_ref::<Float64Array>().unwrap();
+
+    let expected = &[Some(1.0), Some(0.0), None];
+    let expected = Float64Array::from(expected);
+    assert_eq!(c, &expected);
+}
+
+#[test]
+#[should_panic(expected = "Casting from Int32 to Timestamp(Microsecond, None) not supported")]
+fn int32_to_timestamp() {
+    let array = Int32Array::from(&[Some(2), Some(10), None]);
+    cast(&array, &DataType::Timestamp(TimeUnit::Microsecond, None)).unwrap();
+}
+
+#[test]
+fn consistency() {
+    use DataType::*;
+    let datatypes = vec![
+        Null,
+        Boolean,
+        UInt8,
+        UInt16,
+        UInt32,
+        UInt64,
+        Int8,
+        Int16,
+        Int32,
+        Int64,
+        Float32,
+        Float64,
+        Timestamp(TimeUnit::Second, None),
+        Timestamp(TimeUnit::Millisecond, None),
+        Timestamp(TimeUnit::Microsecond, None),
+        Timestamp(TimeUnit::Nanosecond, None),
+        Time64(TimeUnit::Microsecond),
+        Time64(TimeUnit::Nanosecond),
+        Date32,
+        Time32(TimeUnit::Second),
+        Time32(TimeUnit::Millisecond),
+        Date64,
+        Utf8,
+        LargeUtf8,
+        Binary,
+        LargeBinary,
+        Duration(TimeUnit::Second),
+        Duration(TimeUnit::Millisecond),
+        Duration(TimeUnit::Microsecond),
+        Duration(TimeUnit::Nanosecond),
+        List(Box::new(Field::new("a", Utf8, true))),
+        LargeList(Box::new(Field::new("a", Utf8, true))),
+    ];
+    for d1 in &datatypes {
+        for d2 in &datatypes {
+            let array = new_null_array(d1.clone(), 10);
+            if can_cast_types(d1, d2) {
+                let result = cast(array.as_ref(), d2);
+                if let Ok(result) = result {
+                    assert_eq!(result.data_type(), d2, "type not equal: {:?} {:?}", d1, d2);
+                } else {
+                    panic!("Cast should have not failed {:?} {:?}", d1, d2);
+                }
+            } else if cast(array.as_ref(), d2).is_ok() {
+                panic!("Cast should have failed {:?} {:?}", d1, d2);
+            }
+        }
+    }
+}
+
+fn test_primitive_to_primitive<I: NativeType, O: NativeType>(
+    lhs: &[I],
+    lhs_type: DataType,
+    expected: &[O],
+    expected_type: DataType,
+) {
+    let a = PrimitiveArray::<I>::from_slice(lhs).to(lhs_type);
+    let b = cast(&a, &expected_type).unwrap();
+    let b = b.as_any().downcast_ref::<PrimitiveArray<O>>().unwrap();
+    let expected = PrimitiveArray::<O>::from_slice(expected).to(expected_type);
+    assert_eq!(b, &expected);
+}
+
+#[test]
+fn date32_to_date64() {
+    test_primitive_to_primitive(
+        &[10000i32, 17890],
+        DataType::Date32,
+        &[864000000000i64, 1545696000000],
+        DataType::Date64,
+    );
+}
+
+#[test]
+fn date64_to_date32() {
+    test_primitive_to_primitive(
+        &[864000000005i64, 1545696000001],
+        DataType::Date64,
+        &[10000i32, 17890],
+        DataType::Date32,
+    );
+}
+
+#[test]
+fn date32_to_int32() {
+    test_primitive_to_primitive(
+        &[10000i32, 17890],
+        DataType::Date32,
+        &[10000i32, 17890],
+        DataType::Int32,
+    );
+}
+
+#[test]
+fn int32_to_date32() {
+    test_primitive_to_primitive(
+        &[10000i32, 17890],
+        DataType::Int32,
+        &[10000i32, 17890],
+        DataType::Date32,
+    );
+}
+
+#[test]
+fn timestamp_to_date32() {
+    test_primitive_to_primitive(
+        &[864000000005i64, 1545696000001],
+        DataType::Timestamp(TimeUnit::Millisecond, Some(String::from("UTC"))),
+        &[10000i32, 17890],
+        DataType::Date32,
+    );
+}
+
+#[test]
+fn timestamp_to_date64() {
+    test_primitive_to_primitive(
+        &[864000000005i64, 1545696000001],
+        DataType::Timestamp(TimeUnit::Millisecond, Some(String::from("UTC"))),
+        &[864000000005i64, 1545696000001i64],
+        DataType::Date64,
+    );
+}
+
+#[test]
+fn timestamp_to_i64() {
+    test_primitive_to_primitive(
+        &[864000000005i64, 1545696000001],
+        DataType::Timestamp(TimeUnit::Millisecond, Some(String::from("UTC"))),
+        &[864000000005i64, 1545696000001i64],
+        DataType::Int64,
+    );
+}
+
+#[test]
+fn timestamp_to_timestamp() {
+    test_primitive_to_primitive(
+        &[864000003005i64, 1545696002001],
+        DataType::Timestamp(TimeUnit::Millisecond, None),
+        &[864000003i64, 1545696002],
+        DataType::Timestamp(TimeUnit::Second, None),
+    );
+}
+
+#[test]
+fn utf8_to_dict() {
+    let array = Utf8Array::<i32>::from(&[Some("one"), None, Some("three"), Some("one")]);
+
+    // Cast to a dictionary (same value type, Utf8)
+    let cast_type = DataType::Dictionary(Box::new(DataType::UInt8), Box::new(DataType::Utf8));
+    let result = cast(&array, &cast_type).expect("cast failed");
+
+    let mut expected = MutableDictionaryArray::<u8, MutableUtf8Array<i32>>::new();
+    expected
+        .try_extend([Some("one"), None, Some("three"), Some("one")])
+        .unwrap();
+    let expected: DictionaryArray<u8> = expected.into();
+    assert_eq!(expected, result.as_ref());
+}
+
+#[test]
+fn dict_to_utf8() {
+    let mut array = MutableDictionaryArray::<u8, MutableUtf8Array<i32>>::new();
+    array
+        .try_extend([Some("one"), None, Some("three"), Some("one")])
+        .unwrap();
+    let array: DictionaryArray<u8> = array.into();
+
+    let result = cast(&array, &DataType::Utf8).expect("cast failed");
+
+    let expected = Utf8Array::<i32>::from(&[Some("one"), None, Some("three"), Some("one")]);
+
+    assert_eq!(expected, result.as_ref());
+}
+
+#[test]
+fn i32_to_dict() {
+    let array = Int32Array::from(&[Some(1), None, Some(3), Some(1)]);
+
+    // Cast to a dictionary (same value type, Utf8)
+    let cast_type = DataType::Dictionary(Box::new(DataType::UInt8), Box::new(DataType::Int32));
+    let result = cast(&array, &cast_type).expect("cast failed");
+
+    let mut expected = MutableDictionaryArray::<u8, MutablePrimitiveArray<i32>>::new();
+    expected
+        .try_extend([Some(1), None, Some(3), Some(1)])
+        .unwrap();
+    let expected: DictionaryArray<u8> = expected.into();
+    assert_eq!(expected, result.as_ref());
+}
+
+#[test]
+fn list_to_list() {
+    let data = vec![
+        Some(vec![Some(1i32), Some(2), Some(3)]),
+        None,
+        Some(vec![Some(4), None, Some(6)]),
+    ];
+
+    let expected_data = data
+        .iter()
+        .map(|x| x.as_ref().map(|x| x.iter().map(|x| x.map(|x| x as u16))));
+
+    let mut array = MutableListArray::<i32, MutablePrimitiveArray<i32>>::new();
+    array.try_extend(data.clone()).unwrap();
+    let array: ListArray<i32> = array.into();
+
+    let mut expected = MutableListArray::<i32, MutablePrimitiveArray<u16>>::new();
+    expected.try_extend(expected_data).unwrap();
+    let expected: ListArray<i32> = expected.into();
+
+    let result = cast(&array, expected.data_type()).unwrap();
+    assert_eq!(expected, result.as_ref());
+}
+
+/*
+#[test]
+fn dict_to_dict_bad_index_value_primitive() {
+    use DataType::*;
+    // test converting from an array that has indexes of a type
+    // that are out of bounds for a particular other kind of
+    // index.
+
+    let keys_builder = PrimitiveBuilder::<i32>::new(10);
+    let values_builder = PrimitiveBuilder::<i64>::new(10);
+    let mut builder = PrimitiveDictionaryBuilder::new(keys_builder, values_builder);
+
+    // add 200 distinct values (which can be stored by a
+    // dictionary indexed by int32, but not a dictionary indexed
+    // with int8)
+    for i in 0..200 {
+        builder.append(i).unwrap();
+    }
+    let array: ArrayRef = Arc::new(builder.finish());
+
+    let cast_type = Dictionary(Box::new(Int8), Box::new(Utf8));
+    let res = cast_with_options(&array, &cast_type);
+    assert, CastOptions::default())!(res.is_err());
+    let actual_error = format!("{:?}", res);
+    let expected_error = "Could not convert 72 dictionary indexes from Int32 to Int8";
+    assert!(
+        actual_error.contains(expected_error),
+        "did not find expected error '{}' in actual error '{}'",
+        actual_error,
+        expected_error
+    );
+}
+
+#[test]
+fn dict_to_dict_bad_index_value_utf8() {
+    use DataType::*;
+    // Same test as dict_to_dict_bad_index_value but use
+    // string values (and encode the expected behavior here);
+
+    let keys_builder = PrimitiveBuilder::<i32>::new(10);
+    let values_builder = StringBuilder::new(10);
+    let mut builder = StringDictionaryBuilder::new(keys_builder, values_builder);
+
+    // add 200 distinct values (which can be stored by a
+    // dictionary indexed by int32, but not a dictionary indexed
+    // with int8)
+    for i in 0..200 {
+        let val = format!("val{}", i);
+        builder.append(&val).unwrap();
+    }
+    let array: ArrayRef = Arc::new(builder.finish());
+
+    let cast_type = Dictionary(Box::new(Int8), Box::new(Utf8));
+    let res = cast_with_options(&array, &cast_type);
+    assert, CastOptions::default())!(res.is_err());
+    let actual_error = format!("{:?}", res);
+    let expected_error = "Could not convert 72 dictionary indexes from Int32 to Int8";
+    assert!(
+        actual_error.contains(expected_error),
+        "did not find expected error '{}' in actual error '{}'",
+        actual_error,
+        expected_error
+    );
+}
+
+#[test]
+fn utf8_to_date32() {
+    use chrono::NaiveDate;
+    let from_ymd = chrono::NaiveDate::from_ymd;
+    let since = chrono::NaiveDate::signed_duration_since;
+
+    let a = StringArray::from(vec![
+        "2000-01-01",          // valid date with leading 0s
+        "2000-2-2",            // valid date without leading 0s
+        "2000-00-00",          // invalid month and day
+        "2000-01-01T12:00:00", // date + time is invalid
+        "2000",                // just a year is invalid
+    ]);
+    let array = Arc::new(a) as ArrayRef;
+    let b = cast_with_options(&array, &DataType::Date32, CastOptions::default()).unwrap();
+    let c = b.as_any().downcast_ref::<Date32Array>().unwrap();
+
+    // test valid inputs
+    let date_value = since(NaiveDate::from_ymd(2000, 1, 1), from_ymd(1970, 1, 1))
+        .num_days() as i32;
+    assert_eq!(true, c.is_valid(0)); // "2000-01-01"
+    assert_eq!(date_value, c.value(0));
+
+    let date_value = since(NaiveDate::from_ymd(2000, 2, 2), from_ymd(1970, 1, 1))
+        .num_days() as i32;
+    assert_eq!(true, c.is_valid(1)); // "2000-2-2"
+    assert_eq!(date_value, c.value(1));
+
+    // test invalid inputs
+    assert_eq!(false, c.is_valid(2)); // "2000-00-00"
+    assert_eq!(false, c.is_valid(3)); // "2000-01-01T12:00:00"
+    assert_eq!(false, c.is_valid(4)); // "2000"
+}
+
+#[test]
+fn utf8_to_date64() {
+    let a = StringArray::from(vec![
+        "2000-01-01T12:00:00", // date + time valid
+        "2020-12-15T12:34:56", // date + time valid
+        "2020-2-2T12:34:56",   // valid date time without leading 0s
+        "2000-00-00T12:00:00", // invalid month and day
+        "2000-01-01 12:00:00", // missing the 'T'
+        "2000-01-01",          // just a date is invalid
+    ]);
+    let array = Arc::new(a) as ArrayRef;
+    let b = cast_with_options(&array, &DataType::Date64, CastOptions::default()).unwrap();
+    let c = b.as_any().downcast_ref::<Date64Array>().unwrap();
+
+    // test valid inputs
+    assert_eq!(true, c.is_valid(0)); // "2000-01-01T12:00:00"
+    assert_eq!(946728000000, c.value(0));
+    assert_eq!(true, c.is_valid(1)); // "2020-12-15T12:34:56"
+    assert_eq!(1608035696000, c.value(1));
+    assert_eq!(true, c.is_valid(2)); // "2020-2-2T12:34:56"
+    assert_eq!(1580646896000, c.value(2));
+
+    // test invalid inputs
+    assert_eq!(false, c.is_valid(3)); // "2000-00-00T12:00:00"
+    assert_eq!(false, c.is_valid(4)); // "2000-01-01 12:00:00"
+    assert_eq!(false, c.is_valid(5)); // "2000-01-01"
+}
+
+fn make_union_array() -> UnionArray {
+    let mut builder = UnionBuilder::new_dense(7);
+    builder.append::<i32>("a", 1).unwrap();
+    builder.append::<i64>("b", 2).unwrap();
+    builder.build().unwrap()
+}
+*/
diff --git a/tests/it/compute/mod.rs b/tests/it/compute/mod.rs
new file mode 100644
index 0000000000..f64549ca02
--- /dev/null
+++ b/tests/it/compute/mod.rs
@@ -0,0 +1,1 @@
+mod cast;
diff --git a/tests/it/main.rs b/tests/it/main.rs
index 9cd03e6b78..b1501a9f24 100644
--- a/tests/it/main.rs
+++ b/tests/it/main.rs
@@ -6,3 +6,5 @@ mod ffi;
 
 mod io;
 mod test_util;
+
+mod compute;