Repository: jorgecarleitao/arrow2
Pull request #56 (2021-04-25 10:29:49)

Scalar
This PR is an experiment to enable support for scalar values, the dimension 0 version of an `Array`. See README with the design choices.

It does not enforce any in-memory specification, but their struct alignments are semantically equivalent to arrays.

The rational for this scalar is that it allows some of our compute operators to be written as generic functions. The two most relevant use-cases:

1. aggregates, `Fn(Array) -> Scalar`
2. arithmetic with scalars, `Fn(Array, Scalar) -> Array`

This PR demonstrates the usefulness on the aggregates.

codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/56?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#56](https://codecov.io/gh/jorgecarleitao/arrow2/pull/56?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (7cf351c) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/e1fa7e84d6f4003c7be7b42c484c66e09f95f0dc?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (e1fa7e8) will **decrease** coverage by `0.60%`.
> The diff coverage is `5.44%`.

[![Impacted file tree graph](https://codecov.io/gh/jorgecarleitao/arrow2/pull/56/graphs/tree.svg?width=650&height=150&src=pr&token=AgyTF60R3D&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)](https://codecov.io/gh/jorgecarleitao/arrow2/pull/56?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)

```diff
@@            Coverage Diff             @@
##             main      #56      +/-   ##
==========================================
- Coverage   66.82%   66.21%   -0.61%     
==========================================
  Files         180      181       +1     
  Lines       17462    17625     +163     
==========================================
+ Hits        11669    11671       +2     
- Misses       5793     5954     +161     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/56?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/compute/aggregate/min\_max.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/56/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2NvbXB1dGUvYWdncmVnYXRlL21pbl9tYXgucnM=) | `28.14% <0.00%> (-23.91%)` | :arrow_down: |
| [src/scalar/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/56/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL3NjYWxhci9tb2QucnM=) | `0.00% <0.00%> (ø)` | |
| [src/compute/aggregate/sum.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/56/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2NvbXB1dGUvYWdncmVnYXRlL3N1bS5ycw==) | `72.36% <32.00%> (-20.62%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/56?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/56?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Last update [e1fa7e8...7cf351c](https://codecov.io/gh/jorgecarleitao/arrow2/pull/56?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).

jorgecarleitao: @elferherrera what do you think about this?

It is a bit different from DataFusions' current `ScalarValue` because it does not use an `enum`, and has one variation per physical, not logical, data type, to be aligned with arrow2's overall design.

The API would be along the lines of

```rust 
let scalar = max(array);

match scalar.data_type() {
     DataType::Float64 => scalar.as_any().downcast_ref<PrimitiveScalar<f64>>(),
     DataType::Int64 | DataType::Date64 | DataType::Timestamp(_, _) | ... => scalar.as_any().downcast_ref<PrimitiveScalar<i64>>(),
     ...
}
```

which is the same idiom that arrow2 uses for arrays, where each physical type has a unique in-memory representation.
nevi-me: > * aggregates, `Fn(Array) -> Scalar`
> * arithmetic with scalars, `Fn(Array, Scalar) -> Array`

I like this approach, and using a trait instead of an enum offers flexibility with implementing a possible `Datum` like in C++, where kernels can take `Fn(Datum, Datum) -> Array | Scalar | Datum` depending on the kernel function
elferherrera: @jorgecarleitao I also like this approach. It makes `Scalar` a more robust representation. I would guess that this `Scalar` trait object will have to be able to be added to a `PrimitiveArray`, right? We could use the arithmetic traits to help with this operation. 
jorgecarleitao: @elferherrera , I think that we have the exact same problem as `dyn Array`: some kernels support, others do not.
elferherrera: > @elferherrera , I think that we have the exact same problem as `dyn Array`: some kernels support, others do not.

@jorgecarleitao Im lost on this one. Do you mean the same issue we had with `take`?

Patch:
diff --git a/.github/workflows/integration-parquet.yml b/.github/workflows/integration-parquet.yml
index 23d2a27eaf..0bfa715ba3 100644
--- a/.github/workflows/integration-parquet.yml
+++ b/.github/workflows/integration-parquet.yml
@@ -48,5 +48,5 @@ jobs:
           pip install --upgrade pip
           pip install pyarrow pyspark
           python main.py
-          # test delta encoding against spark (pyarrow does not support it)
+          # test against spark
           python main_spark.py
diff --git a/Cargo.toml b/Cargo.toml
index 74ac46afd3..5d8d8bc61f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -33,14 +33,14 @@ serde_json = { version = "^1.0", features = ["preserve_order"], optional = true
 indexmap = { version = "^1.6", optional = true }
 
 # used to print columns in a nice columnar format
-prettytable-rs = { version = "^0.8", optional = true }
+comfy-table = { version = "4.0", optional = true }
 
 flatbuffers = { version = "=2.0.0", optional = true }
 hex = { version = "^0.4", optional = true }
 
 # for IPC compression
 lz4 = { version = "1.23.1", optional = true }
-zstd = { version = "^0.6", optional = true }
+zstd = { version = "0.9", optional = true }
 
 rand = { version = "0.8", optional = true }
 
@@ -58,7 +58,7 @@ futures = { version = "0.3", optional = true }
 # for faster hashing
 ahash = { version = "0.7", optional = true }
 
-parquet2 = { version = "0.1", optional = true, default_features = false, features = ["stream"] }
+parquet2 = { version = "0.3", optional = true, default_features = false, features = ["stream"] }
 
 [dev-dependencies]
 rand = "0.8"
@@ -96,7 +96,7 @@ io_parquet_compression = [
     "parquet2/brotli",
 ]
 io_json_integration = ["io_json", "hex"]
-io_print = ["prettytable-rs"]
+io_print = ["comfy-table"]
 # the compute kernels. Disabling this significantly reduces compile time.
 compute = []
 # base64 + io_ipc because arrow schemas are stored as base64-encoded ipc format.
diff --git a/benches/growable.rs b/benches/growable.rs
index bedb89bd5a..e0a4d2426a 100644
--- a/benches/growable.rs
+++ b/benches/growable.rs
@@ -14,7 +14,7 @@ fn add_benchmark(c: &mut Criterion) {
     let i32_array = create_primitive_array::<i32>(1026 * 10, DataType::Int32, 0.0);
     c.bench_function("growable::primitive::non_null::non_null", |b| {
         b.iter(|| {
-            let mut a = GrowablePrimitive::new(&[&i32_array], false, 1026 * 10);
+            let mut a = GrowablePrimitive::new(vec![&i32_array], false, 1026 * 10);
             values
                 .clone()
                 .into_iter()
@@ -25,7 +25,7 @@ fn add_benchmark(c: &mut Criterion) {
     let i32_array = create_primitive_array::<i32>(1026 * 10, DataType::Int32, 0.0);
     c.bench_function("growable::primitive::non_null::null", |b| {
         b.iter(|| {
-            let mut a = GrowablePrimitive::new(&[&i32_array], true, 1026 * 10);
+            let mut a = GrowablePrimitive::new(vec![&i32_array], true, 1026 * 10);
             values.clone().into_iter().for_each(|start| {
                 if start % 2 == 0 {
                     a.extend_validity(10);
@@ -41,7 +41,7 @@ fn add_benchmark(c: &mut Criterion) {
     let values = values.collect::<Vec<_>>();
     c.bench_function("growable::primitive::null::non_null", |b| {
         b.iter(|| {
-            let mut a = GrowablePrimitive::new(&[&i32_array], false, 1026 * 10);
+            let mut a = GrowablePrimitive::new(vec![&i32_array], false, 1026 * 10);
             values
                 .clone()
                 .into_iter()
@@ -50,7 +50,7 @@ fn add_benchmark(c: &mut Criterion) {
     });
     c.bench_function("growable::primitive::null::null", |b| {
         b.iter(|| {
-            let mut a = GrowablePrimitive::new(&[&i32_array], true, 1026 * 10);
+            let mut a = GrowablePrimitive::new(vec![&i32_array], true, 1026 * 10);
             values.clone().into_iter().for_each(|start| {
                 if start % 2 == 0 {
                     a.extend_validity(10);
diff --git a/benches/sort_kernel.rs b/benches/sort_kernel.rs
index 30d89c441c..e78e4462b0 100644
--- a/benches/sort_kernel.rs
+++ b/benches/sort_kernel.rs
@@ -19,7 +19,7 @@
 extern crate criterion;
 use criterion::Criterion;
 
-use arrow2::compute::sort::{lexsort, sort, SortColumn, SortOptions};
+use arrow2::compute::sort::{lexsort, sort, sort_to_indices, SortColumn, SortOptions};
 use arrow2::util::bench_util::*;
 use arrow2::{array::*, datatypes::*};
 
@@ -42,6 +42,15 @@ fn bench_sort(arr_a: &dyn Array) {
     sort(criterion::black_box(arr_a), &SortOptions::default(), None).unwrap();
 }
 
+fn bench_sort_limit(arr_a: &dyn Array) {
+    let _: PrimitiveArray<u32> = sort_to_indices(
+        criterion::black_box(arr_a),
+        &SortOptions::default(),
+        Some(100),
+    )
+    .unwrap();
+}
+
 fn add_benchmark(c: &mut Criterion) {
     (10..=20).step_by(2).for_each(|log2_size| {
         let size = 2usize.pow(log2_size);
@@ -51,6 +60,10 @@ fn add_benchmark(c: &mut Criterion) {
             b.iter(|| bench_sort(&arr_a))
         });
 
+        c.bench_function(&format!("sort-limit 2^{} f32", log2_size), |b| {
+            b.iter(|| bench_sort_limit(&arr_a))
+        });
+
         let arr_b = create_primitive_array_with_seed::<f32>(size, DataType::Float32, 0.0, 43);
         c.bench_function(&format!("lexsort 2^{} f32", log2_size), |b| {
             b.iter(|| bench_lexsort(&arr_a, &arr_b))
diff --git a/benches/take_kernels.rs b/benches/take_kernels.rs
index 2ab746b443..9303a719f3 100644
--- a/benches/take_kernels.rs
+++ b/benches/take_kernels.rs
@@ -35,7 +35,7 @@ fn create_random_index(size: usize, null_density: f32) -> PrimitiveArray<i32> {
     (0..size)
         .map(|_| {
             if rng.gen::<f32>() > null_density {
-                let value = rng.gen_range::<i32, _, _>(0i32, size as i32);
+                let value = rng.gen_range::<i32, _>(0i32..size as i32);
                 Some(value)
             } else {
                 None
diff --git a/benches/write_parquet.rs b/benches/write_parquet.rs
index df3340a0d4..7cff2c1305 100644
--- a/benches/write_parquet.rs
+++ b/benches/write_parquet.rs
@@ -14,7 +14,7 @@ fn write(array: &dyn Array, encoding: Encoding) -> Result<()> {
 
     let options = WriteOptions {
         write_statistics: false,
-        compression: CompressionCodec::Uncompressed,
+        compression: Compression::Uncompressed,
         version: Version::V1,
     };
 
diff --git a/examples/parquet_write.rs b/examples/parquet_write.rs
index 52e70f417d..af00a6a25a 100644
--- a/examples/parquet_write.rs
+++ b/examples/parquet_write.rs
@@ -7,7 +7,7 @@ use arrow2::{
     datatypes::{Field, Schema},
     error::Result,
     io::parquet::write::{
-        array_to_page, write_file, CompressionCodec, DynIter, Encoding, Version, WriteOptions,
+        array_to_page, write_file, Compression, DynIter, Encoding, Version, WriteOptions,
     },
 };
 
@@ -16,7 +16,7 @@ fn write_single_array(path: &str, array: &dyn Array, field: Field) -> Result<()>
 
     let options = WriteOptions {
         write_statistics: true,
-        compression: CompressionCodec::Uncompressed,
+        compression: Compression::Uncompressed,
         version: Version::V2,
     };
     let encoding = Encoding::Plain;
diff --git a/examples/parquet_write_record.rs b/examples/parquet_write_record.rs
index 9b4deb6879..7528f85a33 100644
--- a/examples/parquet_write_record.rs
+++ b/examples/parquet_write_record.rs
@@ -5,17 +5,18 @@ use arrow2::{
     array::{Array, Int32Array},
     datatypes::{Field, Schema},
     error::Result,
-    io::parquet::write::{write_file, CompressionCodec, RowGroupIterator, Version, WriteOptions},
+    io::parquet::write::{
+        write_file, Compression, Encoding, RowGroupIterator, Version, WriteOptions,
+    },
     record_batch::RecordBatch,
 };
-use parquet2::schema::Encoding;
 
 fn write_batch(path: &str, batch: RecordBatch) -> Result<()> {
     let schema = batch.schema().clone();
 
     let options = WriteOptions {
         write_statistics: true,
-        compression: CompressionCodec::Uncompressed,
+        compression: Compression::Uncompressed,
         version: Version::V2,
     };
 
diff --git a/src/array/mod.rs b/src/array/mod.rs
index ac9dfa79e6..772c49b46c 100644
--- a/src/array/mod.rs
+++ b/src/array/mod.rs
@@ -397,7 +397,7 @@ pub use fixed_size_list::FixedSizeListArray;
 pub use list::{ListArray, MutableListArray};
 pub use null::NullArray;
 pub use primitive::*;
-pub use specification::{Index, Offset};
+pub use specification::Offset;
 pub use struct_::StructArray;
 pub use utf8::{MutableUtf8Array, Utf8Array, Utf8ValuesIter};
 
diff --git a/src/array/specification.rs b/src/array/specification.rs
index 70576cb862..2db3d76293 100644
--- a/src/array/specification.rs
+++ b/src/array/specification.rs
@@ -2,24 +2,13 @@ use std::convert::TryFrom;
 
 use num::Num;
 
-use crate::{
-    buffer::Buffer,
-    types::{NativeType, NaturalDataType},
-};
-
-/// Trait describing any type that can be used to index a slot of an array.
-pub trait Index: NativeType + NaturalDataType {
-    fn to_usize(&self) -> usize;
-    fn from_usize(index: usize) -> Option<Self>;
-}
+use crate::{buffer::Buffer, types::Index};
 
 /// Trait describing types that can be used as offsets as per Arrow specification.
 /// This trait is only implemented for `i32` and `i64`, the two sizes part of the specification.
 /// # Safety
 /// Do not implement.
-pub unsafe trait Offset:
-    Index + Num + Ord + std::ops::AddAssign + std::ops::Sub + num::CheckedAdd
-{
+pub unsafe trait Offset: Index + Num + Ord + num::CheckedAdd {
     fn is_large() -> bool;
 
     fn to_isize(&self) -> isize;
@@ -61,54 +50,6 @@ unsafe impl Offset for i64 {
     }
 }
 
-impl Index for i32 {
-    #[inline]
-    fn to_usize(&self) -> usize {
-        *self as usize
-    }
-
-    #[inline]
-    fn from_usize(value: usize) -> Option<Self> {
-        Self::try_from(value).ok()
-    }
-}
-
-impl Index for i64 {
-    #[inline]
-    fn to_usize(&self) -> usize {
-        *self as usize
-    }
-
-    #[inline]
-    fn from_usize(value: usize) -> Option<Self> {
-        Self::try_from(value).ok()
-    }
-}
-
-impl Index for u32 {
-    #[inline]
-    fn to_usize(&self) -> usize {
-        *self as usize
-    }
-
-    #[inline]
-    fn from_usize(value: usize) -> Option<Self> {
-        Self::try_from(value).ok()
-    }
-}
-
-impl Index for u64 {
-    #[inline]
-    fn to_usize(&self) -> usize {
-        *self as usize
-    }
-
-    #[inline]
-    fn from_usize(value: usize) -> Option<Self> {
-        Self::try_from(value).ok()
-    }
-}
-
 #[inline]
 pub fn check_offsets<O: Offset>(offsets: &Buffer<O>, values_len: usize) -> usize {
     assert!(
diff --git a/src/bitmap/immutable.rs b/src/bitmap/immutable.rs
index 5630e8b4bd..0b1ab34ea4 100644
--- a/src/bitmap/immutable.rs
+++ b/src/bitmap/immutable.rs
@@ -216,7 +216,7 @@ impl Bitmap {
     #[inline]
     pub(crate) fn as_slice(&self) -> &[u8] {
         let start = self.offset / 8;
-        let len = self.length.saturating_add(7) / 8;
+        let len = (self.offset() + self.length).saturating_add(7) / 8;
         &self.bytes[start..start + len]
     }
 }
@@ -261,4 +261,15 @@ mod tests {
 
         assert_eq!(0, b.offset());
     }
+
+    #[test]
+    fn as_slice_offset_middle() {
+        let b = Bitmap::from_u8_slice(&[0, 0, 0, 0b00010101], 27);
+        let b = b.slice(22, 5);
+
+        let slice = b.as_slice();
+        assert_eq!(slice, &[0, 0b00010101]);
+
+        assert_eq!(6, b.offset());
+    }
 }
diff --git a/src/bitmap/utils/chunk_iterator/mod.rs b/src/bitmap/utils/chunk_iterator/mod.rs
index 26df6ee827..b896b1a666 100644
--- a/src/bitmap/utils/chunk_iterator/mod.rs
+++ b/src/bitmap/utils/chunk_iterator/mod.rs
@@ -31,11 +31,9 @@ pub struct BitChunks<'a, T: BitChunk> {
 /// writes `bytes` into `dst`.
 #[inline]
 fn copy_with_merge<T: BitChunk>(dst: &mut T::Bytes, bytes: &[u8], bit_offset: usize) {
-    let mut last = bytes[bytes.len() - 1];
-    last >>= bit_offset;
-    dst[0] = last;
     bytes
         .windows(2)
+        .chain(std::iter::once([bytes[bytes.len() - 1], 0].as_ref()))
         .take(std::mem::size_of::<T>())
         .enumerate()
         .for_each(|(i, w)| {
@@ -340,4 +338,23 @@ mod tests {
         assert_eq!(iter.next(), None);
         assert_eq!(iter.remainder(), 0b1_1111_1111u64);
     }
+
+    #[test]
+    fn remainder_2() {
+        // (i % 3 == 0) in bitmap
+        let input: &[u8] = &[
+            0b01001001, 0b10010010, 0b00100100, 0b01001001, 0b10010010, 0b00100100, 0b01001001,
+            0b10010010, 0b00100100, 0b01001001, /* 73 */
+            0b10010010, /* 146 */
+            0b00100100, 0b00001001,
+        ];
+        let offset = 10; // 8 + 2
+        let length = 90;
+
+        let mut iter = BitChunks::<u64>::new(input, offset, length);
+        let first: u64 = 0b0100100100100100100100100100100100100100100100100100100100100100;
+        assert_eq!(first, iter.next().unwrap());
+        assert_eq!(iter.next(), None);
+        assert_eq!(iter.remainder(), 0b10010010010010010010010010u64);
+    }
 }
diff --git a/src/bitmap/utils/mod.rs b/src/bitmap/utils/mod.rs
index db284b28d7..538c193b0b 100644
--- a/src/bitmap/utils/mod.rs
+++ b/src/bitmap/utils/mod.rs
@@ -65,6 +65,8 @@ pub fn bytes_for(bits: usize) -> usize {
 
 #[inline]
 pub fn null_count(slice: &[u8], offset: usize, len: usize) -> usize {
+    //return BitmapIter::new(slice, offset, len).filter(|x| !*x).count();
+
     // u64 results in optimal performance (verified via benches)
     let mut chunks = chunk_iterator::BitChunks::<u64>::new(slice, offset, len);
 
@@ -138,4 +140,11 @@ mod tests {
         assert_eq!(null_count(input, 13, 2), 1);
         assert_eq!(null_count(input, 14, 2), 1);
     }
+
+    #[test]
+    fn null_count_1() {
+        // offset = 10, len = 90 => remainder
+        let input: &[u8] = &[73, 146, 36, 73, 146, 36, 73, 146, 36, 73, 146, 36, 9];
+        assert_eq!(null_count(input, 10, 90), 60);
+    }
 }
diff --git a/src/bitmap/utils/slice_iterator.rs b/src/bitmap/utils/slice_iterator.rs
index 2dfbe111ee..f2633a1615 100644
--- a/src/bitmap/utils/slice_iterator.rs
+++ b/src/bitmap/utils/slice_iterator.rs
@@ -284,4 +284,13 @@ mod tests {
         // the first "11" in the second byte
         assert_eq!(chunks, vec![(0, 2)]);
     }
+
+    #[test]
+    fn remainder_1() {
+        let values = Bitmap::from_u8_slice(&[0, 0, 0b00000000, 0b00010101], 27);
+        let values = values.slice(22, 5);
+        let iter = SlicesIterator::new(&values);
+        let chunks = iter.collect::<Vec<_>>();
+        assert_eq!(chunks, vec![(2, 1), (4, 1)]);
+    }
 }
diff --git a/src/compute/aggregate/min_max.rs b/src/compute/aggregate/min_max.rs
index fdc08f05a3..2ab1bf1534 100644
--- a/src/compute/aggregate/min_max.rs
+++ b/src/compute/aggregate/min_max.rs
@@ -1,4 +1,7 @@
 use crate::bitmap::utils::{BitChunkIterExact, BitChunksExact};
+use crate::datatypes::{DataType, IntervalUnit};
+use crate::error::{ArrowError, Result};
+use crate::scalar::*;
 use crate::types::simd::*;
 use crate::types::NativeType;
 use crate::{
@@ -281,6 +284,94 @@ pub fn max_boolean(array: &BooleanArray) -> Option<bool> {
         .or(Some(false))
 }
 
+macro_rules! dyn_primitive {
+    ($ty:ty, $array:expr, $f:ident) => {{
+        let array = $array
+            .as_any()
+            .downcast_ref::<PrimitiveArray<$ty>>()
+            .unwrap();
+        Box::new(PrimitiveScalar::<$ty>::new(
+            $array.data_type().clone(),
+            $f::<$ty>(array),
+        ))
+    }};
+}
+
+macro_rules! dyn_generic {
+    ($array_ty:ty, $scalar_ty:ty, $array:expr, $f:ident) => {{
+        let array = $array.as_any().downcast_ref::<$array_ty>().unwrap();
+        Box::new(<$scalar_ty>::new($f(array)))
+    }};
+}
+
+pub fn max(array: &dyn Array) -> Result<Box<dyn Scalar>> {
+    Ok(match array.data_type() {
+        DataType::Boolean => dyn_generic!(BooleanArray, BooleanScalar, array, max_boolean),
+        DataType::Int8 => dyn_primitive!(i8, array, max_primitive),
+        DataType::Int16 => dyn_primitive!(i16, array, max_primitive),
+        DataType::Int32
+        | DataType::Date32
+        | DataType::Time32(_)
+        | DataType::Interval(IntervalUnit::YearMonth) => {
+            dyn_primitive!(i32, array, max_primitive)
+        }
+        DataType::Int64
+        | DataType::Date64
+        | DataType::Time64(_)
+        | DataType::Timestamp(_, _)
+        | DataType::Duration(_) => dyn_primitive!(i64, array, max_primitive),
+        DataType::UInt8 => dyn_primitive!(u8, array, max_primitive),
+        DataType::UInt16 => dyn_primitive!(u16, array, max_primitive),
+        DataType::UInt32 => dyn_primitive!(u32, array, max_primitive),
+        DataType::UInt64 => dyn_primitive!(u64, array, max_primitive),
+        DataType::Float16 => unreachable!(),
+        DataType::Float32 => dyn_primitive!(f32, array, max_primitive),
+        DataType::Float64 => dyn_primitive!(f64, array, max_primitive),
+        DataType::Utf8 => dyn_generic!(Utf8Array<i32>, Utf8Scalar<i32>, array, max_string),
+        DataType::LargeUtf8 => dyn_generic!(Utf8Array<i64>, Utf8Scalar<i64>, array, max_string),
+        _ => {
+            return Err(ArrowError::InvalidArgumentError(format!(
+                "The `max` operator does not support type `{}`",
+                array.data_type(),
+            )))
+        }
+    })
+}
+
+pub fn min(array: &dyn Array) -> Result<Box<dyn Scalar>> {
+    Ok(match array.data_type() {
+        DataType::Boolean => dyn_generic!(BooleanArray, BooleanScalar, array, min_boolean),
+        DataType::Int8 => dyn_primitive!(i8, array, min_primitive),
+        DataType::Int16 => dyn_primitive!(i16, array, min_primitive),
+        DataType::Int32
+        | DataType::Date32
+        | DataType::Time32(_)
+        | DataType::Interval(IntervalUnit::YearMonth) => {
+            dyn_primitive!(i32, array, min_primitive)
+        }
+        DataType::Int64
+        | DataType::Date64
+        | DataType::Time64(_)
+        | DataType::Timestamp(_, _)
+        | DataType::Duration(_) => dyn_primitive!(i64, array, min_primitive),
+        DataType::UInt8 => dyn_primitive!(u8, array, min_primitive),
+        DataType::UInt16 => dyn_primitive!(u16, array, min_primitive),
+        DataType::UInt32 => dyn_primitive!(u32, array, min_primitive),
+        DataType::UInt64 => dyn_primitive!(u64, array, min_primitive),
+        DataType::Float16 => unreachable!(),
+        DataType::Float32 => dyn_primitive!(f32, array, min_primitive),
+        DataType::Float64 => dyn_primitive!(f64, array, min_primitive),
+        DataType::Utf8 => dyn_generic!(Utf8Array<i32>, Utf8Scalar<i32>, array, min_string),
+        DataType::LargeUtf8 => dyn_generic!(Utf8Array<i64>, Utf8Scalar<i64>, array, min_string),
+        _ => {
+            return Err(ArrowError::InvalidArgumentError(format!(
+                "The `max` operator does not support type `{}`",
+                array.data_type(),
+            )))
+        }
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -372,81 +463,79 @@ mod tests {
     }
 
     // todo: convert me
-    /*
     #[test]
     fn test_string_min_max_with_nulls() {
-        let a = StringArray::from(vec![Some("b"), None, None, Some("a"), Some("c")]);
+        let a = Utf8Array::<i32>::from(&[Some("b"), None, None, Some("a"), Some("c")]);
         assert_eq!("a", min_string(&a).unwrap());
         assert_eq!("c", max_string(&a).unwrap());
     }
 
     #[test]
     fn test_string_min_max_all_nulls() {
-        let a = StringArray::from(vec![None, None]);
+        let a = Utf8Array::<i32>::from(&[None::<&str>, None]);
         assert_eq!(None, min_string(&a));
         assert_eq!(None, max_string(&a));
     }
 
     #[test]
     fn test_string_min_max_1() {
-        let a = StringArray::from(vec![None, None, Some("b"), Some("a")]);
+        let a = Utf8Array::<i32>::from(&[None, None, Some("b"), Some("a")]);
         assert_eq!(Some("a"), min_string(&a));
         assert_eq!(Some("b"), max_string(&a));
     }
 
     #[test]
     fn test_boolean_min_max_empty() {
-        let a = BooleanArray::from(vec![] as Vec<Option<bool>>);
+        let a = BooleanArray::new_empty();
         assert_eq!(None, min_boolean(&a));
         assert_eq!(None, max_boolean(&a));
     }
 
     #[test]
     fn test_boolean_min_max_all_null() {
-        let a = BooleanArray::from(vec![None, None]);
+        let a = BooleanArray::from(&[None, None]);
         assert_eq!(None, min_boolean(&a));
         assert_eq!(None, max_boolean(&a));
     }
 
     #[test]
     fn test_boolean_min_max_no_null() {
-        let a = BooleanArray::from(vec![Some(true), Some(false), Some(true)]);
+        let a = BooleanArray::from(&[Some(true), Some(false), Some(true)]);
         assert_eq!(Some(false), min_boolean(&a));
         assert_eq!(Some(true), max_boolean(&a));
     }
 
     #[test]
     fn test_boolean_min_max() {
-        let a = BooleanArray::from(vec![Some(true), Some(true), None, Some(false), None]);
+        let a = BooleanArray::from(&[Some(true), Some(true), None, Some(false), None]);
         assert_eq!(Some(false), min_boolean(&a));
         assert_eq!(Some(true), max_boolean(&a));
 
-        let a = BooleanArray::from(vec![None, Some(true), None, Some(false), None]);
+        let a = BooleanArray::from(&[None, Some(true), None, Some(false), None]);
         assert_eq!(Some(false), min_boolean(&a));
         assert_eq!(Some(true), max_boolean(&a));
 
-        let a = BooleanArray::from(vec![Some(false), Some(true), None, Some(false), None]);
+        let a = BooleanArray::from(&[Some(false), Some(true), None, Some(false), None]);
         assert_eq!(Some(false), min_boolean(&a));
         assert_eq!(Some(true), max_boolean(&a));
     }
 
     #[test]
     fn test_boolean_min_max_smaller() {
-        let a = BooleanArray::from(vec![Some(false)]);
+        let a = BooleanArray::from(&[Some(false)]);
         assert_eq!(Some(false), min_boolean(&a));
         assert_eq!(Some(false), max_boolean(&a));
 
-        let a = BooleanArray::from(vec![None, Some(false)]);
+        let a = BooleanArray::from(&[None, Some(false)]);
         assert_eq!(Some(false), min_boolean(&a));
         assert_eq!(Some(false), max_boolean(&a));
 
-        let a = BooleanArray::from(vec![None, Some(true)]);
+        let a = BooleanArray::from(&[None, Some(true)]);
         assert_eq!(Some(true), min_boolean(&a));
         assert_eq!(Some(true), max_boolean(&a));
 
-        let a = BooleanArray::from(vec![Some(true)]);
+        let a = BooleanArray::from(&[Some(true)]);
         assert_eq!(Some(true), min_boolean(&a));
         assert_eq!(Some(true), max_boolean(&a));
     }
-     */
 }
diff --git a/src/compute/aggregate/sum.rs b/src/compute/aggregate/sum.rs
index b1f1cece78..eb3b3ef4d2 100644
--- a/src/compute/aggregate/sum.rs
+++ b/src/compute/aggregate/sum.rs
@@ -3,6 +3,9 @@ use std::ops::Add;
 use multiversion::multiversion;
 
 use crate::bitmap::utils::{BitChunkIterExact, BitChunksExact};
+use crate::datatypes::{DataType, IntervalUnit};
+use crate::error::{ArrowError, Result};
+use crate::scalar::*;
 use crate::types::simd::*;
 use crate::types::NativeType;
 use crate::{
@@ -85,7 +88,7 @@ where
 /// Returns the sum of values in the array.
 ///
 /// Returns `None` if the array is empty or only contains null values.
-pub fn sum<T>(array: &PrimitiveArray<T>) -> Option<T>
+pub fn sum_primitive<T>(array: &PrimitiveArray<T>) -> Option<T>
 where
     T: NativeType + Simd,
     T::Simd: Add<Output = T::Simd> + Sum<T>,
@@ -102,6 +105,76 @@ where
     }
 }
 
+macro_rules! dyn_sum {
+    ($ty:ty, $array:expr) => {{
+        let array = $array
+            .as_any()
+            .downcast_ref::<PrimitiveArray<$ty>>()
+            .unwrap();
+        Box::new(PrimitiveScalar::<$ty>::new(
+            $array.data_type().clone(),
+            sum_primitive::<$ty>(array),
+        ))
+    }};
+}
+
+pub fn can_sum(data_type: &DataType) -> bool {
+    use DataType::*;
+    matches!(
+        data_type,
+        Int8 | Int16
+            | Date32
+            | Time32(_)
+            | Interval(IntervalUnit::YearMonth)
+            | Int64
+            | Date64
+            | Time64(_)
+            | Timestamp(_, _)
+            | Duration(_)
+            | UInt8
+            | UInt16
+            | UInt32
+            | UInt64
+            | Float32
+            | Float64
+    )
+}
+
+/// Returns the sum of all elements in `array` as a [`Scalar`] of the same physical
+/// and logical types as `array`.
+/// # Error
+/// Errors iff the operation is not supported.
+pub fn sum(array: &dyn Array) -> Result<Box<dyn Scalar>> {
+    Ok(match array.data_type() {
+        DataType::Int8 => dyn_sum!(i8, array),
+        DataType::Int16 => dyn_sum!(i16, array),
+        DataType::Int32
+        | DataType::Date32
+        | DataType::Time32(_)
+        | DataType::Interval(IntervalUnit::YearMonth) => {
+            dyn_sum!(i32, array)
+        }
+        DataType::Int64
+        | DataType::Date64
+        | DataType::Time64(_)
+        | DataType::Timestamp(_, _)
+        | DataType::Duration(_) => dyn_sum!(i64, array),
+        DataType::UInt8 => dyn_sum!(u8, array),
+        DataType::UInt16 => dyn_sum!(u16, array),
+        DataType::UInt32 => dyn_sum!(u32, array),
+        DataType::UInt64 => dyn_sum!(u64, array),
+        DataType::Float16 => unreachable!(),
+        DataType::Float32 => dyn_sum!(f32, array),
+        DataType::Float64 => dyn_sum!(f64, array),
+        _ => {
+            return Err(ArrowError::InvalidArgumentError(format!(
+                "The `sum` operator does not support type `{}`",
+                array.data_type(),
+            )))
+        }
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::super::super::arithmetics;
@@ -111,25 +184,34 @@ mod tests {
     #[test]
     fn test_primitive_array_sum() {
         let a = Int32Array::from_slice(&[1, 2, 3, 4, 5]);
-        assert_eq!(15, sum(&a).unwrap());
+        assert_eq!(
+            &PrimitiveScalar::<i32>::from(Some(15)) as &dyn Scalar,
+            sum(&a).unwrap().as_ref()
+        );
+
+        let a = a.to(DataType::Date32);
+        assert_eq!(
+            &PrimitiveScalar::<i32>::from(Some(15)).to(DataType::Date32) as &dyn Scalar,
+            sum(&a).unwrap().as_ref()
+        );
     }
 
     #[test]
     fn test_primitive_array_float_sum() {
         let a = Float64Array::from_slice(&[1.1f64, 2.2, 3.3, 4.4, 5.5]);
-        assert!((16.5 - sum(&a).unwrap()).abs() < f64::EPSILON);
+        assert!((16.5 - sum_primitive(&a).unwrap()).abs() < f64::EPSILON);
     }
 
     #[test]
     fn test_primitive_array_sum_with_nulls() {
         let a = Int32Array::from(&[None, Some(2), Some(3), None, Some(5)]);
-        assert_eq!(10, sum(&a).unwrap());
+        assert_eq!(10, sum_primitive(&a).unwrap());
     }
 
     #[test]
     fn test_primitive_array_sum_all_nulls() {
         let a = Int32Array::from(&[None, None, None]);
-        assert_eq!(None, sum(&a));
+        assert_eq!(None, sum_primitive(&a));
     }
 
     #[test]
@@ -142,6 +224,9 @@ mod tests {
             .collect();
         // create an array that actually has non-zero values at the invalid indices
         let c = arithmetics::basic::add::add(&a, &b).unwrap();
-        assert_eq!(Some((1..=100).filter(|i| i % 3 == 0).sum()), sum(&c));
+        assert_eq!(
+            Some((1..=100).filter(|i| i % 3 == 0).sum()),
+            sum_primitive(&c)
+        );
     }
 }
diff --git a/src/compute/cast/dictionary_to.rs b/src/compute/cast/dictionary_to.rs
index 2f77a94838..9e053c9d9a 100644
--- a/src/compute/cast/dictionary_to.rs
+++ b/src/compute/cast/dictionary_to.rs
@@ -1,7 +1,7 @@
-use super::{cast, primitive_to_primitive};
+use super::{primitive_as_primitive, primitive_to_primitive, CastOptions};
 use crate::{
     array::{Array, DictionaryArray, DictionaryKey, PrimitiveArray},
-    compute::take::take,
+    compute::{cast::cast_with_options, take::take},
     datatypes::DataType,
     error::{ArrowError, Result},
 };
@@ -32,7 +32,20 @@ pub fn dictionary_to_dictionary_values<K: DictionaryKey>(
     let keys = from.keys();
     let values = from.values();
 
-    let values = cast(values.as_ref(), values_type)?.into();
+    let values = cast_with_options(values.as_ref(), values_type, CastOptions::default())?.into();
+    Ok(DictionaryArray::from_data(keys.clone(), values))
+}
+
+/// Similar to dictionary_to_dictionary_values, but overflowing cast is wrapped
+pub fn wrapping_dictionary_to_dictionary_values<K: DictionaryKey>(
+    from: &DictionaryArray<K>,
+    values_type: &DataType,
+) -> Result<DictionaryArray<K>> {
+    let keys = from.keys();
+    let values = from.values();
+
+    let values =
+        cast_with_options(values.as_ref(), values_type, CastOptions { wrapped: true })?.into();
     Ok(DictionaryArray::from_data(keys.clone(), values))
 }
 
@@ -60,9 +73,30 @@ where
     }
 }
 
+/// Similar to dictionary_to_dictionary_keys, but overflowing cast is wrapped
+pub fn wrapping_dictionary_to_dictionary_keys<K1, K2>(
+    from: &DictionaryArray<K1>,
+) -> Result<DictionaryArray<K2>>
+where
+    K1: DictionaryKey + num::traits::AsPrimitive<K2>,
+    K2: DictionaryKey,
+{
+    let keys = from.keys();
+    let values = from.values();
+
+    let casted_keys = primitive_as_primitive::<K1, K2>(keys, &K2::DATA_TYPE);
+
+    if casted_keys.null_count() > keys.null_count() {
+        Err(ArrowError::KeyOverflowError)
+    } else {
+        Ok(DictionaryArray::from_data(casted_keys, values.clone()))
+    }
+}
+
 pub(super) fn dictionary_cast_dyn<K: DictionaryKey>(
     array: &dyn Array,
     to_type: &DataType,
+    options: CastOptions,
 ) -> Result<Box<dyn Array>> {
     let array = array.as_any().downcast_ref::<DictionaryArray<K>>().unwrap();
     let keys = array.keys();
@@ -70,7 +104,7 @@ pub(super) fn dictionary_cast_dyn<K: DictionaryKey>(
 
     match to_type {
         DataType::Dictionary(to_keys_type, to_values_type) => {
-            let values = cast(values.as_ref(), to_values_type)?.into();
+            let values = cast_with_options(values.as_ref(), to_values_type, options)?.into();
 
             // create the appropriate array type
             match to_keys_type.as_ref() {
@@ -85,7 +119,7 @@ pub(super) fn dictionary_cast_dyn<K: DictionaryKey>(
                 _ => unreachable!(),
             }
         }
-        _ => unpack_dictionary::<K>(keys, values.as_ref(), to_type),
+        _ => unpack_dictionary::<K>(keys, values.as_ref(), to_type, options),
     }
 }
 
@@ -94,13 +128,14 @@ fn unpack_dictionary<K>(
     keys: &PrimitiveArray<K>,
     values: &dyn Array,
     to_type: &DataType,
+    options: CastOptions,
 ) -> Result<Box<dyn Array>>
 where
     K: DictionaryKey,
 {
     // attempt to cast the dict values to the target type
     // use the take kernel to expand out the dictionary
-    let values = cast(values, to_type)?;
+    let values = cast_with_options(values, to_type, options)?;
 
     // take requires first casting i32
     let indices = primitive_to_primitive::<_, i32>(keys, &DataType::Int32);
diff --git a/src/compute/cast/mod.rs b/src/compute/cast/mod.rs
index 14817a14e4..9d09e6bc75 100644
--- a/src/compute/cast/mod.rs
+++ b/src/compute/cast/mod.rs
@@ -36,6 +36,29 @@ pub use primitive_to::*;
 pub use timestamps::*;
 pub use utf8_to::*;
 
+/// options defining how Cast kernels behave
+#[derive(Clone, Copy, Debug)]
+struct CastOptions {
+    /// default to false
+    /// whether an overflowing cast should be converted to `None` (default), or be wrapped (i.e. `256i16 as u8 = 0` vectorized).
+    /// Settings this to `true` is 5-6x faster for numeric types.
+    wrapped: bool,
+}
+
+impl Default for CastOptions {
+    fn default() -> Self {
+        Self { wrapped: false }
+    }
+}
+
+impl CastOptions {
+    fn with_wrapped(&self, v: bool) -> Self {
+        let mut option = *self;
+        option.wrapped = v;
+        option
+    }
+}
+
 /// Returns true if this type is numeric: (UInt*, Unit*, or Float*).
 fn is_numeric(t: &DataType) -> bool {
     use DataType::*;
@@ -239,9 +262,18 @@ pub fn can_cast_types(from_type: &DataType, to_type: &DataType) -> bool {
     }
 }
 
-fn cast_list<O: Offset>(array: &ListArray<O>, to_type: &DataType) -> Result<ListArray<O>> {
+fn cast_list<O: Offset>(
+    array: &ListArray<O>,
+    to_type: &DataType,
+    options: CastOptions,
+) -> Result<ListArray<O>> {
     let values = array.values();
-    let new_values = cast(values.as_ref(), ListArray::<O>::get_child_type(to_type))?.into();
+    let new_values = cast_with_options(
+        values.as_ref(),
+        ListArray::<O>::get_child_type(to_type),
+        options,
+    )?
+    .into();
 
     Ok(ListArray::<O>::from_data(
         to_type.clone(),
@@ -281,6 +313,7 @@ fn cast_large_to_list(array: &ListArray<i64>, to_type: &DataType) -> ListArray<i
 /// type `to_type`, if possible.
 ///
 /// Behavior:
+/// * PrimitiveArray to PrimitiveArray: overflowing cast will be None
 /// * Boolean to Utf8: `true` => '1', `false` => `0`
 /// * Utf8 to numeric: strings that can't be parsed to numbers return null, float strings
 ///   in integer casts return null
@@ -291,13 +324,28 @@ fn cast_large_to_list(array: &ListArray<i64>, to_type: &DataType) -> ListArray<i
 /// * Time32 and Time64: precision lost when going to higher interval
 /// * Timestamp and Date{32|64}: precision lost when going to higher interval
 /// * Temporal to/from backing primitive: zero-copy with data type change
-///
 /// Unsupported Casts
 /// * To or from `StructArray`
 /// * List to primitive
 /// * Utf8 to boolean
 /// * Interval and duration
 pub fn cast(array: &dyn Array, to_type: &DataType) -> Result<Box<dyn Array>> {
+    cast_with_options(array, to_type, CastOptions { wrapped: false })
+}
+
+/// Similar to [`cast`], but overflowing cast is wrapped
+/// Behavior:
+/// * PrimitiveArray to PrimitiveArray: overflowing cast will be wrapped (i.e. `256i16 as u8 = 0` vectorized).
+pub fn wrapping_cast(array: &dyn Array, to_type: &DataType) -> Result<Box<dyn Array>> {
+    cast_with_options(array, to_type, CastOptions { wrapped: true })
+}
+
+#[inline]
+fn cast_with_options(
+    array: &dyn Array,
+    to_type: &DataType,
+    options: CastOptions,
+) -> Result<Box<dyn Array>> {
     use DataType::*;
     let from_type = array.data_type();
 
@@ -305,6 +353,8 @@ pub fn cast(array: &dyn Array, to_type: &DataType) -> Result<Box<dyn Array>> {
     if from_type == to_type {
         return Ok(clone(array));
     }
+
+    let as_options = options.with_wrapped(true);
     match (from_type, to_type) {
         (Struct(_), _) => Err(ArrowError::NotYetImplemented(
             "Cannot cast from struct to other types".to_string(),
@@ -312,10 +362,12 @@ pub fn cast(array: &dyn Array, to_type: &DataType) -> Result<Box<dyn Array>> {
         (_, Struct(_)) => Err(ArrowError::NotYetImplemented(
             "Cannot cast to struct from other types".to_string(),
         )),
-        (List(_), List(_)) => cast_list::<i32>(array.as_any().downcast_ref().unwrap(), to_type)
-            .map(|x| Box::new(x) as Box<dyn Array>),
+        (List(_), List(_)) => {
+            cast_list::<i32>(array.as_any().downcast_ref().unwrap(), to_type, options)
+                .map(|x| Box::new(x) as Box<dyn Array>)
+        }
         (LargeList(_), LargeList(_)) => {
-            cast_list::<i64>(array.as_any().downcast_ref().unwrap(), to_type)
+            cast_list::<i64>(array.as_any().downcast_ref().unwrap(), to_type, options)
                 .map(|x| Box::new(x) as Box<dyn Array>)
         }
         (List(lhs), LargeList(rhs)) if lhs == rhs => Ok(cast_list_to_large_list(
@@ -331,7 +383,7 @@ pub fn cast(array: &dyn Array, to_type: &DataType) -> Result<Box<dyn Array>> {
 
         (_, List(to)) => {
             // cast primitive to list's primitive
-            let values = cast(array, to.data_type())?.into();
+            let values = cast_with_options(array, to.data_type(), options)?.into();
             // create offsets, where if array.len() = 2, we have [0,1,2]
             let offsets =
                 unsafe { Buffer::from_trusted_len_iter_unchecked(0..=array.len() as i32) };
@@ -343,25 +395,25 @@ pub fn cast(array: &dyn Array, to_type: &DataType) -> Result<Box<dyn Array>> {
         }
 
         (Dictionary(index_type, _), _) => match **index_type {
-            DataType::Int8 => dictionary_cast_dyn::<i8>(array, to_type),
-            DataType::Int16 => dictionary_cast_dyn::<i16>(array, to_type),
-            DataType::Int32 => dictionary_cast_dyn::<i32>(array, to_type),
-            DataType::Int64 => dictionary_cast_dyn::<i64>(array, to_type),
-            DataType::UInt8 => dictionary_cast_dyn::<u8>(array, to_type),
-            DataType::UInt16 => dictionary_cast_dyn::<u16>(array, to_type),
-            DataType::UInt32 => dictionary_cast_dyn::<u32>(array, to_type),
-            DataType::UInt64 => dictionary_cast_dyn::<u64>(array, to_type),
+            DataType::Int8 => dictionary_cast_dyn::<i8>(array, to_type, options),
+            DataType::Int16 => dictionary_cast_dyn::<i16>(array, to_type, options),
+            DataType::Int32 => dictionary_cast_dyn::<i32>(array, to_type, options),
+            DataType::Int64 => dictionary_cast_dyn::<i64>(array, to_type, options),
+            DataType::UInt8 => dictionary_cast_dyn::<u8>(array, to_type, options),
+            DataType::UInt16 => dictionary_cast_dyn::<u16>(array, to_type, options),
+            DataType::UInt32 => dictionary_cast_dyn::<u32>(array, to_type, options),
+            DataType::UInt64 => dictionary_cast_dyn::<u64>(array, to_type, options),
             _ => unreachable!(),
         },
         (_, Dictionary(index_type, value_type)) => match **index_type {
-            DataType::Int8 => cast_to_dictionary::<i8>(array, value_type),
-            DataType::Int16 => cast_to_dictionary::<i16>(array, value_type),
-            DataType::Int32 => cast_to_dictionary::<i32>(array, value_type),
-            DataType::Int64 => cast_to_dictionary::<i64>(array, value_type),
-            DataType::UInt8 => cast_to_dictionary::<u8>(array, value_type),
-            DataType::UInt16 => cast_to_dictionary::<u16>(array, value_type),
-            DataType::UInt32 => cast_to_dictionary::<u32>(array, value_type),
-            DataType::UInt64 => cast_to_dictionary::<u64>(array, value_type),
+            DataType::Int8 => cast_to_dictionary::<i8>(array, value_type, options),
+            DataType::Int16 => cast_to_dictionary::<i16>(array, value_type, options),
+            DataType::Int32 => cast_to_dictionary::<i32>(array, value_type, options),
+            DataType::Int64 => cast_to_dictionary::<i64>(array, value_type, options),
+            DataType::UInt8 => cast_to_dictionary::<u8>(array, value_type, options),
+            DataType::UInt16 => cast_to_dictionary::<u16>(array, value_type, options),
+            DataType::UInt32 => cast_to_dictionary::<u32>(array, value_type, options),
+            DataType::UInt64 => cast_to_dictionary::<u64>(array, value_type, options),
             _ => Err(ArrowError::NotYetImplemented(format!(
                 "Casting from type {:?} to dictionary type {:?} not supported",
                 from_type, to_type,
@@ -509,105 +561,105 @@ pub fn cast(array: &dyn Array, to_type: &DataType) -> Result<Box<dyn Array>> {
             .map(|x| Box::new(x) as Box<dyn Array>),
 
         // start numeric casts
-        (UInt8, UInt16) => primitive_to_primitive_dyn::<u8, u16>(array, to_type),
-        (UInt8, UInt32) => primitive_to_primitive_dyn::<u8, u32>(array, to_type),
-        (UInt8, UInt64) => primitive_to_primitive_dyn::<u8, u64>(array, to_type),
-        (UInt8, Int8) => primitive_to_primitive_dyn::<u8, i8>(array, to_type),
-        (UInt8, Int16) => primitive_to_primitive_dyn::<u8, i16>(array, to_type),
-        (UInt8, Int32) => primitive_to_primitive_dyn::<u8, i32>(array, to_type),
-        (UInt8, Int64) => primitive_to_primitive_dyn::<u8, i64>(array, to_type),
-        (UInt8, Float32) => primitive_to_primitive_dyn::<u8, f32>(array, to_type),
-        (UInt8, Float64) => primitive_to_primitive_dyn::<u8, f64>(array, to_type),
-
-        (UInt16, UInt8) => primitive_to_primitive_dyn::<u16, u8>(array, to_type),
-        (UInt16, UInt32) => primitive_to_primitive_dyn::<u16, u32>(array, to_type),
-        (UInt16, UInt64) => primitive_to_primitive_dyn::<u16, u64>(array, to_type),
-        (UInt16, Int8) => primitive_to_primitive_dyn::<u16, i8>(array, to_type),
-        (UInt16, Int16) => primitive_to_primitive_dyn::<u16, i16>(array, to_type),
-        (UInt16, Int32) => primitive_to_primitive_dyn::<u16, i32>(array, to_type),
-        (UInt16, Int64) => primitive_to_primitive_dyn::<u16, i64>(array, to_type),
-        (UInt16, Float32) => primitive_to_primitive_dyn::<u16, f32>(array, to_type),
-        (UInt16, Float64) => primitive_to_primitive_dyn::<u16, f64>(array, to_type),
-
-        (UInt32, UInt8) => primitive_to_primitive_dyn::<u32, u8>(array, to_type),
-        (UInt32, UInt16) => primitive_to_primitive_dyn::<u32, u16>(array, to_type),
-        (UInt32, UInt64) => primitive_to_primitive_dyn::<u32, u64>(array, to_type),
-        (UInt32, Int8) => primitive_to_primitive_dyn::<u32, i8>(array, to_type),
-        (UInt32, Int16) => primitive_to_primitive_dyn::<u32, i16>(array, to_type),
-        (UInt32, Int32) => primitive_to_primitive_dyn::<u32, i32>(array, to_type),
-        (UInt32, Int64) => primitive_to_primitive_dyn::<u32, i64>(array, to_type),
-        (UInt32, Float32) => primitive_to_primitive_dyn::<u32, f32>(array, to_type),
-        (UInt32, Float64) => primitive_to_primitive_dyn::<u32, f64>(array, to_type),
-
-        (UInt64, UInt8) => primitive_to_primitive_dyn::<u64, u8>(array, to_type),
-        (UInt64, UInt16) => primitive_to_primitive_dyn::<u64, u16>(array, to_type),
-        (UInt64, UInt32) => primitive_to_primitive_dyn::<u64, u32>(array, to_type),
-        (UInt64, Int8) => primitive_to_primitive_dyn::<u64, i8>(array, to_type),
-        (UInt64, Int16) => primitive_to_primitive_dyn::<u64, i16>(array, to_type),
-        (UInt64, Int32) => primitive_to_primitive_dyn::<u64, i32>(array, to_type),
-        (UInt64, Int64) => primitive_to_primitive_dyn::<u64, i64>(array, to_type),
-        (UInt64, Float32) => primitive_to_primitive_dyn::<u64, f32>(array, to_type),
-        (UInt64, Float64) => primitive_to_primitive_dyn::<u64, f64>(array, to_type),
-
-        (Int8, UInt8) => primitive_to_primitive_dyn::<i8, u8>(array, to_type),
-        (Int8, UInt16) => primitive_to_primitive_dyn::<i8, u16>(array, to_type),
-        (Int8, UInt32) => primitive_to_primitive_dyn::<i8, u32>(array, to_type),
-        (Int8, UInt64) => primitive_to_primitive_dyn::<i8, u64>(array, to_type),
-        (Int8, Int16) => primitive_to_primitive_dyn::<i8, i16>(array, to_type),
-        (Int8, Int32) => primitive_to_primitive_dyn::<i8, i32>(array, to_type),
-        (Int8, Int64) => primitive_to_primitive_dyn::<i8, i64>(array, to_type),
-        (Int8, Float32) => primitive_to_primitive_dyn::<i8, f32>(array, to_type),
-        (Int8, Float64) => primitive_to_primitive_dyn::<i8, f64>(array, to_type),
-
-        (Int16, UInt8) => primitive_to_primitive_dyn::<i16, u8>(array, to_type),
-        (Int16, UInt16) => primitive_to_primitive_dyn::<i16, u16>(array, to_type),
-        (Int16, UInt32) => primitive_to_primitive_dyn::<i16, u32>(array, to_type),
-        (Int16, UInt64) => primitive_to_primitive_dyn::<i16, u64>(array, to_type),
-        (Int16, Int8) => primitive_to_primitive_dyn::<i16, i8>(array, to_type),
-        (Int16, Int32) => primitive_to_primitive_dyn::<i16, i32>(array, to_type),
-        (Int16, Int64) => primitive_to_primitive_dyn::<i16, i64>(array, to_type),
-        (Int16, Float32) => primitive_to_primitive_dyn::<i16, f32>(array, to_type),
-        (Int16, Float64) => primitive_to_primitive_dyn::<i16, f64>(array, to_type),
-
-        (Int32, UInt8) => primitive_to_primitive_dyn::<i32, u8>(array, to_type),
-        (Int32, UInt16) => primitive_to_primitive_dyn::<i32, u16>(array, to_type),
-        (Int32, UInt32) => primitive_to_primitive_dyn::<i32, u32>(array, to_type),
-        (Int32, UInt64) => primitive_to_primitive_dyn::<i32, u64>(array, to_type),
-        (Int32, Int8) => primitive_to_primitive_dyn::<i32, i8>(array, to_type),
-        (Int32, Int16) => primitive_to_primitive_dyn::<i32, i16>(array, to_type),
-        (Int32, Int64) => primitive_to_primitive_dyn::<i32, i64>(array, to_type),
-        (Int32, Float32) => primitive_to_primitive_dyn::<i32, f32>(array, to_type),
-        (Int32, Float64) => primitive_to_primitive_dyn::<i32, f64>(array, to_type),
-
-        (Int64, UInt8) => primitive_to_primitive_dyn::<i64, u8>(array, to_type),
-        (Int64, UInt16) => primitive_to_primitive_dyn::<i64, u16>(array, to_type),
-        (Int64, UInt32) => primitive_to_primitive_dyn::<i64, u32>(array, to_type),
-        (Int64, UInt64) => primitive_to_primitive_dyn::<i64, u64>(array, to_type),
-        (Int64, Int8) => primitive_to_primitive_dyn::<i64, i8>(array, to_type),
-        (Int64, Int16) => primitive_to_primitive_dyn::<i64, i16>(array, to_type),
-        (Int64, Int32) => primitive_to_primitive_dyn::<i64, i32>(array, to_type),
-        (Int64, Float32) => primitive_to_primitive_dyn::<i64, f32>(array, to_type),
-        (Int64, Float64) => primitive_to_primitive_dyn::<i64, f64>(array, to_type),
-
-        (Float32, UInt8) => primitive_to_primitive_dyn::<f32, u8>(array, to_type),
-        (Float32, UInt16) => primitive_to_primitive_dyn::<f32, u16>(array, to_type),
-        (Float32, UInt32) => primitive_to_primitive_dyn::<f32, u32>(array, to_type),
-        (Float32, UInt64) => primitive_to_primitive_dyn::<f32, u64>(array, to_type),
-        (Float32, Int8) => primitive_to_primitive_dyn::<f32, i8>(array, to_type),
-        (Float32, Int16) => primitive_to_primitive_dyn::<f32, i16>(array, to_type),
-        (Float32, Int32) => primitive_to_primitive_dyn::<f32, i32>(array, to_type),
-        (Float32, Int64) => primitive_to_primitive_dyn::<f32, i64>(array, to_type),
-        (Float32, Float64) => primitive_to_primitive_dyn::<f32, f64>(array, to_type),
-
-        (Float64, UInt8) => primitive_to_primitive_dyn::<f64, u8>(array, to_type),
-        (Float64, UInt16) => primitive_to_primitive_dyn::<f64, u16>(array, to_type),
-        (Float64, UInt32) => primitive_to_primitive_dyn::<f64, u32>(array, to_type),
-        (Float64, UInt64) => primitive_to_primitive_dyn::<f64, u64>(array, to_type),
-        (Float64, Int8) => primitive_to_primitive_dyn::<f64, i8>(array, to_type),
-        (Float64, Int16) => primitive_to_primitive_dyn::<f64, i16>(array, to_type),
-        (Float64, Int32) => primitive_to_primitive_dyn::<f64, i32>(array, to_type),
-        (Float64, Int64) => primitive_to_primitive_dyn::<f64, i64>(array, to_type),
-        (Float64, Float32) => primitive_to_primitive_dyn::<f64, f32>(array, to_type),
+        (UInt8, UInt16) => primitive_to_primitive_dyn::<u8, u16>(array, to_type, as_options),
+        (UInt8, UInt32) => primitive_to_primitive_dyn::<u8, u32>(array, to_type, as_options),
+        (UInt8, UInt64) => primitive_to_primitive_dyn::<u8, u64>(array, to_type, as_options),
+        (UInt8, Int8) => primitive_to_primitive_dyn::<u8, i8>(array, to_type, options),
+        (UInt8, Int16) => primitive_to_primitive_dyn::<u8, i16>(array, to_type, options),
+        (UInt8, Int32) => primitive_to_primitive_dyn::<u8, i32>(array, to_type, options),
+        (UInt8, Int64) => primitive_to_primitive_dyn::<u8, i64>(array, to_type, options),
+        (UInt8, Float32) => primitive_to_primitive_dyn::<u8, f32>(array, to_type, as_options),
+        (UInt8, Float64) => primitive_to_primitive_dyn::<u8, f64>(array, to_type, as_options),
+
+        (UInt16, UInt8) => primitive_to_primitive_dyn::<u16, u8>(array, to_type, options),
+        (UInt16, UInt32) => primitive_to_primitive_dyn::<u16, u32>(array, to_type, as_options),
+        (UInt16, UInt64) => primitive_to_primitive_dyn::<u16, u64>(array, to_type, as_options),
+        (UInt16, Int8) => primitive_to_primitive_dyn::<u16, i8>(array, to_type, options),
+        (UInt16, Int16) => primitive_to_primitive_dyn::<u16, i16>(array, to_type, options),
+        (UInt16, Int32) => primitive_to_primitive_dyn::<u16, i32>(array, to_type, options),
+        (UInt16, Int64) => primitive_to_primitive_dyn::<u16, i64>(array, to_type, options),
+        (UInt16, Float32) => primitive_to_primitive_dyn::<u16, f32>(array, to_type, as_options),
+        (UInt16, Float64) => primitive_to_primitive_dyn::<u16, f64>(array, to_type, as_options),
+
+        (UInt32, UInt8) => primitive_to_primitive_dyn::<u32, u8>(array, to_type, options),
+        (UInt32, UInt16) => primitive_to_primitive_dyn::<u32, u16>(array, to_type, options),
+        (UInt32, UInt64) => primitive_to_primitive_dyn::<u32, u64>(array, to_type, as_options),
+        (UInt32, Int8) => primitive_to_primitive_dyn::<u32, i8>(array, to_type, options),
+        (UInt32, Int16) => primitive_to_primitive_dyn::<u32, i16>(array, to_type, options),
+        (UInt32, Int32) => primitive_to_primitive_dyn::<u32, i32>(array, to_type, options),
+        (UInt32, Int64) => primitive_to_primitive_dyn::<u32, i64>(array, to_type, options),
+        (UInt32, Float32) => primitive_to_primitive_dyn::<u32, f32>(array, to_type, as_options),
+        (UInt32, Float64) => primitive_to_primitive_dyn::<u32, f64>(array, to_type, as_options),
+
+        (UInt64, UInt8) => primitive_to_primitive_dyn::<u64, u8>(array, to_type, options),
+        (UInt64, UInt16) => primitive_to_primitive_dyn::<u64, u16>(array, to_type, options),
+        (UInt64, UInt32) => primitive_to_primitive_dyn::<u64, u32>(array, to_type, options),
+        (UInt64, Int8) => primitive_to_primitive_dyn::<u64, i8>(array, to_type, options),
+        (UInt64, Int16) => primitive_to_primitive_dyn::<u64, i16>(array, to_type, options),
+        (UInt64, Int32) => primitive_to_primitive_dyn::<u64, i32>(array, to_type, options),
+        (UInt64, Int64) => primitive_to_primitive_dyn::<u64, i64>(array, to_type, options),
+        (UInt64, Float32) => primitive_to_primitive_dyn::<u64, f32>(array, to_type, as_options),
+        (UInt64, Float64) => primitive_to_primitive_dyn::<u64, f64>(array, to_type, as_options),
+
+        (Int8, UInt8) => primitive_to_primitive_dyn::<i8, u8>(array, to_type, options),
+        (Int8, UInt16) => primitive_to_primitive_dyn::<i8, u16>(array, to_type, options),
+        (Int8, UInt32) => primitive_to_primitive_dyn::<i8, u32>(array, to_type, options),
+        (Int8, UInt64) => primitive_to_primitive_dyn::<i8, u64>(array, to_type, options),
+        (Int8, Int16) => primitive_to_primitive_dyn::<i8, i16>(array, to_type, as_options),
+        (Int8, Int32) => primitive_to_primitive_dyn::<i8, i32>(array, to_type, as_options),
+        (Int8, Int64) => primitive_to_primitive_dyn::<i8, i64>(array, to_type, as_options),
+        (Int8, Float32) => primitive_to_primitive_dyn::<i8, f32>(array, to_type, as_options),
+        (Int8, Float64) => primitive_to_primitive_dyn::<i8, f64>(array, to_type, as_options),
+
+        (Int16, UInt8) => primitive_to_primitive_dyn::<i16, u8>(array, to_type, options),
+        (Int16, UInt16) => primitive_to_primitive_dyn::<i16, u16>(array, to_type, options),
+        (Int16, UInt32) => primitive_to_primitive_dyn::<i16, u32>(array, to_type, options),
+        (Int16, UInt64) => primitive_to_primitive_dyn::<i16, u64>(array, to_type, options),
+        (Int16, Int8) => primitive_to_primitive_dyn::<i16, i8>(array, to_type, options),
+        (Int16, Int32) => primitive_to_primitive_dyn::<i16, i32>(array, to_type, as_options),
+        (Int16, Int64) => primitive_to_primitive_dyn::<i16, i64>(array, to_type, as_options),
+        (Int16, Float32) => primitive_to_primitive_dyn::<i16, f32>(array, to_type, as_options),
+        (Int16, Float64) => primitive_to_primitive_dyn::<i16, f64>(array, to_type, as_options),
+
+        (Int32, UInt8) => primitive_to_primitive_dyn::<i32, u8>(array, to_type, options),
+        (Int32, UInt16) => primitive_to_primitive_dyn::<i32, u16>(array, to_type, options),
+        (Int32, UInt32) => primitive_to_primitive_dyn::<i32, u32>(array, to_type, options),
+        (Int32, UInt64) => primitive_to_primitive_dyn::<i32, u64>(array, to_type, options),
+        (Int32, Int8) => primitive_to_primitive_dyn::<i32, i8>(array, to_type, options),
+        (Int32, Int16) => primitive_to_primitive_dyn::<i32, i16>(array, to_type, options),
+        (Int32, Int64) => primitive_to_primitive_dyn::<i32, i64>(array, to_type, as_options),
+        (Int32, Float32) => primitive_to_primitive_dyn::<i32, f32>(array, to_type, as_options),
+        (Int32, Float64) => primitive_to_primitive_dyn::<i32, f64>(array, to_type, as_options),
+
+        (Int64, UInt8) => primitive_to_primitive_dyn::<i64, u8>(array, to_type, options),
+        (Int64, UInt16) => primitive_to_primitive_dyn::<i64, u16>(array, to_type, options),
+        (Int64, UInt32) => primitive_to_primitive_dyn::<i64, u32>(array, to_type, options),
+        (Int64, UInt64) => primitive_to_primitive_dyn::<i64, u64>(array, to_type, options),
+        (Int64, Int8) => primitive_to_primitive_dyn::<i64, i8>(array, to_type, options),
+        (Int64, Int16) => primitive_to_primitive_dyn::<i64, i16>(array, to_type, options),
+        (Int64, Int32) => primitive_to_primitive_dyn::<i64, i32>(array, to_type, options),
+        (Int64, Float32) => primitive_to_primitive_dyn::<i64, f32>(array, to_type, options),
+        (Int64, Float64) => primitive_to_primitive_dyn::<i64, f64>(array, to_type, as_options),
+
+        (Float32, UInt8) => primitive_to_primitive_dyn::<f32, u8>(array, to_type, options),
+        (Float32, UInt16) => primitive_to_primitive_dyn::<f32, u16>(array, to_type, options),
+        (Float32, UInt32) => primitive_to_primitive_dyn::<f32, u32>(array, to_type, options),
+        (Float32, UInt64) => primitive_to_primitive_dyn::<f32, u64>(array, to_type, options),
+        (Float32, Int8) => primitive_to_primitive_dyn::<f32, i8>(array, to_type, options),
+        (Float32, Int16) => primitive_to_primitive_dyn::<f32, i16>(array, to_type, options),
+        (Float32, Int32) => primitive_to_primitive_dyn::<f32, i32>(array, to_type, options),
+        (Float32, Int64) => primitive_to_primitive_dyn::<f32, i64>(array, to_type, options),
+        (Float32, Float64) => primitive_to_primitive_dyn::<f32, f64>(array, to_type, as_options),
+
+        (Float64, UInt8) => primitive_to_primitive_dyn::<f64, u8>(array, to_type, options),
+        (Float64, UInt16) => primitive_to_primitive_dyn::<f64, u16>(array, to_type, options),
+        (Float64, UInt32) => primitive_to_primitive_dyn::<f64, u32>(array, to_type, options),
+        (Float64, UInt64) => primitive_to_primitive_dyn::<f64, u64>(array, to_type, options),
+        (Float64, Int8) => primitive_to_primitive_dyn::<f64, i8>(array, to_type, options),
+        (Float64, Int16) => primitive_to_primitive_dyn::<f64, i16>(array, to_type, options),
+        (Float64, Int32) => primitive_to_primitive_dyn::<f64, i32>(array, to_type, options),
+        (Float64, Int64) => primitive_to_primitive_dyn::<f64, i64>(array, to_type, options),
+        (Float64, Float32) => primitive_to_primitive_dyn::<f64, f32>(array, to_type, options),
         // end numeric casts
 
         // temporal casts
@@ -677,8 +729,9 @@ pub fn cast(array: &dyn Array, to_type: &DataType) -> Result<Box<dyn Array>> {
 fn cast_to_dictionary<K: DictionaryKey>(
     array: &dyn Array,
     dict_value_type: &DataType,
+    options: CastOptions,
 ) -> Result<Box<dyn Array>> {
-    let array = cast(array, dict_value_type)?;
+    let array = cast_with_options(array, dict_value_type, options)?;
     let array = array.as_ref();
     match *dict_value_type {
         DataType::Int8 => primitive_to_dictionary_dyn::<i8, K>(array),
@@ -701,9 +754,10 @@ fn cast_to_dictionary<K: DictionaryKey>(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::types::NativeType;
 
     #[test]
-    fn test_cast_i32_to_f64() {
+    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();
@@ -715,7 +769,140 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_i32_to_u8() {
+    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]);
@@ -724,7 +911,7 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_i32_to_u8_sliced() {
+    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();
@@ -734,7 +921,7 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_i32_to_i32() {
+    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();
@@ -745,7 +932,7 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_i32_to_list_i32() {
+    fn i32_to_list_i32() {
         let array = Int32Array::from_slice(&[5, 6, 7, 8, 9]);
         let b = cast(
             &array,
@@ -766,7 +953,7 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_i32_to_list_i32_nullable() {
+    fn i32_to_list_i32_nullable() {
         let input = [Some(5), None, Some(7), Some(8), Some(9)];
 
         let array = Int32Array::from(input);
@@ -787,7 +974,7 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_i32_to_list_f64_nullable_sliced() {
+    fn i32_to_list_f64_nullable_sliced() {
         let input = [Some(5), None, Some(7), Some(8), None, Some(10)];
 
         let array = Int32Array::from(input);
@@ -810,7 +997,7 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_utf8_to_i32() {
+    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();
@@ -821,7 +1008,7 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_bool_to_i32() {
+    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();
@@ -832,7 +1019,7 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_bool_to_f64() {
+    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();
@@ -844,7 +1031,7 @@ mod tests {
 
     #[test]
     #[should_panic(expected = "Casting from Int32 to Timestamp(Microsecond, None) not supported")]
-    fn test_cast_int32_to_timestamp() {
+    fn int32_to_timestamp() {
         let array = Int32Array::from(&[Some(2), Some(10), None]);
         cast(&array, &DataType::Timestamp(TimeUnit::Microsecond, None)).unwrap();
     }
@@ -884,6 +1071,7 @@ mod tests {
             Duration(TimeUnit::Microsecond),
             Duration(TimeUnit::Nanosecond),
             List(Box::new(Field::new("a", Utf8, true))),
+            LargeList(Box::new(Field::new("a", Utf8, true))),
         ];
         datatypes
             .clone()
@@ -899,1206 +1087,180 @@ mod tests {
                         panic!("Cast should have not failed")
                     }
                 } else {
-                    assert!(cast(array.as_ref(), &d2).is_err());
+                    assert!(
+                        cast_with_options(array.as_ref(), &d2, CastOptions::default()).is_err()
+                    );
                 }
             });
     }
 
-    /*
-    #[test]
-    fn test_cast_list_i32_to_list_u16() {
-        // Construct a value array
-        let value_data = Int32Array::from(vec![0, 0, 0, -1, -2, -1, 2, 100000000]).data();
-
-        let value_offsets = Buffer::from_slice_ref(&[0, 3, 6, 8]);
-
-        // Construct a list array from the above two
-        let list_data_type =
-            DataType::List(Box::new(Field::new("item", DataType::Int32, true)));
-        let list_data = ArrayData::builder(list_data_type)
-            .len(3)
-            .add_buffer(value_offsets)
-            .add_child_data(value_data)
-            .build();
-        let list_array = Arc::new(ListArray::from(list_data)) as ArrayRef;
-
-        let cast_array = cast(
-            &list_array,
-            &DataType::List(Box::new(Field::new("item", DataType::UInt16, true))),
-        )
-        .unwrap();
-        // 3 negative values should get lost when casting to unsigned,
-        // 1 value should overflow
-        assert_eq!(4, cast_array.null_count());
-        // offsets should be the same
-        assert_eq!(
-            list_array.data().buffers().to_vec(),
-            cast_array.data().buffers().to_vec()
-        );
-        let array = cast_array
-            .as_ref()
-            .as_any()
-            .downcast_ref::<ListArray>()
-            .unwrap();
-        assert_eq!(DataType::UInt16, array.value_type());
-        assert_eq!(4, array.values().null_count());
-        assert_eq!(3, array.value_length(0));
-        assert_eq!(3, array.value_length(1));
-        assert_eq!(2, array.value_length(2));
-        let values = array.values();
-        let u16arr = values.as_any().downcast_ref::<UInt16Array>().unwrap();
-        assert_eq!(8, u16arr.len());
-        assert_eq!(4, u16arr.null_count());
-
-        assert_eq!(0, u16arr.value(0));
-        assert_eq!(0, u16arr.value(1));
-        assert_eq!(0, u16arr.value(2));
-        assert_eq!(false, u16arr.is_valid(3));
-        assert_eq!(false, u16arr.is_valid(4));
-        assert_eq!(false, u16arr.is_valid(5));
-        assert_eq!(2, u16arr.value(6));
-        assert_eq!(false, u16arr.is_valid(7));
-    }
-
-    #[test]
-    #[should_panic(
-        expected = "Casting from Int32 to Timestamp(Microsecond, None) not supported"
-    )]
-    fn test_cast_list_i32_to_list_timestamp() {
-        // Construct a value array
-        let value_data =
-            Int32Array::from(vec![0, 0, 0, -1, -2, -1, 2, 8, 100000000]).data();
-
-        let value_offsets = Buffer::from_slice_ref(&[0, 3, 6, 9]);
-
-        // Construct a list array from the above two
-        let list_data_type =
-            DataType::List(Box::new(Field::new("item", DataType::Int32, true)));
-        let list_data = ArrayData::builder(list_data_type)
-            .len(3)
-            .add_buffer(value_offsets)
-            .add_child_data(value_data)
-            .build();
-        let list_array = Arc::new(ListArray::from(list_data)) as ArrayRef;
-
-        cast(
-            &list_array,
-            &DataType::List(Box::new(Field::new(
-                "item",
-                DataType::Timestamp(TimeUnit::Microsecond, None),
-                true,
-            ))),
-        )
-        .unwrap();
-    }
-
-    #[test]
-    fn test_cast_date32_to_date64() {
-        let a = Date32Array::from(vec![10000, 17890]);
-        let array = Arc::new(a) as ArrayRef;
-        let b = cast(&array, &DataType::Date64).unwrap();
-        let c = b.as_any().downcast_ref::<Date64Array>().unwrap();
-        assert_eq!(864000000000, c.value(0));
-        assert_eq!(1545696000000, c.value(1));
-    }
-
-    #[test]
-    fn test_cast_date64_to_date32() {
-        let a = Date64Array::from(vec![Some(864000000005), Some(1545696000001), None]);
-        let array = Arc::new(a) as ArrayRef;
-        let b = cast(&array, &DataType::Date32).unwrap();
-        let c = b.as_any().downcast_ref::<Date32Array>().unwrap();
-        assert_eq!(10000, c.value(0));
-        assert_eq!(17890, c.value(1));
-        assert!(c.is_null(2));
-    }
-
-    #[test]
-    fn test_cast_date32_to_int32() {
-        let a = Date32Array::from(vec![10000, 17890]);
-        let array = Arc::new(a) as ArrayRef;
-        let b = cast(&array, &DataType::Int32).unwrap();
-        let c = b.as_any().downcast_ref::<Int32Array>().unwrap();
-        assert_eq!(10000, c.value(0));
-        assert_eq!(17890, c.value(1));
-    }
-
-    #[test]
-    fn test_cast_int32_to_date32() {
-        let a = Int32Array::from(vec![10000, 17890]);
-        let array = Arc::new(a) as ArrayRef;
-        let b = cast(&array, &DataType::Date32).unwrap();
-        let c = b.as_any().downcast_ref::<Date32Array>().unwrap();
-        assert_eq!(10000, c.value(0));
-        assert_eq!(17890, c.value(1));
+    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 test_cast_timestamp_to_date32() {
-        let a = TimestampMillisecondArray::from_opt_vec(
-            vec![Some(864000000005), Some(1545696000001), None],
-            Some(String::from("UTC")),
+    fn date32_to_date64() {
+        test_primitive_to_primitive(
+            &[10000i32, 17890],
+            DataType::Date32,
+            &[864000000000i64, 1545696000000],
+            DataType::Date64,
         );
-        let array = Arc::new(a) as ArrayRef;
-        let b = cast(&array, &DataType::Date32).unwrap();
-        let c = b.as_any().downcast_ref::<Date32Array>().unwrap();
-        assert_eq!(10000, c.value(0));
-        assert_eq!(17890, c.value(1));
-        assert!(c.is_null(2));
     }
 
     #[test]
-    fn test_cast_timestamp_to_date64() {
-        let a = TimestampMillisecondArray::from_opt_vec(
-            vec![Some(864000000005), Some(1545696000001), None],
-            None,
+    fn date64_to_date32() {
+        test_primitive_to_primitive(
+            &[864000000005i64, 1545696000001],
+            DataType::Date64,
+            &[10000i32, 17890],
+            DataType::Date32,
         );
-        let array = Arc::new(a) as ArrayRef;
-        let b = cast(&array, &DataType::Date64).unwrap();
-        let c = b.as_any().downcast_ref::<Date64Array>().unwrap();
-        assert_eq!(864000000005, c.value(0));
-        assert_eq!(1545696000001, c.value(1));
-        assert!(c.is_null(2));
     }
 
     #[test]
-    fn test_cast_timestamp_to_i64() {
-        let a = TimestampMillisecondArray::from_opt_vec(
-            vec![Some(864000000005), Some(1545696000001), None],
-            Some("UTC".to_string()),
+    fn date32_to_int32() {
+        test_primitive_to_primitive(
+            &[10000i32, 17890],
+            DataType::Date32,
+            &[10000i32, 17890],
+            DataType::Int32,
         );
-        let array = Arc::new(a) as ArrayRef;
-        let b = cast(&array, &DataType::Int64).unwrap();
-        let c = b.as_any().downcast_ref::<Int64Array>().unwrap();
-        assert_eq!(&DataType::Int64, c.data_type());
-        assert_eq!(864000000005, c.value(0));
-        assert_eq!(1545696000001, c.value(1));
-        assert!(c.is_null(2));
     }
 
     #[test]
-    fn test_cast_between_timestamps() {
-        let a = TimestampMillisecondArray::from_opt_vec(
-            vec![Some(864000003005), Some(1545696002001), None],
-            None,
+    fn int32_to_date32() {
+        test_primitive_to_primitive(
+            &[10000i32, 17890],
+            DataType::Int32,
+            &[10000i32, 17890],
+            DataType::Date32,
         );
-        let array = Arc::new(a) as ArrayRef;
-        let b = cast(&array, &DataType::Timestamp(TimeUnit::Second, None)).unwrap();
-        let c = b.as_any().downcast_ref::<TimestampSecondArray>().unwrap();
-        assert_eq!(864000003, c.value(0));
-        assert_eq!(1545696002, c.value(1));
-        assert!(c.is_null(2));
     }
 
     #[test]
-    fn test_cast_from_f64() {
-        let f64_values: Vec<f64> = vec![
-            std::i64::MIN as f64,
-            std::i32::MIN as f64,
-            std::i16::MIN as f64,
-            std::i8::MIN as f64,
-            0_f64,
-            std::u8::MAX as f64,
-            std::u16::MAX as f64,
-            std::u32::MAX as f64,
-            std::u64::MAX as f64,
-        ];
-        let f64_array: ArrayRef = Arc::new(Float64Array::from(f64_values));
-
-        let f64_expected = vec![
-            "-9223372036854776000.0",
-            "-2147483648.0",
-            "-32768.0",
-            "-128.0",
-            "0.0",
-            "255.0",
-            "65535.0",
-            "4294967295.0",
-            "18446744073709552000.0",
-        ];
-        assert_eq!(
-            f64_expected,
-            get_cast_values::<f64>(&f64_array, &DataType::Float64)
-        );
-
-        let f32_expected = vec![
-            "-9223372000000000000.0",
-            "-2147483600.0",
-            "-32768.0",
-            "-128.0",
-            "0.0",
-            "255.0",
-            "65535.0",
-            "4294967300.0",
-            "18446744000000000000.0",
-        ];
-        assert_eq!(
-            f32_expected,
-            get_cast_values::<f32>(&f64_array, &DataType::Float32)
-        );
-
-        let i64_expected = vec![
-            "-9223372036854775808",
-            "-2147483648",
-            "-32768",
-            "-128",
-            "0",
-            "255",
-            "65535",
-            "4294967295",
-            "null",
-        ];
-        assert_eq!(
-            i64_expected,
-            get_cast_values::<i64>(&f64_array, &DataType::Int64)
-        );
-
-        let i32_expected = vec![
-            "null",
-            "-2147483648",
-            "-32768",
-            "-128",
-            "0",
-            "255",
-            "65535",
-            "null",
-            "null",
-        ];
-        assert_eq!(
-            i32_expected,
-            get_cast_values::<i32>(&f64_array, &DataType::Int32)
-        );
-
-        let i16_expected = vec![
-            "null", "null", "-32768", "-128", "0", "255", "null", "null", "null",
-        ];
-        assert_eq!(
-            i16_expected,
-            get_cast_values::<i16>(&f64_array, &DataType::Int16)
-        );
-
-        let i8_expected = vec![
-            "null", "null", "null", "-128", "0", "null", "null", "null", "null",
-        ];
-        assert_eq!(
-            i8_expected,
-            get_cast_values::<i8>(&f64_array, &DataType::Int8)
-        );
-
-        let u64_expected = vec![
-            "null",
-            "null",
-            "null",
-            "null",
-            "0",
-            "255",
-            "65535",
-            "4294967295",
-            "null",
-        ];
-        assert_eq!(
-            u64_expected,
-            get_cast_values::<u64>(&f64_array, &DataType::UInt64)
-        );
-
-        let u32_expected = vec![
-            "null",
-            "null",
-            "null",
-            "null",
-            "0",
-            "255",
-            "65535",
-            "4294967295",
-            "null",
-        ];
-        assert_eq!(
-            u32_expected,
-            get_cast_values::<u32>(&f64_array, &DataType::UInt32)
-        );
-
-        let u16_expected = vec![
-            "null", "null", "null", "null", "0", "255", "65535", "null", "null",
-        ];
-        assert_eq!(
-            u16_expected,
-            get_cast_values::<u16>(&f64_array, &DataType::UInt16)
-        );
-
-        let u8_expected = vec![
-            "null", "null", "null", "null", "0", "255", "null", "null", "null",
-        ];
-        assert_eq!(
-            u8_expected,
-            get_cast_values::<u8>(&f64_array, &DataType::UInt8)
+    fn timestamp_to_date32() {
+        test_primitive_to_primitive(
+            &[864000000005i64, 1545696000001],
+            DataType::Timestamp(TimeUnit::Millisecond, Some(String::from("UTC"))),
+            &[10000i32, 17890],
+            DataType::Date32,
         );
     }
 
     #[test]
-    fn test_cast_from_f32() {
-        let f32_values: Vec<f32> = vec![
-            std::i32::MIN as f32,
-            std::i32::MIN as f32,
-            std::i16::MIN as f32,
-            std::i8::MIN as f32,
-            0_f32,
-            std::u8::MAX as f32,
-            std::u16::MAX as f32,
-            std::u32::MAX as f32,
-            std::u32::MAX as f32,
-        ];
-        let f32_array: ArrayRef = Arc::new(Float32Array::from(f32_values));
-
-        let f64_expected = vec![
-            "-2147483648.0",
-            "-2147483648.0",
-            "-32768.0",
-            "-128.0",
-            "0.0",
-            "255.0",
-            "65535.0",
-            "4294967296.0",
-            "4294967296.0",
-        ];
-        assert_eq!(
-            f64_expected,
-            get_cast_values::<f64>(&f32_array, &DataType::Float64)
-        );
-
-        let f32_expected = vec![
-            "-2147483600.0",
-            "-2147483600.0",
-            "-32768.0",
-            "-128.0",
-            "0.0",
-            "255.0",
-            "65535.0",
-            "4294967300.0",
-            "4294967300.0",
-        ];
-        assert_eq!(
-            f32_expected,
-            get_cast_values::<f32>(&f32_array, &DataType::Float32)
-        );
-
-        let i64_expected = vec![
-            "-2147483648",
-            "-2147483648",
-            "-32768",
-            "-128",
-            "0",
-            "255",
-            "65535",
-            "4294967296",
-            "4294967296",
-        ];
-        assert_eq!(
-            i64_expected,
-            get_cast_values::<i64>(&f32_array, &DataType::Int64)
-        );
-
-        let i32_expected = vec![
-            "-2147483648",
-            "-2147483648",
-            "-32768",
-            "-128",
-            "0",
-            "255",
-            "65535",
-            "null",
-            "null",
-        ];
-        assert_eq!(
-            i32_expected,
-            get_cast_values::<i32>(&f32_array, &DataType::Int32)
-        );
-
-        let i16_expected = vec![
-            "null", "null", "-32768", "-128", "0", "255", "null", "null", "null",
-        ];
-        assert_eq!(
-            i16_expected,
-            get_cast_values::<i16>(&f32_array, &DataType::Int16)
-        );
-
-        let i8_expected = vec![
-            "null", "null", "null", "-128", "0", "null", "null", "null", "null",
-        ];
-        assert_eq!(
-            i8_expected,
-            get_cast_values::<i8>(&f32_array, &DataType::Int8)
-        );
-
-        let u64_expected = vec![
-            "null",
-            "null",
-            "null",
-            "null",
-            "0",
-            "255",
-            "65535",
-            "4294967296",
-            "4294967296",
-        ];
-        assert_eq!(
-            u64_expected,
-            get_cast_values::<u64>(&f32_array, &DataType::UInt64)
-        );
-
-        let u32_expected = vec![
-            "null", "null", "null", "null", "0", "255", "65535", "null", "null",
-        ];
-        assert_eq!(
-            u32_expected,
-            get_cast_values::<u32>(&f32_array, &DataType::UInt32)
-        );
-
-        let u16_expected = vec![
-            "null", "null", "null", "null", "0", "255", "65535", "null", "null",
-        ];
-        assert_eq!(
-            u16_expected,
-            get_cast_values::<u16>(&f32_array, &DataType::UInt16)
-        );
-
-        let u8_expected = vec![
-            "null", "null", "null", "null", "0", "255", "null", "null", "null",
-        ];
-        assert_eq!(
-            u8_expected,
-            get_cast_values::<u8>(&f32_array, &DataType::UInt8)
+    fn timestamp_to_date64() {
+        test_primitive_to_primitive(
+            &[864000000005i64, 1545696000001],
+            DataType::Timestamp(TimeUnit::Millisecond, Some(String::from("UTC"))),
+            &[864000000005i64, 1545696000001i64],
+            DataType::Date64,
         );
     }
 
     #[test]
-    fn test_cast_from_uint64() {
-        let u64_values: Vec<u64> = vec![
-            0,
-            std::u8::MAX as u64,
-            std::u16::MAX as u64,
-            std::u32::MAX as u64,
-            std::u64::MAX,
-        ];
-        let u64_array: ArrayRef = Arc::new(UInt64Array::from(u64_values));
-
-        let f64_expected = vec![
-            "0.0",
-            "255.0",
-            "65535.0",
-            "4294967295.0",
-            "18446744073709552000.0",
-        ];
-        assert_eq!(
-            f64_expected,
-            get_cast_values::<f64>(&u64_array, &DataType::Float64)
-        );
-
-        let f32_expected = vec![
-            "0.0",
-            "255.0",
-            "65535.0",
-            "4294967300.0",
-            "18446744000000000000.0",
-        ];
-        assert_eq!(
-            f32_expected,
-            get_cast_values::<f32>(&u64_array, &DataType::Float32)
-        );
-
-        let i64_expected = vec!["0", "255", "65535", "4294967295", "null"];
-        assert_eq!(
-            i64_expected,
-            get_cast_values::<i64>(&u64_array, &DataType::Int64)
-        );
-
-        let i32_expected = vec!["0", "255", "65535", "null", "null"];
-        assert_eq!(
-            i32_expected,
-            get_cast_values::<i32>(&u64_array, &DataType::Int32)
-        );
-
-        let i16_expected = vec!["0", "255", "null", "null", "null"];
-        assert_eq!(
-            i16_expected,
-            get_cast_values::<i16>(&u64_array, &DataType::Int16)
-        );
-
-        let i8_expected = vec!["0", "null", "null", "null", "null"];
-        assert_eq!(
-            i8_expected,
-            get_cast_values::<i8>(&u64_array, &DataType::Int8)
-        );
-
-        let u64_expected =
-            vec!["0", "255", "65535", "4294967295", "18446744073709551615"];
-        assert_eq!(
-            u64_expected,
-            get_cast_values::<u64>(&u64_array, &DataType::UInt64)
-        );
-
-        let u32_expected = vec!["0", "255", "65535", "4294967295", "null"];
-        assert_eq!(
-            u32_expected,
-            get_cast_values::<u32>(&u64_array, &DataType::UInt32)
-        );
-
-        let u16_expected = vec!["0", "255", "65535", "null", "null"];
-        assert_eq!(
-            u16_expected,
-            get_cast_values::<u16>(&u64_array, &DataType::UInt16)
-        );
-
-        let u8_expected = vec!["0", "255", "null", "null", "null"];
-        assert_eq!(
-            u8_expected,
-            get_cast_values::<u8>(&u64_array, &DataType::UInt8)
+    fn timestamp_to_i64() {
+        test_primitive_to_primitive(
+            &[864000000005i64, 1545696000001],
+            DataType::Timestamp(TimeUnit::Millisecond, Some(String::from("UTC"))),
+            &[864000000005i64, 1545696000001i64],
+            DataType::Int64,
         );
     }
 
     #[test]
-    fn test_cast_from_uint32() {
-        let u32_values: Vec<u32> = vec![
-            0,
-            std::u8::MAX as u32,
-            std::u16::MAX as u32,
-            std::u32::MAX as u32,
-        ];
-        let u32_array: ArrayRef = Arc::new(UInt32Array::from(u32_values));
-
-        let f64_expected = vec!["0.0", "255.0", "65535.0", "4294967295.0"];
-        assert_eq!(
-            f64_expected,
-            get_cast_values::<f64>(&u32_array, &DataType::Float64)
-        );
-
-        let f32_expected = vec!["0.0", "255.0", "65535.0", "4294967300.0"];
-        assert_eq!(
-            f32_expected,
-            get_cast_values::<f32>(&u32_array, &DataType::Float32)
-        );
-
-        let i64_expected = vec!["0", "255", "65535", "4294967295"];
-        assert_eq!(
-            i64_expected,
-            get_cast_values::<i64>(&u32_array, &DataType::Int64)
-        );
-
-        let i32_expected = vec!["0", "255", "65535", "null"];
-        assert_eq!(
-            i32_expected,
-            get_cast_values::<i32>(&u32_array, &DataType::Int32)
-        );
-
-        let i16_expected = vec!["0", "255", "null", "null"];
-        assert_eq!(
-            i16_expected,
-            get_cast_values::<i16>(&u32_array, &DataType::Int16)
-        );
-
-        let i8_expected = vec!["0", "null", "null", "null"];
-        assert_eq!(
-            i8_expected,
-            get_cast_values::<i8>(&u32_array, &DataType::Int8)
-        );
-
-        let u64_expected = vec!["0", "255", "65535", "4294967295"];
-        assert_eq!(
-            u64_expected,
-            get_cast_values::<u64>(&u32_array, &DataType::UInt64)
-        );
-
-        let u32_expected = vec!["0", "255", "65535", "4294967295"];
-        assert_eq!(
-            u32_expected,
-            get_cast_values::<u32>(&u32_array, &DataType::UInt32)
-        );
-
-        let u16_expected = vec!["0", "255", "65535", "null"];
-        assert_eq!(
-            u16_expected,
-            get_cast_values::<u16>(&u32_array, &DataType::UInt16)
-        );
-
-        let u8_expected = vec!["0", "255", "null", "null"];
-        assert_eq!(
-            u8_expected,
-            get_cast_values::<u8>(&u32_array, &DataType::UInt8)
-        );
-    }
-
-    #[test]
-    fn test_cast_from_uint16() {
-        let u16_values: Vec<u16> = vec![0, std::u8::MAX as u16, std::u16::MAX as u16];
-        let u16_array: ArrayRef = Arc::new(UInt16Array::from(u16_values));
-
-        let f64_expected = vec!["0.0", "255.0", "65535.0"];
-        assert_eq!(
-            f64_expected,
-            get_cast_values::<f64>(&u16_array, &DataType::Float64)
-        );
-
-        let f32_expected = vec!["0.0", "255.0", "65535.0"];
-        assert_eq!(
-            f32_expected,
-            get_cast_values::<f32>(&u16_array, &DataType::Float32)
-        );
-
-        let i64_expected = vec!["0", "255", "65535"];
-        assert_eq!(
-            i64_expected,
-            get_cast_values::<i64>(&u16_array, &DataType::Int64)
-        );
-
-        let i32_expected = vec!["0", "255", "65535"];
-        assert_eq!(
-            i32_expected,
-            get_cast_values::<i32>(&u16_array, &DataType::Int32)
-        );
-
-        let i16_expected = vec!["0", "255", "null"];
-        assert_eq!(
-            i16_expected,
-            get_cast_values::<i16>(&u16_array, &DataType::Int16)
-        );
-
-        let i8_expected = vec!["0", "null", "null"];
-        assert_eq!(
-            i8_expected,
-            get_cast_values::<i8>(&u16_array, &DataType::Int8)
-        );
-
-        let u64_expected = vec!["0", "255", "65535"];
-        assert_eq!(
-            u64_expected,
-            get_cast_values::<u64>(&u16_array, &DataType::UInt64)
-        );
-
-        let u32_expected = vec!["0", "255", "65535"];
-        assert_eq!(
-            u32_expected,
-            get_cast_values::<u32>(&u16_array, &DataType::UInt32)
-        );
-
-        let u16_expected = vec!["0", "255", "65535"];
-        assert_eq!(
-            u16_expected,
-            get_cast_values::<u16>(&u16_array, &DataType::UInt16)
-        );
-
-        let u8_expected = vec!["0", "255", "null"];
-        assert_eq!(
-            u8_expected,
-            get_cast_values::<u8>(&u16_array, &DataType::UInt8)
+    fn timestamp_to_timestamp() {
+        test_primitive_to_primitive(
+            &[864000003005i64, 1545696002001],
+            DataType::Timestamp(TimeUnit::Millisecond, None),
+            &[864000003i64, 1545696002],
+            DataType::Timestamp(TimeUnit::Second, None),
         );
     }
 
     #[test]
-    fn test_cast_from_uint8() {
-        let u8_values: Vec<u8> = vec![0, std::u8::MAX];
-        let u8_array: ArrayRef = Arc::new(UInt8Array::from(u8_values));
-
-        let f64_expected = vec!["0.0", "255.0"];
-        assert_eq!(
-            f64_expected,
-            get_cast_values::<f64>(&u8_array, &DataType::Float64)
-        );
-
-        let f32_expected = vec!["0.0", "255.0"];
-        assert_eq!(
-            f32_expected,
-            get_cast_values::<f32>(&u8_array, &DataType::Float32)
-        );
-
-        let i64_expected = vec!["0", "255"];
-        assert_eq!(
-            i64_expected,
-            get_cast_values::<i64>(&u8_array, &DataType::Int64)
-        );
-
-        let i32_expected = vec!["0", "255"];
-        assert_eq!(
-            i32_expected,
-            get_cast_values::<i32>(&u8_array, &DataType::Int32)
-        );
-
-        let i16_expected = vec!["0", "255"];
-        assert_eq!(
-            i16_expected,
-            get_cast_values::<i16>(&u8_array, &DataType::Int16)
-        );
-
-        let i8_expected = vec!["0", "null"];
-        assert_eq!(
-            i8_expected,
-            get_cast_values::<i8>(&u8_array, &DataType::Int8)
-        );
+    fn utf8_to_dict() {
+        let array = Utf8Array::<i32>::from(&[Some("one"), None, Some("three"), Some("one")]);
 
-        let u64_expected = vec!["0", "255"];
-        assert_eq!(
-            u64_expected,
-            get_cast_values::<u64>(&u8_array, &DataType::UInt64)
-        );
-
-        let u32_expected = vec!["0", "255"];
-        assert_eq!(
-            u32_expected,
-            get_cast_values::<u32>(&u8_array, &DataType::UInt32)
-        );
-
-        let u16_expected = vec!["0", "255"];
-        assert_eq!(
-            u16_expected,
-            get_cast_values::<u16>(&u8_array, &DataType::UInt16)
-        );
+        // 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 u8_expected = vec!["0", "255"];
-        assert_eq!(
-            u8_expected,
-            get_cast_values::<u8>(&u8_array, &DataType::UInt8)
-        );
+        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 test_cast_from_int64() {
-        let i64_values: Vec<i64> = vec![
-            std::i64::MIN,
-            std::i32::MIN as i64,
-            std::i16::MIN as i64,
-            std::i8::MIN as i64,
-            0,
-            std::i8::MAX as i64,
-            std::i16::MAX as i64,
-            std::i32::MAX as i64,
-            std::i64::MAX,
-        ];
-        let i64_array: ArrayRef = Arc::new(Int64Array::from(i64_values));
-
-        let f64_expected = vec![
-            "-9223372036854776000.0",
-            "-2147483648.0",
-            "-32768.0",
-            "-128.0",
-            "0.0",
-            "127.0",
-            "32767.0",
-            "2147483647.0",
-            "9223372036854776000.0",
-        ];
-        assert_eq!(
-            f64_expected,
-            get_cast_values::<f64>(&i64_array, &DataType::Float64)
-        );
-
-        let f32_expected = vec![
-            "-9223372000000000000.0",
-            "-2147483600.0",
-            "-32768.0",
-            "-128.0",
-            "0.0",
-            "127.0",
-            "32767.0",
-            "2147483600.0",
-            "9223372000000000000.0",
-        ];
-        assert_eq!(
-            f32_expected,
-            get_cast_values::<f32>(&i64_array, &DataType::Float32)
-        );
-
-        let i64_expected = vec![
-            "-9223372036854775808",
-            "-2147483648",
-            "-32768",
-            "-128",
-            "0",
-            "127",
-            "32767",
-            "2147483647",
-            "9223372036854775807",
-        ];
-        assert_eq!(
-            i64_expected,
-            get_cast_values::<i64>(&i64_array, &DataType::Int64)
-        );
-
-        let i32_expected = vec![
-            "null",
-            "-2147483648",
-            "-32768",
-            "-128",
-            "0",
-            "127",
-            "32767",
-            "2147483647",
-            "null",
-        ];
-        assert_eq!(
-            i32_expected,
-            get_cast_values::<i32>(&i64_array, &DataType::Int32)
-        );
-
-        let i16_expected = vec![
-            "null", "null", "-32768", "-128", "0", "127", "32767", "null", "null",
-        ];
-        assert_eq!(
-            i16_expected,
-            get_cast_values::<i16>(&i64_array, &DataType::Int16)
-        );
-
-        let i8_expected = vec![
-            "null", "null", "null", "-128", "0", "127", "null", "null", "null",
-        ];
-        assert_eq!(
-            i8_expected,
-            get_cast_values::<i8>(&i64_array, &DataType::Int8)
-        );
-
-        let u64_expected = vec![
-            "null",
-            "null",
-            "null",
-            "null",
-            "0",
-            "127",
-            "32767",
-            "2147483647",
-            "9223372036854775807",
-        ];
-        assert_eq!(
-            u64_expected,
-            get_cast_values::<u64>(&i64_array, &DataType::UInt64)
-        );
+    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 u32_expected = vec![
-            "null",
-            "null",
-            "null",
-            "null",
-            "0",
-            "127",
-            "32767",
-            "2147483647",
-            "null",
-        ];
-        assert_eq!(
-            u32_expected,
-            get_cast_values::<u32>(&i64_array, &DataType::UInt32)
-        );
+        let result = cast(&array, &DataType::Utf8).expect("cast failed");
 
-        let u16_expected = vec![
-            "null", "null", "null", "null", "0", "127", "32767", "null", "null",
-        ];
-        assert_eq!(
-            u16_expected,
-            get_cast_values::<u16>(&i64_array, &DataType::UInt16)
-        );
+        let expected = Utf8Array::<i32>::from(&[Some("one"), None, Some("three"), Some("one")]);
 
-        let u8_expected = vec![
-            "null", "null", "null", "null", "0", "127", "null", "null", "null",
-        ];
-        assert_eq!(
-            u8_expected,
-            get_cast_values::<u8>(&i64_array, &DataType::UInt8)
-        );
+        assert_eq!(expected, result.as_ref());
     }
 
     #[test]
-    fn test_cast_from_int32() {
-        let i32_values: Vec<i32> = vec![
-            std::i32::MIN as i32,
-            std::i16::MIN as i32,
-            std::i8::MIN as i32,
-            0,
-            std::i8::MAX as i32,
-            std::i16::MAX as i32,
-            std::i32::MAX as i32,
-        ];
-        let i32_array: ArrayRef = Arc::new(Int32Array::from(i32_values));
-
-        let f64_expected = vec![
-            "-2147483648.0",
-            "-32768.0",
-            "-128.0",
-            "0.0",
-            "127.0",
-            "32767.0",
-            "2147483647.0",
-        ];
-        assert_eq!(
-            f64_expected,
-            get_cast_values::<f64>(&i32_array, &DataType::Float64)
-        );
-
-        let f32_expected = vec![
-            "-2147483600.0",
-            "-32768.0",
-            "-128.0",
-            "0.0",
-            "127.0",
-            "32767.0",
-            "2147483600.0",
-        ];
-        assert_eq!(
-            f32_expected,
-            get_cast_values::<f32>(&i32_array, &DataType::Float32)
-        );
-
-        let i16_expected = vec!["null", "-32768", "-128", "0", "127", "32767", "null"];
-        assert_eq!(
-            i16_expected,
-            get_cast_values::<i16>(&i32_array, &DataType::Int16)
-        );
-
-        let i8_expected = vec!["null", "null", "-128", "0", "127", "null", "null"];
-        assert_eq!(
-            i8_expected,
-            get_cast_values::<i8>(&i32_array, &DataType::Int8)
-        );
-
-        let u64_expected =
-            vec!["null", "null", "null", "0", "127", "32767", "2147483647"];
-        assert_eq!(
-            u64_expected,
-            get_cast_values::<u64>(&i32_array, &DataType::UInt64)
-        );
-
-        let u32_expected =
-            vec!["null", "null", "null", "0", "127", "32767", "2147483647"];
-        assert_eq!(
-            u32_expected,
-            get_cast_values::<u32>(&i32_array, &DataType::UInt32)
-        );
+    fn i32_to_dict() {
+        let array = Int32Array::from(&[Some(1), None, Some(3), Some(1)]);
 
-        let u16_expected = vec!["null", "null", "null", "0", "127", "32767", "null"];
-        assert_eq!(
-            u16_expected,
-            get_cast_values::<u16>(&i32_array, &DataType::UInt16)
-        );
+        // 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 u8_expected = vec!["null", "null", "null", "0", "127", "null", "null"];
-        assert_eq!(
-            u8_expected,
-            get_cast_values::<u8>(&i32_array, &DataType::UInt8)
-        );
+        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 test_cast_from_int16() {
-        let i16_values: Vec<i16> = vec![
-            std::i16::MIN,
-            std::i8::MIN as i16,
-            0,
-            std::i8::MAX as i16,
-            std::i16::MAX,
+    fn list_to_list() {
+        let data = vec![
+            Some(vec![Some(1i32), Some(2), Some(3)]),
+            None,
+            Some(vec![Some(4), None, Some(6)]),
         ];
-        let i16_array: ArrayRef = Arc::new(Int16Array::from(i16_values));
-
-        let f64_expected = vec!["-32768.0", "-128.0", "0.0", "127.0", "32767.0"];
-        assert_eq!(
-            f64_expected,
-            get_cast_values::<f64>(&i16_array, &DataType::Float64)
-        );
-
-        let f32_expected = vec!["-32768.0", "-128.0", "0.0", "127.0", "32767.0"];
-        assert_eq!(
-            f32_expected,
-            get_cast_values::<f32>(&i16_array, &DataType::Float32)
-        );
-
-        let i64_expected = vec!["-32768", "-128", "0", "127", "32767"];
-        assert_eq!(
-            i64_expected,
-            get_cast_values::<i64>(&i16_array, &DataType::Int64)
-        );
-
-        let i32_expected = vec!["-32768", "-128", "0", "127", "32767"];
-        assert_eq!(
-            i32_expected,
-            get_cast_values::<i32>(&i16_array, &DataType::Int32)
-        );
-
-        let i16_expected = vec!["-32768", "-128", "0", "127", "32767"];
-        assert_eq!(
-            i16_expected,
-            get_cast_values::<i16>(&i16_array, &DataType::Int16)
-        );
-
-        let i8_expected = vec!["null", "-128", "0", "127", "null"];
-        assert_eq!(
-            i8_expected,
-            get_cast_values::<i8>(&i16_array, &DataType::Int8)
-        );
 
-        let u64_expected = vec!["null", "null", "0", "127", "32767"];
-        assert_eq!(
-            u64_expected,
-            get_cast_values::<u64>(&i16_array, &DataType::UInt64)
-        );
+        let expected_data = data
+            .iter()
+            .map(|x| x.as_ref().map(|x| x.iter().map(|x| x.map(|x| x as u16))));
 
-        let u32_expected = vec!["null", "null", "0", "127", "32767"];
-        assert_eq!(
-            u32_expected,
-            get_cast_values::<u32>(&i16_array, &DataType::UInt32)
-        );
+        let mut array = MutableListArray::<i32, MutablePrimitiveArray<i32>>::new();
+        array.try_extend(data.clone()).unwrap();
+        let array: ListArray<i32> = array.into();
 
-        let u16_expected = vec!["null", "null", "0", "127", "32767"];
-        assert_eq!(
-            u16_expected,
-            get_cast_values::<u16>(&i16_array, &DataType::UInt16)
-        );
+        let mut expected = MutableListArray::<i32, MutablePrimitiveArray<u16>>::new();
+        expected.try_extend(expected_data).unwrap();
+        let expected: ListArray<i32> = expected.into();
 
-        let u8_expected = vec!["null", "null", "0", "127", "null"];
-        assert_eq!(
-            u8_expected,
-            get_cast_values::<u8>(&i16_array, &DataType::UInt8)
-        );
+        let result = cast(&array, expected.data_type()).unwrap();
+        assert_eq!(expected, result.as_ref());
     }
 
+    /*
     #[test]
-    fn test_cast_from_int8() {
-        let i8_values: Vec<i8> = vec![std::i8::MIN, 0, std::i8::MAX];
-        let i8_array: ArrayRef = Arc::new(Int8Array::from(i8_values));
-
-        let f64_expected = vec!["-128.0", "0.0", "127.0"];
-        assert_eq!(
-            f64_expected,
-            get_cast_values::<f64>(&i8_array, &DataType::Float64)
-        );
-
-        let f32_expected = vec!["-128.0", "0.0", "127.0"];
-        assert_eq!(
-            f32_expected,
-            get_cast_values::<f32>(&i8_array, &DataType::Float32)
-        );
-
-        let i64_expected = vec!["-128", "0", "127"];
-        assert_eq!(
-            i64_expected,
-            get_cast_values::<i64>(&i8_array, &DataType::Int64)
-        );
-
-        let i32_expected = vec!["-128", "0", "127"];
-        assert_eq!(
-            i32_expected,
-            get_cast_values::<i32>(&i8_array, &DataType::Int32)
-        );
-
-        let i16_expected = vec!["-128", "0", "127"];
-        assert_eq!(
-            i16_expected,
-            get_cast_values::<i16>(&i8_array, &DataType::Int16)
-        );
-
-        let i8_expected = vec!["-128", "0", "127"];
-        assert_eq!(
-            i8_expected,
-            get_cast_values::<i8>(&i8_array, &DataType::Int8)
-        );
-
-        let u64_expected = vec!["null", "0", "127"];
-        assert_eq!(
-            u64_expected,
-            get_cast_values::<u64>(&i8_array, &DataType::UInt64)
-        );
-
-        let u32_expected = vec!["null", "0", "127"];
-        assert_eq!(
-            u32_expected,
-            get_cast_values::<u32>(&i8_array, &DataType::UInt32)
-        );
-
-        let u16_expected = vec!["null", "0", "127"];
-        assert_eq!(
-            u16_expected,
-            get_cast_values::<u16>(&i8_array, &DataType::UInt16)
-        );
-
-        let u8_expected = vec!["null", "0", "127"];
-        assert_eq!(
-            u8_expected,
-            get_cast_values::<u8>(&i8_array, &DataType::UInt8)
-        );
-    }
-
-    /// Convert `array` into a vector of strings by casting to data type dt
-    fn get_cast_values<T>(array: &ArrayRef, dt: &DataType) -> Vec<String>
-    where
-        T: ArrowNumericType,
-    {
-        let c = cast(&array, dt).unwrap();
-        let a = c.as_any().downcast_ref::<PrimitiveArray<T>>().unwrap();
-        let mut v: Vec<String> = vec![];
-        for i in 0..array.len() {
-            if a.is_null(i) {
-                v.push("null".to_string())
-            } else {
-                v.push(format!("{:?}", a.value(i)));
-            }
-        }
-        v
-    }
-
-    #[test]
-    fn test_cast_utf8_dict() {
-        // FROM a dictionary with of Utf8 values
-        use DataType::*;
-
-        let keys_builder = PrimitiveBuilder::<i8>::new(10);
-        let values_builder = StringBuilder::new(10);
-        let mut builder = StringDictionaryBuilder::new(keys_builder, values_builder);
-        builder.append("one").unwrap();
-        builder.append_null().unwrap();
-        builder.append("three").unwrap();
-        let array: ArrayRef = Arc::new(builder.finish());
-
-        let expected = vec!["one", "null", "three"];
-
-        // Test casting TO StringArray
-        let cast_type = Utf8;
-        let cast_array = cast(&array, &cast_type).expect("cast to UTF-8 failed");
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(array_to_strings(&cast_array), expected);
-
-        // Test casting TO Dictionary (with different index sizes)
-
-        let cast_type = Dictionary(Box::new(Int16), Box::new(Utf8));
-        let cast_array = cast(&array, &cast_type).expect("cast failed");
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(array_to_strings(&cast_array), expected);
-
-        let cast_type = Dictionary(Box::new(Int32), Box::new(Utf8));
-        let cast_array = cast(&array, &cast_type).expect("cast failed");
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(array_to_strings(&cast_array), expected);
-
-        let cast_type = Dictionary(Box::new(Int64), Box::new(Utf8));
-        let cast_array = cast(&array, &cast_type).expect("cast failed");
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(array_to_strings(&cast_array), expected);
-
-        let cast_type = Dictionary(Box::new(UInt8), Box::new(Utf8));
-        let cast_array = cast(&array, &cast_type).expect("cast failed");
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(array_to_strings(&cast_array), expected);
-
-        let cast_type = Dictionary(Box::new(UInt16), Box::new(Utf8));
-        let cast_array = cast(&array, &cast_type).expect("cast failed");
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(array_to_strings(&cast_array), expected);
-
-        let cast_type = Dictionary(Box::new(UInt32), Box::new(Utf8));
-        let cast_array = cast(&array, &cast_type).expect("cast failed");
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(array_to_strings(&cast_array), expected);
-
-        let cast_type = Dictionary(Box::new(UInt64), Box::new(Utf8));
-        let cast_array = cast(&array, &cast_type).expect("cast failed");
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(array_to_strings(&cast_array), expected);
-    }
-
-    #[test]
-    fn test_cast_dict_to_dict_bad_index_value_primitive() {
+    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
@@ -2117,8 +1279,8 @@ mod tests {
         let array: ArrayRef = Arc::new(builder.finish());
 
         let cast_type = Dictionary(Box::new(Int8), Box::new(Utf8));
-        let res = cast(&array, &cast_type);
-        assert!(res.is_err());
+        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!(
@@ -2130,9 +1292,9 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_dict_to_dict_bad_index_value_utf8() {
+    fn dict_to_dict_bad_index_value_utf8() {
         use DataType::*;
-        // Same test as test_cast_dict_to_dict_bad_index_value but use
+        // 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);
@@ -2149,8 +1311,8 @@ mod tests {
         let array: ArrayRef = Arc::new(builder.finish());
 
         let cast_type = Dictionary(Box::new(Int8), Box::new(Utf8));
-        let res = cast(&array, &cast_type);
-        assert!(res.is_err());
+        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!(
@@ -2162,100 +1324,7 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_primitive_dict() {
-        // FROM a dictionary with of INT32 values
-        use DataType::*;
-
-        let keys_builder = PrimitiveBuilder::<i8>::new(10);
-        let values_builder = PrimitiveBuilder::<i32>::new(10);
-        let mut builder = PrimitiveDictionaryBuilder::new(keys_builder, values_builder);
-        builder.append(1).unwrap();
-        builder.append_null().unwrap();
-        builder.append(3).unwrap();
-        let array: ArrayRef = Arc::new(builder.finish());
-
-        let expected = vec!["1", "null", "3"];
-
-        // Test casting TO PrimitiveArray, different dictionary type
-        let cast_array = cast(&array, &Utf8).expect("cast to UTF-8 failed");
-        assert_eq!(array_to_strings(&cast_array), expected);
-        assert_eq!(cast_array.data_type(), &Utf8);
-
-        let cast_array = cast(&array, &Int64).expect("cast to int64 failed");
-        assert_eq!(array_to_strings(&cast_array), expected);
-        assert_eq!(cast_array.data_type(), &Int64);
-    }
-
-    #[test]
-    fn test_cast_primitive_array_to_dict() {
-        use DataType::*;
-
-        let mut builder = PrimitiveBuilder::<i32>::new(10);
-        builder.append_value(1).unwrap();
-        builder.append_null().unwrap();
-        builder.append_value(3).unwrap();
-        let array: ArrayRef = Arc::new(builder.finish());
-
-        let expected = vec!["1", "null", "3"];
-
-        // Cast to a dictionary (same value type, Int32)
-        let cast_type = Dictionary(Box::new(UInt8), Box::new(Int32));
-        let cast_array = cast(&array, &cast_type).expect("cast failed");
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(array_to_strings(&cast_array), expected);
-
-        // Cast to a dictionary (different value type, Int8)
-        let cast_type = Dictionary(Box::new(UInt8), Box::new(Int8));
-        let cast_array = cast(&array, &cast_type).expect("cast failed");
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(array_to_strings(&cast_array), expected);
-    }
-
-    #[test]
-    fn test_cast_string_array_to_dict() {
-        use DataType::*;
-
-        let array = Arc::new(StringArray::from(vec![Some("one"), None, Some("three")]))
-            as ArrayRef;
-
-        let expected = vec!["one", "null", "three"];
-
-        // Cast to a dictionary (same value type, Utf8)
-        let cast_type = Dictionary(Box::new(UInt8), Box::new(Utf8));
-        let cast_array = cast(&array, &cast_type).expect("cast failed");
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(array_to_strings(&cast_array), expected);
-    }
-
-    #[test]
-    fn test_cast_null_array_to_int32() {
-        let array = Arc::new(NullArray::new(6)) as ArrayRef;
-
-        let expected = Int32Array::from(vec![None; 6]);
-
-        // Cast to a dictionary (same value type, Utf8)
-        let cast_type = DataType::Int32;
-        let cast_array = cast(&array, &cast_type).expect("cast failed");
-        let cast_array = as_primitive_array::<i32>(&cast_array);
-        assert_eq!(cast_array.data_type(), &cast_type);
-        assert_eq!(cast_array, &expected);
-    }
-
-    /// Print the `DictionaryArray` `array` as a vector of strings
-    fn array_to_strings(array: &ArrayRef) -> Vec<String> {
-        (0..array.len())
-            .map(|i| {
-                if array.is_null(i) {
-                    "null".to_string()
-                } else {
-                    array_value_to_string(array, i).expect("Convert array to String")
-                }
-            })
-            .collect()
-    }
-
-    #[test]
-    fn test_cast_utf8_to_date32() {
+    fn utf8_to_date32() {
         use chrono::NaiveDate;
         let from_ymd = chrono::NaiveDate::from_ymd;
         let since = chrono::NaiveDate::signed_duration_since;
@@ -2268,7 +1337,7 @@ mod tests {
             "2000",                // just a year is invalid
         ]);
         let array = Arc::new(a) as ArrayRef;
-        let b = cast(&array, &DataType::Date32).unwrap();
+        let b = cast_with_options(&array, &DataType::Date32, CastOptions::default()).unwrap();
         let c = b.as_any().downcast_ref::<Date32Array>().unwrap();
 
         // test valid inputs
@@ -2289,7 +1358,7 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_utf8_to_date64() {
+    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
@@ -2299,7 +1368,7 @@ mod tests {
             "2000-01-01",          // just a date is invalid
         ]);
         let array = Arc::new(a) as ArrayRef;
-        let b = cast(&array, &DataType::Date64).unwrap();
+        let b = cast_with_options(&array, &DataType::Date64, CastOptions::default()).unwrap();
         let c = b.as_any().downcast_ref::<Date64Array>().unwrap();
 
         // test valid inputs
@@ -2316,291 +1385,11 @@ mod tests {
         assert_eq!(false, c.is_valid(5)); // "2000-01-01"
     }
 
-    #[test]
-    fn test_can_cast_types() {
-        // this function attempts to ensure that can_cast_types stays
-        // in sync with cast.  It simply tries all combinations of
-        // types and makes sure that if `can_cast_types` returns
-        // true, so does `cast`
-
-        let all_types = get_all_types();
-
-        for array in get_arrays_of_all_types() {
-            for to_type in &all_types {
-                println!("Test casting {:?} --> {:?}", array.data_type(), to_type);
-                let cast_result = cast(&array, &to_type);
-                let reported_cast_ability = can_cast_types(array.data_type(), to_type);
-
-                // check for mismatch
-                match (cast_result, reported_cast_ability) {
-                    (Ok(_), false) => {
-                        panic!("Was able to cast array from {:?} to {:?} but can_cast_types reported false",
-                               array.data_type(), to_type)
-                    }
-                    (Err(e), true) => {
-                        panic!("Was not able to cast array from {:?} to {:?} but can_cast_types reported true. \
-                                Error was {:?}",
-                               array.data_type(), to_type, e)
-                    }
-                    // otherwise it was a match
-                    _ => {}
-                };
-            }
-        }
-    }
-
-    /// Create instances of arrays with varying types for cast tests
-    fn get_arrays_of_all_types() -> Vec<ArrayRef> {
-        let tz_name = String::from("America/New_York");
-        let binary_data: Vec<&[u8]> = vec![b"foo", b"bar"];
-        vec![
-            Arc::new(BinaryArray::from(binary_data.clone())),
-            Arc::new(LargeBinaryArray::from(binary_data.clone())),
-            make_dictionary_primitive_to::<i8>(),
-            make_dictionary_primitive_to::<i16>(),
-            make_dictionary_primitive_to::<i32>(),
-            make_dictionary_primitive_to::<i64>(),
-            make_dictionary_primitive_to::<u8>(),
-            make_dictionary_primitive_to::<u16>(),
-            make_dictionary_primitive_to::<u32>(),
-            make_dictionary_primitive_to::<u64>(),
-            make_dictionary_utf8::<i8>(),
-            make_dictionary_utf8::<i16>(),
-            make_dictionary_utf8::<i32>(),
-            make_dictionary_utf8::<i64>(),
-            make_dictionary_utf8::<u8>(),
-            make_dictionary_utf8::<u16>(),
-            make_dictionary_utf8::<u32>(),
-            make_dictionary_utf8::<u64>(),
-            Arc::new(make_list_array()),
-            Arc::new(make_large_list_array()),
-            Arc::new(make_fixed_size_list_array()),
-            Arc::new(make_fixed_size_binary_array()),
-            Arc::new(StructArray::from(vec![
-                (
-                    Field::new("a", DataType::Boolean, false),
-                    Arc::new(BooleanArray::from(vec![false, false, true, true]))
-                        as Arc<Array>,
-                ),
-                (
-                    Field::new("b", DataType::Int32, false),
-                    Arc::new(Int32Array::from(vec![42, 28, 19, 31])),
-                ),
-            ])),
-            //Arc::new(make_union_array()),
-            Arc::new(NullArray::new(10)),
-            Arc::new(StringArray::from(vec!["foo", "bar"])),
-            Arc::new(LargeStringArray::from(vec!["foo", "bar"])),
-            Arc::new(BooleanArray::from(vec![true, false])),
-            Arc::new(Int8Array::from(vec![1, 2])),
-            Arc::new(Int16Array::from(vec![1, 2])),
-            Arc::new(Int32Array::from(vec![1, 2])),
-            Arc::new(Int64Array::from(vec![1, 2])),
-            Arc::new(UInt8Array::from(vec![1, 2])),
-            Arc::new(UInt16Array::from(vec![1, 2])),
-            Arc::new(UInt32Array::from(vec![1, 2])),
-            Arc::new(UInt64Array::from(vec![1, 2])),
-            Arc::new(Float32Array::from(vec![1.0, 2.0])),
-            Arc::new(Float64Array::from(vec![1.0, 2.0])),
-            Arc::new(TimestampSecondArray::from_vec(vec![1000, 2000], None)),
-            Arc::new(TimestampMillisecondArray::from_vec(vec![1000, 2000], None)),
-            Arc::new(TimestampMicrosecondArray::from_vec(vec![1000, 2000], None)),
-            Arc::new(TimestampNanosecondArray::from_vec(vec![1000, 2000], None)),
-            Arc::new(TimestampSecondArray::from_vec(
-                vec![1000, 2000],
-                Some(tz_name.clone()),
-            )),
-            Arc::new(TimestampMillisecondArray::from_vec(
-                vec![1000, 2000],
-                Some(tz_name.clone()),
-            )),
-            Arc::new(TimestampMicrosecondArray::from_vec(
-                vec![1000, 2000],
-                Some(tz_name.clone()),
-            )),
-            Arc::new(TimestampNanosecondArray::from_vec(
-                vec![1000, 2000],
-                Some(tz_name),
-            )),
-            Arc::new(Date32Array::from(vec![1000, 2000])),
-            Arc::new(Date64Array::from(vec![1000, 2000])),
-            Arc::new(Time32SecondArray::from(vec![1000, 2000])),
-            Arc::new(Time32MillisecondArray::from(vec![1000, 2000])),
-            Arc::new(Time64MicrosecondArray::from(vec![1000, 2000])),
-            Arc::new(Time64NanosecondArray::from(vec![1000, 2000])),
-            Arc::new(IntervalYearMonthArray::from(vec![1000, 2000])),
-            Arc::new(IntervalDayTimeArray::from(vec![1000, 2000])),
-            Arc::new(DurationSecondArray::from(vec![1000, 2000])),
-            Arc::new(DurationMillisecondArray::from(vec![1000, 2000])),
-            Arc::new(DurationMicrosecondArray::from(vec![1000, 2000])),
-            Arc::new(DurationNanosecondArray::from(vec![1000, 2000])),
-        ]
-    }
-
-    fn make_list_array() -> ListArray {
-        // Construct a value array
-        let value_data = ArrayData::builder(DataType::Int32)
-            .len(8)
-            .add_buffer(Buffer::from_slice_ref(&[0, 1, 2, 3, 4, 5, 6, 7]))
-            .build();
-
-        // Construct a buffer for value offsets, for the nested array:
-        //  [[0, 1, 2], [3, 4, 5], [6, 7]]
-        let value_offsets = Buffer::from_slice_ref(&[0, 3, 6, 8]);
-
-        // Construct a list array from the above two
-        let list_data_type =
-            DataType::List(Box::new(Field::new("item", DataType::Int32, true)));
-        let list_data = ArrayData::builder(list_data_type)
-            .len(3)
-            .add_buffer(value_offsets)
-            .add_child_data(value_data)
-            .build();
-        ListArray::from(list_data)
-    }
-
-    fn make_large_list_array() -> LargeListArray {
-        // Construct a value array
-        let value_data = ArrayData::builder(DataType::Int32)
-            .len(8)
-            .add_buffer(Buffer::from_slice_ref(&[0, 1, 2, 3, 4, 5, 6, 7]))
-            .build();
-
-        // Construct a buffer for value offsets, for the nested array:
-        //  [[0, 1, 2], [3, 4, 5], [6, 7]]
-        let value_offsets = Buffer::from_slice_ref(&[0i64, 3, 6, 8]);
-
-        // Construct a list array from the above two
-        let list_data_type =
-            DataType::LargeList(Box::new(Field::new("item", DataType::Int32, true)));
-        let list_data = ArrayData::builder(list_data_type)
-            .len(3)
-            .add_buffer(value_offsets)
-            .add_child_data(value_data)
-            .build();
-        LargeListArray::from(list_data)
-    }
-
-    fn make_fixed_size_list_array() -> FixedSizeListArray {
-        // Construct a value array
-        let value_data = ArrayData::builder(DataType::Int32)
-            .len(10)
-            .add_buffer(Buffer::from_slice_ref(&[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]))
-            .build();
-
-        // Construct a fixed size list array from the above two
-        let list_data_type = DataType::FixedSizeList(
-            Box::new(Field::new("item", DataType::Int32, true)),
-            2,
-        );
-        let list_data = ArrayData::builder(list_data_type)
-            .len(5)
-            .add_child_data(value_data)
-            .build();
-        FixedSizeListArray::from(list_data)
-    }
-
-    fn make_fixed_size_binary_array() -> FixedSizeBinaryArray {
-        let values: [u8; 15] = *b"hellotherearrow";
-
-        let array_data = ArrayData::builder(DataType::FixedSizeBinary(5))
-            .len(3)
-            .add_buffer(Buffer::from(&values[..]))
-            .build();
-        FixedSizeBinaryArray::from(array_data)
-    }
-
     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()
     }
-
-    /// Creates a dictionary with primitive dictionary values, and keys of type K
-    fn make_dictionary_primitive<K: ArrowDictionaryKeyType>() -> ArrayRef {
-        let keys_builder = PrimitiveBuilder::<K>::new(2);
-        // Pick Int32 arbitrarily for dictionary values
-        let values_builder = PrimitiveBuilder::<i32>::new(2);
-        let mut b = PrimitiveDictionaryBuilder::new(keys_builder, values_builder);
-        b.append(1).unwrap();
-        b.append(2).unwrap();
-        Arc::new(b.finish())
-    }
-
-    /// Creates a dictionary with utf8 values, and keys of type K
-    fn make_dictionary_utf8<K: ArrowDictionaryKeyType>() -> ArrayRef {
-        let keys_builder = PrimitiveBuilder::<K>::new(2);
-        // Pick Int32 arbitrarily for dictionary values
-        let values_builder = StringBuilder::new(2);
-        let mut b = StringDictionaryBuilder::new(keys_builder, values_builder);
-        b.append("foo").unwrap();
-        b.append("bar").unwrap();
-        Arc::new(b.finish())
-    }
-
-    // Get a selection of datatypes to try and cast to
-    fn get_all_types() -> Vec<DataType> {
-        use DataType::*;
-        let tz_name = String::from("America/New_York");
-
-        vec![
-            Null,
-            Boolean,
-            Int8,
-            Int16,
-            Int32,
-            UInt64,
-            UInt8,
-            UInt16,
-            UInt32,
-            UInt64,
-            Float16,
-            Float32,
-            Float64,
-            Timestamp(TimeUnit::Second, None),
-            Timestamp(TimeUnit::Millisecond, None),
-            Timestamp(TimeUnit::Microsecond, None),
-            Timestamp(TimeUnit::Nanosecond, None),
-            Timestamp(TimeUnit::Second, Some(tz_name.clone())),
-            Timestamp(TimeUnit::Millisecond, Some(tz_name.clone())),
-            Timestamp(TimeUnit::Microsecond, Some(tz_name.clone())),
-            Timestamp(TimeUnit::Nanosecond, Some(tz_name)),
-            Date32,
-            Date64,
-            Time32(TimeUnit::Second),
-            Time32(TimeUnit::Millisecond),
-            Time64(TimeUnit::Microsecond),
-            Time64(TimeUnit::Nanosecond),
-            Duration(TimeUnit::Second),
-            Duration(TimeUnit::Millisecond),
-            Duration(TimeUnit::Microsecond),
-            Duration(TimeUnit::Nanosecond),
-            Interval(IntervalUnit::YearMonth),
-            Interval(IntervalUnit::DayTime),
-            Binary,
-            FixedSizeBinary(10),
-            LargeBinary,
-            Utf8,
-            LargeUtf8,
-            List(Box::new(Field::new("item", DataType::Int8, true))),
-            List(Box::new(Field::new("item", DataType::Utf8, true))),
-            FixedSizeList(Box::new(Field::new("item", DataType::Int8, true)), 10),
-            FixedSizeList(Box::new(Field::new("item", DataType::Utf8, false)), 10),
-            LargeList(Box::new(Field::new("item", DataType::Int8, true))),
-            LargeList(Box::new(Field::new("item", DataType::Utf8, false))),
-            Struct(vec![
-                Field::new("f1", DataType::Int32, false),
-                Field::new("f2", DataType::Utf8, true),
-            ]),
-            Union(vec![
-                Field::new("f1", DataType::Int32, false),
-                Field::new("f2", DataType::Utf8, true),
-            ]),
-            Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
-            Dictionary(Box::new(DataType::Int16), Box::new(DataType::Utf8)),
-            Dictionary(Box::new(DataType::UInt32), Box::new(DataType::Utf8)),
-        ]
-    }
     */
 }
diff --git a/src/compute/cast/primitive_to.rs b/src/compute/cast/primitive_to.rs
index dd34d93461..6b6bb90f3f 100644
--- a/src/compute/cast/primitive_to.rs
+++ b/src/compute/cast/primitive_to.rs
@@ -10,6 +10,8 @@ use crate::{
 };
 use crate::{error::Result, util::lexical_to_string};
 
+use super::CastOptions;
+
 /// Returns a [`BooleanArray`] where every element is different from zero.
 /// Validity is preserved.
 pub fn primitive_to_boolean<T: NativeType>(from: &PrimitiveArray<T>) -> BooleanArray {
@@ -48,13 +50,18 @@ where
 pub(super) fn primitive_to_primitive_dyn<I, O>(
     from: &dyn Array,
     to_type: &DataType,
+    options: CastOptions,
 ) -> Result<Box<dyn Array>>
 where
-    I: NativeType + num::NumCast,
+    I: NativeType + num::NumCast + num::traits::AsPrimitive<O>,
     O: NativeType + num::NumCast,
 {
     let from = from.as_any().downcast_ref::<PrimitiveArray<I>>().unwrap();
-    Ok(Box::new(primitive_to_primitive::<I, O>(from, to_type)))
+    if options.wrapped {
+        Ok(Box::new(primitive_as_primitive::<I, O>(from, to_type)))
+    } else {
+        Ok(Box::new(primitive_to_primitive::<I, O>(from, to_type)))
+    }
 }
 
 /// Cast [`PrimitiveArray`] to a [`PrimitiveArray`] of another physical type via numeric conversion.
@@ -72,6 +79,19 @@ where
     PrimitiveArray::<O>::from_trusted_len_iter(iter).to(to_type.clone())
 }
 
+/// Cast [`PrimitiveArray`] as a [`PrimitiveArray`]
+/// Same as `number as to_number_type` in rust
+pub fn primitive_as_primitive<I, O>(
+    from: &PrimitiveArray<I>,
+    to_type: &DataType,
+) -> PrimitiveArray<O>
+where
+    I: NativeType + num::traits::AsPrimitive<O>,
+    O: NativeType,
+{
+    unary(from, num::traits::AsPrimitive::<O>::as_, to_type.clone())
+}
+
 /// Cast [`PrimitiveArray`] to a [`PrimitiveArray`] of the same physical type.
 /// This is O(1).
 pub fn primitive_to_same_primitive<T>(
diff --git a/src/compute/merge_sort/mod.rs b/src/compute/merge_sort/mod.rs
index e3c2d4fedd..3f4736a24a 100644
--- a/src/compute/merge_sort/mod.rs
+++ b/src/compute/merge_sort/mod.rs
@@ -291,6 +291,29 @@ where
             None => self.right = None,
         }
     }
+
+    /// Collect the MergeSortSlices to be a vec for reusing
+    #[warn(dead_code)]
+    pub fn to_vec(self, limit: Option<usize>) -> Vec<MergeSlice> {
+        match limit {
+            Some(limit) => {
+                let mut v = Vec::with_capacity(limit);
+                let mut current_len = 0;
+                for (index, start, len) in self {
+                    if len + current_len >= limit {
+                        v.push((index, start, limit - current_len));
+                        break;
+                    } else {
+                        v.push((index, start, len));
+                    }
+                    current_len += len;
+                }
+
+                v
+            }
+            None => self.into_iter().collect(),
+        }
+    }
 }
 
 impl<'a, L, R> Iterator for MergeSortSlices<'a, L, R>
@@ -562,6 +585,22 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_merge_slices_to_vec() -> Result<()> {
+        let a0: &dyn Array = &Int32Array::from_slice(&[0, 2, 4, 6, 8]);
+        let a1: &dyn Array = &Int32Array::from_slice(&[1, 3, 5, 7, 9]);
+
+        let options = SortOptions::default();
+        let arrays = vec![a0, a1];
+        let pairs = vec![(arrays.as_ref(), &options)];
+        let comparator = build_comparator(&pairs)?;
+
+        let slices = merge_sort_slices(once(&(0, 0, 5)), once(&(1, 0, 5)), &comparator);
+        let vec = slices.to_vec(Some(5));
+        assert_eq!(vec, [(0, 0, 1), (1, 0, 1), (0, 1, 1), (1, 1, 1), (0, 2, 1)]);
+        Ok(())
+    }
+
     #[test]
     fn test_merge_4_i32() -> Result<()> {
         let a0: &dyn Array = &Int32Array::from_slice(&[0, 1]);
diff --git a/src/compute/sort/boolean.rs b/src/compute/sort/boolean.rs
index 2e7082a590..79ad428576 100644
--- a/src/compute/sort/boolean.rs
+++ b/src/compute/sort/boolean.rs
@@ -1,6 +1,7 @@
 use crate::{
-    array::{Array, BooleanArray, Index, PrimitiveArray},
+    array::{Array, BooleanArray, PrimitiveArray},
     buffer::MutableBuffer,
+    types::Index,
 };
 
 use super::SortOptions;
@@ -26,7 +27,7 @@ pub fn sort_boolean<I: Index>(
     if !descending {
         valids.sort_by(|a, b| a.1.cmp(&b.1));
     } else {
-        valids.sort_by(|a, b| a.1.cmp(&b.1).reverse());
+        valids.sort_by(|a, b| b.1.cmp(&a.1));
         // reverse to keep a stable ordering
         nulls.reverse();
     }
@@ -45,6 +46,7 @@ pub fn sort_boolean<I: Index>(
     // un-efficient; there are much more performant ways of sorting nulls above, anyways.
     if let Some(limit) = limit {
         values.truncate(limit);
+        values.shrink_to_fit();
     }
 
     PrimitiveArray::<I>::from_data(I::DATA_TYPE, values.into(), None)
diff --git a/src/compute/sort/common.rs b/src/compute/sort/common.rs
index c36763831b..fc3ea0c667 100644
--- a/src/compute/sort/common.rs
+++ b/src/compute/sort/common.rs
@@ -1,8 +1,4 @@
-use crate::{
-    array::{Index, PrimitiveArray},
-    bitmap::Bitmap,
-    buffer::MutableBuffer,
-};
+use crate::{array::PrimitiveArray, bitmap::Bitmap, buffer::MutableBuffer, types::Index};
 
 use super::SortOptions;
 
@@ -22,31 +18,21 @@ fn k_element_sort_inner<I: Index, T, G, F>(
     F: FnMut(&T, &T) -> std::cmp::Ordering,
 {
     if descending {
-        let compare = |lhs: &I, rhs: &I| {
+        let mut compare = |lhs: &I, rhs: &I| {
             let lhs = get(lhs.to_usize());
             let rhs = get(rhs.to_usize());
-            cmp(&lhs, &rhs).reverse()
+            cmp(&rhs, &lhs)
         };
-        let (before, _, _) = indices.select_nth_unstable_by(limit, compare);
-        let compare = |lhs: &I, rhs: &I| {
-            let lhs = get(lhs.to_usize());
-            let rhs = get(rhs.to_usize());
-            cmp(&lhs, &rhs).reverse()
-        };
-        before.sort_unstable_by(compare);
+        let (before, _, _) = indices.select_nth_unstable_by(limit, &mut compare);
+        before.sort_unstable_by(&mut compare);
     } else {
-        let compare = |lhs: &I, rhs: &I| {
-            let lhs = get(lhs.to_usize());
-            let rhs = get(rhs.to_usize());
-            cmp(&lhs, &rhs)
-        };
-        let (before, _, _) = indices.select_nth_unstable_by(limit, compare);
-        let compare = |lhs: &I, rhs: &I| {
+        let mut compare = |lhs: &I, rhs: &I| {
             let lhs = get(lhs.to_usize());
             let rhs = get(rhs.to_usize());
             cmp(&lhs, &rhs)
         };
-        before.sort_unstable_by(compare);
+        let (before, _, _) = indices.select_nth_unstable_by(limit, &mut compare);
+        before.sort_unstable_by(&mut compare);
     }
 }
 
@@ -74,7 +60,7 @@ fn sort_unstable_by<I, T, G, F>(
         indices.sort_unstable_by(|lhs, rhs| {
             let lhs = get(lhs.to_usize());
             let rhs = get(rhs.to_usize());
-            cmp(&lhs, &rhs).reverse()
+            cmp(&rhs, &lhs)
         })
     } else {
         indices.sort_unstable_by(|lhs, rhs| {
@@ -111,19 +97,18 @@ where
 
     let indices = if let Some(validity) = validity {
         let mut indices = MutableBuffer::<I>::from_len_zeroed(length);
-
         if options.nulls_first {
             let mut nulls = 0;
             let mut valids = 0;
             validity
                 .iter()
-                .zip(0..length)
+                .zip(I::range(0, length).unwrap())
                 .for_each(|(is_valid, index)| {
                     if is_valid {
-                        indices[validity.null_count() + valids] = I::from_usize(index).unwrap();
+                        indices[validity.null_count() + valids] = index;
                         valids += 1;
                     } else {
-                        indices[nulls] = I::from_usize(index).unwrap();
+                        indices[nulls] = index;
                         nulls += 1;
                     }
                 });
@@ -134,23 +119,26 @@ where
                 // Soundness:
                 // all indices in `indices` are by construction `< array.len() == values.len()`
                 // limit is by construction < indices.len()
-                let limit = limit - validity.null_count();
+                let limit = limit.saturating_sub(validity.null_count());
                 let indices = &mut indices.as_mut_slice()[validity.null_count()..];
                 sort_unstable_by(indices, get, cmp, options.descending, limit)
             }
         } else {
-            let last_valid_index = length - validity.null_count();
+            let last_valid_index = length.saturating_sub(validity.null_count());
             let mut nulls = 0;
             let mut valids = 0;
-            validity.iter().zip(0..length).for_each(|(x, index)| {
-                if x {
-                    indices[valids] = I::from_usize(index).unwrap();
-                    valids += 1;
-                } else {
-                    indices[last_valid_index + nulls] = I::from_usize(index).unwrap();
-                    nulls += 1;
-                }
-            });
+            validity
+                .iter()
+                .zip(I::range(0, length).unwrap())
+                .for_each(|(x, index)| {
+                    if x {
+                        indices[valids] = index;
+                        valids += 1;
+                    } else {
+                        indices[last_valid_index + nulls] = index;
+                        nulls += 1;
+                    }
+                });
 
             // Soundness:
             // all indices in `indices` are by construction `< array.len() == values.len()`
@@ -165,21 +153,16 @@ where
 
         indices
     } else {
-        let mut indices = unsafe {
-            MutableBuffer::from_trusted_len_iter_unchecked(
-                (0..length).map(|x| I::from_usize(x).unwrap()),
-            )
-        };
+        let mut indices = MutableBuffer::from_trusted_len_iter(I::range(0, length).unwrap());
 
         // Soundness:
         // indices are by construction `< values.len()`
         // limit is by construction `< values.len()`
         sort_unstable_by(&mut indices, get, cmp, descending, limit);
-
         indices.truncate(limit);
         indices.shrink_to_fit();
-
         indices
     };
+
     PrimitiveArray::<I>::from_data(I::DATA_TYPE, indices.into(), None)
 }
diff --git a/src/compute/sort/lex_sort.rs b/src/compute/sort/lex_sort.rs
index 0a596bfd06..4e8797eba4 100644
--- a/src/compute/sort/lex_sort.rs
+++ b/src/compute/sort/lex_sort.rs
@@ -3,8 +3,9 @@ use std::cmp::Ordering;
 use crate::compute::take;
 use crate::error::{ArrowError, Result};
 use crate::{
-    array::{ord, Array, Index, PrimitiveArray},
+    array::{ord, Array, PrimitiveArray},
     buffer::MutableBuffer,
+    types::Index,
 };
 
 use super::{sort_to_indices, SortOptions};
@@ -177,6 +178,7 @@ pub fn lexsort_to_indices<I: Index>(
         let (before, _, _) = values.select_nth_unstable_by(limit, lex_comparator);
         before.sort_unstable_by(lex_comparator);
         values.truncate(limit);
+        values.shrink_to_fit();
     } else {
         values.sort_unstable_by(lex_comparator);
     }
diff --git a/src/compute/sort/mod.rs b/src/compute/sort/mod.rs
index 32cf63166c..d1fcd213ad 100644
--- a/src/compute/sort/mod.rs
+++ b/src/compute/sort/mod.rs
@@ -6,7 +6,7 @@ use crate::datatypes::*;
 use crate::error::{ArrowError, Result};
 use crate::{
     array::*,
-    types::{days_ms, NativeType},
+    types::{days_ms, Index, NativeType},
 };
 
 use crate::buffer::MutableBuffer;
@@ -387,7 +387,7 @@ where
     if !options.descending {
         valids.sort_by(|a, b| cmp_array(a.1.as_ref(), b.1.as_ref()))
     } else {
-        valids.sort_by(|a, b| cmp_array(a.1.as_ref(), b.1.as_ref()).reverse())
+        valids.sort_by(|a, b| cmp_array(b.1.as_ref(), a.1.as_ref()))
     }
 
     let values = valids.iter().map(|tuple| tuple.0);
diff --git a/src/compute/sort/primitive/indices.rs b/src/compute/sort/primitive/indices.rs
index e7c79947b4..8b229f1c8b 100644
--- a/src/compute/sort/primitive/indices.rs
+++ b/src/compute/sort/primitive/indices.rs
@@ -1,6 +1,6 @@
 use crate::{
-    array::{Array, Index, PrimitiveArray},
-    types::NativeType,
+    array::{Array, PrimitiveArray},
+    types::{Index, NativeType},
 };
 
 use super::super::common;
@@ -18,10 +18,11 @@ where
     T: NativeType,
     F: Fn(&T, &T) -> std::cmp::Ordering,
 {
+    let values = array.values().as_slice();
     unsafe {
         common::indices_sorted_unstable_by(
             array.validity(),
-            |x: usize| *array.values().as_slice().get_unchecked(x),
+            |x: usize| *values.get_unchecked(x),
             cmp,
             array.len(),
             options,
diff --git a/src/compute/sort/primitive/sort.rs b/src/compute/sort/primitive/sort.rs
index 197ea29941..de6aa58867 100644
--- a/src/compute/sort/primitive/sort.rs
+++ b/src/compute/sort/primitive/sort.rs
@@ -34,7 +34,7 @@ where
     F: FnMut(&T, &T) -> std::cmp::Ordering,
 {
     if descending {
-        let (before, _, _) = values.select_nth_unstable_by(limit, |x, y| cmp(x, y).reverse());
+        let (before, _, _) = values.select_nth_unstable_by(limit, |x, y| cmp(y, x));
         before.sort_unstable_by(|x, y| cmp(x, y));
     } else {
         let (before, _, _) = values.select_nth_unstable_by(limit, |x, y| cmp(x, y));
@@ -52,7 +52,7 @@ where
     }
 
     if descending {
-        values.sort_unstable_by(|x, y| cmp(x, y).reverse());
+        values.sort_unstable_by(|x, y| cmp(y, x));
     } else {
         values.sort_unstable_by(cmp);
     };
@@ -125,6 +125,7 @@ where
     };
     // values are sorted, we can now truncate the remaining.
     buffer.truncate(limit);
+    buffer.shrink_to_fit();
 
     (buffer.into(), new_validity.into())
 }
@@ -154,6 +155,7 @@ where
 
         sort_values(&mut buffer.as_mut_slice(), cmp, options.descending, limit);
         buffer.truncate(limit);
+        buffer.shrink_to_fit();
 
         (buffer.into(), None)
     };
diff --git a/src/compute/sort/utf8.rs b/src/compute/sort/utf8.rs
index 20f2d65fea..1776b6b022 100644
--- a/src/compute/sort/utf8.rs
+++ b/src/compute/sort/utf8.rs
@@ -1,5 +1,6 @@
-use crate::array::{Array, Index, Offset, PrimitiveArray, Utf8Array};
+use crate::array::{Array, Offset, PrimitiveArray, Utf8Array};
 use crate::array::{DictionaryArray, DictionaryKey};
+use crate::types::Index;
 
 use super::common;
 use super::SortOptions;
@@ -32,6 +33,7 @@ pub(super) fn indices_sorted_unstable_by_dictionary<I: Index, K: DictionaryKey,
         // Note: there is no check that the keys are within bounds of the dictionary.
         dict.value(index.to_usize().unwrap())
     };
+
     let cmp = |lhs: &&str, rhs: &&str| lhs.cmp(rhs);
     common::indices_sorted_unstable_by(array.validity(), get, cmp, array.len(), options, limit)
 }
diff --git a/src/compute/take/mod.rs b/src/compute/take/mod.rs
index 1517f37e40..87d3cfaa9d 100644
--- a/src/compute/take/mod.rs
+++ b/src/compute/take/mod.rs
@@ -21,11 +21,9 @@ use crate::{
     array::{new_empty_array, Array, NullArray, PrimitiveArray},
     datatypes::{DataType, IntervalUnit},
     error::Result,
-    types::days_ms,
+    types::{days_ms, Index},
 };
 
-pub use crate::array::Index;
-
 mod binary;
 mod boolean;
 mod dict;
diff --git a/src/io/ipc/write/mod.rs b/src/io/ipc/write/mod.rs
index e2be9700e0..7bdf9089fe 100644
--- a/src/io/ipc/write/mod.rs
+++ b/src/io/ipc/write/mod.rs
@@ -22,6 +22,7 @@ mod stream;
 mod writer;
 
 pub use super::gen::Schema::MetadataVersion;
+pub use common::IpcWriteOptions;
 pub use schema::schema_to_bytes;
 pub use serialize::{write, write_dictionary};
 pub use stream::StreamWriter;
diff --git a/src/io/parquet/mod.rs b/src/io/parquet/mod.rs
index 1150fec4db..0a4e80f0b5 100644
--- a/src/io/parquet/mod.rs
+++ b/src/io/parquet/mod.rs
@@ -250,14 +250,19 @@ mod tests {
                     .iter()
                     .map(|x| x.map(|x| x as u32))
                     .collect::<Vec<_>>();
-                Box::new(PrimitiveArray::<u32>::from(values).to(DataType::UInt32))
+                Box::new(PrimitiveArray::<u32>::from(values))
+            }
+            6 => {
+                let keys = PrimitiveArray::<i32>::from([Some(0), Some(1), None, Some(1)]);
+                let values = Arc::new(PrimitiveArray::<i32>::from_slice([10, 200]));
+                Box::new(DictionaryArray::<i32>::from_data(keys, values))
             }
             _ => unreachable!(),
         }
     }
 
-    pub fn pyarrow_nullable_statistics(column: usize) -> Box<dyn Statistics> {
-        match column {
+    pub fn pyarrow_nullable_statistics(column: usize) -> Option<Box<dyn Statistics>> {
+        Some(match column {
             0 => Box::new(PrimitiveStatistics::<i64> {
                 data_type: DataType::Int64,
                 distinct_count: None,
@@ -300,8 +305,9 @@ mod tests {
                 min_value: Some(0),
                 max_value: Some(9),
             }),
+            6 => return None,
             _ => unreachable!(),
-        }
+        })
     }
 
     // these values match the values in `integration`
@@ -331,8 +337,8 @@ mod tests {
         }
     }
 
-    pub fn pyarrow_required_statistics(column: usize) -> Box<dyn Statistics> {
-        match column {
+    pub fn pyarrow_required_statistics(column: usize) -> Option<Box<dyn Statistics>> {
+        Some(match column {
             0 => Box::new(PrimitiveStatistics::<i64> {
                 data_type: DataType::Int64,
                 null_count: Some(0),
@@ -353,11 +359,11 @@ mod tests {
                 max_value: Some("def".to_string()),
             }),
             _ => unreachable!(),
-        }
+        })
     }
 
-    pub fn pyarrow_nested_nullable_statistics(column: usize) -> Box<dyn Statistics> {
-        match column {
+    pub fn pyarrow_nested_nullable_statistics(column: usize) -> Option<Box<dyn Statistics>> {
+        Some(match column {
             3 => Box::new(PrimitiveStatistics::<i16> {
                 data_type: DataType::Int16,
                 distinct_count: None,
@@ -390,7 +396,7 @@ mod tests {
                 min_value: Some(0),
                 max_value: Some(9),
             }),
-        }
+        })
     }
 }
 
@@ -399,7 +405,7 @@ mod tests {
 mod tests_integration {
     use std::sync::Arc;
 
-    use super::write::CompressionCodec;
+    use super::write::Compression;
     use crate::array::{Array, PrimitiveArray, Utf8Array};
     use crate::datatypes::DataType;
     use crate::datatypes::TimeUnit;
@@ -415,7 +421,7 @@ mod tests_integration {
     fn integration_write(schema: &Schema, batches: &[RecordBatch]) -> Result<Vec<u8>> {
         let options = WriteOptions {
             write_statistics: true,
-            compression: CompressionCodec::Uncompressed,
+            compression: Compression::Uncompressed,
             version: Version::V1,
         };
 
diff --git a/src/io/parquet/read/binary/basic.rs b/src/io/parquet/read/binary/basic.rs
index 535741798a..8ceca3ebbb 100644
--- a/src/io/parquet/read/binary/basic.rs
+++ b/src/io/parquet/read/binary/basic.rs
@@ -1,7 +1,8 @@
 use parquet2::{
     encoding::{bitpacking, delta_length_byte_array, hybrid_rle, uleb128, Encoding},
     metadata::{ColumnChunkMetaData, ColumnDescriptor},
-    read::{levels, BinaryPageDict, Page, PageHeader, StreamingIterator},
+    page::{BinaryPageDict, DataPage, DataPageHeader, DataPageHeaderExt},
+    read::{levels, StreamingIterator},
 };
 
 use crate::{
@@ -202,7 +203,7 @@ pub(super) fn read_plain_required<O: Offset>(
 }
 
 fn extend_from_page<O: Offset>(
-    page: &Page,
+    page: &DataPage,
     descriptor: &ColumnDescriptor,
     offsets: &mut MutableBuffer<O>,
     values: &mut MutableBuffer<u8>,
@@ -212,22 +213,24 @@ fn extend_from_page<O: Offset>(
     assert!(descriptor.max_def_level() <= 1);
     let is_optional = descriptor.max_def_level() == 1;
     match page.header() {
-        PageHeader::V1(header) => {
-            assert_eq!(header.definition_level_encoding, Encoding::Rle);
+        DataPageHeader::V1(header) => {
+            assert_eq!(header.definition_level_encoding(), Encoding::Rle);
 
             let (_, validity_buffer, values_buffer) =
                 levels::split_buffer_v1(page.buffer(), false, is_optional);
 
             match (&page.encoding(), page.dictionary_page(), is_optional) {
-                (Encoding::PlainDictionary, Some(dict), true) => read_dict_buffer::<O>(
-                    validity_buffer,
-                    values_buffer,
-                    page.num_values() as u32,
-                    dict.as_any().downcast_ref().unwrap(),
-                    offsets,
-                    values,
-                    validity,
-                ),
+                (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true) => {
+                    read_dict_buffer::<O>(
+                        validity_buffer,
+                        values_buffer,
+                        page.num_values() as u32,
+                        dict.as_any().downcast_ref().unwrap(),
+                        offsets,
+                        values,
+                        validity,
+                    )
+                }
                 (Encoding::DeltaLengthByteArray, None, true) => read_delta_optional::<O>(
                     validity_buffer,
                     values_buffer,
@@ -258,22 +261,24 @@ fn extend_from_page<O: Offset>(
                 }
             }
         }
-        PageHeader::V2(header) => {
+        DataPageHeader::V2(header) => {
             let def_level_buffer_length = header.definition_levels_byte_length as usize;
 
             let (_, validity_buffer, values_buffer) =
                 levels::split_buffer_v2(page.buffer(), 0, def_level_buffer_length);
 
             match (page.encoding(), page.dictionary_page(), is_optional) {
-                (Encoding::PlainDictionary, Some(dict), true) => read_dict_buffer::<O>(
-                    validity_buffer,
-                    values_buffer,
-                    page.num_values() as u32,
-                    dict.as_any().downcast_ref().unwrap(),
-                    offsets,
-                    values,
-                    validity,
-                ),
+                (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true) => {
+                    read_dict_buffer::<O>(
+                        validity_buffer,
+                        values_buffer,
+                        page.num_values() as u32,
+                        dict.as_any().downcast_ref().unwrap(),
+                        offsets,
+                        values,
+                        validity,
+                    )
+                }
                 (Encoding::Plain, None, true) => read_plain_optional::<O>(
                     validity_buffer,
                     values_buffer,
@@ -317,7 +322,7 @@ where
     ArrowError: From<E>,
     O: Offset,
     E: Clone,
-    I: StreamingIterator<Item = std::result::Result<Page, E>>,
+    I: StreamingIterator<Item = std::result::Result<DataPage, E>>,
 {
     let capacity = metadata.num_values() as usize;
     let mut values = MutableBuffer::<u8>::with_capacity(0);
diff --git a/src/io/parquet/read/binary/nested.rs b/src/io/parquet/read/binary/nested.rs
index d3835a521d..e20af80426 100644
--- a/src/io/parquet/read/binary/nested.rs
+++ b/src/io/parquet/read/binary/nested.rs
@@ -3,9 +3,10 @@ use std::sync::Arc;
 use parquet2::{
     encoding::Encoding,
     metadata::{ColumnChunkMetaData, ColumnDescriptor},
+    page::{DataPage, DataPageHeader, DataPageHeaderExt},
     read::{
         levels::{get_bit_width, split_buffer_v1, split_buffer_v2, RLEDecoder},
-        Page, PageHeader, StreamingIterator,
+        StreamingIterator,
     },
 };
 
@@ -108,7 +109,7 @@ fn read<O: Offset>(
 }
 
 fn extend_from_page<O: Offset>(
-    page: &Page,
+    page: &DataPage,
     descriptor: &ColumnDescriptor,
     is_nullable: bool,
     nested: &mut Vec<Box<dyn Nested>>,
@@ -119,9 +120,9 @@ fn extend_from_page<O: Offset>(
     let additional = page.num_values();
 
     match page.header() {
-        PageHeader::V1(header) => {
-            assert_eq!(header.definition_level_encoding, Encoding::Rle);
-            assert_eq!(header.repetition_level_encoding, Encoding::Rle);
+        DataPageHeader::V1(header) => {
+            assert_eq!(header.definition_level_encoding(), Encoding::Rle);
+            assert_eq!(header.repetition_level_encoding(), Encoding::Rle);
 
             match (&page.encoding(), page.dictionary_page()) {
                 (Encoding::Plain, None) => {
@@ -136,11 +137,11 @@ fn extend_from_page<O: Offset>(
                         values_buffer,
                         additional,
                         (
-                            &header.repetition_level_encoding,
+                            &header.repetition_level_encoding(),
                             descriptor.max_rep_level(),
                         ),
                         (
-                            &header.definition_level_encoding,
+                            &header.definition_level_encoding(),
                             descriptor.max_def_level(),
                         ),
                         is_nullable,
@@ -161,7 +162,7 @@ fn extend_from_page<O: Offset>(
                 }
             }
         }
-        PageHeader::V2(header) => match (&page.encoding(), page.dictionary_page()) {
+        DataPageHeader::V2(header) => match (&page.encoding(), page.dictionary_page()) {
             (Encoding::Plain, None) => {
                 let def_level_buffer_length = header.definition_levels_byte_length as usize;
                 let rep_level_buffer_length = header.repetition_levels_byte_length as usize;
@@ -207,7 +208,7 @@ where
     O: Offset,
     ArrowError: From<E>,
     E: Clone,
-    I: StreamingIterator<Item = std::result::Result<Page, E>>,
+    I: StreamingIterator<Item = std::result::Result<DataPage, E>>,
 {
     let capacity = metadata.num_values() as usize;
     let mut values = MutableBuffer::<u8>::with_capacity(0);
diff --git a/src/io/parquet/read/boolean/basic.rs b/src/io/parquet/read/boolean/basic.rs
index 29f959e49e..8e3e155d79 100644
--- a/src/io/parquet/read/boolean/basic.rs
+++ b/src/io/parquet/read/boolean/basic.rs
@@ -8,7 +8,8 @@ use super::super::utils;
 use parquet2::{
     encoding::{hybrid_rle, Encoding},
     metadata::{ColumnChunkMetaData, ColumnDescriptor},
-    read::{levels, Page, PageHeader, StreamingIterator},
+    page::{DataPage, DataPageHeader, DataPageHeaderExt},
+    read::{levels, StreamingIterator},
 };
 
 pub(super) fn read_required(buffer: &[u8], additional: usize, values: &mut MutableBitmap) {
@@ -68,7 +69,7 @@ pub fn iter_to_array<I, E>(mut iter: I, metadata: &ColumnChunkMetaData) -> Resul
 where
     ArrowError: From<E>,
     E: Clone,
-    I: StreamingIterator<Item = std::result::Result<Page, E>>,
+    I: StreamingIterator<Item = std::result::Result<DataPage, E>>,
 {
     let capacity = metadata.num_values() as usize;
     let mut values = MutableBitmap::with_capacity(capacity);
@@ -86,7 +87,7 @@ where
 }
 
 fn extend_from_page(
-    page: &Page,
+    page: &DataPage,
     descriptor: &ColumnDescriptor,
     values: &mut MutableBitmap,
     validity: &mut MutableBitmap,
@@ -95,8 +96,8 @@ fn extend_from_page(
     assert!(descriptor.max_def_level() <= 1);
     let is_optional = descriptor.max_def_level() == 1;
     match page.header() {
-        PageHeader::V1(header) => {
-            assert_eq!(header.definition_level_encoding, Encoding::Rle);
+        DataPageHeader::V1(header) => {
+            assert_eq!(header.definition_level_encoding(), Encoding::Rle);
 
             match (&page.encoding(), page.dictionary_page(), is_optional) {
                 (Encoding::Plain, None, true) => {
@@ -124,7 +125,7 @@ fn extend_from_page(
                 }
             }
         }
-        PageHeader::V2(header) => {
+        DataPageHeader::V2(header) => {
             let def_level_buffer_length = header.definition_levels_byte_length as usize;
             match (page.encoding(), page.dictionary_page(), is_optional) {
                 (Encoding::Plain, None, true) => {
diff --git a/src/io/parquet/read/boolean/nested.rs b/src/io/parquet/read/boolean/nested.rs
index 154d08852f..2accb1e352 100644
--- a/src/io/parquet/read/boolean/nested.rs
+++ b/src/io/parquet/read/boolean/nested.rs
@@ -3,9 +3,10 @@ use std::sync::Arc;
 use parquet2::{
     encoding::Encoding,
     metadata::{ColumnChunkMetaData, ColumnDescriptor},
+    page::{DataPage, DataPageHeader, DataPageHeaderExt},
     read::{
         levels::{get_bit_width, split_buffer_v1, split_buffer_v2, RLEDecoder},
-        Page, PageHeader, StreamingIterator,
+        StreamingIterator,
     },
 };
 
@@ -95,7 +96,7 @@ fn read(
 }
 
 fn extend_from_page(
-    page: &Page,
+    page: &DataPage,
     descriptor: &ColumnDescriptor,
     is_nullable: bool,
     nested: &mut Vec<Box<dyn Nested>>,
@@ -105,9 +106,9 @@ fn extend_from_page(
     let additional = page.num_values();
 
     match page.header() {
-        PageHeader::V1(header) => {
-            assert_eq!(header.definition_level_encoding, Encoding::Rle);
-            assert_eq!(header.repetition_level_encoding, Encoding::Rle);
+        DataPageHeader::V1(header) => {
+            assert_eq!(header.definition_level_encoding(), Encoding::Rle);
+            assert_eq!(header.repetition_level_encoding(), Encoding::Rle);
 
             match (&page.encoding(), page.dictionary_page()) {
                 (Encoding::Plain, None) => {
@@ -122,11 +123,11 @@ fn extend_from_page(
                         values_buffer,
                         additional,
                         (
-                            &header.repetition_level_encoding,
+                            &header.repetition_level_encoding(),
                             descriptor.max_rep_level(),
                         ),
                         (
-                            &header.definition_level_encoding,
+                            &header.definition_level_encoding(),
                             descriptor.max_def_level(),
                         ),
                         is_nullable,
@@ -146,7 +147,7 @@ fn extend_from_page(
                 }
             }
         }
-        PageHeader::V2(header) => match (&page.encoding(), page.dictionary_page()) {
+        DataPageHeader::V2(header) => match (&page.encoding(), page.dictionary_page()) {
             (Encoding::Plain, None) => {
                 let def_level_buffer_length = header.definition_levels_byte_length as usize;
                 let rep_level_buffer_length = header.repetition_levels_byte_length as usize;
@@ -190,7 +191,7 @@ pub fn iter_to_array<I, E>(
 where
     ArrowError: From<E>,
     E: Clone,
-    I: StreamingIterator<Item = std::result::Result<Page, E>>,
+    I: StreamingIterator<Item = std::result::Result<DataPage, E>>,
 {
     let capacity = metadata.num_values() as usize;
     let mut values = MutableBitmap::with_capacity(capacity);
diff --git a/src/io/parquet/read/fixed_size_binary.rs b/src/io/parquet/read/fixed_size_binary.rs
index 9408aaa7d2..e92dd0619b 100644
--- a/src/io/parquet/read/fixed_size_binary.rs
+++ b/src/io/parquet/read/fixed_size_binary.rs
@@ -1,6 +1,7 @@
 use parquet2::{
     encoding::{bitpacking, hybrid_rle, uleb128, Encoding},
-    read::{levels, FixedLenByteArrayPageDict, Page, PageHeader, StreamingIterator},
+    page::{DataPage, DataPageHeader, DataPageHeaderExt, FixedLenByteArrayPageDict},
+    read::{levels, StreamingIterator},
 };
 
 use super::{ColumnChunkMetaData, ColumnDescriptor};
@@ -136,7 +137,7 @@ pub fn iter_to_array<I, E>(
 where
     ArrowError: From<E>,
     E: Clone,
-    I: StreamingIterator<Item = std::result::Result<Page, E>>,
+    I: StreamingIterator<Item = std::result::Result<DataPage, E>>,
 {
     let capacity = metadata.num_values() as usize;
     let mut values = MutableBuffer::<u8>::with_capacity(capacity * size as usize);
@@ -159,7 +160,7 @@ where
 }
 
 pub(crate) fn extend_from_page(
-    page: &Page,
+    page: &DataPage,
     size: i32,
     descriptor: &ColumnDescriptor,
     values: &mut MutableBuffer<u8>,
@@ -169,8 +170,8 @@ pub(crate) fn extend_from_page(
     assert!(descriptor.max_def_level() <= 1);
     let is_optional = descriptor.max_def_level() == 1;
     match page.header() {
-        PageHeader::V1(header) => {
-            assert_eq!(header.definition_level_encoding, Encoding::Rle);
+        DataPageHeader::V1(header) => {
+            assert_eq!(header.definition_level_encoding(), Encoding::Rle);
 
             let (_, validity_buffer, values_buffer) =
                 levels::split_buffer_v1(page.buffer(), false, is_optional);
@@ -207,7 +208,7 @@ pub(crate) fn extend_from_page(
                 }
             }
         }
-        PageHeader::V2(header) => {
+        DataPageHeader::V2(header) => {
             let def_level_buffer_length = header.definition_levels_byte_length as usize;
 
             match (page.encoding(), page.dictionary_page(), is_optional) {
diff --git a/src/io/parquet/read/mod.rs b/src/io/parquet/read/mod.rs
index ba029ffe5a..8cddeb220b 100644
--- a/src/io/parquet/read/mod.rs
+++ b/src/io/parquet/read/mod.rs
@@ -12,7 +12,7 @@ mod utils;
 
 use crate::{
     array::Array,
-    datatypes::{DataType, TimeUnit},
+    datatypes::{DataType, IntervalUnit, TimeUnit},
     error::{ArrowError, Result},
 };
 
@@ -22,19 +22,15 @@ pub use schema::{get_schema, is_type_nullable, FileMetaData};
 pub use parquet2::{
     error::ParquetError,
     metadata::{ColumnChunkMetaData, ColumnDescriptor, RowGroupMetaData},
+    page::{CompressedDataPage, DataPage, DataPageHeader},
     read::{
-        decompress, streaming_iterator, CompressedPage, Decompressor, Page, PageHeader,
-        StreamingIterator,
+        decompress, get_page_iterator as _get_page_iterator, read_metadata as _read_metadata,
+        streaming_iterator, Decompressor, PageIterator, StreamingIterator,
     },
-    schema::{
-        types::{LogicalType, ParquetType, PhysicalType, PrimitiveConvertedType},
+    schema::types::{
+        LogicalType, ParquetType, PhysicalType, PrimitiveConvertedType,
         TimeUnit as ParquetTimeUnit, TimestampType,
     },
-};
-use parquet2::{
-    read::{
-        get_page_iterator as _get_page_iterator, read_metadata as _read_metadata, PageIterator,
-    },
     types::int96_to_i64_ns,
 };
 
@@ -47,7 +43,7 @@ pub fn get_page_iterator<'b, RR: Read + Seek>(
     buffer: Vec<u8>,
 ) -> Result<PageIterator<'b, RR>> {
     Ok(_get_page_iterator(
-        metadata, row_group, column, reader, buffer,
+        metadata, row_group, column, reader, None, buffer,
     )?)
 }
 
@@ -56,185 +52,114 @@ pub fn read_metadata<R: Read + Seek>(reader: &mut R) -> Result<FileMetaData> {
     Ok(_read_metadata(reader)?)
 }
 
-fn page_iter_i64<I: StreamingIterator<Item = std::result::Result<Page, ParquetError>>>(
-    iter: I,
+pub fn page_iter_to_array<
+    I: StreamingIterator<Item = std::result::Result<DataPage, ParquetError>>,
+>(
+    iter: &mut I,
     metadata: &ColumnChunkMetaData,
     data_type: DataType,
 ) -> Result<Box<dyn Array>> {
-    if metadata.descriptor().max_rep_level() > 0 {
-        let inner = match data_type {
-            DataType::List(ref inner) => inner.data_type(),
-            DataType::LargeList(ref inner) => inner.data_type(),
-            _ => unimplemented!(),
-        };
-
-        match inner {
-            DataType::UInt64 => {
-                primitive::iter_to_array_nested(iter, metadata, data_type, |x: i64| x as u64)
-            }
-            _ => primitive::iter_to_array_nested(iter, metadata, data_type, |x: i64| x as i64),
+    use DataType::*;
+    match data_type {
+        // 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),
+        UInt32 => primitive::iter_to_array(iter, metadata, data_type, |x: i32| x as u32),
+        Int8 => primitive::iter_to_array(iter, metadata, data_type, |x: i32| x as i8),
+        Int16 => primitive::iter_to_array(iter, metadata, data_type, |x: i32| x as i16),
+        Int32 | Date32 | Time32(_) | Interval(IntervalUnit::YearMonth) => {
+            primitive::iter_to_array(iter, metadata, data_type, |x: i32| x as i32)
         }
-    } else {
-        match data_type {
-            DataType::UInt64 => {
-                primitive::iter_to_array(iter, metadata, data_type, |x: i64| x as u64)
-            }
-            _ => primitive::iter_to_array(iter, metadata, data_type, |x: i64| x as i64),
+
+        Timestamp(TimeUnit::Nanosecond, None) => match metadata.descriptor().type_() {
+            ParquetType::PrimitiveType { physical_type, .. } => match physical_type {
+                PhysicalType::Int96 => primitive::iter_to_array(
+                    iter,
+                    metadata,
+                    DataType::Timestamp(TimeUnit::Nanosecond, None),
+                    int96_to_i64_ns,
+                ),
+                _ => primitive::iter_to_array(iter, metadata, data_type, |x: i64| x),
+            },
+            _ => unreachable!(),
+        },
+
+        // INT64
+        Int64 | Date64 | Time64(_) | Duration(_) | Timestamp(_, _) => {
+            primitive::iter_to_array(iter, metadata, data_type, |x: i64| x)
         }
-    }
-}
+        UInt64 => primitive::iter_to_array(iter, metadata, data_type, |x: i64| x as u64),
 
-fn page_iter_i32<I: StreamingIterator<Item = std::result::Result<Page, ParquetError>>>(
-    iter: I,
-    metadata: &ColumnChunkMetaData,
-    data_type: DataType,
-) -> Result<Box<dyn Array>> {
-    use DataType::*;
+        Float32 => primitive::iter_to_array(iter, metadata, data_type, |x: f32| x),
+        Float64 => primitive::iter_to_array(iter, metadata, data_type, |x: f64| x),
 
-    if metadata.descriptor().max_rep_level() > 0 {
-        let inner = match data_type {
-            List(ref inner) => inner.data_type(),
-            LargeList(ref inner) => inner.data_type(),
-            _ => unimplemented!(),
-        };
+        Boolean => Ok(Box::new(boolean::iter_to_array(iter, metadata)?)),
+
+        Binary | Utf8 => binary::iter_to_array::<i32, _, _>(iter, metadata, &data_type),
+        LargeBinary | LargeUtf8 => binary::iter_to_array::<i64, _, _>(iter, metadata, &data_type),
+        FixedSizeBinary(size) => Ok(Box::new(fixed_size_binary::iter_to_array(
+            iter, size, metadata,
+        )?)),
 
-        match inner {
+        List(ref inner) => match inner.data_type() {
             UInt8 => primitive::iter_to_array_nested(iter, metadata, data_type, |x: i32| x as u8),
             UInt16 => primitive::iter_to_array_nested(iter, metadata, data_type, |x: i32| x as u16),
             UInt32 => primitive::iter_to_array_nested(iter, metadata, data_type, |x: i32| x as u32),
             Int8 => primitive::iter_to_array_nested(iter, metadata, data_type, |x: i32| x as i8),
             Int16 => primitive::iter_to_array_nested(iter, metadata, data_type, |x: i32| x as i16),
-            _ => primitive::iter_to_array_nested(iter, metadata, data_type, |x: i32| x),
-        }
-    } else {
-        match data_type {
-            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),
-            UInt32 => primitive::iter_to_array(iter, metadata, data_type, |x: i32| x as u32),
-            Int8 => primitive::iter_to_array(iter, metadata, data_type, |x: i32| x as i8),
-            Int16 => primitive::iter_to_array(iter, metadata, data_type, |x: i32| x as i16),
-            _ => primitive::iter_to_array(iter, metadata, data_type, |x: i32| x),
-        }
-    }
-}
+            Int32 => primitive::iter_to_array_nested(iter, metadata, data_type, |x: i32| x as i32),
+
+            Timestamp(TimeUnit::Nanosecond, None) => match metadata.descriptor().type_() {
+                ParquetType::PrimitiveType { physical_type, .. } => match physical_type {
+                    PhysicalType::Int96 => primitive::iter_to_array_nested(
+                        iter,
+                        metadata,
+                        DataType::Timestamp(TimeUnit::Nanosecond, None),
+                        int96_to_i64_ns,
+                    ),
+                    _ => primitive::iter_to_array(iter, metadata, data_type, |x: i64| x),
+                },
+                _ => unreachable!(),
+            },
 
-fn page_iter_byte_array<I: StreamingIterator<Item = std::result::Result<Page, ParquetError>>>(
-    iter: I,
-    metadata: &ColumnChunkMetaData,
-    data_type: DataType,
-) -> Result<Box<dyn Array>> {
-    use DataType::*;
+            // INT64
+            Int64 | Date64 | Time64(_) | Duration(_) | Timestamp(_, _) => {
+                primitive::iter_to_array_nested(iter, metadata, data_type, |x: i64| x)
+            }
+            UInt64 => primitive::iter_to_array_nested(iter, metadata, data_type, |x: i64| x as u64),
 
-    if metadata.descriptor().max_rep_level() > 0 {
-        let inner = match data_type {
-            List(ref inner) => inner.data_type(),
-            LargeList(ref inner) => inner.data_type(),
-            _ => unimplemented!(),
-        };
+            Float32 => primitive::iter_to_array_nested(iter, metadata, data_type, |x: f32| x),
+            Float64 => primitive::iter_to_array_nested(iter, metadata, data_type, |x: f64| x),
+
+            Boolean => boolean::iter_to_array_nested(iter, metadata, data_type),
 
-        match inner {
             Binary | Utf8 => binary::iter_to_array_nested::<i32, _, _>(iter, metadata, data_type),
             LargeBinary | LargeUtf8 => {
                 binary::iter_to_array_nested::<i64, _, _>(iter, metadata, data_type)
             }
             other => Err(ArrowError::NotYetImplemented(format!(
-                "Can't read {:?} from parquet",
+                "The conversion of {:?} to arrow still not implemented",
                 other
             ))),
-        }
-    } else {
-        match data_type {
-            Binary | Utf8 => binary::iter_to_array::<i32, _, _>(iter, metadata, &data_type),
-            LargeBinary | LargeUtf8 => {
-                binary::iter_to_array::<i64, _, _>(iter, metadata, &data_type)
-            }
-            other => Err(ArrowError::NotYetImplemented(format!(
-                "Can't read {:?} from parquet",
-                other
-            ))),
-        }
-    }
-}
-
-fn page_iter_fixed_len_byte_array<
-    I: StreamingIterator<Item = std::result::Result<Page, ParquetError>>,
->(
-    iter: I,
-    metadata: &ColumnChunkMetaData,
-    data_type: DataType,
-) -> Result<Box<dyn Array>> {
-    use DataType::*;
-    Ok(match data_type {
-        FixedSizeBinary(size) => Box::new(fixed_size_binary::iter_to_array(iter, size, metadata)?),
-        other => {
-            return Err(ArrowError::NotYetImplemented(format!(
-                "Can't read {:?} from parquet",
-                other
-            )))
-        }
-    })
-}
+        },
 
-pub fn page_iter_to_array<I: StreamingIterator<Item = std::result::Result<Page, ParquetError>>>(
-    iter: &mut I,
-    metadata: &ColumnChunkMetaData,
-    data_type: DataType,
-) -> Result<Box<dyn Array>> {
-    match metadata.descriptor().base_type() {
-        ParquetType::PrimitiveType {
-            physical_type,
-            converted_type,
-            logical_type,
-            ..
-        } => match (physical_type, converted_type, logical_type) {
-            (PhysicalType::Int32, _, _) => page_iter_i32(iter, metadata, data_type),
-            (PhysicalType::Int64, _, _) => page_iter_i64(iter, metadata, data_type),
-            (PhysicalType::Float, None, None) => {
-                primitive::iter_to_array(iter, metadata, DataType::Float32, |x: f32| x)
-            }
-            (PhysicalType::Double, None, None) => {
-                primitive::iter_to_array(iter, metadata, DataType::Float64, |x: f64| x)
-            }
-            (PhysicalType::Boolean, None, None) => {
-                Ok(Box::new(boolean::iter_to_array(iter, metadata)?))
-            }
-            (PhysicalType::ByteArray, _, _) => page_iter_byte_array(iter, metadata, data_type),
-            (PhysicalType::FixedLenByteArray(_), _, _) => {
-                page_iter_fixed_len_byte_array(iter, metadata, data_type)
-            }
-            (PhysicalType::Int96, _, _) => primitive::iter_to_array(
-                iter,
-                metadata,
-                DataType::Timestamp(TimeUnit::Nanosecond, None),
-                int96_to_i64_ns,
-            ),
-            (p, c, l) => Err(ArrowError::NotYetImplemented(format!(
-                "The conversion of ({:?}, {:?}, {:?}) to arrow still not implemented",
-                p, c, l
-            ))),
+        Dictionary(ref key, ref values) => match key.as_ref() {
+            Int32 => match values.as_ref() {
+                Int32 => primitive::iter_to_dict_array::<i32, _, _, _, _, _>(
+                    iter,
+                    metadata,
+                    data_type,
+                    |x: i32| x as i32,
+                ),
+                _ => todo!(),
+            },
+            _ => todo!(),
         },
-        _ => {
-            // get all primitives in the type_ and their max rep levels
-            match metadata.descriptor().type_() {
-                ParquetType::PrimitiveType {
-                    physical_type,
-                    converted_type,
-                    logical_type,
-                    ..
-                } => match (physical_type, converted_type, logical_type) {
-                    (PhysicalType::Boolean, None, None) => {
-                        boolean::iter_to_array_nested(iter, metadata, data_type)
-                    }
-                    (PhysicalType::Int64, _, _) => page_iter_i64(iter, metadata, data_type),
-                    (PhysicalType::Int32, _, _) => page_iter_i32(iter, metadata, data_type),
-                    (PhysicalType::ByteArray, _, _) => {
-                        page_iter_byte_array(iter, metadata, data_type)
-                    }
-                    _ => todo!(),
-                },
-                _ => unreachable!(),
-            }
-        }
+
+        other => Err(ArrowError::NotYetImplemented(format!(
+            "The conversion of {:?} to arrow still not implemented",
+            other
+        ))),
     }
 }
 
@@ -285,7 +210,7 @@ mod tests {
         };
 
         assert_eq!(expected.as_ref(), array.as_ref());
-        assert_eq!(expected_statistics.as_ref(), statistics.unwrap().as_ref());
+        assert_eq!(expected_statistics, statistics);
 
         Ok(())
     }
diff --git a/src/io/parquet/read/primitive/basic.rs b/src/io/parquet/read/primitive/basic.rs
index 9cfaa7ced2..6cf55868cf 100644
--- a/src/io/parquet/read/primitive/basic.rs
+++ b/src/io/parquet/read/primitive/basic.rs
@@ -1,6 +1,7 @@
 use parquet2::{
     encoding::{bitpacking, hybrid_rle, uleb128, Encoding},
-    read::{levels, Page, PageHeader, PrimitivePageDict},
+    page::{DataPage, DataPageHeader, DataPageHeaderExt, PrimitivePageDict},
+    read::levels,
     types::NativeType,
 };
 
@@ -142,7 +143,7 @@ fn read_required<T, A, F>(
 }
 
 pub fn extend_from_page<T, A, F>(
-    page: &Page,
+    page: &DataPage,
     descriptor: &ColumnDescriptor,
     values: &mut MutableBuffer<A>,
     validity: &mut MutableBitmap,
@@ -158,22 +159,24 @@ where
     assert_eq!(descriptor.max_rep_level(), 0);
     let is_optional = descriptor.max_def_level() == 1;
     match page.header() {
-        PageHeader::V1(header) => {
-            assert_eq!(header.definition_level_encoding, Encoding::Rle);
+        DataPageHeader::V1(header) => {
+            assert_eq!(header.definition_level_encoding(), Encoding::Rle);
 
             let (_, validity_buffer, values_buffer) =
                 levels::split_buffer_v1(page.buffer(), false, is_optional);
 
             match (&page.encoding(), page.dictionary_page(), is_optional) {
-                (Encoding::PlainDictionary, Some(dict), true) => read_dict_buffer_optional(
-                    validity_buffer,
-                    values_buffer,
-                    additional,
-                    dict.as_any().downcast_ref().unwrap(),
-                    values,
-                    validity,
-                    op,
-                ),
+                (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true) => {
+                    read_dict_buffer_optional(
+                        validity_buffer,
+                        values_buffer,
+                        additional,
+                        dict.as_any().downcast_ref().unwrap(),
+                        values,
+                        validity,
+                        op,
+                    )
+                }
                 (Encoding::Plain, None, true) => read_nullable(
                     validity_buffer,
                     values_buffer,
@@ -196,21 +199,23 @@ where
                 }
             }
         }
-        PageHeader::V2(header) => {
+        DataPageHeader::V2(header) => {
             let def_level_buffer_length = header.definition_levels_byte_length as usize;
 
             let (_, validity_buffer, values_buffer) =
                 levels::split_buffer_v2(page.buffer(), 0, def_level_buffer_length);
             match (&page.encoding(), page.dictionary_page(), is_optional) {
-                (Encoding::PlainDictionary, Some(dict), true) => read_dict_buffer_optional(
-                    validity_buffer,
-                    values_buffer,
-                    additional,
-                    dict.as_any().downcast_ref().unwrap(),
-                    values,
-                    validity,
-                    op,
-                ),
+                (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true) => {
+                    read_dict_buffer_optional(
+                        validity_buffer,
+                        values_buffer,
+                        additional,
+                        dict.as_any().downcast_ref().unwrap(),
+                        values,
+                        validity,
+                        op,
+                    )
+                }
                 (Encoding::Plain, None, true) => read_nullable(
                     validity_buffer,
                     values_buffer,
diff --git a/src/io/parquet/read/primitive/dictionary.rs b/src/io/parquet/read/primitive/dictionary.rs
new file mode 100644
index 0000000000..640534b619
--- /dev/null
+++ b/src/io/parquet/read/primitive/dictionary.rs
@@ -0,0 +1,200 @@
+use std::sync::Arc;
+
+use parquet2::{
+    encoding::{bitpacking, hybrid_rle, uleb128, Encoding},
+    page::{DataPage, DataPageHeader, DataPageHeaderExt, PrimitivePageDict},
+    read::{levels, StreamingIterator},
+    types::NativeType,
+};
+
+use super::super::utils as other_utils;
+use super::{ColumnChunkMetaData, ColumnDescriptor};
+use crate::{
+    array::{Array, DictionaryArray, DictionaryKey, PrimitiveArray},
+    bitmap::{utils::BitmapIter, MutableBitmap},
+    buffer::MutableBuffer,
+    datatypes::DataType,
+    error::{ArrowError, Result},
+    types::NativeType as ArrowNativeType,
+};
+
+fn read_dict_optional<K, T, A, F>(
+    validity_buffer: &[u8],
+    indices_buffer: &[u8],
+    additional: usize,
+    dict: &PrimitivePageDict<T>,
+    indices: &mut MutableBuffer<K>,
+    values: &mut MutableBuffer<A>,
+    validity: &mut MutableBitmap,
+    op: F,
+) where
+    K: DictionaryKey,
+    T: NativeType,
+    A: ArrowNativeType,
+    F: Fn(T) -> A,
+{
+    let dict_values = dict.values();
+    values.extend_from_trusted_len_iter(dict_values.iter().map(|x| op(*x)));
+
+    // SPEC: Data page format: the bit width used to encode the entry ids stored as 1 byte (max bit width = 32),
+    // SPEC: followed by the values encoded using RLE/Bit packed described above (with the given bit width).
+    let bit_width = indices_buffer[0];
+    let indices_buffer = &indices_buffer[1..];
+
+    let (_, consumed) = uleb128::decode(indices_buffer);
+    let indices_buffer = &indices_buffer[consumed..];
+
+    let non_null_indices_len = indices_buffer.len() * 8 / bit_width as usize;
+
+    let mut new_indices = bitpacking::Decoder::new(indices_buffer, bit_width, non_null_indices_len);
+
+    let validity_iterator = hybrid_rle::Decoder::new(validity_buffer, 1);
+
+    for run in validity_iterator {
+        match run {
+            hybrid_rle::HybridEncoded::Bitpacked(packed) => {
+                let remaining = additional - indices.len();
+                let len = std::cmp::min(packed.len() * 8, remaining);
+                for is_valid in BitmapIter::new(packed, 0, len) {
+                    let value = if is_valid {
+                        K::from_u32(new_indices.next().unwrap()).unwrap()
+                    } else {
+                        K::default()
+                    };
+                    indices.push(value);
+                }
+                validity.extend_from_slice(packed, 0, len);
+            }
+            hybrid_rle::HybridEncoded::Rle(value, additional) => {
+                let is_set = value[0] == 1;
+                validity.extend_constant(additional, is_set);
+                if is_set {
+                    (0..additional).for_each(|_| {
+                        let index = K::from_u32(new_indices.next().unwrap()).unwrap();
+                        indices.push(index)
+                    })
+                } else {
+                    values.extend_constant(additional, A::default())
+                }
+            }
+        }
+    }
+}
+
+fn extend_from_page<K, T, A, F>(
+    page: &DataPage,
+    descriptor: &ColumnDescriptor,
+    indices: &mut MutableBuffer<K>,
+    values: &mut MutableBuffer<A>,
+    validity: &mut MutableBitmap,
+    op: F,
+) -> Result<()>
+where
+    K: DictionaryKey,
+    T: NativeType,
+    A: ArrowNativeType,
+    F: Fn(T) -> A,
+{
+    let additional = page.num_values();
+
+    assert_eq!(descriptor.max_rep_level(), 0);
+    let is_optional = descriptor.max_def_level() == 1;
+    match page.header() {
+        DataPageHeader::V1(header) => {
+            assert_eq!(header.definition_level_encoding(), Encoding::Rle);
+
+            let (_, validity_buffer, values_buffer) =
+                levels::split_buffer_v1(page.buffer(), false, is_optional);
+
+            match (&page.encoding(), page.dictionary_page(), is_optional) {
+                (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true) => {
+                    read_dict_optional(
+                        validity_buffer,
+                        values_buffer,
+                        additional,
+                        dict.as_any().downcast_ref().unwrap(),
+                        indices,
+                        values,
+                        validity,
+                        op,
+                    )
+                }
+                _ => {
+                    return Err(other_utils::not_implemented(
+                        &page.encoding(),
+                        is_optional,
+                        page.dictionary_page().is_some(),
+                        "V1",
+                        "primitive",
+                    ))
+                }
+            }
+        }
+        DataPageHeader::V2(header) => {
+            let def_level_buffer_length = header.definition_levels_byte_length as usize;
+
+            let (_, validity_buffer, values_buffer) =
+                levels::split_buffer_v2(page.buffer(), 0, def_level_buffer_length);
+            match (&page.encoding(), page.dictionary_page(), is_optional) {
+                (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true) => {
+                    read_dict_optional(
+                        validity_buffer,
+                        values_buffer,
+                        additional,
+                        dict.as_any().downcast_ref().unwrap(),
+                        indices,
+                        values,
+                        validity,
+                        op,
+                    )
+                }
+                _ => {
+                    return Err(other_utils::not_implemented(
+                        &page.encoding(),
+                        is_optional,
+                        page.dictionary_page().is_some(),
+                        "V2",
+                        "primitive",
+                    ))
+                }
+            }
+        }
+    };
+    Ok(())
+}
+
+pub fn iter_to_array<K, T, A, I, E, F>(
+    mut iter: I,
+    metadata: &ColumnChunkMetaData,
+    data_type: DataType,
+    op: F,
+) -> Result<Box<dyn Array>>
+where
+    ArrowError: From<E>,
+    T: NativeType,
+    K: DictionaryKey,
+    E: Clone,
+    A: ArrowNativeType,
+    F: Copy + Fn(T) -> A,
+    I: StreamingIterator<Item = std::result::Result<DataPage, E>>,
+{
+    let capacity = metadata.num_values() as usize;
+    let mut indices = MutableBuffer::<K>::with_capacity(capacity);
+    let mut values = MutableBuffer::<A>::with_capacity(capacity);
+    let mut validity = MutableBitmap::with_capacity(capacity);
+    while let Some(page) = iter.next() {
+        extend_from_page(
+            page.as_ref().map_err(|x| x.clone())?,
+            metadata.descriptor(),
+            &mut indices,
+            &mut values,
+            &mut validity,
+            op,
+        )?
+    }
+
+    let keys = PrimitiveArray::from_data(K::DATA_TYPE, indices.into(), validity.into());
+    let data_type = DictionaryArray::<K>::get_child(&data_type).clone();
+    let values = Arc::new(PrimitiveArray::from_data(data_type, values.into(), None));
+    Ok(Box::new(DictionaryArray::<K>::from_data(keys, values)))
+}
diff --git a/src/io/parquet/read/primitive/mod.rs b/src/io/parquet/read/primitive/mod.rs
index 0771b73989..de4332b807 100644
--- a/src/io/parquet/read/primitive/mod.rs
+++ b/src/io/parquet/read/primitive/mod.rs
@@ -1,13 +1,11 @@
 mod basic;
+mod dictionary;
 mod nested;
 mod utils;
 
 use std::sync::Arc;
 
-use parquet2::{
-    read::{Page, StreamingIterator},
-    types::NativeType,
-};
+use parquet2::{page::DataPage, read::StreamingIterator, types::NativeType};
 
 use super::nested_utils::*;
 use super::{ColumnChunkMetaData, ColumnDescriptor};
@@ -20,6 +18,8 @@ use crate::{
     types::NativeType as ArrowNativeType,
 };
 
+pub use dictionary::iter_to_array as iter_to_dict_array;
+
 pub fn iter_to_array<T, A, I, E, F>(
     mut iter: I,
     metadata: &ColumnChunkMetaData,
@@ -32,7 +32,7 @@ where
     E: Clone,
     A: ArrowNativeType,
     F: Copy + Fn(T) -> A,
-    I: StreamingIterator<Item = std::result::Result<Page, E>>,
+    I: StreamingIterator<Item = std::result::Result<DataPage, E>>,
 {
     let capacity = metadata.num_values() as usize;
     let mut values = MutableBuffer::<A>::with_capacity(capacity);
@@ -47,6 +47,11 @@ where
         )?
     }
 
+    let data_type = match data_type {
+        DataType::Dictionary(_, values) => values.as_ref().clone(),
+        _ => data_type,
+    };
+
     Ok(Box::new(PrimitiveArray::from_data(
         data_type,
         values.into(),
@@ -66,7 +71,7 @@ where
     E: Clone,
     A: ArrowNativeType,
     F: Copy + Fn(T) -> A,
-    I: StreamingIterator<Item = std::result::Result<Page, E>>,
+    I: StreamingIterator<Item = std::result::Result<DataPage, E>>,
 {
     let capacity = metadata.num_values() as usize;
     let mut values = MutableBuffer::<A>::with_capacity(capacity);
diff --git a/src/io/parquet/read/primitive/nested.rs b/src/io/parquet/read/primitive/nested.rs
index acba41b9bb..da10354681 100644
--- a/src/io/parquet/read/primitive/nested.rs
+++ b/src/io/parquet/read/primitive/nested.rs
@@ -1,9 +1,7 @@
 use parquet2::{
     encoding::Encoding,
-    read::{
-        levels::{get_bit_width, split_buffer_v1, split_buffer_v2, RLEDecoder},
-        Page, PageHeader,
-    },
+    page::{DataPage, DataPageHeader, DataPageHeaderExt},
+    read::levels::{get_bit_width, split_buffer_v1, split_buffer_v2, RLEDecoder},
     types::NativeType,
 };
 
@@ -112,7 +110,7 @@ fn read<T, A, F>(
 }
 
 pub fn extend_from_page<T, A, F>(
-    page: &Page,
+    page: &DataPage,
     descriptor: &ColumnDescriptor,
     is_nullable: bool,
     nested: &mut Vec<Box<dyn Nested>>,
@@ -128,9 +126,9 @@ where
     let additional = page.num_values();
 
     match page.header() {
-        PageHeader::V1(header) => {
-            assert_eq!(header.definition_level_encoding, Encoding::Rle);
-            assert_eq!(header.repetition_level_encoding, Encoding::Rle);
+        DataPageHeader::V1(header) => {
+            assert_eq!(header.definition_level_encoding(), Encoding::Rle);
+            assert_eq!(header.repetition_level_encoding(), Encoding::Rle);
 
             match (&page.encoding(), page.dictionary_page()) {
                 (Encoding::Plain, None) => {
@@ -145,11 +143,11 @@ where
                         values_buffer,
                         additional,
                         (
-                            &header.repetition_level_encoding,
+                            &header.repetition_level_encoding(),
                             descriptor.max_rep_level(),
                         ),
                         (
-                            &header.definition_level_encoding,
+                            &header.definition_level_encoding(),
                             descriptor.max_def_level(),
                         ),
                         is_nullable,
@@ -170,7 +168,7 @@ where
                 }
             }
         }
-        PageHeader::V2(header) => match (&page.encoding(), page.dictionary_page()) {
+        DataPageHeader::V2(header) => match (&page.encoding(), page.dictionary_page()) {
             (Encoding::Plain, None) => {
                 let def_level_buffer_length = header.definition_levels_byte_length as usize;
                 let rep_level_buffer_length = header.repetition_levels_byte_length as usize;
diff --git a/src/io/parquet/read/schema/convert.rs b/src/io/parquet/read/schema/convert.rs
index 4fa9d4999a..1c92683b16 100644
--- a/src/io/parquet/read/schema/convert.rs
+++ b/src/io/parquet/read/schema/convert.rs
@@ -6,9 +6,9 @@ use parquet2::{
     schema::{
         types::{
             BasicTypeInfo, GroupConvertedType, LogicalType, ParquetType, PhysicalType,
-            PrimitiveConvertedType, TimeUnit as ParquetTimeUnit,
+            PrimitiveConvertedType, TimeUnit as ParquetTimeUnit, TimestampType,
         },
-        Repetition, TimestampType,
+        Repetition,
     },
 };
 
diff --git a/src/io/parquet/read/statistics/mod.rs b/src/io/parquet/read/statistics/mod.rs
index c217ba48d9..975ec99582 100644
--- a/src/io/parquet/read/statistics/mod.rs
+++ b/src/io/parquet/read/statistics/mod.rs
@@ -24,6 +24,12 @@ impl PartialEq for &dyn Statistics {
     }
 }
 
+impl PartialEq for Box<dyn Statistics> {
+    fn eq(&self, other: &Self) -> bool {
+        self.data_type() == other.data_type()
+    }
+}
+
 pub fn deserialize_statistics(stats: &dyn ParquetStatistics) -> Result<Box<dyn Statistics>> {
     match stats.physical_type() {
         PhysicalType::Int32 => {
diff --git a/src/io/parquet/read/utils.rs b/src/io/parquet/read/utils.rs
index 2ae9ab8793..e4c8417d7f 100644
--- a/src/io/parquet/read/utils.rs
+++ b/src/io/parquet/read/utils.rs
@@ -1,4 +1,4 @@
-use parquet2::{encoding::get_length, schema::Encoding};
+use parquet2::encoding::{get_length, Encoding};
 
 use crate::error::ArrowError;
 
diff --git a/src/io/parquet/write/binary/basic.rs b/src/io/parquet/write/binary/basic.rs
index b9da6ce5cf..4752d72e1d 100644
--- a/src/io/parquet/write/binary/basic.rs
+++ b/src/io/parquet/write/binary/basic.rs
@@ -1,7 +1,7 @@
 use parquet2::{
     encoding::{delta_bitpacked, Encoding},
     metadata::ColumnDescriptor,
-    read::CompressedPage,
+    page::CompressedDataPage,
     statistics::{serialize_statistics, BinaryStatistics, ParquetStatistics, Statistics},
     write::WriteOptions,
 };
@@ -14,7 +14,7 @@ use crate::{
     io::parquet::read::is_type_nullable,
 };
 
-pub(super) fn encode_plain<O: Offset>(
+pub(crate) fn encode_plain<O: Offset>(
     array: &BinaryArray<O>,
     is_optional: bool,
     buffer: &mut Vec<u8>,
@@ -44,7 +44,7 @@ pub fn array_to_page<O: Offset>(
     options: WriteOptions,
     descriptor: ColumnDescriptor,
     encoding: Encoding,
-) -> Result<CompressedPage> {
+) -> Result<CompressedDataPage> {
     let validity = array.validity();
     let is_optional = is_type_nullable(descriptor.type_());
 
diff --git a/src/io/parquet/write/binary/mod.rs b/src/io/parquet/write/binary/mod.rs
index f3c414b319..8d9e94cd0f 100644
--- a/src/io/parquet/write/binary/mod.rs
+++ b/src/io/parquet/write/binary/mod.rs
@@ -2,5 +2,6 @@ mod basic;
 mod nested;
 
 pub use basic::array_to_page;
+pub(crate) use basic::encode_plain;
 pub(super) use basic::{encode_delta, ord_binary};
 pub use nested::array_to_page as nested_array_to_page;
diff --git a/src/io/parquet/write/binary/nested.rs b/src/io/parquet/write/binary/nested.rs
index 6cefd9b45f..cf92180a3c 100644
--- a/src/io/parquet/write/binary/nested.rs
+++ b/src/io/parquet/write/binary/nested.rs
@@ -1,5 +1,6 @@
-use parquet2::schema::Encoding;
-use parquet2::{metadata::ColumnDescriptor, read::CompressedPage, write::WriteOptions};
+use parquet2::{
+    encoding::Encoding, metadata::ColumnDescriptor, page::CompressedDataPage, write::WriteOptions,
+};
 
 use super::super::{levels, utils};
 use super::basic::{build_statistics, encode_plain};
@@ -14,7 +15,7 @@ pub fn array_to_page<O, OO>(
     options: WriteOptions,
     descriptor: ColumnDescriptor,
     nested: levels::NestedInfo<OO>,
-) -> Result<CompressedPage>
+) -> Result<CompressedDataPage>
 where
     OO: Offset,
     O: Offset,
diff --git a/src/io/parquet/write/boolean/basic.rs b/src/io/parquet/write/boolean/basic.rs
index ca1fca9663..604aa49d3c 100644
--- a/src/io/parquet/write/boolean/basic.rs
+++ b/src/io/parquet/write/boolean/basic.rs
@@ -1,8 +1,7 @@
 use parquet2::{
-    encoding::hybrid_rle::bitpacked_encode,
+    encoding::{hybrid_rle::bitpacked_encode, Encoding},
     metadata::ColumnDescriptor,
-    read::CompressedPage,
-    schema::Encoding,
+    page::CompressedDataPage,
     statistics::{serialize_statistics, BooleanStatistics, ParquetStatistics, Statistics},
     write::WriteOptions,
 };
@@ -43,7 +42,7 @@ pub fn array_to_page(
     array: &BooleanArray,
     options: WriteOptions,
     descriptor: ColumnDescriptor,
-) -> Result<CompressedPage> {
+) -> Result<CompressedDataPage> {
     let is_optional = is_type_nullable(descriptor.type_());
 
     let validity = array.validity();
diff --git a/src/io/parquet/write/boolean/nested.rs b/src/io/parquet/write/boolean/nested.rs
index 0bc6170541..b9726f93b6 100644
--- a/src/io/parquet/write/boolean/nested.rs
+++ b/src/io/parquet/write/boolean/nested.rs
@@ -1,5 +1,6 @@
-use parquet2::schema::Encoding;
-use parquet2::{metadata::ColumnDescriptor, read::CompressedPage, write::WriteOptions};
+use parquet2::{
+    encoding::Encoding, metadata::ColumnDescriptor, page::CompressedDataPage, write::WriteOptions,
+};
 
 use super::super::{levels, utils};
 use super::basic::{build_statistics, encode_plain};
@@ -14,7 +15,7 @@ pub fn array_to_page<O>(
     options: WriteOptions,
     descriptor: ColumnDescriptor,
     nested: levels::NestedInfo<O>,
-) -> Result<CompressedPage>
+) -> Result<CompressedDataPage>
 where
     O: Offset,
 {
diff --git a/src/io/parquet/write/dictionary.rs b/src/io/parquet/write/dictionary.rs
new file mode 100644
index 0000000000..5a7e5ad16b
--- /dev/null
+++ b/src/io/parquet/write/dictionary.rs
@@ -0,0 +1,197 @@
+use parquet2::{
+    encoding::{hybrid_rle::encode_u32, Encoding},
+    metadata::ColumnDescriptor,
+    page::{CompressedDictPage, CompressedPage},
+    write::{DynIter, WriteOptions},
+};
+
+use super::binary::encode_plain as binary_encode_plain;
+use super::primitive::encode_plain as primitive_encode_plain;
+use super::utf8::encode_plain as utf8_encode_plain;
+use crate::array::{Array, DictionaryArray, DictionaryKey, PrimitiveArray};
+use crate::bitmap::Bitmap;
+use crate::datatypes::DataType;
+use crate::error::{ArrowError, Result};
+use crate::io::parquet::read::is_type_nullable;
+use crate::io::parquet::write::utils;
+
+fn encode_keys<K: DictionaryKey>(
+    array: &PrimitiveArray<K>,
+    // todo: merge this to not discard values' validity
+    validity: &Option<Bitmap>,
+    descriptor: ColumnDescriptor,
+    options: WriteOptions,
+) -> Result<CompressedPage> {
+    let is_optional = is_type_nullable(descriptor.type_());
+
+    let mut buffer = vec![];
+
+    if let Some(validity) = validity {
+        let projected_val = array.iter().map(|x| {
+            x.map(|x| validity.get_bit(x.to_usize().unwrap()))
+                .unwrap_or(false)
+        });
+        let projected_val = Bitmap::from_trusted_len_iter(projected_val);
+
+        utils::write_def_levels(
+            &mut buffer,
+            is_optional,
+            &Some(projected_val),
+            array.len(),
+            options.version,
+        )?;
+    } else {
+        utils::write_def_levels(
+            &mut buffer,
+            is_optional,
+            array.validity(),
+            array.len(),
+            options.version,
+        )?;
+    }
+
+    let definition_levels_byte_length = buffer.len();
+
+    // encode indices
+    // compute the required number of bits
+    if let Some(validity) = validity {
+        let keys = array
+            .iter()
+            .flatten()
+            .map(|x| {
+                let index = x.to_usize().unwrap();
+                // discard indices whose values are null, since they are part of the def levels.
+                if validity.get_bit(index) {
+                    Some(index as u32)
+                } else {
+                    None
+                }
+            })
+            .flatten();
+        let num_bits = utils::get_bit_width(keys.clone().max().unwrap_or(0) as u64) as u8;
+
+        let keys = utils::ExactSizedIter::new(keys, array.len() - array.null_count());
+
+        // num_bits as a single byte
+        buffer.push(num_bits);
+
+        // followed by the encoded indices.
+        encode_u32(&mut buffer, keys, num_bits)?;
+    } else {
+        let keys = array.iter().flatten().map(|x| x.to_usize().unwrap() as u32);
+        let num_bits = utils::get_bit_width(keys.clone().max().unwrap_or(0) as u64) as u8;
+
+        let keys = utils::ExactSizedIter::new(keys, array.len() - array.null_count());
+
+        // num_bits as a single byte
+        buffer.push(num_bits);
+
+        // followed by the encoded indices.
+        encode_u32(&mut buffer, keys, num_bits)?;
+    }
+
+    let uncompressed_page_size = buffer.len();
+
+    let buffer = utils::compress(buffer, options, definition_levels_byte_length)?;
+
+    utils::build_plain_page(
+        buffer,
+        array.len(),
+        array.null_count(),
+        uncompressed_page_size,
+        0,
+        definition_levels_byte_length,
+        None,
+        descriptor,
+        options,
+        Encoding::PlainDictionary,
+    )
+    .map(CompressedPage::Data)
+}
+
+macro_rules! dyn_prim {
+    ($from:ty, $to:ty, $array:expr) => {{
+        let values = $array.values().as_any().downcast_ref().unwrap();
+
+        let mut buffer = vec![];
+        primitive_encode_plain::<$from, $to>(values, false, &mut buffer);
+
+        CompressedPage::Dict(CompressedDictPage::new(buffer, values.len()))
+    }};
+}
+
+pub fn array_to_pages<K: DictionaryKey>(
+    array: &DictionaryArray<K>,
+    descriptor: ColumnDescriptor,
+    options: WriteOptions,
+    encoding: Encoding,
+) -> Result<DynIter<'static, Result<CompressedPage>>>
+where
+    PrimitiveArray<K>: std::fmt::Display,
+{
+    match encoding {
+        Encoding::PlainDictionary | Encoding::RleDictionary => {
+            // write DictPage
+            let dict_page = match array.values().data_type() {
+                DataType::Int8 => dyn_prim!(i8, i32, array),
+                DataType::Int16 => dyn_prim!(i16, i32, array),
+                DataType::Int32 | DataType::Date32 | DataType::Time32(_) => {
+                    dyn_prim!(i32, i32, array)
+                }
+                DataType::Int64
+                | DataType::Date64
+                | DataType::Time64(_)
+                | DataType::Timestamp(_, _)
+                | DataType::Duration(_) => dyn_prim!(i64, i64, array),
+                DataType::UInt8 => dyn_prim!(u8, i32, array),
+                DataType::UInt16 => dyn_prim!(u16, i32, array),
+                DataType::UInt32 => dyn_prim!(u32, i32, array),
+                DataType::UInt64 => dyn_prim!(i64, i64, array),
+                DataType::Utf8 => {
+                    let values = array.values().as_any().downcast_ref().unwrap();
+
+                    let mut buffer = vec![];
+                    utf8_encode_plain::<i32>(values, false, &mut buffer);
+                    CompressedPage::Dict(CompressedDictPage::new(buffer, values.len()))
+                }
+                DataType::LargeUtf8 => {
+                    let values = array.values().as_any().downcast_ref().unwrap();
+
+                    let mut buffer = vec![];
+                    utf8_encode_plain::<i64>(values, false, &mut buffer);
+                    CompressedPage::Dict(CompressedDictPage::new(buffer, values.len()))
+                }
+                DataType::Binary => {
+                    let values = array.values().as_any().downcast_ref().unwrap();
+
+                    let mut buffer = vec![];
+                    binary_encode_plain::<i32>(values, false, &mut buffer);
+                    CompressedPage::Dict(CompressedDictPage::new(buffer, values.len()))
+                }
+                DataType::LargeBinary => {
+                    let values = array.values().as_any().downcast_ref().unwrap();
+
+                    let mut buffer = vec![];
+                    binary_encode_plain::<i64>(values, false, &mut buffer);
+                    CompressedPage::Dict(CompressedDictPage::new(buffer, values.len()))
+                }
+                other => {
+                    return Err(ArrowError::NotYetImplemented(format!(
+                        "Writing dictionary arrays to parquet only support data type {:?}",
+                        other
+                    )))
+                }
+            };
+
+            // write DataPage pointing to DictPage
+            let data_page =
+                encode_keys(array.keys(), array.values().validity(), descriptor, options)?;
+
+            let iter = std::iter::once(Ok(dict_page)).chain(std::iter::once(Ok(data_page)));
+            Ok(DynIter::new(Box::new(iter)))
+        }
+        _ => Err(ArrowError::NotYetImplemented(
+            "Dictionary arrays only support dictionary encoding".to_string(),
+        )),
+    }
+}
diff --git a/src/io/parquet/write/fixed_len_bytes.rs b/src/io/parquet/write/fixed_len_bytes.rs
index 3e4d011bd1..9d2cc2c188 100644
--- a/src/io/parquet/write/fixed_len_bytes.rs
+++ b/src/io/parquet/write/fixed_len_bytes.rs
@@ -1,6 +1,6 @@
 use parquet2::{
-    compression::create_codec, metadata::ColumnDescriptor, read::CompressedPage, schema::Encoding,
-    write::WriteOptions,
+    compression::create_codec, encoding::Encoding, metadata::ColumnDescriptor,
+    page::CompressedDataPage, write::WriteOptions,
 };
 
 use super::utils;
@@ -14,7 +14,7 @@ pub fn array_to_page(
     array: &FixedSizeBinaryArray,
     options: WriteOptions,
     descriptor: ColumnDescriptor,
-) -> Result<CompressedPage> {
+) -> Result<CompressedDataPage> {
     let is_optional = is_type_nullable(descriptor.type_());
     let validity = array.validity();
 
diff --git a/src/io/parquet/write/mod.rs b/src/io/parquet/write/mod.rs
index 612c9f7b0f..2b4c0f906f 100644
--- a/src/io/parquet/write/mod.rs
+++ b/src/io/parquet/write/mod.rs
@@ -1,5 +1,6 @@
 mod binary;
 mod boolean;
+mod dictionary;
 mod fixed_len_bytes;
 mod levels;
 mod primitive;
@@ -10,6 +11,8 @@ mod utils;
 
 pub mod stream;
 
+use std::sync::Arc;
+
 use crate::array::*;
 use crate::bitmap::Bitmap;
 use crate::buffer::{Buffer, MutableBuffer};
@@ -20,17 +23,13 @@ use crate::io::parquet::write::levels::NestedInfo;
 use crate::types::days_ms;
 use crate::types::NativeType;
 
-use parquet2::metadata::ColumnDescriptor;
 pub use parquet2::{
-    compression::CompressionCodec,
-    read::CompressedPage,
+    compression::Compression,
+    encoding::Encoding,
+    metadata::{ColumnDescriptor, KeyValue, SchemaDescriptor},
+    page::{CompressedDataPage, CompressedPage},
     schema::types::ParquetType,
-    schema::Encoding,
-    write::{DynIter, RowGroupIter},
-    write::{Version, WriteOptions},
-};
-use parquet2::{
-    metadata::SchemaDescriptor, schema::KeyValue, write::write_file as parquet_write_file,
+    write::{write_file as parquet_write_file, DynIter, RowGroupIter, Version, WriteOptions},
 };
 pub use record_batch::RowGroupIterator;
 use schema::schema_to_metadata_key;
@@ -89,7 +88,7 @@ where
 }
 
 /// Checks whether the `data_type` can be encoded as `encoding`.
-/// Note that this is whether this implementation supports it, which is not necessarily
+/// Note that this is whether this implementation supports it, which is a subset of
 /// what the parquet spec allows.
 pub fn can_encode(data_type: &DataType, encoding: Encoding) -> bool {
     matches!(
@@ -99,9 +98,54 @@ pub fn can_encode(data_type: &DataType, encoding: Encoding) -> bool {
                 Encoding::DeltaLengthByteArray,
                 DataType::Binary | DataType::LargeBinary | DataType::Utf8 | DataType::LargeUtf8,
             )
+            | (Encoding::RleDictionary, DataType::Dictionary(_, _))
+            | (Encoding::PlainDictionary, DataType::Dictionary(_, _))
     )
 }
 
+/// Returns an iterator of compressed pages,
+pub fn array_to_pages(
+    array: Arc<dyn Array>,
+    descriptor: ColumnDescriptor,
+    options: WriteOptions,
+    encoding: Encoding,
+) -> Result<DynIter<'static, Result<CompressedPage>>> {
+    match array.data_type() {
+        DataType::Dictionary(key, _) => match key.as_ref() {
+            DataType::Int8 => dictionary::array_to_pages::<i8>(
+                array.as_any().downcast_ref().unwrap(),
+                descriptor,
+                options,
+                encoding,
+            ),
+            DataType::Int16 => dictionary::array_to_pages::<i16>(
+                array.as_any().downcast_ref().unwrap(),
+                descriptor,
+                options,
+                encoding,
+            ),
+            DataType::Int32 => dictionary::array_to_pages::<i32>(
+                array.as_any().downcast_ref().unwrap(),
+                descriptor,
+                options,
+                encoding,
+            ),
+            DataType::Int64 => dictionary::array_to_pages::<i64>(
+                array.as_any().downcast_ref().unwrap(),
+                descriptor,
+                options,
+                encoding,
+            ),
+            other => Err(ArrowError::NotYetImplemented(format!(
+                "Writing parquet pages for data type {:?}",
+                other
+            ))),
+        },
+        _ => array_to_page(array.as_ref(), descriptor, options, encoding)
+            .map(|page| DynIter::new(std::iter::once(Ok(page)))),
+    }
+}
+
 pub fn array_to_page(
     array: &dyn Array,
     descriptor: ColumnDescriptor,
@@ -295,6 +339,7 @@ pub fn array_to_page(
             other
         ))),
     }
+    .map(CompressedPage::Data)
 }
 
 macro_rules! dyn_nested_prim {
@@ -316,7 +361,7 @@ fn list_array_to_page<O: Offset>(
     values: &dyn Array,
     descriptor: ColumnDescriptor,
     options: WriteOptions,
-) -> Result<CompressedPage> {
+) -> Result<CompressedDataPage> {
     use DataType::*;
     let is_optional = is_type_nullable(descriptor.type_());
     let nested = NestedInfo::new(offsets, validity, is_optional);
@@ -395,7 +440,7 @@ fn nested_array_to_page(
     array: &dyn Array,
     descriptor: ColumnDescriptor,
     options: WriteOptions,
-) -> Result<CompressedPage> {
+) -> Result<CompressedDataPage> {
     match array.data_type() {
         DataType::List(_) => {
             let array = array.as_any().downcast_ref::<ListArray<i32>>().unwrap();
@@ -438,7 +483,7 @@ fn nested_array_to_page(
 mod tests {
     use super::*;
 
-    use crate::error::Result;
+    use crate::{error::Result, record_batch::RecordBatch};
     use std::io::Cursor;
 
     use super::super::tests::*;
@@ -448,7 +493,7 @@ mod tests {
         nullable: bool,
         nested: bool,
         version: Version,
-        compression: CompressionCodec,
+        compression: Compression,
         encoding: Encoding,
     ) -> Result<()> {
         let (array, statistics) = if nested {
@@ -467,6 +512,7 @@ mod tests {
                 pyarrow_required_statistics(column),
             )
         };
+        let array: Arc<dyn Array> = array.into();
 
         let field = Field::new("a1", array.data_type().clone(), nullable);
         let schema = Schema::new(vec![field]);
@@ -479,15 +525,13 @@ mod tests {
 
         let parquet_schema = to_parquet_schema(&schema)?;
 
-        // one row group
-        // one column chunk
-        // one page
+        let iter = vec![RecordBatch::try_new(
+            Arc::new(schema.clone()),
+            vec![array.clone()],
+        )];
+
         let row_groups =
-            std::iter::once(Result::Ok(DynIter::new(std::iter::once(Ok(DynIter::new(
-                std::iter::once(array.as_ref())
-                    .zip(parquet_schema.columns().to_vec().into_iter())
-                    .map(|(array, descriptor)| array_to_page(array, descriptor, options, encoding)),
-            ))))));
+            RowGroupIterator::try_new(iter.into_iter(), &schema, options, vec![encoding])?;
 
         let mut writer = Cursor::new(vec![]);
         write_file(
@@ -503,7 +547,7 @@ mod tests {
 
         let (result, stats) = read_column(&mut Cursor::new(data), 0, 0)?;
         assert_eq!(array.as_ref(), result.as_ref());
-        assert_eq!(statistics.as_ref(), stats.unwrap().as_ref());
+        assert_eq!(statistics.as_ref(), stats.as_ref());
         Ok(())
     }
 
@@ -514,7 +558,7 @@ mod tests {
             true,
             false,
             Version::V1,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -526,7 +570,7 @@ mod tests {
             false,
             false,
             Version::V1,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -538,7 +582,7 @@ mod tests {
             true,
             false,
             Version::V2,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -550,7 +594,7 @@ mod tests {
             true,
             false,
             Version::V2,
-            CompressionCodec::Snappy,
+            Compression::Snappy,
             Encoding::Plain,
         )
     }
@@ -562,7 +606,7 @@ mod tests {
             true,
             false,
             Version::V1,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -574,7 +618,7 @@ mod tests {
             false,
             false,
             Version::V1,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -586,7 +630,7 @@ mod tests {
             true,
             false,
             Version::V2,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -598,7 +642,7 @@ mod tests {
             false,
             false,
             Version::V2,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -610,7 +654,7 @@ mod tests {
             true,
             false,
             Version::V2,
-            CompressionCodec::Snappy,
+            Compression::Snappy,
             Encoding::Plain,
         )
     }
@@ -622,7 +666,7 @@ mod tests {
             false,
             false,
             Version::V2,
-            CompressionCodec::Snappy,
+            Compression::Snappy,
             Encoding::Plain,
         )
     }
@@ -634,7 +678,7 @@ mod tests {
             true,
             false,
             Version::V1,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -646,7 +690,7 @@ mod tests {
             false,
             false,
             Version::V1,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -658,7 +702,7 @@ mod tests {
             true,
             false,
             Version::V2,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -670,7 +714,7 @@ mod tests {
             false,
             false,
             Version::V2,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -682,7 +726,7 @@ mod tests {
             false,
             false,
             Version::V2,
-            CompressionCodec::Snappy,
+            Compression::Snappy,
             Encoding::Plain,
         )
     }
@@ -694,7 +738,7 @@ mod tests {
             true,
             true,
             Version::V2,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -706,7 +750,7 @@ mod tests {
             true,
             true,
             Version::V1,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -718,7 +762,7 @@ mod tests {
             true,
             true,
             Version::V2,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -730,7 +774,7 @@ mod tests {
             true,
             true,
             Version::V1,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -742,7 +786,7 @@ mod tests {
             true,
             true,
             Version::V2,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -754,7 +798,7 @@ mod tests {
             true,
             true,
             Version::V1,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -766,7 +810,7 @@ mod tests {
             true,
             true,
             Version::V2,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -778,7 +822,7 @@ mod tests {
             true,
             true,
             Version::V1,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::Plain,
         )
     }
@@ -790,8 +834,20 @@ mod tests {
             true,
             false,
             Version::V2,
-            CompressionCodec::Uncompressed,
+            Compression::Uncompressed,
             Encoding::DeltaLengthByteArray,
         )
     }
+
+    #[test]
+    fn test_i32_optional_v2_dict() -> Result<()> {
+        round_trip(
+            6,
+            true,
+            false,
+            Version::V2,
+            Compression::Uncompressed,
+            Encoding::RleDictionary,
+        )
+    }
 }
diff --git a/src/io/parquet/write/primitive/basic.rs b/src/io/parquet/write/primitive/basic.rs
index 94be5d72e6..796df77c08 100644
--- a/src/io/parquet/write/primitive/basic.rs
+++ b/src/io/parquet/write/primitive/basic.rs
@@ -1,7 +1,7 @@
 use parquet2::{
+    encoding::Encoding,
     metadata::ColumnDescriptor,
-    read::CompressedPage,
-    schema::Encoding,
+    page::CompressedDataPage,
     statistics::{serialize_statistics, ParquetStatistics, PrimitiveStatistics, Statistics},
     types::NativeType,
     write::WriteOptions,
@@ -15,7 +15,7 @@ use crate::{
     types::NativeType as ArrowNativeType,
 };
 
-pub(super) fn encode_plain<T, R>(array: &PrimitiveArray<T>, is_optional: bool, buffer: &mut Vec<u8>)
+pub(crate) fn encode_plain<T, R>(array: &PrimitiveArray<T>, is_optional: bool, buffer: &mut Vec<u8>)
 where
     T: ArrowNativeType,
     R: NativeType,
@@ -42,7 +42,7 @@ pub fn array_to_page<T, R>(
     array: &PrimitiveArray<T>,
     options: WriteOptions,
     descriptor: ColumnDescriptor,
-) -> Result<CompressedPage>
+) -> Result<CompressedDataPage>
 where
     T: ArrowNativeType,
     R: NativeType,
diff --git a/src/io/parquet/write/primitive/mod.rs b/src/io/parquet/write/primitive/mod.rs
index 280e2ff9ef..ddeb654160 100644
--- a/src/io/parquet/write/primitive/mod.rs
+++ b/src/io/parquet/write/primitive/mod.rs
@@ -2,4 +2,5 @@ mod basic;
 mod nested;
 
 pub use basic::array_to_page;
+pub(crate) use basic::encode_plain;
 pub use nested::array_to_page as nested_array_to_page;
diff --git a/src/io/parquet/write/primitive/nested.rs b/src/io/parquet/write/primitive/nested.rs
index fafed4ac2c..36ff01a723 100644
--- a/src/io/parquet/write/primitive/nested.rs
+++ b/src/io/parquet/write/primitive/nested.rs
@@ -1,6 +1,6 @@
-use parquet2::schema::Encoding;
 use parquet2::{
-    metadata::ColumnDescriptor, read::CompressedPage, types::NativeType, write::WriteOptions,
+    encoding::Encoding, metadata::ColumnDescriptor, page::CompressedDataPage, types::NativeType,
+    write::WriteOptions,
 };
 
 use super::super::levels;
@@ -18,7 +18,7 @@ pub fn array_to_page<T, R, O>(
     options: WriteOptions,
     descriptor: ColumnDescriptor,
     nested: levels::NestedInfo<O>,
-) -> Result<CompressedPage>
+) -> Result<CompressedDataPage>
 where
     T: ArrowNativeType,
     R: NativeType,
diff --git a/src/io/parquet/write/record_batch.rs b/src/io/parquet/write/record_batch.rs
index 117cdc009d..3db125f815 100644
--- a/src/io/parquet/write/record_batch.rs
+++ b/src/io/parquet/write/record_batch.rs
@@ -1,5 +1,5 @@
 use super::{
-    array_to_page, to_parquet_schema, DynIter, Encoding, RowGroupIter, SchemaDescriptor,
+    array_to_pages, to_parquet_schema, DynIter, Encoding, RowGroupIter, SchemaDescriptor,
     WriteOptions,
 };
 use crate::{
@@ -59,12 +59,7 @@ impl<I: Iterator<Item = Result<RecordBatch>>> Iterator for RowGroupIterator<I> {
                     .zip(self.parquet_schema.columns().to_vec().into_iter())
                     .zip(encodings.into_iter())
                     .map(move |((array, type_), encoding)| {
-                        Ok(DynIter::new(std::iter::once(array_to_page(
-                            array.as_ref(),
-                            type_,
-                            options,
-                            encoding,
-                        ))))
+                        array_to_pages(array, type_, options, encoding)
                     }),
             ))
         })
diff --git a/src/io/parquet/write/schema.rs b/src/io/parquet/write/schema.rs
index b3169d8785..efd03955cf 100644
--- a/src/io/parquet/write/schema.rs
+++ b/src/io/parquet/write/schema.rs
@@ -1,6 +1,12 @@
-use parquet2::schema::{
-    types::{ParquetType, PhysicalType, PrimitiveConvertedType, TimeUnit as ParquetTimeUnit},
-    DecimalType, FieldRepetitionType, IntType, KeyValue, LogicalType, TimeType, TimestampType,
+use parquet2::{
+    metadata::KeyValue,
+    schema::{
+        types::{
+            DecimalType, IntType, LogicalType, ParquetType, PhysicalType, PrimitiveConvertedType,
+            TimeType, TimeUnit as ParquetTimeUnit, TimestampType,
+        },
+        Repetition,
+    },
 };
 
 use crate::{
@@ -34,9 +40,9 @@ pub fn schema_to_metadata_key(schema: &Schema) -> KeyValue {
 pub fn to_parquet_type(field: &Field) -> Result<ParquetType> {
     let name = field.name().clone();
     let repetition = if field.is_nullable() {
-        FieldRepetitionType::Optional
+        Repetition::Optional
     } else {
-        FieldRepetitionType::Required
+        Repetition::Required
     };
     // create type from field
     match field.data_type() {
@@ -326,7 +332,7 @@ pub fn to_parquet_type(field: &Field) -> Result<ParquetType> {
                 Some(LogicalType::LIST(Default::default())),
                 vec![ParquetType::try_from_group(
                     "list".to_string(),
-                    FieldRepetitionType::Repeated,
+                    Repetition::Repeated,
                     None,
                     None,
                     vec![to_parquet_type(f)?],
@@ -335,14 +341,6 @@ pub fn to_parquet_type(field: &Field) -> Result<ParquetType> {
                 None,
             )?)
         }
-        /*
-        DataType::FixedSizeBinary(length) => {
-            Type::primitive_type_builder(name, PhysicalType::FIXED_LEN_BYTE_ARRAY)
-                .with_repetition(repetition)
-                .with_length(*length)
-                .build()
-        }
-        */
         other => Err(ArrowError::NotYetImplemented(format!(
             "Writing the data type {:?} is not yet implemented",
             other
diff --git a/src/io/parquet/write/stream.rs b/src/io/parquet/write/stream.rs
index fd9ea69a04..8383ffb1ed 100644
--- a/src/io/parquet/write/stream.rs
+++ b/src/io/parquet/write/stream.rs
@@ -2,7 +2,7 @@ use futures::stream::Stream;
 
 use parquet2::write::RowGroupIter;
 use parquet2::{
-    metadata::SchemaDescriptor, schema::KeyValue,
+    metadata::{KeyValue, SchemaDescriptor},
     write::stream::write_stream as parquet_write_stream,
 };
 
diff --git a/src/io/parquet/write/utf8/basic.rs b/src/io/parquet/write/utf8/basic.rs
index af501f1847..ab1c074f21 100644
--- a/src/io/parquet/write/utf8/basic.rs
+++ b/src/io/parquet/write/utf8/basic.rs
@@ -1,7 +1,7 @@
 use parquet2::{
     encoding::Encoding,
     metadata::ColumnDescriptor,
-    read::CompressedPage,
+    page::CompressedDataPage,
     statistics::{serialize_statistics, BinaryStatistics, ParquetStatistics, Statistics},
     write::WriteOptions,
 };
@@ -14,7 +14,7 @@ use crate::{
     io::parquet::read::is_type_nullable,
 };
 
-pub(super) fn encode_plain<O: Offset>(
+pub(crate) fn encode_plain<O: Offset>(
     array: &Utf8Array<O>,
     is_optional: bool,
     buffer: &mut Vec<u8>,
@@ -43,7 +43,7 @@ pub fn array_to_page<O: Offset>(
     options: WriteOptions,
     descriptor: ColumnDescriptor,
     encoding: Encoding,
-) -> Result<CompressedPage> {
+) -> Result<CompressedDataPage> {
     let validity = array.validity();
     let is_optional = is_type_nullable(descriptor.type_());
 
diff --git a/src/io/parquet/write/utf8/mod.rs b/src/io/parquet/write/utf8/mod.rs
index 280e2ff9ef..ddeb654160 100644
--- a/src/io/parquet/write/utf8/mod.rs
+++ b/src/io/parquet/write/utf8/mod.rs
@@ -2,4 +2,5 @@ mod basic;
 mod nested;
 
 pub use basic::array_to_page;
+pub(crate) use basic::encode_plain;
 pub use nested::array_to_page as nested_array_to_page;
diff --git a/src/io/parquet/write/utf8/nested.rs b/src/io/parquet/write/utf8/nested.rs
index efc75b8aa8..65e7d3d13b 100644
--- a/src/io/parquet/write/utf8/nested.rs
+++ b/src/io/parquet/write/utf8/nested.rs
@@ -1,5 +1,5 @@
-use parquet2::schema::Encoding;
-use parquet2::{metadata::ColumnDescriptor, read::CompressedPage, write::WriteOptions};
+use parquet2::encoding::Encoding;
+use parquet2::{metadata::ColumnDescriptor, page::CompressedDataPage, write::WriteOptions};
 
 use super::super::{levels, utils};
 use super::basic::{build_statistics, encode_plain};
@@ -14,7 +14,7 @@ pub fn array_to_page<O, OO>(
     options: WriteOptions,
     descriptor: ColumnDescriptor,
     nested: levels::NestedInfo<OO>,
-) -> Result<CompressedPage>
+) -> Result<CompressedDataPage>
 where
     OO: Offset,
     O: Offset,
diff --git a/src/io/parquet/write/utils.rs b/src/io/parquet/write/utils.rs
index bdf247f671..7850ea724e 100644
--- a/src/io/parquet/write/utils.rs
+++ b/src/io/parquet/write/utils.rs
@@ -1,11 +1,10 @@
 use crate::bitmap::Bitmap;
 
 use parquet2::{
-    compression::create_codec,
+    compression::{create_codec, Compression},
     encoding::{hybrid_rle::encode_bool, Encoding},
     metadata::ColumnDescriptor,
-    read::{CompressedPage, PageHeader},
-    schema::{CompressionCodec, DataPageHeader, DataPageHeaderV2},
+    page::{CompressedDataPage, DataPageHeader, DataPageHeaderV1, DataPageHeaderV2},
     statistics::ParquetStatistics,
     write::WriteOptions,
 };
@@ -70,18 +69,18 @@ pub fn build_plain_page(
     descriptor: ColumnDescriptor,
     options: WriteOptions,
     encoding: Encoding,
-) -> Result<CompressedPage> {
+) -> Result<CompressedDataPage> {
     match options.version {
         Version::V1 => {
-            let header = PageHeader::V1(DataPageHeader {
+            let header = DataPageHeader::V1(DataPageHeaderV1 {
                 num_values: len as i32,
-                encoding,
-                definition_level_encoding: Encoding::Rle,
-                repetition_level_encoding: Encoding::Rle,
+                encoding: encoding.into(),
+                definition_level_encoding: Encoding::Rle.into(),
+                repetition_level_encoding: Encoding::Rle.into(),
                 statistics,
             });
 
-            Ok(CompressedPage::new(
+            Ok(CompressedDataPage::new(
                 header,
                 buffer,
                 options.compression,
@@ -91,18 +90,18 @@ pub fn build_plain_page(
             ))
         }
         Version::V2 => {
-            let header = PageHeader::V2(DataPageHeaderV2 {
+            let header = DataPageHeader::V2(DataPageHeaderV2 {
                 num_values: len as i32,
-                encoding,
+                encoding: encoding.into(),
                 num_nulls: null_count as i32,
                 num_rows: len as i32,
                 definition_levels_byte_length: definition_levels_byte_length as i32,
                 repetition_levels_byte_length: repetition_levels_byte_length as i32,
-                is_compressed: Some(options.compression != CompressionCodec::Uncompressed),
+                is_compressed: Some(options.compression != Compression::Uncompressed),
                 statistics,
             });
 
-            Ok(CompressedPage::new(
+            Ok(CompressedDataPage::new(
                 header,
                 buffer,
                 options.compression,
@@ -184,3 +183,9 @@ impl<T, I: Iterator<Item = T>> Iterator for ExactSizedIter<T, I> {
         (self.remaining, Some(self.remaining))
     }
 }
+
+/// Returns the number of bits needed to bitpack `max`
+#[inline]
+pub fn get_bit_width(max: u64) -> u32 {
+    64 - max.leading_zeros()
+}
diff --git a/src/io/print.rs b/src/io/print.rs
index cc4b9ac26a..102ab35595 100644
--- a/src/io/print.rs
+++ b/src/io/print.rs
@@ -17,8 +17,7 @@
 
 use crate::{array::*, error::Result, record_batch::RecordBatch};
 
-use prettytable::format;
-use prettytable::{Cell, Row, Table};
+use comfy_table::{Cell, Table};
 
 /// Returns a visual representation of multiple [`RecordBatch`]es.
 pub fn write(batches: &[RecordBatch]) -> Result<String> {
@@ -27,14 +26,14 @@ pub fn write(batches: &[RecordBatch]) -> Result<String> {
 
 /// Prints a visual representation of record batches to stdout
 pub fn print(results: &[RecordBatch]) -> Result<()> {
-    create_table(results)?.printstd();
+    println!("{}", create_table(results)?);
     Ok(())
 }
 
 /// Convert a series of record batches into a table
 fn create_table(results: &[RecordBatch]) -> Result<Table> {
     let mut table = Table::new();
-    table.set_format(*format::consts::FORMAT_NO_LINESEP_WITH_TITLE);
+    table.load_preset("||--+-++|    ++++++");
 
     if results.is_empty() {
         return Ok(table);
@@ -46,7 +45,7 @@ fn create_table(results: &[RecordBatch]) -> Result<Table> {
     for field in schema.fields() {
         header.push(Cell::new(field.name()));
     }
-    table.set_titles(Row::new(header));
+    table.set_header(header);
 
     for batch in results {
         let displayes = batch
@@ -61,7 +60,7 @@ fn create_table(results: &[RecordBatch]) -> Result<Table> {
                 let string = displayes[col](row);
                 cells.push(Cell::new(&string));
             });
-            table.add_row(Row::new(cells));
+            table.add_row(cells);
         }
     }
 
diff --git a/src/lib.rs b/src/lib.rs
index 060b627888..f0b43d91fa 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -4,6 +4,7 @@ pub mod bitmap;
 pub mod buffer;
 mod endianess;
 pub mod error;
+pub mod scalar;
 pub mod trusted_len;
 pub mod types;
 
diff --git a/src/scalar/README.md b/src/scalar/README.md
new file mode 100644
index 0000000000..2bac790873
--- /dev/null
+++ b/src/scalar/README.md
@@ -0,0 +1,28 @@
+# Scalar API
+
+Design choices:
+
+### `Scalar` is trait object
+
+There are three reasons:
+
+* a scalar should have a small memory footprint, which an enum would not ensure given the different physical types available.
+* forward-compatibility: a new entry on an `enum` is backward-incompatible
+* do not expose implementation details to users (reduce the surface of the public API)
+
+### `Scalar` MUST contain nullability information
+
+This is to be aligned with the general notion of arrow's `Array`.
+
+This API is a companion to the `Array`, and follows the same design as `Array`.
+Specifically, a `Scalar` is a trait object that can be downcasted to concrete implementations.
+
+Like `Array`, `Scalar` implements
+
+* `data_type`, which is used to perform the correct downcast
+* `is_valid`, to tell whether the scalar is null or not
+
+### There is one implementation per arrows' physical type
+
+* Reduces the number of `match` that users need to write
+* Allows casting of logical types without changing the underlying physical type
diff --git a/src/scalar/binary.rs b/src/scalar/binary.rs
new file mode 100644
index 0000000000..fbd769f9d5
--- /dev/null
+++ b/src/scalar/binary.rs
@@ -0,0 +1,95 @@
+use crate::{array::*, buffer::Buffer, datatypes::DataType};
+
+use super::Scalar;
+
+#[derive(Debug, Clone)]
+pub struct BinaryScalar<O: Offset> {
+    value: Buffer<u8>,
+    is_valid: bool,
+    phantom: std::marker::PhantomData<O>,
+}
+
+impl<O: Offset> PartialEq for BinaryScalar<O> {
+    fn eq(&self, other: &Self) -> bool {
+        self.is_valid == other.is_valid && ((!self.is_valid) | (self.value == other.value))
+    }
+}
+
+impl<O: Offset> BinaryScalar<O> {
+    #[inline]
+    pub fn new<P: AsRef<[u8]>>(v: Option<P>) -> Self {
+        let is_valid = v.is_some();
+        O::from_usize(v.as_ref().map(|x| x.as_ref().len()).unwrap_or_default()).expect("Too large");
+        let value = Buffer::from(v.as_ref().map(|x| x.as_ref()).unwrap_or(&[]));
+        Self {
+            value,
+            is_valid,
+            phantom: std::marker::PhantomData,
+        }
+    }
+
+    #[inline]
+    pub fn value(&self) -> &[u8] {
+        self.value.as_slice()
+    }
+}
+
+impl<O: Offset, P: AsRef<[u8]>> From<Option<P>> for BinaryScalar<O> {
+    #[inline]
+    fn from(v: Option<P>) -> Self {
+        Self::new(v)
+    }
+}
+
+impl<O: Offset> Scalar for BinaryScalar<O> {
+    #[inline]
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    #[inline]
+    fn is_valid(&self) -> bool {
+        self.is_valid
+    }
+
+    #[inline]
+    fn data_type(&self) -> &DataType {
+        if O::is_large() {
+            &DataType::LargeBinary
+        } else {
+            &DataType::Binary
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[allow(clippy::eq_op)]
+    #[test]
+    fn equal() {
+        let a = BinaryScalar::<i32>::from(Some("a"));
+        let b = BinaryScalar::<i32>::from(None::<&str>);
+        assert_eq!(a, a);
+        assert_eq!(b, b);
+        assert!(a != b);
+        let b = BinaryScalar::<i32>::from(Some("b"));
+        assert!(a != b);
+        assert_eq!(b, b);
+    }
+
+    #[test]
+    fn basics() {
+        let a = BinaryScalar::<i32>::from(Some("a"));
+
+        assert_eq!(a.value(), b"a");
+        assert_eq!(a.data_type(), &DataType::Binary);
+        assert!(a.is_valid());
+
+        let a = BinaryScalar::<i64>::from(None::<&str>);
+
+        assert_eq!(a.data_type(), &DataType::LargeBinary);
+        assert!(!a.is_valid());
+    }
+}
diff --git a/src/scalar/boolean.rs b/src/scalar/boolean.rs
new file mode 100644
index 0000000000..67137c752f
--- /dev/null
+++ b/src/scalar/boolean.rs
@@ -0,0 +1,82 @@
+use crate::datatypes::DataType;
+
+use super::Scalar;
+
+#[derive(Debug, Clone)]
+pub struct BooleanScalar {
+    value: bool,
+    is_valid: bool,
+}
+
+impl PartialEq for BooleanScalar {
+    fn eq(&self, other: &Self) -> bool {
+        self.is_valid == other.is_valid && ((!self.is_valid) | (self.value == other.value))
+    }
+}
+
+impl BooleanScalar {
+    #[inline]
+    pub fn new(v: Option<bool>) -> Self {
+        let is_valid = v.is_some();
+        Self {
+            value: v.unwrap_or_default(),
+            is_valid,
+        }
+    }
+
+    #[inline]
+    pub fn value(&self) -> bool {
+        self.value
+    }
+}
+
+impl Scalar for BooleanScalar {
+    #[inline]
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    #[inline]
+    fn is_valid(&self) -> bool {
+        self.is_valid
+    }
+
+    #[inline]
+    fn data_type(&self) -> &DataType {
+        &DataType::Boolean
+    }
+}
+
+impl From<Option<bool>> for BooleanScalar {
+    #[inline]
+    fn from(v: Option<bool>) -> Self {
+        Self::new(v)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[allow(clippy::eq_op)]
+    #[test]
+    fn equal() {
+        let a = BooleanScalar::from(Some(true));
+        let b = BooleanScalar::from(None);
+        assert_eq!(a, a);
+        assert_eq!(b, b);
+        assert!(a != b);
+        let b = BooleanScalar::from(Some(false));
+        assert!(a != b);
+        assert_eq!(b, b);
+    }
+
+    #[test]
+    fn basics() {
+        let a = BooleanScalar::new(Some(true));
+
+        assert!(a.value());
+        assert_eq!(a.data_type(), &DataType::Boolean);
+        assert!(a.is_valid());
+    }
+}
diff --git a/src/scalar/equal.rs b/src/scalar/equal.rs
new file mode 100644
index 0000000000..503f9fc685
--- /dev/null
+++ b/src/scalar/equal.rs
@@ -0,0 +1,115 @@
+use super::*;
+
+impl PartialEq for dyn Scalar {
+    fn eq(&self, other: &Self) -> bool {
+        equal(self, other)
+    }
+}
+
+macro_rules! dyn_eq {
+    ($ty:ty, $lhs:expr, $rhs:expr) => {{
+        let lhs = $lhs
+            .as_any()
+            .downcast_ref::<PrimitiveScalar<$ty>>()
+            .unwrap();
+        let rhs = $rhs
+            .as_any()
+            .downcast_ref::<PrimitiveScalar<$ty>>()
+            .unwrap();
+        lhs == rhs
+    }};
+}
+
+fn equal(lhs: &dyn Scalar, rhs: &dyn Scalar) -> bool {
+    if lhs.data_type() != rhs.data_type() {
+        return false;
+    }
+
+    match lhs.data_type() {
+        DataType::Null => {
+            let lhs = lhs.as_any().downcast_ref::<NullScalar>().unwrap();
+            let rhs = rhs.as_any().downcast_ref::<NullScalar>().unwrap();
+            lhs == rhs
+        }
+        DataType::Boolean => {
+            let lhs = lhs.as_any().downcast_ref::<BooleanScalar>().unwrap();
+            let rhs = rhs.as_any().downcast_ref::<BooleanScalar>().unwrap();
+            lhs == rhs
+        }
+        DataType::UInt8 => {
+            dyn_eq!(u8, lhs, rhs)
+        }
+        DataType::UInt16 => {
+            dyn_eq!(u16, lhs, rhs)
+        }
+        DataType::UInt32 => {
+            dyn_eq!(u32, lhs, rhs)
+        }
+        DataType::UInt64 => {
+            dyn_eq!(u64, lhs, rhs)
+        }
+        DataType::Int8 => {
+            dyn_eq!(i8, lhs, rhs)
+        }
+        DataType::Int16 => {
+            dyn_eq!(i16, lhs, rhs)
+        }
+        DataType::Int32
+        | DataType::Date32
+        | DataType::Time32(_)
+        | DataType::Interval(IntervalUnit::YearMonth) => {
+            dyn_eq!(i32, lhs, rhs)
+        }
+        DataType::Int64
+        | DataType::Date64
+        | DataType::Time64(_)
+        | DataType::Timestamp(_, _)
+        | DataType::Duration(_) => {
+            dyn_eq!(i64, lhs, rhs)
+        }
+        DataType::Decimal(_, _) => {
+            dyn_eq!(i128, lhs, rhs)
+        }
+        DataType::Interval(IntervalUnit::DayTime) => {
+            dyn_eq!(days_ms, lhs, rhs)
+        }
+        DataType::Float16 => unreachable!(),
+        DataType::Float32 => {
+            dyn_eq!(f32, lhs, rhs)
+        }
+        DataType::Float64 => {
+            dyn_eq!(f64, lhs, rhs)
+        }
+        DataType::Utf8 => {
+            let lhs = lhs.as_any().downcast_ref::<Utf8Scalar<i32>>().unwrap();
+            let rhs = rhs.as_any().downcast_ref::<Utf8Scalar<i32>>().unwrap();
+            lhs == rhs
+        }
+        DataType::LargeUtf8 => {
+            let lhs = lhs.as_any().downcast_ref::<Utf8Scalar<i64>>().unwrap();
+            let rhs = rhs.as_any().downcast_ref::<Utf8Scalar<i64>>().unwrap();
+            lhs == rhs
+        }
+        DataType::Binary => {
+            let lhs = lhs.as_any().downcast_ref::<BinaryScalar<i32>>().unwrap();
+            let rhs = rhs.as_any().downcast_ref::<BinaryScalar<i32>>().unwrap();
+            lhs == rhs
+        }
+        DataType::LargeBinary => {
+            let lhs = lhs.as_any().downcast_ref::<BinaryScalar<i64>>().unwrap();
+            let rhs = rhs.as_any().downcast_ref::<BinaryScalar<i64>>().unwrap();
+            lhs == rhs
+        }
+        DataType::List(_) => {
+            let lhs = lhs.as_any().downcast_ref::<ListScalar<i32>>().unwrap();
+            let rhs = rhs.as_any().downcast_ref::<ListScalar<i32>>().unwrap();
+            lhs == rhs
+        }
+        DataType::LargeList(_) => {
+            let lhs = lhs.as_any().downcast_ref::<ListScalar<i64>>().unwrap();
+            let rhs = rhs.as_any().downcast_ref::<ListScalar<i64>>().unwrap();
+            lhs == rhs
+        }
+        _ => unimplemented!(),
+    }
+}
diff --git a/src/scalar/list.rs b/src/scalar/list.rs
new file mode 100644
index 0000000000..4623711a5b
--- /dev/null
+++ b/src/scalar/list.rs
@@ -0,0 +1,66 @@
+use std::any::Any;
+use std::sync::Arc;
+
+use crate::{array::*, datatypes::DataType};
+
+use super::Scalar;
+
+/// The scalar equivalent of [`ListArray`]. Like [`ListArray`], this struct holds a dynamically-typed
+/// [`Array`]. The only difference is that this has only one element.
+#[derive(Debug, Clone)]
+pub struct ListScalar<O: Offset> {
+    values: Arc<dyn Array>,
+    is_valid: bool,
+    phantom: std::marker::PhantomData<O>,
+    data_type: DataType,
+}
+
+impl<O: Offset> PartialEq for ListScalar<O> {
+    fn eq(&self, other: &Self) -> bool {
+        (self.data_type == other.data_type)
+            && (self.is_valid == other.is_valid)
+            && ((!self.is_valid) | (self.values.as_ref() == other.values.as_ref()))
+    }
+}
+
+impl<O: Offset> ListScalar<O> {
+    /// # Panics
+    /// iff
+    /// * the `data_type` is not `List` or `LargeList` (depending on this scalar's offset `O`)
+    /// * the child of the `data_type` is not equal to the `values`
+    #[inline]
+    pub fn new(data_type: DataType, values: Option<Arc<dyn Array>>) -> Self {
+        let inner_data_type = ListArray::<O>::get_child_type(&data_type);
+        let (is_valid, values) = match values {
+            Some(values) => {
+                assert_eq!(inner_data_type, values.data_type());
+                (true, values)
+            }
+            None => (false, new_empty_array(inner_data_type.clone()).into()),
+        };
+        Self {
+            values,
+            is_valid,
+            phantom: std::marker::PhantomData,
+            data_type,
+        }
+    }
+
+    pub fn values(&self) -> &Arc<dyn Array> {
+        &self.values
+    }
+}
+
+impl<O: Offset> Scalar for ListScalar<O> {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn is_valid(&self) -> bool {
+        self.is_valid
+    }
+
+    fn data_type(&self) -> &DataType {
+        &self.data_type
+    }
+}
diff --git a/src/scalar/mod.rs b/src/scalar/mod.rs
new file mode 100644
index 0000000000..9ec6417ad6
--- /dev/null
+++ b/src/scalar/mod.rs
@@ -0,0 +1,136 @@
+use std::any::Any;
+
+use crate::{array::*, datatypes::*, types::days_ms};
+
+mod equal;
+mod primitive;
+pub use primitive::*;
+mod utf8;
+pub use utf8::*;
+mod binary;
+pub use binary::*;
+mod boolean;
+pub use boolean::*;
+mod list;
+pub use list::*;
+mod null;
+pub use null::*;
+mod struct_;
+pub use struct_::*;
+
+pub trait Scalar: std::fmt::Debug {
+    fn as_any(&self) -> &dyn Any;
+
+    fn is_valid(&self) -> bool;
+
+    fn data_type(&self) -> &DataType;
+}
+
+macro_rules! dyn_new {
+    ($array:expr, $index:expr, $type:ty) => {{
+        let array = $array
+            .as_any()
+            .downcast_ref::<PrimitiveArray<$type>>()
+            .unwrap();
+        let value = if array.is_valid($index) {
+            Some(array.value($index))
+        } else {
+            None
+        };
+        Box::new(PrimitiveScalar::new(array.data_type().clone(), value))
+    }};
+}
+
+macro_rules! dyn_new_utf8 {
+    ($array:expr, $index:expr, $type:ty) => {{
+        let array = $array.as_any().downcast_ref::<Utf8Array<$type>>().unwrap();
+        let value = if array.is_valid($index) {
+            Some(array.value($index))
+        } else {
+            None
+        };
+        Box::new(Utf8Scalar::<$type>::new(value))
+    }};
+}
+
+macro_rules! dyn_new_binary {
+    ($array:expr, $index:expr, $type:ty) => {{
+        let array = $array
+            .as_any()
+            .downcast_ref::<BinaryArray<$type>>()
+            .unwrap();
+        let value = if array.is_valid($index) {
+            Some(array.value($index))
+        } else {
+            None
+        };
+        Box::new(BinaryScalar::<$type>::new(value))
+    }};
+}
+
+macro_rules! dyn_new_list {
+    ($array:expr, $index:expr, $type:ty) => {{
+        let array = $array.as_any().downcast_ref::<ListArray<$type>>().unwrap();
+        let value = if array.is_valid($index) {
+            Some(array.value($index).into())
+        } else {
+            None
+        };
+        Box::new(ListScalar::<$type>::new(array.data_type().clone(), value))
+    }};
+}
+
+/// creates a new [`Scalar`] from an [`Array`].
+pub fn new_scalar(array: &dyn Array, index: usize) -> Box<dyn Scalar> {
+    use DataType::*;
+    match array.data_type() {
+        Null => Box::new(NullScalar::new()),
+        Boolean => {
+            let array = array.as_any().downcast_ref::<BooleanArray>().unwrap();
+            let value = if array.is_valid(index) {
+                Some(array.value(index))
+            } else {
+                None
+            };
+            Box::new(BooleanScalar::new(value))
+        }
+        Int8 => dyn_new!(array, index, i8),
+        Int16 => dyn_new!(array, index, i16),
+        Int32 | Date32 | Time32(_) | Interval(IntervalUnit::YearMonth) => {
+            dyn_new!(array, index, i32)
+        }
+        Int64 | Date64 | Time64(_) | Duration(_) | Timestamp(_, _) => dyn_new!(array, index, i64),
+        Interval(IntervalUnit::DayTime) => dyn_new!(array, index, days_ms),
+        UInt8 => dyn_new!(array, index, u8),
+        UInt16 => dyn_new!(array, index, u16),
+        UInt32 => dyn_new!(array, index, u32),
+        UInt64 => dyn_new!(array, index, u64),
+        Decimal(_, _) => dyn_new!(array, index, i128),
+        Float16 => unreachable!(),
+        Float32 => dyn_new!(array, index, f32),
+        Float64 => dyn_new!(array, index, f64),
+        Utf8 => dyn_new_utf8!(array, index, i32),
+        LargeUtf8 => dyn_new_utf8!(array, index, i64),
+        Binary => dyn_new_binary!(array, index, i32),
+        LargeBinary => dyn_new_binary!(array, index, i64),
+        List(_) => dyn_new_list!(array, index, i32),
+        LargeList(_) => dyn_new_list!(array, index, i64),
+        Struct(_) => {
+            let array = array.as_any().downcast_ref::<StructArray>().unwrap();
+            if array.is_valid(index) {
+                let values = array
+                    .values()
+                    .iter()
+                    .map(|x| new_scalar(x.as_ref(), index).into())
+                    .collect();
+                Box::new(StructScalar::new(array.data_type().clone(), Some(values)))
+            } else {
+                Box::new(StructScalar::new(array.data_type().clone(), None))
+            }
+        }
+        FixedSizeBinary(_) => todo!(),
+        FixedSizeList(_, _) => todo!(),
+        Union(_) => todo!(),
+        Dictionary(_, _) => todo!(),
+    }
+}
diff --git a/src/scalar/null.rs b/src/scalar/null.rs
new file mode 100644
index 0000000000..3751c6cfbd
--- /dev/null
+++ b/src/scalar/null.rs
@@ -0,0 +1,36 @@
+use crate::datatypes::DataType;
+
+use super::Scalar;
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct NullScalar {}
+
+impl NullScalar {
+    #[inline]
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+impl Default for NullScalar {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Scalar for NullScalar {
+    #[inline]
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    #[inline]
+    fn is_valid(&self) -> bool {
+        false
+    }
+
+    #[inline]
+    fn data_type(&self) -> &DataType {
+        &DataType::Null
+    }
+}
diff --git a/src/scalar/primitive.rs b/src/scalar/primitive.rs
new file mode 100644
index 0000000000..b204a774ef
--- /dev/null
+++ b/src/scalar/primitive.rs
@@ -0,0 +1,113 @@
+use crate::{
+    datatypes::DataType,
+    types::{NativeType, NaturalDataType},
+};
+
+use super::Scalar;
+
+#[derive(Debug, Clone)]
+pub struct PrimitiveScalar<T: NativeType> {
+    // Not Option<T> because this offers a stabler pointer offset on the struct
+    value: T,
+    is_valid: bool,
+    data_type: DataType,
+}
+
+impl<T: NativeType> PartialEq for PrimitiveScalar<T> {
+    fn eq(&self, other: &Self) -> bool {
+        self.data_type == other.data_type
+            && self.is_valid == other.is_valid
+            && ((!self.is_valid) | (self.value == other.value))
+    }
+}
+
+impl<T: NativeType> PrimitiveScalar<T> {
+    #[inline]
+    pub fn new(data_type: DataType, v: Option<T>) -> Self {
+        let is_valid = v.is_some();
+        Self {
+            value: v.unwrap_or_default(),
+            is_valid,
+            data_type,
+        }
+    }
+
+    #[inline]
+    pub fn value(&self) -> T {
+        self.value
+    }
+
+    /// Returns a new `PrimitiveScalar` with the same value but different [`DataType`]
+    /// # Panic
+    /// This function panics if the `data_type` is not valid for self's physical type `T`.
+    pub fn to(self, data_type: DataType) -> Self {
+        let v = if self.is_valid {
+            Some(self.value)
+        } else {
+            None
+        };
+        Self::new(data_type, v)
+    }
+}
+
+impl<T: NativeType + NaturalDataType> From<Option<T>> for PrimitiveScalar<T> {
+    #[inline]
+    fn from(v: Option<T>) -> Self {
+        Self::new(T::DATA_TYPE, v)
+    }
+}
+
+impl<T: NativeType> Scalar for PrimitiveScalar<T> {
+    #[inline]
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    #[inline]
+    fn is_valid(&self) -> bool {
+        self.is_valid
+    }
+
+    #[inline]
+    fn data_type(&self) -> &DataType {
+        &self.data_type
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[allow(clippy::eq_op)]
+    #[test]
+    fn equal() {
+        let a = PrimitiveScalar::from(Some(2i32));
+        let b = PrimitiveScalar::<i32>::from(None);
+        assert_eq!(a, a);
+        assert_eq!(b, b);
+        assert!(a != b);
+        let b = PrimitiveScalar::<i32>::from(Some(1i32));
+        assert!(a != b);
+        assert_eq!(b, b);
+    }
+
+    #[test]
+    fn basics() {
+        let a = PrimitiveScalar::from(Some(2i32));
+
+        assert_eq!(a.value(), 2i32);
+        assert_eq!(a.data_type(), &DataType::Int32);
+        assert!(a.is_valid());
+
+        let a = a.to(DataType::Date32);
+        assert_eq!(a.data_type(), &DataType::Date32);
+
+        let a = PrimitiveScalar::<i32>::from(None);
+
+        assert_eq!(a.data_type(), &DataType::Int32);
+        assert!(!a.is_valid());
+
+        let a = a.to(DataType::Date32);
+        assert_eq!(a.data_type(), &DataType::Date32);
+    }
+}
diff --git a/src/scalar/struct_.rs b/src/scalar/struct_.rs
new file mode 100644
index 0000000000..eab4671f1d
--- /dev/null
+++ b/src/scalar/struct_.rs
@@ -0,0 +1,54 @@
+use std::sync::Arc;
+
+use crate::datatypes::DataType;
+
+use super::Scalar;
+
+#[derive(Debug, Clone)]
+pub struct StructScalar {
+    values: Vec<Arc<dyn Scalar>>,
+    is_valid: bool,
+    data_type: DataType,
+}
+
+impl PartialEq for StructScalar {
+    fn eq(&self, other: &Self) -> bool {
+        (self.data_type == other.data_type)
+            && (self.is_valid == other.is_valid)
+            && ((!self.is_valid) | (self.values == other.values))
+    }
+}
+
+impl StructScalar {
+    #[inline]
+    pub fn new(data_type: DataType, values: Option<Vec<Arc<dyn Scalar>>>) -> Self {
+        let is_valid = values.is_some();
+        Self {
+            values: values.unwrap_or_default(),
+            is_valid,
+            data_type,
+        }
+    }
+
+    #[inline]
+    pub fn values(&self) -> &[Arc<dyn Scalar>] {
+        &self.values
+    }
+}
+
+impl Scalar for StructScalar {
+    #[inline]
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    #[inline]
+    fn is_valid(&self) -> bool {
+        self.is_valid
+    }
+
+    #[inline]
+    fn data_type(&self) -> &DataType {
+        &self.data_type
+    }
+}
diff --git a/src/scalar/utf8.rs b/src/scalar/utf8.rs
new file mode 100644
index 0000000000..32fe5d7f1b
--- /dev/null
+++ b/src/scalar/utf8.rs
@@ -0,0 +1,95 @@
+use crate::{array::*, buffer::Buffer, datatypes::DataType};
+
+use super::Scalar;
+
+#[derive(Debug, Clone)]
+pub struct Utf8Scalar<O: Offset> {
+    value: Buffer<u8>,
+    is_valid: bool,
+    phantom: std::marker::PhantomData<O>,
+}
+
+impl<O: Offset> PartialEq for Utf8Scalar<O> {
+    fn eq(&self, other: &Self) -> bool {
+        self.is_valid == other.is_valid && ((!self.is_valid) | (self.value == other.value))
+    }
+}
+
+impl<O: Offset> Utf8Scalar<O> {
+    #[inline]
+    pub fn new<P: AsRef<str>>(v: Option<P>) -> Self {
+        let is_valid = v.is_some();
+        O::from_usize(v.as_ref().map(|x| x.as_ref().len()).unwrap_or_default()).expect("Too large");
+        let value = Buffer::from(v.as_ref().map(|x| x.as_ref().as_bytes()).unwrap_or(&[]));
+        Self {
+            value,
+            is_valid,
+            phantom: std::marker::PhantomData,
+        }
+    }
+
+    #[inline]
+    pub fn value(&self) -> &str {
+        unsafe { std::str::from_utf8_unchecked(self.value.as_slice()) }
+    }
+}
+
+impl<O: Offset, P: AsRef<str>> From<Option<P>> for Utf8Scalar<O> {
+    #[inline]
+    fn from(v: Option<P>) -> Self {
+        Self::new(v)
+    }
+}
+
+impl<O: Offset> Scalar for Utf8Scalar<O> {
+    #[inline]
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    #[inline]
+    fn is_valid(&self) -> bool {
+        self.is_valid
+    }
+
+    #[inline]
+    fn data_type(&self) -> &DataType {
+        if O::is_large() {
+            &DataType::LargeUtf8
+        } else {
+            &DataType::Utf8
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[allow(clippy::eq_op)]
+    #[test]
+    fn equal() {
+        let a = Utf8Scalar::<i32>::from(Some("a"));
+        let b = Utf8Scalar::<i32>::from(None::<&str>);
+        assert_eq!(a, a);
+        assert_eq!(b, b);
+        assert!(a != b);
+        let b = Utf8Scalar::<i32>::from(Some("b"));
+        assert!(a != b);
+        assert_eq!(b, b);
+    }
+
+    #[test]
+    fn basics() {
+        let a = Utf8Scalar::<i32>::from(Some("a"));
+
+        assert_eq!(a.value(), "a");
+        assert_eq!(a.data_type(), &DataType::Utf8);
+        assert!(a.is_valid());
+
+        let a = Utf8Scalar::<i64>::from(None::<&str>);
+
+        assert_eq!(a.data_type(), &DataType::LargeUtf8);
+        assert!(!a.is_valid());
+    }
+}
diff --git a/src/types/index.rs b/src/types/index.rs
new file mode 100644
index 0000000000..3a6103f002
--- /dev/null
+++ b/src/types/index.rs
@@ -0,0 +1,113 @@
+use std::convert::TryFrom;
+
+use crate::{
+    trusted_len::TrustedLen,
+    types::{NativeType, NaturalDataType},
+};
+
+/// iterator of [`Index`] equivalent to `(a..b)`.
+// `Step` is unstable in Rust which does not allow (a..b) for generic `Index`.
+pub struct IndexRange<I: Index> {
+    start: I,
+    end: I,
+}
+
+impl<I: Index> IndexRange<I> {
+    pub fn new(start: I, end: I) -> Self {
+        assert!(end >= start);
+        Self { start, end }
+    }
+}
+
+impl<I: Index> Iterator for IndexRange<I> {
+    type Item = I;
+
+    #[inline]
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.start == self.end {
+            return None;
+        }
+        let old = self.start;
+        self.start += I::one();
+        Some(old)
+    }
+
+    #[inline]
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let len = (self.end - self.start).to_usize();
+        (len, Some(len))
+    }
+}
+
+/// Safety: a range is always of known length
+unsafe impl<I: Index> TrustedLen for IndexRange<I> {}
+
+/// Trait describing any type that can be used to index a slot of an array.
+pub trait Index:
+    NativeType
+    + NaturalDataType
+    + std::ops::AddAssign
+    + std::ops::Sub<Output = Self>
+    + num::One
+    + PartialOrd
+{
+    fn to_usize(&self) -> usize;
+    fn from_usize(index: usize) -> Option<Self>;
+
+    fn range(start: usize, end: usize) -> Option<IndexRange<Self>> {
+        let start = Self::from_usize(start);
+        let end = Self::from_usize(end);
+        match (start, end) {
+            (Some(start), Some(end)) => Some(IndexRange::new(start, end)),
+            _ => None,
+        }
+    }
+}
+
+impl Index for i32 {
+    #[inline]
+    fn to_usize(&self) -> usize {
+        *self as usize
+    }
+
+    #[inline]
+    fn from_usize(value: usize) -> Option<Self> {
+        Self::try_from(value).ok()
+    }
+}
+
+impl Index for i64 {
+    #[inline]
+    fn to_usize(&self) -> usize {
+        *self as usize
+    }
+
+    #[inline]
+    fn from_usize(value: usize) -> Option<Self> {
+        Self::try_from(value).ok()
+    }
+}
+
+impl Index for u32 {
+    #[inline]
+    fn to_usize(&self) -> usize {
+        *self as usize
+    }
+
+    #[inline]
+    fn from_usize(value: usize) -> Option<Self> {
+        Self::try_from(value).ok()
+    }
+}
+
+impl Index for u64 {
+    #[inline]
+    fn to_usize(&self) -> usize {
+        *self as usize
+    }
+
+    #[inline]
+    fn from_usize(value: usize) -> Option<Self> {
+        Self::try_from(value).ok()
+    }
+}
diff --git a/src/types/mod.rs b/src/types/mod.rs
index 45167baceb..fb96b099b2 100644
--- a/src/types/mod.rs
+++ b/src/types/mod.rs
@@ -14,7 +14,9 @@ use std::{
 
 mod bit_chunk;
 pub use bit_chunk::{BitChunk, BitChunkIter};
+mod index;
 pub mod simd;
+pub use index::*;
 
 use crate::datatypes::{DataType, IntervalUnit, TimeUnit};
 


Test patch:
diff --git a/arrow-parquet-integration-testing/main.py b/arrow-parquet-integration-testing/main.py
index 41e7cdefde..62a477d398 100644
--- a/arrow-parquet-integration-testing/main.py
+++ b/arrow-parquet-integration-testing/main.py
@@ -65,7 +65,8 @@ def variations():
             "generated_datetime",
             "generated_decimal",
             "generated_interval",
-            # requires writing Dictionary
+            # see https://issues.apache.org/jira/browse/ARROW-13486 and
+            # https://issues.apache.org/jira/browse/ARROW-13487
             # "generated_dictionary",
             # requires writing Struct
             # "generated_duplicate_fieldnames",
diff --git a/arrow-parquet-integration-testing/main_spark.py b/arrow-parquet-integration-testing/main_spark.py
index 3268d0613d..a51dcbaf65 100644
--- a/arrow-parquet-integration-testing/main_spark.py
+++ b/arrow-parquet-integration-testing/main_spark.py
@@ -6,27 +6,40 @@ import pyspark.sql
 
 from main import _prepare, _expected
 
-_file = "generated_primitive"
-_version = "2"
-_encoding = "delta"
-column = ("utf8_nullable", 24)
 
-expected = _expected(_file)
-expected = next(c for i, c in enumerate(expected) if i == column[1])
-expected = expected.combine_chunks().tolist()
+def test(file: str, version: str, column, encoding: str):
+    """
+    Tests that pyspark can read a parquet file written by arrow2.
 
-path = _prepare(_file, _version, _encoding, [column[1]])
+    In arrow2: read IPC, write parquet
+    In pyarrow: read (same) IPC to Python
+    In pyspark: read (written) parquet to Python
+    assert that they are equal
+    """
+    # write parquet
+    path = _prepare(file, version, encoding, [column[1]])
 
-spark = pyspark.sql.SparkSession.builder.config(
-    # see https://stackoverflow.com/a/62024670/931303
-    "spark.sql.parquet.enableVectorizedReader",
-    "false",
-).getOrCreate()
+    # read IPC to Python
+    expected = _expected(file)
+    expected = next(c for i, c in enumerate(expected) if i == column[1])
+    expected = expected.combine_chunks().tolist()
 
-df = spark.read.parquet(path)
+    # read parquet to Python
+    spark = pyspark.sql.SparkSession.builder.config(
+        # see https://stackoverflow.com/a/62024670/931303
+        "spark.sql.parquet.enableVectorizedReader",
+        "false",
+    ).getOrCreate()
 
-r = df.select(column[0]).collect()
-os.remove(path)
+    result = spark.read.parquet(path).select(column[0]).collect()
+    result = [r[column[0]] for r in result]
+    os.remove(path)
 
-result = [r[column[0]] for r in r]
-assert expected == result
+    # assert equality
+    assert expected == result
+
+
+test("generated_primitive", "2", ("utf8_nullable", 24), "delta")
+
+test("generated_dictionary", "1", ("dict0", 0), "")
+test("generated_dictionary", "2", ("dict0", 0), "")
diff --git a/arrow-parquet-integration-testing/src/main.rs b/arrow-parquet-integration-testing/src/main.rs
index 7304eb0dd4..6c8c6ea105 100644
--- a/arrow-parquet-integration-testing/src/main.rs
+++ b/arrow-parquet-integration-testing/src/main.rs
@@ -2,14 +2,17 @@ use std::fs::File;
 use std::sync::Arc;
 use std::{collections::HashMap, convert::TryFrom, io::Read};
 
-use arrow2::datatypes::DataType;
-use arrow2::error::Result;
-use arrow2::io::parquet::write::{Encoding, RowGroupIterator};
-use arrow2::io::{
-    json_integration::ArrowJson,
-    parquet::write::{write_file, CompressionCodec, Version, WriteOptions},
+use arrow2::{
+    datatypes::{DataType, Schema},
+    error::Result,
+    io::{
+        json_integration::{to_record_batch, ArrowJson},
+        parquet::write::{
+            write_file, Compression, Encoding, RowGroupIterator, Version, WriteOptions,
+        },
+    },
+    record_batch::RecordBatch,
 };
-use arrow2::{datatypes::Schema, io::json_integration::to_record_batch, record_batch::RecordBatch};
 
 use clap::{App, Arg};
 
@@ -154,7 +157,7 @@ fn main() -> Result<()> {
 
     let options = WriteOptions {
         write_statistics: true,
-        compression: CompressionCodec::Uncompressed,
+        compression: Compression::Uncompressed,
         version,
     };
 
@@ -162,6 +165,7 @@ fn main() -> Result<()> {
         .fields()
         .iter()
         .map(|x| match x.data_type() {
+            DataType::Dictionary(_, _) => Encoding::RleDictionary,
             DataType::Utf8 | DataType::LargeUtf8 => {
                 if utf8_encoding == "delta" {
                     Encoding::DeltaLengthByteArray
diff --git a/integration-testing/src/flight_client_scenarios/integration_test.rs b/integration-testing/src/flight_client_scenarios/integration_test.rs
index 9c4179e7fd..7cb128cdf6 100644
--- a/integration-testing/src/flight_client_scenarios/integration_test.rs
+++ b/integration-testing/src/flight_client_scenarios/integration_test.rs
@@ -76,7 +76,7 @@ async fn upload_data(
 ) -> Result {
     let (mut upload_tx, upload_rx) = mpsc::channel(10);
 
-    let options = write::common::IpcWriteOptions::default();
+    let options = write::IpcWriteOptions::default();
     let mut schema_flight_data =
         arrow_flight::utils::flight_data_from_arrow_schema(&schema, &options);
     schema_flight_data.flight_descriptor = Some(descriptor.clone());
@@ -128,7 +128,7 @@ async fn send_batch(
     upload_tx: &mut mpsc::Sender<FlightData>,
     metadata: &[u8],
     batch: &RecordBatch,
-    options: &write::common::IpcWriteOptions,
+    options: &write::IpcWriteOptions,
 ) -> Result {
     let (dictionary_flight_data, mut batch_flight_data) =
         arrow_flight::utils::flight_data_from_arrow_batch(batch, &options);
diff --git a/integration-testing/src/flight_server_scenarios/integration_test.rs b/integration-testing/src/flight_server_scenarios/integration_test.rs
index b10df3dc9b..118564cf8a 100644
--- a/integration-testing/src/flight_server_scenarios/integration_test.rs
+++ b/integration-testing/src/flight_server_scenarios/integration_test.rs
@@ -106,7 +106,7 @@ impl FlightService for FlightServiceImpl {
             .get(&key)
             .ok_or_else(|| Status::not_found(format!("Could not find flight. {}", key)))?;
 
-        let options = ipc::write::common::IpcWriteOptions::default();
+        let options = ipc::write::IpcWriteOptions::default();
 
         let schema = std::iter::once({
             Ok(arrow_flight::utils::flight_data_from_arrow_schema(
@@ -174,7 +174,7 @@ impl FlightService for FlightServiceImpl {
 
                 let total_records: usize = flight.chunks.iter().map(|chunk| chunk.num_rows()).sum();
 
-                let options = ipc::write::common::IpcWriteOptions::default();
+                let options = ipc::write::IpcWriteOptions::default();
                 let schema =
                     arrow_flight::utils::ipc_message_from_arrow_schema(&flight.schema, &options)
                         .expect(