Repository: jorgecarleitao/arrow2
Pull request #267 (2021-08-10 20:30:25)

Added support for FFI of dictionary-encoded arrays
out of abundance of caution, I have also ran `cargo valgrind test --lib --no-default-features`, which runs the tests behind valgrind, to make sure no memory leaks are detected.

codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#267](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (f774364) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/bfb4910bf7f9385e13e255f95bd6d1dd7b0d1464?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (bfb4910) will **increase** coverage by `0.17%`.
> The diff coverage is `63.43%`.

[![Impacted file tree graph](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/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/267?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     #267      +/-   ##
==========================================
+ Coverage   77.37%   77.54%   +0.17%     
==========================================
  Files         232      234       +2     
  Lines       20108    20167      +59     
==========================================
+ Hits        15558    15638      +80     
+ Misses       4550     4529      -21     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/array/boolean/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2Jvb2xlYW4vZmZpLnJz) | `0.00% <0.00%> (ø)` | |
| [src/array/dictionary/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2RpY3Rpb25hcnkvbW9kLnJz) | `65.00% <ø> (+5.90%)` | :arrow_up: |
| [src/array/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L21vZC5ycw==) | `53.15% <ø> (ø)` | |
| [src/array/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2ZmaS5ycw==) | `43.24% <33.33%> (+8.10%)` | :arrow_up: |
| [src/array/struct\_.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L3N0cnVjdF8ucnM=) | `49.18% <50.00%> (ø)` | |
| [src/ffi/schema.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2ZmaS9zY2hlbWEucnM=) | `61.53% <61.53%> (ø)` | |
| [src/ffi/array.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2ZmaS9hcnJheS5ycw==) | `80.00% <64.70%> (-3.53%)` | :arrow_down: |
| [src/array/dictionary/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2RpY3Rpb25hcnkvZmZpLnJz) | `75.00% <75.00%> (ø)` | |
| [src/ffi/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2ZmaS9mZmkucnM=) | `86.45% <86.36%> (+15.91%)` | :arrow_up: |
| [src/array/binary/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2JpbmFyeS9mZmkucnM=) | `81.81% <100.00%> (ø)` | |
| ... and [6 more](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267?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/267?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Last update [bfb4910...f774364](https://codecov.io/gh/jorgecarleitao/arrow2/pull/267?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).


Patch:
diff --git a/benches/read_parquet.rs b/benches/read_parquet.rs
index 7c567c7205..f13f3f0da3 100644
--- a/benches/read_parquet.rs
+++ b/benches/read_parquet.rs
@@ -21,7 +21,7 @@ fn read_decompressed_pages(buffer: &[u8], size: usize, column: usize) -> Result<
     let file = Cursor::new(buffer);
 
     let reader =
-        read::RecordReader::try_new(file, Some(vec![column]), None, Arc::new(|_, _| true))?;
+        read::RecordReader::try_new(file, Some(vec![column]), None, Arc::new(|_, _| true), None)?;
 
     for maybe_batch in reader {
         let batch = maybe_batch?;
diff --git a/examples/parquet_read.rs b/examples/parquet_read.rs
index e3b82b4714..a5fb965879 100644
--- a/examples/parquet_read.rs
+++ b/examples/parquet_read.rs
@@ -19,7 +19,8 @@ fn read_column_chunk(path: &str, row_group: usize, column: usize) -> Result<Box<
     // Construct an iterator over pages. This binds `file` to this iterator, and each iteration
     // is IO intensive as it will read a compressed page into memory. There is almost no CPU work
     // on this operation
-    let pages = read::get_page_iterator(&file_metadata, row_group, column, &mut file, vec![])?;
+    let pages =
+        read::get_page_iterator(&file_metadata, row_group, column, &mut file, None, vec![])?;
 
     // get the columns' metadata
     let metadata = file_metadata.row_groups[row_group].column(column);
diff --git a/examples/parquet_read_parallel.rs b/examples/parquet_read_parallel.rs
index ae18ea0f8f..9efc3ba0b5 100644
--- a/examples/parquet_read_parallel.rs
+++ b/examples/parquet_read_parallel.rs
@@ -30,6 +30,7 @@ fn parallel_read(path: &str) -> Result<Vec<Box<dyn Array>>> {
                     row_group,
                     column,
                     &mut file,
+                    None,
                     vec![],
                 )
                 .unwrap()
diff --git a/examples/parquet_read_record.rs b/examples/parquet_read_record.rs
index 6d74629a64..d55065125b 100644
--- a/examples/parquet_read_record.rs
+++ b/examples/parquet_read_record.rs
@@ -11,7 +11,7 @@ fn main() -> Result<()> {
     let file_path = &args[1];
 
     let reader = File::open(file_path)?;
-    let reader = read::RecordReader::try_new(reader, None, None, Arc::new(|_, _| true))?;
+    let reader = read::RecordReader::try_new(reader, None, None, Arc::new(|_, _| true), None)?;
 
     for maybe_batch in reader {
         let batch = maybe_batch?;
diff --git a/examples/s3/Cargo.toml b/examples/s3/Cargo.toml
new file mode 100644
index 0000000000..a238257f90
--- /dev/null
+++ b/examples/s3/Cargo.toml
@@ -0,0 +1,10 @@
+[package]
+name = "s3"
+version = "0.1.0"
+edition = "2018"
+
+[dependencies]
+arrow2 = { path = "../../", default-features = false, features = ["io_parquet", "io_parquet_compression"] }
+rust-s3 = { version = "0.27.0-rc4", features = ["tokio"] }
+futures = "0.3"
+tokio = { version = "1.0.0", features = ["macros", "rt-multi-thread"] }
diff --git a/examples/s3/src/main.rs b/examples/s3/src/main.rs
new file mode 100644
index 0000000000..db98240f40
--- /dev/null
+++ b/examples/s3/src/main.rs
@@ -0,0 +1,67 @@
+use arrow2::array::{Array, Int64Array};
+use arrow2::datatypes::DataType;
+use arrow2::error::Result;
+use arrow2::io::parquet::read::{
+    decompress, get_page_stream, page_stream_to_array, read_metadata_async,
+};
+use futures::{future::BoxFuture, StreamExt};
+use s3::Bucket;
+
+mod stream;
+use stream::{RangedStreamer, SeekOutput};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let bucket_name = "dev-jorgecardleitao";
+    let region = "eu-central-1".parse().unwrap();
+    let bucket = Bucket::new_public(bucket_name, region).unwrap();
+    let path = "benches_65536.parquet".to_string();
+
+    let (data, _) = bucket.head_object(&path).await.unwrap();
+    let length = data.content_length.unwrap() as usize;
+    println!("total size in bytes: {}", length);
+
+    let range_get = Box::new(move |start: u64, length: usize| {
+        let bucket = bucket.clone();
+        let path = path.clone();
+        Box::pin(async move {
+            let bucket = bucket.clone();
+            let path = path.clone();
+            // to get a sense of what is being queried in s3
+            println!("getting {} bytes starting at {}", length, start);
+            let (mut data, _) = bucket
+                // -1 because ranges are inclusive in `get_object_range`
+                .get_object_range(&path, start, Some(start + length as u64 - 1))
+                .await
+                .map_err(|x| std::io::Error::new(std::io::ErrorKind::Other, x.to_string()))?;
+            println!("got {}/{} bytes starting at {}", data.len(), length, start);
+            data.truncate(length);
+            Ok(SeekOutput { start, data })
+        }) as BoxFuture<'static, std::io::Result<SeekOutput>>
+    });
+
+    // at least 4kb per s3 request. Adjust as you like.
+    let mut reader = RangedStreamer::new(length, 4 * 1024, range_get);
+
+    let metadata = read_metadata_async(&mut reader).await?;
+
+    // metadata
+    println!("{}", metadata.num_rows);
+
+    // pages of the first row group and first column
+    // This is IO bounded and SHOULD be done in a shared thread pool (e.g. Tokio)
+    let pages = get_page_stream(&metadata, 0, 0, &mut reader, vec![]).await?;
+
+    // decompress the pages. This is CPU bounded and SHOULD be done in a dedicated thread pool (e.g. Rayon)
+    let pages = pages.map(|compressed_page| decompress(compressed_page?, &mut vec![]));
+
+    // deserialize the pages. This is CPU bounded and SHOULD be done in a dedicated thread pool (e.g. Rayon)
+    let array =
+        page_stream_to_array(pages, &metadata.row_groups[0].columns()[0], DataType::Int64).await?;
+
+    let array = array.as_any().downcast_ref::<Int64Array>().unwrap();
+    // ... and have fun with it.
+    println!("len: {}", array.len());
+    println!("null_count: {}", array.null_count());
+    Ok(())
+}
diff --git a/examples/s3/src/stream.rs b/examples/s3/src/stream.rs
new file mode 100644
index 0000000000..d09d26e7dd
--- /dev/null
+++ b/examples/s3/src/stream.rs
@@ -0,0 +1,113 @@
+// Special thanks to Alice for the help: https://users.rust-lang.org/t/63019/6
+use std::io::{Result, SeekFrom};
+use std::pin::Pin;
+
+use futures::{
+    future::BoxFuture,
+    io::{AsyncRead, AsyncSeek},
+    Future,
+};
+
+pub struct RangedStreamer {
+    pos: u64,
+    length: u64, // total size
+    state: State,
+    range_get: F,
+    min_request_size: usize, // requests have at least this size
+}
+
+enum State {
+    HasChunk(SeekOutput),
+    Seeking(BoxFuture<'static, std::io::Result<SeekOutput>>),
+}
+
+pub struct SeekOutput {
+    pub start: u64,
+    pub data: Vec<u8>,
+}
+
+pub type F = Box<
+    dyn Fn(u64, usize) -> BoxFuture<'static, std::io::Result<SeekOutput>> + Send + Sync,
+>;
+
+impl RangedStreamer {
+    pub fn new(length: usize, min_request_size: usize, range_get: F) -> Self {
+        let length = length as u64;
+        Self {
+            pos: 0,
+            length,
+            state: State::HasChunk(SeekOutput {
+                start: 0,
+                data: vec![],
+            }),
+            range_get,
+            min_request_size,
+        }
+    }
+}
+
+// whether `test_interval` is inside `a` (start, length).
+fn range_includes(a: (usize, usize), test_interval: (usize, usize)) -> bool {
+    if test_interval.0 < a.0 {
+        return false;
+    }
+    let test_end = test_interval.0 + test_interval.1;
+    let a_end = a.0 + a.1;
+    if test_end > a_end {
+        return false;
+    }
+    true
+}
+
+impl AsyncRead for RangedStreamer {
+    fn poll_read(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut [u8],
+    ) -> std::task::Poll<Result<usize>> {
+        let requested_range = (self.pos as usize, buf.len());
+        let min_request_size = self.min_request_size;
+        match &mut self.state {
+            State::HasChunk(output) => {
+                let existing_range = (output.start as usize, output.data.len());
+                if range_includes(existing_range, requested_range) {
+                    let offset = requested_range.0 - existing_range.0;
+                    buf.copy_from_slice(&output.data[offset..offset + buf.len()]);
+                    self.pos += buf.len() as u64;
+                    std::task::Poll::Ready(Ok(buf.len()))
+                } else {
+                    let start = requested_range.0 as u64;
+                    let length = std::cmp::max(min_request_size, requested_range.1);
+                    let future = (self.range_get)(start, length);
+                    self.state = State::Seeking(Box::pin(future));
+                    self.poll_read(cx, buf)
+                }
+            }
+            State::Seeking(ref mut future) => match Pin::new(future).poll(cx) {
+                std::task::Poll::Ready(v) => {
+                    match v {
+                        Ok(output) => self.state = State::HasChunk(output),
+                        Err(e) => return std::task::Poll::Ready(Err(e)),
+                    };
+                    self.poll_read(cx, buf)
+                }
+                std::task::Poll::Pending => std::task::Poll::Pending,
+            },
+        }
+    }
+}
+
+impl AsyncSeek for RangedStreamer {
+    fn poll_seek(
+        mut self: std::pin::Pin<&mut Self>,
+        _: &mut std::task::Context<'_>,
+        pos: std::io::SeekFrom,
+    ) -> std::task::Poll<Result<u64>> {
+        match pos {
+            SeekFrom::Start(pos) => self.pos = pos,
+            SeekFrom::End(pos) => self.pos = (self.length as i64 + pos) as u64,
+            SeekFrom::Current(pos) => self.pos = (self.pos as i64 + pos) as u64,
+        };
+        std::task::Poll::Ready(Ok(self.pos))
+    }
+}
diff --git a/src/array/binary/ffi.rs b/src/array/binary/ffi.rs
index 2771ee9a27..a216be7b42 100644
--- a/src/array/binary/ffi.rs
+++ b/src/array/binary/ffi.rs
@@ -25,7 +25,7 @@ unsafe impl<O: Offset> ToFfi for BinaryArray<O> {
 
 unsafe impl<O: Offset, A: ffi::ArrowArrayRef> FromFfi<A> for BinaryArray<O> {
     fn try_from_ffi(array: A) -> Result<Self> {
-        let data_type = array.data_type()?;
+        let data_type = array.field()?.data_type().clone();
         let expected = if O::is_large() {
             DataType::LargeBinary
         } else {
diff --git a/src/array/boolean/ffi.rs b/src/array/boolean/ffi.rs
index a3b9b651fe..b141c3a3a6 100644
--- a/src/array/boolean/ffi.rs
+++ b/src/array/boolean/ffi.rs
@@ -1,5 +1,6 @@
 use crate::{
     array::{FromFfi, ToFfi},
+    datatypes::DataType,
     ffi,
 };
 
@@ -22,6 +23,8 @@ unsafe impl ToFfi for BooleanArray {
 
 unsafe impl<A: ffi::ArrowArrayRef> FromFfi<A> for BooleanArray {
     fn try_from_ffi(array: A) -> Result<Self> {
+        let data_type = array.field()?.data_type().clone();
+        assert_eq!(data_type, DataType::Boolean);
         let length = array.array().len();
         let offset = array.array().offset();
         let mut validity = unsafe { array.validity() }?;
diff --git a/src/array/dictionary/ffi.rs b/src/array/dictionary/ffi.rs
new file mode 100644
index 0000000000..8fe0e49917
--- /dev/null
+++ b/src/array/dictionary/ffi.rs
@@ -0,0 +1,39 @@
+use crate::{
+    array::{FromFfi, PrimitiveArray, ToFfi},
+    error::Result,
+    ffi,
+};
+
+use super::{DictionaryArray, DictionaryKey};
+
+unsafe impl<K: DictionaryKey> ToFfi for DictionaryArray<K> {
+    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+        self.keys.buffers()
+    }
+
+    #[inline]
+    fn offset(&self) -> usize {
+        self.offset
+    }
+}
+
+unsafe impl<K: DictionaryKey, A: ffi::ArrowArrayRef> FromFfi<A> for DictionaryArray<K> {
+    fn try_from_ffi(array: A) -> Result<Self> {
+        // keys: similar to PrimitiveArray, but the datatype is the inner one
+        let length = array.array().len();
+        let offset = array.array().offset();
+        let mut validity = unsafe { array.validity() }?;
+        let mut values = unsafe { array.buffer::<K>(0) }?;
+
+        if offset > 0 {
+            values = values.slice(offset, length);
+            validity = validity.map(|x| x.slice(offset, length))
+        }
+        let keys = PrimitiveArray::<K>::from_data(K::DATA_TYPE, values, validity);
+        // values
+        let values = array.dictionary()?.unwrap();
+        let values = ffi::try_from(values)?.into();
+
+        Ok(DictionaryArray::<K>::from_data(keys, values))
+    }
+}
diff --git a/src/array/dictionary/mod.rs b/src/array/dictionary/mod.rs
index 21bc374a20..d63d82bccb 100644
--- a/src/array/dictionary/mod.rs
+++ b/src/array/dictionary/mod.rs
@@ -6,12 +6,13 @@ use crate::{
     types::{NativeType, NaturalDataType},
 };
 
+mod ffi;
 mod iterator;
 mod mutable;
 pub use iterator::*;
 pub use mutable::*;
 
-use super::{ffi::ToFfi, new_empty_array, primitive::PrimitiveArray, Array};
+use super::{new_empty_array, primitive::PrimitiveArray, Array};
 
 /// Trait denoting [`NativeType`]s that can be used as keys of a dictionary.
 pub trait DictionaryKey: NativeType + NaturalDataType + num::NumCast + num::FromPrimitive {}
@@ -143,14 +144,3 @@ where
         write!(f, "}}")
     }
 }
-
-unsafe impl<K: DictionaryKey> ToFfi for DictionaryArray<K> {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
-        vec![self.keys.validity().as_ref().map(|x| x.as_ptr())]
-    }
-
-    #[inline]
-    fn offset(&self) -> usize {
-        self.offset
-    }
-}
diff --git a/src/array/ffi.rs b/src/array/ffi.rs
index 123d925d36..e707f98d6b 100644
--- a/src/array/ffi.rs
+++ b/src/array/ffi.rs
@@ -29,13 +29,28 @@ pub unsafe trait FromFfi<T: ffi::ArrowArrayRef>: Sized {
 macro_rules! ffi_dyn {
     ($array:expr, $ty:ty) => {{
         let array = $array.as_any().downcast_ref::<$ty>().unwrap();
-        (array.buffers(), array.children())
+        (array.buffers(), array.children(), None)
     }};
 }
 
-type BuffersChildren = (Vec<Option<std::ptr::NonNull<u8>>>, Vec<Arc<dyn Array>>);
+macro_rules! ffi_dict_dyn {
+    ($array:expr, $ty:ty) => {{
+        let array = $array.as_any().downcast_ref::<$ty>().unwrap();
+        (
+            array.buffers(),
+            array.children(),
+            Some(array.values().clone()),
+        )
+    }};
+}
+
+type BuffersChildren = (
+    Vec<Option<std::ptr::NonNull<u8>>>,
+    Vec<Arc<dyn Array>>,
+    Option<Arc<dyn Array>>,
+);
 
-pub fn buffers_children(array: &dyn Array) -> BuffersChildren {
+pub fn buffers_children_dictionary(array: &dyn Array) -> BuffersChildren {
     match array.data_type() {
         DataType::Null => ffi_dyn!(array, NullArray),
         DataType::Boolean => ffi_dyn!(array, BooleanArray),
@@ -72,14 +87,14 @@ pub fn buffers_children(array: &dyn Array) -> BuffersChildren {
         DataType::Struct(_) => ffi_dyn!(array, StructArray),
         DataType::Union(_) => unimplemented!(),
         DataType::Dictionary(key_type, _) => match key_type.as_ref() {
-            DataType::Int8 => ffi_dyn!(array, DictionaryArray::<i8>),
-            DataType::Int16 => ffi_dyn!(array, DictionaryArray::<i16>),
-            DataType::Int32 => ffi_dyn!(array, DictionaryArray::<i32>),
-            DataType::Int64 => ffi_dyn!(array, DictionaryArray::<i64>),
-            DataType::UInt8 => ffi_dyn!(array, DictionaryArray::<u8>),
-            DataType::UInt16 => ffi_dyn!(array, DictionaryArray::<u16>),
-            DataType::UInt32 => ffi_dyn!(array, DictionaryArray::<u32>),
-            DataType::UInt64 => ffi_dyn!(array, DictionaryArray::<u64>),
+            DataType::Int8 => ffi_dict_dyn!(array, DictionaryArray::<i8>),
+            DataType::Int16 => ffi_dict_dyn!(array, DictionaryArray::<i16>),
+            DataType::Int32 => ffi_dict_dyn!(array, DictionaryArray::<i32>),
+            DataType::Int64 => ffi_dict_dyn!(array, DictionaryArray::<i64>),
+            DataType::UInt8 => ffi_dict_dyn!(array, DictionaryArray::<u8>),
+            DataType::UInt16 => ffi_dict_dyn!(array, DictionaryArray::<u16>),
+            DataType::UInt32 => ffi_dict_dyn!(array, DictionaryArray::<u32>),
+            DataType::UInt64 => ffi_dict_dyn!(array, DictionaryArray::<u64>),
             _ => unreachable!(),
         },
     }
diff --git a/src/array/list/ffi.rs b/src/array/list/ffi.rs
index cd794ff5cc..63d6e8f586 100644
--- a/src/array/list/ffi.rs
+++ b/src/array/list/ffi.rs
@@ -24,7 +24,7 @@ unsafe impl<O: Offset> ToFfi for ListArray<O> {
 
 unsafe impl<O: Offset, A: ffi::ArrowArrayRef> FromFfi<A> for ListArray<O> {
     fn try_from_ffi(array: A) -> Result<Self> {
-        let data_type = array.data_type()?;
+        let data_type = array.field()?.data_type().clone();
         let length = array.array().len();
         let offset = array.array().offset();
         let mut validity = unsafe { array.validity() }?;
diff --git a/src/array/mod.rs b/src/array/mod.rs
index 772c49b46c..3505a6204a 100644
--- a/src/array/mod.rs
+++ b/src/array/mod.rs
@@ -401,7 +401,7 @@ pub use specification::Offset;
 pub use struct_::StructArray;
 pub use utf8::{MutableUtf8Array, Utf8Array, Utf8ValuesIter};
 
-pub(crate) use self::ffi::buffers_children;
+pub(crate) use self::ffi::buffers_children_dictionary;
 pub use self::ffi::FromFfi;
 pub use self::ffi::ToFfi;
 
diff --git a/src/array/primitive/ffi.rs b/src/array/primitive/ffi.rs
index 87980219c0..5168f322b7 100644
--- a/src/array/primitive/ffi.rs
+++ b/src/array/primitive/ffi.rs
@@ -24,7 +24,7 @@ unsafe impl<T: NativeType> ToFfi for PrimitiveArray<T> {
 
 unsafe impl<T: NativeType, A: ffi::ArrowArrayRef> FromFfi<A> for PrimitiveArray<T> {
     fn try_from_ffi(array: A) -> Result<Self> {
-        let data_type = array.data_type()?;
+        let data_type = array.field()?.data_type().clone();
         let length = array.array().len();
         let offset = array.array().offset();
         let mut validity = unsafe { array.validity() }?;
diff --git a/src/array/struct_.rs b/src/array/struct_.rs
index e1770a1290..86c2b09210 100644
--- a/src/array/struct_.rs
+++ b/src/array/struct_.rs
@@ -77,8 +77,8 @@ impl StructArray {
 }
 
 impl StructArray {
-    pub fn get_fields(datatype: &DataType) -> &[Field] {
-        if let DataType::Struct(fields) = datatype {
+    pub fn get_fields(data_type: &DataType) -> &[Field] {
+        if let DataType::Struct(fields) = data_type {
             fields
         } else {
             panic!("Wrong datatype passed to Struct.")
@@ -139,8 +139,8 @@ unsafe impl ToFfi for StructArray {
 
 unsafe impl<A: ffi::ArrowArrayRef> FromFfi<A> for StructArray {
     fn try_from_ffi(array: A) -> Result<Self> {
-        let data_type = array.data_type()?;
-        let fields = Self::get_fields(&data_type).to_vec();
+        let field = array.field()?;
+        let fields = Self::get_fields(field.data_type()).to_vec();
 
         let length = array.array().len();
         let offset = array.array().offset();
diff --git a/src/compute/aggregate/min_max.rs b/src/compute/aggregate/min_max.rs
index 8e9c0577e6..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::*;
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/ffi/array.rs b/src/ffi/array.rs
index 0929aaebc9..88575b60f5 100644
--- a/src/ffi/array.rs
+++ b/src/ffi/array.rs
@@ -22,7 +22,7 @@ use crate::array::{BooleanArray, FromFfi};
 use crate::error::{ArrowError, Result};
 use crate::types::days_ms;
 use crate::{
-    array::{Array, BinaryArray, ListArray, PrimitiveArray, StructArray, Utf8Array},
+    array::*,
     datatypes::{DataType, IntervalUnit},
 };
 
@@ -32,8 +32,8 @@ use crate::{
 /// * the data type is not supported
 /// * the interface is not valid (e.g. a null pointer)
 pub fn try_from<A: ArrowArrayRef>(array: A) -> Result<Box<dyn Array>> {
-    let data_type = array.data_type()?;
-    let array: Box<dyn Array> = match data_type {
+    let field = array.field()?;
+    let array: Box<dyn Array> = match field.data_type() {
         DataType::Boolean => Box::new(BooleanArray::try_from_ffi(array)?),
         DataType::Int8 => Box::new(PrimitiveArray::<i8>::try_from_ffi(array)?),
         DataType::Int16 => Box::new(PrimitiveArray::<i16>::try_from_ffi(array)?),
@@ -66,6 +66,17 @@ pub fn try_from<A: ArrowArrayRef>(array: A) -> Result<Box<dyn Array>> {
         DataType::List(_) => Box::new(ListArray::<i32>::try_from_ffi(array)?),
         DataType::LargeList(_) => Box::new(ListArray::<i64>::try_from_ffi(array)?),
         DataType::Struct(_) => Box::new(StructArray::try_from_ffi(array)?),
+        DataType::Dictionary(keys, _) => match keys.as_ref() {
+            DataType::Int8 => Box::new(DictionaryArray::<i8>::try_from_ffi(array)?),
+            DataType::Int16 => Box::new(DictionaryArray::<i16>::try_from_ffi(array)?),
+            DataType::Int32 => Box::new(DictionaryArray::<i32>::try_from_ffi(array)?),
+            DataType::Int64 => Box::new(DictionaryArray::<i64>::try_from_ffi(array)?),
+            DataType::UInt8 => Box::new(DictionaryArray::<u8>::try_from_ffi(array)?),
+            DataType::UInt16 => Box::new(DictionaryArray::<u16>::try_from_ffi(array)?),
+            DataType::UInt32 => Box::new(DictionaryArray::<u32>::try_from_ffi(array)?),
+            DataType::UInt64 => Box::new(DictionaryArray::<u64>::try_from_ffi(array)?),
+            _ => unreachable!(),
+        },
         data_type => {
             return Err(ArrowError::NotYetImplemented(format!(
                 "Reading DataType \"{}\" is not yet supported.",
@@ -80,7 +91,6 @@ pub fn try_from<A: ArrowArrayRef>(array: A) -> Result<Box<dyn Array>> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::array::*;
     use crate::datatypes::TimeUnit;
     use crate::{error::Result, ffi};
     use std::sync::Arc;
@@ -199,4 +209,16 @@ mod tests {
 
         test_round_trip(array)
     }
+
+    #[test]
+    fn test_dict() -> Result<()> {
+        let data = vec![Some("a"), Some("a"), None, Some("b")];
+
+        let mut array = MutableDictionaryArray::<i32, MutableUtf8Array<i32>>::new();
+        array.try_extend(data)?;
+
+        let array: DictionaryArray<i32> = array.into();
+
+        test_round_trip(array)
+    }
 }
diff --git a/src/ffi/ffi.rs b/src/ffi/ffi.rs
index f4c1f77669..756b0b04da 100644
--- a/src/ffi/ffi.rs
+++ b/src/ffi/ffi.rs
@@ -15,311 +15,21 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::{
-    ffi::CStr,
-    ffi::CString,
-    ptr::{self, NonNull},
-    sync::Arc,
-};
+use std::{ptr::NonNull, sync::Arc};
 
+use super::schema::{to_field, Ffi_ArrowSchema};
 use crate::{
-    array::{buffers_children, Array},
+    array::{buffers_children_dictionary, Array},
     bitmap::{utils::bytes_for, Bitmap},
     buffer::{
         bytes::{Bytes, Deallocation},
         Buffer,
     },
-    datatypes::{DataType, Field, IntervalUnit, TimeUnit},
+    datatypes::{DataType, Field},
     error::{ArrowError, Result},
     types::NativeType,
 };
 
-#[allow(dead_code)]
-struct SchemaPrivateData {
-    field: Field,
-    children_ptr: Box<[*mut Ffi_ArrowSchema]>,
-}
-
-/// ABI-compatible struct for `ArrowSchema` from C Data Interface
-/// See <https://arrow.apache.org/docs/format/CDataInterface.html#structure-definitions>
-/// This was created by bindgen
-#[repr(C)]
-#[derive(Debug)]
-pub struct Ffi_ArrowSchema {
-    format: *const ::std::os::raw::c_char,
-    name: *const ::std::os::raw::c_char,
-    metadata: *const ::std::os::raw::c_char,
-    flags: i64,
-    n_children: i64,
-    children: *mut *mut Ffi_ArrowSchema,
-    dictionary: *mut Ffi_ArrowSchema,
-    release: ::std::option::Option<unsafe extern "C" fn(arg1: *mut Ffi_ArrowSchema)>,
-    private_data: *mut ::std::os::raw::c_void,
-}
-
-// callback used to drop [Ffi_ArrowSchema] when it is exported.
-unsafe extern "C" fn c_release_schema(schema: *mut Ffi_ArrowSchema) {
-    if schema.is_null() {
-        return;
-    }
-    let schema = &mut *schema;
-
-    // take ownership back to release it.
-    CString::from_raw(schema.format as *mut std::os::raw::c_char);
-    CString::from_raw(schema.name as *mut std::os::raw::c_char);
-    let private = Box::from_raw(schema.private_data as *mut SchemaPrivateData);
-    for child in private.children_ptr.iter() {
-        let _ = Box::from_raw(*child);
-    }
-
-    schema.release = None;
-}
-
-impl Ffi_ArrowSchema {
-    /// create a new [`Ffi_ArrowSchema`]. This fails if the fields' [`DataType`] is not supported.
-    fn try_new(field: Field) -> Result<Ffi_ArrowSchema> {
-        let format = to_format(field.data_type())?;
-        let name = field.name().clone();
-
-        // allocate (and hold) the children
-        let children_vec = match field.data_type() {
-            DataType::List(field) => {
-                vec![Box::new(Ffi_ArrowSchema::try_new(field.as_ref().clone())?)]
-            }
-            DataType::LargeList(field) => {
-                vec![Box::new(Ffi_ArrowSchema::try_new(field.as_ref().clone())?)]
-            }
-            DataType::Struct(fields) => fields
-                .iter()
-                .map(|field| Ok(Box::new(Ffi_ArrowSchema::try_new(field.clone())?)))
-                .collect::<Result<Vec<_>>>()?,
-            _ => vec![],
-        };
-        // note: this cannot be done along with the above because the above is fallible and this op leaks.
-        let children_ptr = children_vec
-            .into_iter()
-            .map(Box::into_raw)
-            .collect::<Box<_>>();
-        let n_children = children_ptr.len() as i64;
-
-        let flags = field.is_nullable() as i64 * 2;
-
-        let mut private = Box::new(SchemaPrivateData {
-            field,
-            children_ptr,
-        });
-
-        // <https://arrow.apache.org/docs/format/CDataInterface.html#c.ArrowSchema>
-        Ok(Ffi_ArrowSchema {
-            format: CString::new(format).unwrap().into_raw(),
-            name: CString::new(name).unwrap().into_raw(),
-            metadata: std::ptr::null_mut(),
-            flags,
-            n_children,
-            children: private.children_ptr.as_mut_ptr(),
-            dictionary: std::ptr::null_mut(),
-            release: Some(c_release_schema),
-            private_data: Box::into_raw(private) as *mut ::std::os::raw::c_void,
-        })
-    }
-
-    /// create an empty [Ffi_ArrowSchema]
-    fn empty() -> Self {
-        Self {
-            format: std::ptr::null_mut(),
-            name: std::ptr::null_mut(),
-            metadata: std::ptr::null_mut(),
-            flags: 0,
-            n_children: 0,
-            children: ptr::null_mut(),
-            dictionary: std::ptr::null_mut(),
-            release: None,
-            private_data: std::ptr::null_mut(),
-        }
-    }
-
-    /// returns the format of this schema.
-    pub fn format(&self) -> &str {
-        assert!(!self.format.is_null());
-        // safe because the lifetime of `self.format` equals `self`
-        unsafe { CStr::from_ptr(self.format) }
-            .to_str()
-            .expect("The external API has a non-utf8 as format")
-    }
-
-    /// returns the name of this schema.
-    pub fn name(&self) -> &str {
-        assert!(!self.name.is_null());
-        // safe because the lifetime of `self.name` equals `self`
-        unsafe { CStr::from_ptr(self.name) }.to_str().unwrap()
-    }
-
-    pub fn child(&self, index: usize) -> &Self {
-        assert!(index < self.n_children as usize);
-        assert!(!self.name.is_null());
-        unsafe { self.children.add(index).as_ref().unwrap().as_ref().unwrap() }
-    }
-
-    pub fn nullable(&self) -> bool {
-        (self.flags / 2) & 1 == 1
-    }
-}
-
-impl Drop for Ffi_ArrowSchema {
-    fn drop(&mut self) {
-        match self.release {
-            None => (),
-            Some(release) => unsafe { release(self) },
-        };
-    }
-}
-
-/// See https://arrow.apache.org/docs/format/CDataInterface.html#data-type-description-format-strings
-fn to_field(schema: &Ffi_ArrowSchema) -> Result<Field> {
-    let data_type = match schema.format() {
-        "n" => DataType::Null,
-        "b" => DataType::Boolean,
-        "c" => DataType::Int8,
-        "C" => DataType::UInt8,
-        "s" => DataType::Int16,
-        "S" => DataType::UInt16,
-        "i" => DataType::Int32,
-        "I" => DataType::UInt32,
-        "l" => DataType::Int64,
-        "L" => DataType::UInt64,
-        "e" => DataType::Float16,
-        "f" => DataType::Float32,
-        "g" => DataType::Float64,
-        "z" => DataType::Binary,
-        "Z" => DataType::LargeBinary,
-        "u" => DataType::Utf8,
-        "U" => DataType::LargeUtf8,
-        "tdD" => DataType::Date32,
-        "tdm" => DataType::Date64,
-        "tts" => DataType::Time32(TimeUnit::Second),
-        "ttm" => DataType::Time32(TimeUnit::Millisecond),
-        "ttu" => DataType::Time64(TimeUnit::Microsecond),
-        "ttn" => DataType::Time64(TimeUnit::Nanosecond),
-        "tDs" => DataType::Duration(TimeUnit::Second),
-        "tDm" => DataType::Duration(TimeUnit::Millisecond),
-        "tDu" => DataType::Duration(TimeUnit::Microsecond),
-        "tDn" => DataType::Duration(TimeUnit::Nanosecond),
-        "tiM" => DataType::Interval(IntervalUnit::YearMonth),
-        "tiD" => DataType::Interval(IntervalUnit::DayTime),
-        "+l" => {
-            let child = schema.child(0);
-            DataType::List(Box::new(to_field(child)?))
-        }
-        "+L" => {
-            let child = schema.child(0);
-            DataType::LargeList(Box::new(to_field(child)?))
-        }
-        "+s" => {
-            let children = (0..schema.n_children as usize)
-                .map(|x| to_field(schema.child(x)))
-                .collect::<Result<Vec<_>>>()?;
-            DataType::Struct(children)
-        }
-        other => {
-            let parts = other.split(':').collect::<Vec<_>>();
-            if parts.len() == 2 && parts[0] == "tss" {
-                DataType::Timestamp(TimeUnit::Second, Some(parts[1].to_string()))
-            } else if parts.len() == 2 && parts[0] == "tsm" {
-                DataType::Timestamp(TimeUnit::Millisecond, Some(parts[1].to_string()))
-            } else if parts.len() == 2 && parts[0] == "tsu" {
-                DataType::Timestamp(TimeUnit::Microsecond, Some(parts[1].to_string()))
-            } else if parts.len() == 2 && parts[0] == "tsn" {
-                DataType::Timestamp(TimeUnit::Nanosecond, Some(parts[1].to_string()))
-            } else if parts.len() == 2 && parts[0] == "d" {
-                let parts = parts[1].split(',').collect::<Vec<_>>();
-                if parts.len() < 2 || parts.len() > 3 {
-                    return Err(ArrowError::Ffi(
-                        "Decimal must contain 2 or 3 comma-separated values".to_string(),
-                    ));
-                };
-                if parts.len() == 3 {
-                    let bit_width = parts[0].parse::<usize>().map_err(|_| {
-                        ArrowError::Ffi("Decimal bit width is not a valid integer".to_string())
-                    })?;
-                    if bit_width != 128 {
-                        return Err(ArrowError::Ffi("Decimal256 is not supported".to_string()));
-                    }
-                }
-                let precision = parts[0].parse::<usize>().map_err(|_| {
-                    ArrowError::Ffi("Decimal precision is not a valid integer".to_string())
-                })?;
-                let scale = parts[1].parse::<usize>().map_err(|_| {
-                    ArrowError::Ffi("Decimal scale is not a valid integer".to_string())
-                })?;
-                DataType::Decimal(precision, scale)
-            } else {
-                return Err(ArrowError::Ffi(format!(
-                    "The datatype \"{}\" is still not supported in Rust implementation",
-                    other
-                )));
-            }
-        }
-    };
-    Ok(Field::new(schema.name(), data_type, schema.nullable()))
-}
-
-/// the inverse of [to_datatype]
-fn to_format(data_type: &DataType) -> Result<String> {
-    Ok(match data_type {
-        DataType::Null => "n",
-        DataType::Boolean => "b",
-        DataType::Int8 => "c",
-        DataType::UInt8 => "C",
-        DataType::Int16 => "s",
-        DataType::UInt16 => "S",
-        DataType::Int32 => "i",
-        DataType::UInt32 => "I",
-        DataType::Int64 => "l",
-        DataType::UInt64 => "L",
-        DataType::Float16 => "e",
-        DataType::Float32 => "f",
-        DataType::Float64 => "g",
-        DataType::Binary => "z",
-        DataType::LargeBinary => "Z",
-        DataType::Utf8 => "u",
-        DataType::LargeUtf8 => "U",
-        DataType::Date32 => "tdD",
-        DataType::Date64 => "tdm",
-        DataType::Time32(TimeUnit::Second) => "tts",
-        DataType::Time32(TimeUnit::Millisecond) => "ttm",
-        DataType::Time64(TimeUnit::Microsecond) => "ttu",
-        DataType::Time64(TimeUnit::Nanosecond) => "ttn",
-        DataType::Duration(TimeUnit::Second) => "tDs",
-        DataType::Duration(TimeUnit::Millisecond) => "tDm",
-        DataType::Duration(TimeUnit::Microsecond) => "tDu",
-        DataType::Duration(TimeUnit::Nanosecond) => "tDn",
-        DataType::Interval(IntervalUnit::YearMonth) => "tiM",
-        DataType::Interval(IntervalUnit::DayTime) => "tiD",
-        DataType::Timestamp(unit, tz) => {
-            let unit = match unit {
-                TimeUnit::Second => "s",
-                TimeUnit::Millisecond => "m",
-                TimeUnit::Microsecond => "u",
-                TimeUnit::Nanosecond => "n",
-            };
-            return Ok(format!(
-                "ts{}:{}",
-                unit,
-                tz.as_ref().map(|x| x.as_ref()).unwrap_or("")
-            ));
-        }
-        DataType::Decimal(precision, scale) => return Ok(format!("d:{},{}", precision, scale)),
-        DataType::List(_) => "+l",
-        DataType::LargeList(_) => "+L",
-        DataType::Struct(_) => "+s",
-        DataType::FixedSizeBinary(size) => return Ok(format!("w{}", size)),
-        DataType::FixedSizeList(_, size) => return Ok(format!("+w:{}", size)),
-        DataType::Union(_) => todo!(),
-        _ => todo!(),
-    }
-    .to_string())
-}
-
 /// ABI-compatible struct for ArrowArray from C Data Interface
 /// See <https://arrow.apache.org/docs/format/CDataInterface.html#structure-definitions>
 /// This was created by bindgen
@@ -365,6 +75,10 @@ unsafe extern "C" fn c_release_array(array: *mut Ffi_ArrowArray) {
         let _ = Box::from_raw(*child);
     }
 
+    if let Some(ptr) = private.dictionary_ptr {
+        let _ = Box::from_raw(ptr);
+    }
+
     array.release = None;
 }
 
@@ -373,6 +87,7 @@ struct PrivateData {
     array: Arc<dyn Array>,
     buffers_ptr: Box<[*const std::os::raw::c_void]>,
     children_ptr: Box<[*mut Ffi_ArrowArray]>,
+    dictionary_ptr: Option<*mut Ffi_ArrowArray>,
 }
 
 impl Ffi_ArrowArray {
@@ -381,7 +96,7 @@ impl Ffi_ArrowArray {
     /// This method releases `buffers`. Consumers of this struct *must* call `release` before
     /// releasing this struct, or contents in `buffers` leak.
     fn new(array: Arc<dyn Array>) -> Self {
-        let (buffers, children) = buffers_children(array.as_ref());
+        let (buffers, children, dictionary) = buffers_children_dictionary(array.as_ref());
 
         let buffers_ptr = buffers
             .iter()
@@ -399,6 +114,9 @@ impl Ffi_ArrowArray {
             .collect::<Box<_>>();
         let n_children = children_ptr.len() as i64;
 
+        let dictionary_ptr =
+            dictionary.map(|array| Box::into_raw(Box::new(Ffi_ArrowArray::new(array))));
+
         let length = array.len() as i64;
         let null_count = array.null_count() as i64;
 
@@ -406,6 +124,7 @@ impl Ffi_ArrowArray {
             array,
             buffers_ptr,
             children_ptr,
+            dictionary_ptr,
         });
 
         Self {
@@ -416,7 +135,7 @@ impl Ffi_ArrowArray {
             n_children,
             buffers: private_data.buffers_ptr.as_mut_ptr(),
             children: private_data.children_ptr.as_mut_ptr(),
-            dictionary: std::ptr::null_mut(),
+            dictionary: private_data.dictionary_ptr.unwrap_or(std::ptr::null_mut()),
             release: Some(c_release_array),
             private_data: Box::into_raw(private_data) as *mut ::std::os::raw::c_void,
         }
@@ -570,18 +289,30 @@ fn create_child(
 ) -> Result<ArrowArrayChild<'static>> {
     assert!(index < array.n_children as usize);
     assert!(!array.children.is_null());
-    assert!(!array.children.is_null());
     unsafe {
         let arr_ptr = *array.children.add(index);
-        let schema_ptr = *schema.children.add(index);
+        let schema_ptr = schema.child(index);
         assert!(!arr_ptr.is_null());
-        assert!(!schema_ptr.is_null());
         let arr_ptr = &*arr_ptr;
-        let schema_ptr = &*schema_ptr;
         Ok(ArrowArrayChild::from_raw(arr_ptr, schema_ptr, parent))
     }
 }
 
+fn create_dictionary(
+    array: &Ffi_ArrowArray,
+    schema: &Ffi_ArrowSchema,
+    parent: Arc<ArrowArray>,
+) -> Result<Option<ArrowArrayChild<'static>>> {
+    let schema = schema.dictionary();
+    if let Some(schema) = schema {
+        assert!(!array.dictionary.is_null());
+        let array = unsafe { &*array.dictionary };
+        Ok(Some(ArrowArrayChild::from_raw(array, schema, parent)))
+    } else {
+        Ok(None)
+    }
+}
+
 pub trait ArrowArrayRef {
     fn deallocation(&self) -> Deallocation {
         Deallocation::Foreign(self.parent().clone())
@@ -608,7 +339,7 @@ pub trait ArrowArrayRef {
         // +1 to ignore null bitmap
         create_buffer::<T>(
             self.array(),
-            &self.data_type()?,
+            self.field()?.data_type(),
             self.deallocation(),
             index + 1,
         )
@@ -626,10 +357,14 @@ pub trait ArrowArrayRef {
         create_child(self.array(), self.schema(), self.parent().clone(), index)
     }
 
+    fn dictionary(&self) -> Result<Option<ArrowArrayChild>> {
+        create_dictionary(self.array(), self.schema(), self.parent().clone())
+    }
+
     fn parent(&self) -> &Arc<ArrowArray>;
     fn array(&self) -> &Ffi_ArrowArray;
     fn schema(&self) -> &Ffi_ArrowSchema;
-    fn data_type(&self) -> Result<DataType>;
+    fn field(&self) -> Result<Field>;
 }
 
 /// Struct used to move an Array from and to the C Data Interface.
@@ -659,8 +394,8 @@ pub struct ArrowArray {
 
 impl ArrowArrayRef for Arc<ArrowArray> {
     /// the data_type as declared in the schema
-    fn data_type(&self) -> Result<DataType> {
-        to_field(&self.schema).map(|x| x.data_type().clone())
+    fn field(&self) -> Result<Field> {
+        to_field(&self.schema)
     }
 
     fn parent(&self) -> &Arc<ArrowArray> {
@@ -685,8 +420,8 @@ pub struct ArrowArrayChild<'a> {
 
 impl<'a> ArrowArrayRef for ArrowArrayChild<'a> {
     /// the data_type as declared in the schema
-    fn data_type(&self) -> Result<DataType> {
-        to_field(self.schema).map(|x| x.data_type().clone())
+    fn field(&self) -> Result<Field> {
+        to_field(self.schema)
     }
 
     fn parent(&self) -> &Arc<ArrowArray> {
diff --git a/src/ffi/mod.rs b/src/ffi/mod.rs
index 6d3fad44ea..19fa83a8fb 100644
--- a/src/ffi/mod.rs
+++ b/src/ffi/mod.rs
@@ -3,6 +3,7 @@
 mod array;
 #[allow(clippy::module_inception)]
 mod ffi;
+mod schema;
 
 pub use array::try_from;
 pub use ffi::{create_empty, export_to_c, ArrowArray, ArrowArrayRef};
diff --git a/src/ffi/schema.rs b/src/ffi/schema.rs
new file mode 100644
index 0000000000..b9edf4d544
--- /dev/null
+++ b/src/ffi/schema.rs
@@ -0,0 +1,324 @@
+use std::{ffi::CStr, ffi::CString, ptr};
+
+use crate::{
+    datatypes::{DataType, Field, IntervalUnit, TimeUnit},
+    error::{ArrowError, Result},
+};
+
+#[allow(dead_code)]
+struct SchemaPrivateData {
+    field: Field,
+    children_ptr: Box<[*mut Ffi_ArrowSchema]>,
+    dictionary: Option<*mut Ffi_ArrowSchema>,
+}
+
+/// ABI-compatible struct for `ArrowSchema` from C Data Interface
+/// See <https://arrow.apache.org/docs/format/CDataInterface.html#structure-definitions>
+/// This was created by bindgen
+#[repr(C)]
+#[derive(Debug)]
+pub struct Ffi_ArrowSchema {
+    format: *const ::std::os::raw::c_char,
+    name: *const ::std::os::raw::c_char,
+    metadata: *const ::std::os::raw::c_char,
+    flags: i64,
+    n_children: i64,
+    children: *mut *mut Ffi_ArrowSchema,
+    dictionary: *mut Ffi_ArrowSchema,
+    release: ::std::option::Option<unsafe extern "C" fn(arg1: *mut Ffi_ArrowSchema)>,
+    private_data: *mut ::std::os::raw::c_void,
+}
+
+// callback used to drop [Ffi_ArrowSchema] when it is exported.
+unsafe extern "C" fn c_release_schema(schema: *mut Ffi_ArrowSchema) {
+    if schema.is_null() {
+        return;
+    }
+    let schema = &mut *schema;
+
+    // take ownership back to release it.
+    CString::from_raw(schema.format as *mut std::os::raw::c_char);
+    CString::from_raw(schema.name as *mut std::os::raw::c_char);
+    let private = Box::from_raw(schema.private_data as *mut SchemaPrivateData);
+    for child in private.children_ptr.iter() {
+        let _ = Box::from_raw(*child);
+    }
+
+    if let Some(ptr) = private.dictionary {
+        let _ = Box::from_raw(ptr);
+    }
+
+    schema.release = None;
+}
+
+impl Ffi_ArrowSchema {
+    /// create a new [`Ffi_ArrowSchema`]. This fails if the fields' [`DataType`] is not supported.
+    pub fn try_new(field: Field) -> Result<Ffi_ArrowSchema> {
+        let format = to_format(field.data_type())?;
+        let name = field.name().clone();
+
+        // allocate (and hold) the children
+        let children_vec = match field.data_type() {
+            DataType::List(field) => {
+                vec![Box::new(Ffi_ArrowSchema::try_new(field.as_ref().clone())?)]
+            }
+            DataType::LargeList(field) => {
+                vec![Box::new(Ffi_ArrowSchema::try_new(field.as_ref().clone())?)]
+            }
+            DataType::Struct(fields) => fields
+                .iter()
+                .map(|field| Ok(Box::new(Ffi_ArrowSchema::try_new(field.clone())?)))
+                .collect::<Result<Vec<_>>>()?,
+            _ => vec![],
+        };
+        // note: this cannot be done along with the above because the above is fallible and this op leaks.
+        let children_ptr = children_vec
+            .into_iter()
+            .map(Box::into_raw)
+            .collect::<Box<_>>();
+        let n_children = children_ptr.len() as i64;
+
+        let flags = field.is_nullable() as i64 * 2;
+
+        let dictionary = if let DataType::Dictionary(_, values) = field.data_type() {
+            // we do not store field info in the dict values, so can't recover it all :(
+            let field = Field::new("item", values.as_ref().clone(), true);
+            Some(Box::new(Ffi_ArrowSchema::try_new(field)?))
+        } else {
+            None
+        };
+
+        let mut private = Box::new(SchemaPrivateData {
+            field,
+            children_ptr,
+            dictionary: dictionary.map(Box::into_raw),
+        });
+
+        // <https://arrow.apache.org/docs/format/CDataInterface.html#c.ArrowSchema>
+        Ok(Ffi_ArrowSchema {
+            format: CString::new(format).unwrap().into_raw(),
+            name: CString::new(name).unwrap().into_raw(),
+            metadata: std::ptr::null_mut(),
+            flags,
+            n_children,
+            children: private.children_ptr.as_mut_ptr(),
+            dictionary: private.dictionary.unwrap_or(std::ptr::null_mut()),
+            release: Some(c_release_schema),
+            private_data: Box::into_raw(private) as *mut ::std::os::raw::c_void,
+        })
+    }
+
+    /// create an empty [Ffi_ArrowSchema]
+    pub fn empty() -> Self {
+        Self {
+            format: std::ptr::null_mut(),
+            name: std::ptr::null_mut(),
+            metadata: std::ptr::null_mut(),
+            flags: 0,
+            n_children: 0,
+            children: ptr::null_mut(),
+            dictionary: std::ptr::null_mut(),
+            release: None,
+            private_data: std::ptr::null_mut(),
+        }
+    }
+
+    /// returns the format of this schema.
+    pub fn format(&self) -> &str {
+        assert!(!self.format.is_null());
+        // safe because the lifetime of `self.format` equals `self`
+        unsafe { CStr::from_ptr(self.format) }
+            .to_str()
+            .expect("The external API has a non-utf8 as format")
+    }
+
+    /// returns the name of this schema.
+    pub fn name(&self) -> &str {
+        assert!(!self.name.is_null());
+        // safe because the lifetime of `self.name` equals `self`
+        unsafe { CStr::from_ptr(self.name) }.to_str().unwrap()
+    }
+
+    pub fn child(&self, index: usize) -> &'static Self {
+        assert!(index < self.n_children as usize);
+        assert!(!self.name.is_null());
+        unsafe { self.children.add(index).as_ref().unwrap().as_ref().unwrap() }
+    }
+
+    pub fn dictionary(&self) -> Option<&'static Self> {
+        if self.dictionary.is_null() {
+            return None;
+        };
+        Some(unsafe { self.dictionary.as_ref().unwrap() })
+    }
+
+    pub fn nullable(&self) -> bool {
+        (self.flags / 2) & 1 == 1
+    }
+}
+
+impl Drop for Ffi_ArrowSchema {
+    fn drop(&mut self) {
+        match self.release {
+            None => (),
+            Some(release) => unsafe { release(self) },
+        };
+    }
+}
+
+pub fn to_field(schema: &Ffi_ArrowSchema) -> Result<Field> {
+    let dictionary = schema.dictionary();
+    let data_type = if let Some(dictionary) = dictionary {
+        let indices_data_type = to_data_type(schema)?;
+        let values_data_type = to_data_type(dictionary)?;
+        DataType::Dictionary(Box::new(indices_data_type), Box::new(values_data_type))
+    } else {
+        to_data_type(schema)?
+    };
+    Ok(Field::new(schema.name(), data_type, schema.nullable()))
+}
+
+fn to_data_type(schema: &Ffi_ArrowSchema) -> Result<DataType> {
+    Ok(match schema.format() {
+        "n" => DataType::Null,
+        "b" => DataType::Boolean,
+        "c" => DataType::Int8,
+        "C" => DataType::UInt8,
+        "s" => DataType::Int16,
+        "S" => DataType::UInt16,
+        "i" => DataType::Int32,
+        "I" => DataType::UInt32,
+        "l" => DataType::Int64,
+        "L" => DataType::UInt64,
+        "e" => DataType::Float16,
+        "f" => DataType::Float32,
+        "g" => DataType::Float64,
+        "z" => DataType::Binary,
+        "Z" => DataType::LargeBinary,
+        "u" => DataType::Utf8,
+        "U" => DataType::LargeUtf8,
+        "tdD" => DataType::Date32,
+        "tdm" => DataType::Date64,
+        "tts" => DataType::Time32(TimeUnit::Second),
+        "ttm" => DataType::Time32(TimeUnit::Millisecond),
+        "ttu" => DataType::Time64(TimeUnit::Microsecond),
+        "ttn" => DataType::Time64(TimeUnit::Nanosecond),
+        "tDs" => DataType::Duration(TimeUnit::Second),
+        "tDm" => DataType::Duration(TimeUnit::Millisecond),
+        "tDu" => DataType::Duration(TimeUnit::Microsecond),
+        "tDn" => DataType::Duration(TimeUnit::Nanosecond),
+        "tiM" => DataType::Interval(IntervalUnit::YearMonth),
+        "tiD" => DataType::Interval(IntervalUnit::DayTime),
+        "+l" => {
+            let child = schema.child(0);
+            DataType::List(Box::new(to_field(child)?))
+        }
+        "+L" => {
+            let child = schema.child(0);
+            DataType::LargeList(Box::new(to_field(child)?))
+        }
+        "+s" => {
+            let children = (0..schema.n_children as usize)
+                .map(|x| to_field(schema.child(x)))
+                .collect::<Result<Vec<_>>>()?;
+            DataType::Struct(children)
+        }
+        other => {
+            let parts = other.split(':').collect::<Vec<_>>();
+            if parts.len() == 2 && parts[0] == "tss" {
+                DataType::Timestamp(TimeUnit::Second, Some(parts[1].to_string()))
+            } else if parts.len() == 2 && parts[0] == "tsm" {
+                DataType::Timestamp(TimeUnit::Millisecond, Some(parts[1].to_string()))
+            } else if parts.len() == 2 && parts[0] == "tsu" {
+                DataType::Timestamp(TimeUnit::Microsecond, Some(parts[1].to_string()))
+            } else if parts.len() == 2 && parts[0] == "tsn" {
+                DataType::Timestamp(TimeUnit::Nanosecond, Some(parts[1].to_string()))
+            } else if parts.len() == 2 && parts[0] == "d" {
+                let parts = parts[1].split(',').collect::<Vec<_>>();
+                if parts.len() < 2 || parts.len() > 3 {
+                    return Err(ArrowError::Ffi(
+                        "Decimal must contain 2 or 3 comma-separated values".to_string(),
+                    ));
+                };
+                if parts.len() == 3 {
+                    let bit_width = parts[0].parse::<usize>().map_err(|_| {
+                        ArrowError::Ffi("Decimal bit width is not a valid integer".to_string())
+                    })?;
+                    if bit_width != 128 {
+                        return Err(ArrowError::Ffi("Decimal256 is not supported".to_string()));
+                    }
+                }
+                let precision = parts[0].parse::<usize>().map_err(|_| {
+                    ArrowError::Ffi("Decimal precision is not a valid integer".to_string())
+                })?;
+                let scale = parts[1].parse::<usize>().map_err(|_| {
+                    ArrowError::Ffi("Decimal scale is not a valid integer".to_string())
+                })?;
+                DataType::Decimal(precision, scale)
+            } else {
+                return Err(ArrowError::Ffi(format!(
+                    "The datatype \"{}\" is still not supported in Rust implementation",
+                    other
+                )));
+            }
+        }
+    })
+}
+
+/// the inverse of [to_field]
+fn to_format(data_type: &DataType) -> Result<String> {
+    Ok(match data_type {
+        DataType::Null => "n",
+        DataType::Boolean => "b",
+        DataType::Int8 => "c",
+        DataType::UInt8 => "C",
+        DataType::Int16 => "s",
+        DataType::UInt16 => "S",
+        DataType::Int32 => "i",
+        DataType::UInt32 => "I",
+        DataType::Int64 => "l",
+        DataType::UInt64 => "L",
+        DataType::Float16 => "e",
+        DataType::Float32 => "f",
+        DataType::Float64 => "g",
+        DataType::Binary => "z",
+        DataType::LargeBinary => "Z",
+        DataType::Utf8 => "u",
+        DataType::LargeUtf8 => "U",
+        DataType::Date32 => "tdD",
+        DataType::Date64 => "tdm",
+        DataType::Time32(TimeUnit::Second) => "tts",
+        DataType::Time32(TimeUnit::Millisecond) => "ttm",
+        DataType::Time64(TimeUnit::Microsecond) => "ttu",
+        DataType::Time64(TimeUnit::Nanosecond) => "ttn",
+        DataType::Duration(TimeUnit::Second) => "tDs",
+        DataType::Duration(TimeUnit::Millisecond) => "tDm",
+        DataType::Duration(TimeUnit::Microsecond) => "tDu",
+        DataType::Duration(TimeUnit::Nanosecond) => "tDn",
+        DataType::Interval(IntervalUnit::YearMonth) => "tiM",
+        DataType::Interval(IntervalUnit::DayTime) => "tiD",
+        DataType::Timestamp(unit, tz) => {
+            let unit = match unit {
+                TimeUnit::Second => "s",
+                TimeUnit::Millisecond => "m",
+                TimeUnit::Microsecond => "u",
+                TimeUnit::Nanosecond => "n",
+            };
+            return Ok(format!(
+                "ts{}:{}",
+                unit,
+                tz.as_ref().map(|x| x.as_ref()).unwrap_or("")
+            ));
+        }
+        DataType::Decimal(precision, scale) => return Ok(format!("d:{},{}", precision, scale)),
+        DataType::List(_) => "+l",
+        DataType::LargeList(_) => "+L",
+        DataType::Struct(_) => "+s",
+        DataType::FixedSizeBinary(size) => return Ok(format!("w{}", size)),
+        DataType::FixedSizeList(_, size) => return Ok(format!("+w:{}", size)),
+        DataType::Union(_) => todo!(),
+        DataType::Dictionary(index, _) => return to_format(index.as_ref()),
+        _ => todo!(),
+    }
+    .to_string())
+}
diff --git a/src/io/parquet/mod.rs b/src/io/parquet/mod.rs
index 0a4e80f0b5..0ac846cabb 100644
--- a/src/io/parquet/mod.rs
+++ b/src/io/parquet/mod.rs
@@ -36,8 +36,13 @@ mod tests {
     ) -> Result<ArrayStats> {
         let metadata = read::read_metadata(&mut reader)?;
 
-        let mut reader =
-            read::RecordReader::try_new(reader, Some(vec![column]), None, Arc::new(|_, _| true))?;
+        let mut reader = read::RecordReader::try_new(
+            reader,
+            Some(vec![column]),
+            None,
+            Arc::new(|_, _| true),
+            None,
+        )?;
 
         let statistics = metadata.row_groups[row_group]
             .column(column)
@@ -458,7 +463,7 @@ mod tests_integration {
 
     fn integration_read(data: &[u8]) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
         let reader = Cursor::new(data);
-        let reader = read::RecordReader::try_new(reader, None, None, Arc::new(|_, _| true))?;
+        let reader = read::RecordReader::try_new(reader, None, None, Arc::new(|_, _| true), None)?;
         let schema = reader.schema().clone();
         let batches = reader.collect::<Result<Vec<_>>>()?;
 
diff --git a/src/io/parquet/read/binary/basic.rs b/src/io/parquet/read/binary/basic.rs
index 8ceca3ebbb..a418c00ac5 100644
--- a/src/io/parquet/read/binary/basic.rs
+++ b/src/io/parquet/read/binary/basic.rs
@@ -1,3 +1,4 @@
+use futures::{pin_mut, Stream, StreamExt};
 use parquet2::{
     encoding::{bitpacking, delta_length_byte_array, hybrid_rle, uleb128, Encoding},
     metadata::{ColumnChunkMetaData, ColumnDescriptor},
@@ -353,3 +354,47 @@ where
         _ => unreachable!(),
     })
 }
+
+pub async fn stream_to_array<O, I, E>(
+    pages: I,
+    metadata: &ColumnChunkMetaData,
+    data_type: &DataType,
+) -> Result<Box<dyn Array>>
+where
+    ArrowError: From<E>,
+    O: Offset,
+    E: Clone,
+    I: Stream<Item = std::result::Result<DataPage, E>>,
+{
+    let capacity = metadata.num_values() as usize;
+    let mut values = MutableBuffer::<u8>::with_capacity(0);
+    let mut offsets = MutableBuffer::<O>::with_capacity(1 + capacity);
+    offsets.push(O::default());
+    let mut validity = MutableBitmap::with_capacity(capacity);
+
+    pin_mut!(pages); // needed for iteration
+
+    while let Some(page) = pages.next().await {
+        extend_from_page(
+            page.as_ref().map_err(|x| x.clone())?,
+            metadata.descriptor(),
+            &mut offsets,
+            &mut values,
+            &mut validity,
+        )?
+    }
+
+    Ok(match data_type {
+        DataType::LargeBinary | DataType::Binary => Box::new(BinaryArray::from_data(
+            offsets.into(),
+            values.into(),
+            validity.into(),
+        )),
+        DataType::LargeUtf8 | DataType::Utf8 => Box::new(Utf8Array::from_data(
+            offsets.into(),
+            values.into(),
+            validity.into(),
+        )),
+        _ => unreachable!(),
+    })
+}
diff --git a/src/io/parquet/read/binary/mod.rs b/src/io/parquet/read/binary/mod.rs
index 13b4715437..912f1bb026 100644
--- a/src/io/parquet/read/binary/mod.rs
+++ b/src/io/parquet/read/binary/mod.rs
@@ -2,4 +2,5 @@ mod basic;
 mod nested;
 
 pub use basic::iter_to_array;
+pub use basic::stream_to_array;
 pub use nested::iter_to_array as iter_to_array_nested;
diff --git a/src/io/parquet/read/boolean/basic.rs b/src/io/parquet/read/boolean/basic.rs
index 8e3e155d79..221da32be0 100644
--- a/src/io/parquet/read/boolean/basic.rs
+++ b/src/io/parquet/read/boolean/basic.rs
@@ -5,6 +5,8 @@ use crate::{
 };
 
 use super::super::utils;
+
+use futures::{pin_mut, Stream, StreamExt};
 use parquet2::{
     encoding::{hybrid_rle, Encoding},
     metadata::{ColumnChunkMetaData, ColumnDescriptor},
@@ -86,6 +88,30 @@ where
     Ok(BooleanArray::from_data(values.into(), validity.into()))
 }
 
+pub async fn stream_to_array<I, E>(pages: I, metadata: &ColumnChunkMetaData) -> Result<BooleanArray>
+where
+    ArrowError: From<E>,
+    E: Clone,
+    I: Stream<Item = std::result::Result<DataPage, E>>,
+{
+    let capacity = metadata.num_values() as usize;
+    let mut values = MutableBitmap::with_capacity(capacity);
+    let mut validity = MutableBitmap::with_capacity(capacity);
+
+    pin_mut!(pages); // needed for iteration
+
+    while let Some(page) = pages.next().await {
+        extend_from_page(
+            page.as_ref().map_err(|x| x.clone())?,
+            metadata.descriptor(),
+            &mut values,
+            &mut validity,
+        )?
+    }
+
+    Ok(BooleanArray::from_data(values.into(), validity.into()))
+}
+
 fn extend_from_page(
     page: &DataPage,
     descriptor: &ColumnDescriptor,
diff --git a/src/io/parquet/read/boolean/mod.rs b/src/io/parquet/read/boolean/mod.rs
index 13b4715437..912f1bb026 100644
--- a/src/io/parquet/read/boolean/mod.rs
+++ b/src/io/parquet/read/boolean/mod.rs
@@ -2,4 +2,5 @@ mod basic;
 mod nested;
 
 pub use basic::iter_to_array;
+pub use basic::stream_to_array;
 pub use nested::iter_to_array as iter_to_array_nested;
diff --git a/src/io/parquet/read/fixed_size_binary.rs b/src/io/parquet/read/fixed_size_binary.rs
index e92dd0619b..6ea5e78250 100644
--- a/src/io/parquet/read/fixed_size_binary.rs
+++ b/src/io/parquet/read/fixed_size_binary.rs
@@ -1,3 +1,4 @@
+use futures::{pin_mut, Stream, StreamExt};
 use parquet2::{
     encoding::{bitpacking, hybrid_rle, uleb128, Encoding},
     page::{DataPage, DataPageHeader, DataPageHeaderExt, FixedLenByteArrayPageDict},
@@ -159,6 +160,39 @@ where
     ))
 }
 
+pub async fn stream_to_array<I, E>(
+    pages: I,
+    size: i32,
+    metadata: &ColumnChunkMetaData,
+) -> Result<FixedSizeBinaryArray>
+where
+    ArrowError: From<E>,
+    E: Clone,
+    I: Stream<Item = std::result::Result<DataPage, E>>,
+{
+    let capacity = metadata.num_values() as usize;
+    let mut values = MutableBuffer::<u8>::with_capacity(capacity * size as usize);
+    let mut validity = MutableBitmap::with_capacity(capacity);
+
+    pin_mut!(pages); // needed for iteration
+
+    while let Some(page) = pages.next().await {
+        extend_from_page(
+            page.as_ref().map_err(|x| x.clone())?,
+            size,
+            metadata.descriptor(),
+            &mut values,
+            &mut validity,
+        )?
+    }
+
+    Ok(FixedSizeBinaryArray::from_data(
+        DataType::FixedSizeBinary(size),
+        values.into(),
+        validity.into(),
+    ))
+}
+
 pub(crate) fn extend_from_page(
     page: &DataPage,
     size: i32,
diff --git a/src/io/parquet/read/mod.rs b/src/io/parquet/read/mod.rs
index 8cddeb220b..8d86ab05d0 100644
--- a/src/io/parquet/read/mod.rs
+++ b/src/io/parquet/read/mod.rs
@@ -1,31 +1,17 @@
-use std::io::{Read, Seek};
-
-mod binary;
-mod boolean;
-mod fixed_size_binary;
-mod nested_utils;
-mod primitive;
-mod record_batch;
-pub mod schema;
-pub mod statistics;
-mod utils;
-
-use crate::{
-    array::Array,
-    datatypes::{DataType, IntervalUnit, TimeUnit},
-    error::{ArrowError, Result},
+use std::{
+    io::{Read, Seek},
+    sync::Arc,
 };
 
-pub use record_batch::RecordReader;
-pub use schema::{get_schema, is_type_nullable, FileMetaData};
-
+use futures::{AsyncRead, AsyncSeek, Stream};
 pub use parquet2::{
     error::ParquetError,
     metadata::{ColumnChunkMetaData, ColumnDescriptor, RowGroupMetaData},
     page::{CompressedDataPage, DataPage, DataPageHeader},
     read::{
-        decompress, get_page_iterator as _get_page_iterator, read_metadata as _read_metadata,
-        streaming_iterator, Decompressor, PageIterator, StreamingIterator,
+        decompress, get_page_iterator as _get_page_iterator, get_page_stream as _get_page_stream,
+        read_metadata as _read_metadata, read_metadata_async as _read_metadata_async,
+        streaming_iterator, Decompressor, PageFilter, PageIterator, StreamingIterator,
     },
     schema::types::{
         LogicalType, ParquetType, PhysicalType, PrimitiveConvertedType,
@@ -34,24 +20,69 @@ pub use parquet2::{
     types::int96_to_i64_ns,
 };
 
+use crate::{
+    array::Array,
+    datatypes::{DataType, IntervalUnit, TimeUnit},
+    error::{ArrowError, Result},
+};
+
+mod binary;
+mod boolean;
+mod fixed_size_binary;
+mod nested_utils;
+mod primitive;
+mod record_batch;
+pub mod schema;
+pub mod statistics;
+mod utils;
+
+pub use record_batch::RecordReader;
+pub use schema::{get_schema, is_type_nullable, FileMetaData};
+
 /// Creates a new iterator of compressed pages.
 pub fn get_page_iterator<'b, RR: Read + Seek>(
     metadata: &FileMetaData,
     row_group: usize,
     column: usize,
     reader: &'b mut RR,
+    pages_filter: Option<PageFilter>,
     buffer: Vec<u8>,
 ) -> Result<PageIterator<'b, RR>> {
     Ok(_get_page_iterator(
-        metadata, row_group, column, reader, None, buffer,
+        metadata,
+        row_group,
+        column,
+        reader,
+        pages_filter,
+        buffer,
     )?)
 }
 
-/// Reads parquets' metadata.
+/// Creates a new iterator of compressed pages.
+pub async fn get_page_stream<'a, RR: AsyncRead + Unpin + Send + AsyncSeek>(
+    metadata: &'a FileMetaData,
+    row_group: usize,
+    column: usize,
+    reader: &'a mut RR,
+    pages_filter: Option<PageFilter>,
+    buffer: Vec<u8>,
+) -> Result<impl Stream<Item = std::result::Result<CompressedDataPage, ParquetError>> + 'a> {
+    let pages_filter = pages_filter.unwrap_or_else(|| Arc::new(|_, _| true));
+    Ok(_get_page_stream(metadata, row_group, column, reader, buffer, pages_filter).await?)
+}
+
+/// Reads parquets' metadata syncronously.
 pub fn read_metadata<R: Read + Seek>(reader: &mut R) -> Result<FileMetaData> {
     Ok(_read_metadata(reader)?)
 }
 
+/// Reads parquets' metadata asynchronously.
+pub async fn read_metadata_async<R: AsyncRead + AsyncSeek + Send + Unpin>(
+    reader: &mut R,
+) -> Result<FileMetaData> {
+    Ok(_read_metadata_async(reader).await?)
+}
+
 pub fn page_iter_to_array<
     I: StreamingIterator<Item = std::result::Result<DataPage, ParquetError>>,
 >(
@@ -163,6 +194,65 @@ pub fn page_iter_to_array<
     }
 }
 
+// Converts an async stream of compressed data pages into an [`Array`].
+pub async fn page_stream_to_array<I: Stream<Item = std::result::Result<DataPage, ParquetError>>>(
+    pages: I,
+    metadata: &ColumnChunkMetaData,
+    data_type: DataType,
+) -> Result<Box<dyn Array>> {
+    use DataType::*;
+    match data_type {
+        // INT32
+        UInt8 => primitive::stream_to_array(pages, metadata, data_type, |x: i32| x as u8).await,
+        UInt16 => primitive::stream_to_array(pages, metadata, data_type, |x: i32| x as u16).await,
+        UInt32 => primitive::stream_to_array(pages, metadata, data_type, |x: i32| x as u32).await,
+        Int8 => primitive::stream_to_array(pages, metadata, data_type, |x: i32| x as i8).await,
+        Int16 => primitive::stream_to_array(pages, metadata, data_type, |x: i32| x as i16).await,
+        Int32 | Date32 | Time32(_) | Interval(IntervalUnit::YearMonth) => {
+            primitive::stream_to_array(pages, metadata, data_type, |x: i32| x as i32).await
+        }
+
+        Timestamp(TimeUnit::Nanosecond, None) => match metadata.descriptor().type_() {
+            ParquetType::PrimitiveType { physical_type, .. } => match physical_type {
+                PhysicalType::Int96 => {
+                    primitive::stream_to_array(
+                        pages,
+                        metadata,
+                        DataType::Timestamp(TimeUnit::Nanosecond, None),
+                        int96_to_i64_ns,
+                    )
+                    .await
+                }
+                _ => primitive::stream_to_array(pages, metadata, data_type, |x: i64| x).await,
+            },
+            _ => unreachable!(),
+        },
+
+        // INT64
+        Int64 | Date64 | Time64(_) | Duration(_) | Timestamp(_, _) => {
+            primitive::stream_to_array(pages, metadata, data_type, |x: i64| x).await
+        }
+        UInt64 => primitive::stream_to_array(pages, metadata, data_type, |x: i64| x as u64).await,
+
+        Float32 => primitive::stream_to_array(pages, metadata, data_type, |x: f32| x).await,
+        Float64 => primitive::stream_to_array(pages, metadata, data_type, |x: f64| x).await,
+
+        Boolean => Ok(Box::new(boolean::stream_to_array(pages, metadata).await?)),
+
+        Binary | Utf8 => binary::stream_to_array::<i32, _, _>(pages, metadata, &data_type).await,
+        LargeBinary | LargeUtf8 => {
+            binary::stream_to_array::<i64, _, _>(pages, metadata, &data_type).await
+        }
+        FixedSizeBinary(size) => Ok(Box::new(
+            fixed_size_binary::stream_to_array(pages, size, metadata).await?,
+        )),
+        other => Err(ArrowError::NotYetImplemented(format!(
+            "Async conversion of {:?}",
+            other
+        ))),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::fs::File;
@@ -394,7 +484,7 @@ mod tests_integration {
         let path = "testing/parquet-testing/data/alltypes_plain.parquet";
         let reader = std::fs::File::open(path)?;
 
-        let reader = RecordReader::try_new(reader, None, None, Arc::new(|_, _| true))?;
+        let reader = RecordReader::try_new(reader, None, None, Arc::new(|_, _| true), None)?;
 
         let batches = reader.collect::<Result<Vec<_>>>()?;
         assert_eq!(batches.len(), 1);
diff --git a/src/io/parquet/read/primitive/mod.rs b/src/io/parquet/read/primitive/mod.rs
index de4332b807..3ea354e656 100644
--- a/src/io/parquet/read/primitive/mod.rs
+++ b/src/io/parquet/read/primitive/mod.rs
@@ -5,6 +5,7 @@ mod utils;
 
 use std::sync::Arc;
 
+use futures::{pin_mut, Stream, StreamExt};
 use parquet2::{page::DataPage, read::StreamingIterator, types::NativeType};
 
 use super::nested_utils::*;
@@ -59,6 +60,48 @@ where
     )))
 }
 
+pub async fn stream_to_array<T, A, I, E, F>(
+    pages: I,
+    metadata: &ColumnChunkMetaData,
+    data_type: DataType,
+    op: F,
+) -> Result<Box<dyn Array>>
+where
+    ArrowError: From<E>,
+    T: NativeType,
+    E: Clone,
+    A: ArrowNativeType,
+    F: Copy + Fn(T) -> A,
+    I: Stream<Item = std::result::Result<DataPage, E>>,
+{
+    let capacity = metadata.num_values() as usize;
+    let mut values = MutableBuffer::<A>::with_capacity(capacity);
+    let mut validity = MutableBitmap::with_capacity(capacity);
+
+    pin_mut!(pages); // needed for iteration
+
+    while let Some(page) = pages.next().await {
+        basic::extend_from_page(
+            page.as_ref().map_err(|x| x.clone())?,
+            metadata.descriptor(),
+            &mut values,
+            &mut validity,
+            op,
+        )?
+    }
+
+    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(),
+        validity.into(),
+    )))
+}
+
 pub fn iter_to_array_nested<T, A, I, E, F>(
     mut iter: I,
     metadata: &ColumnChunkMetaData,
diff --git a/src/io/parquet/read/record_batch.rs b/src/io/parquet/read/record_batch.rs
index 4d3f75fdfa..b88f3703ba 100644
--- a/src/io/parquet/read/record_batch.rs
+++ b/src/io/parquet/read/record_batch.rs
@@ -12,7 +12,7 @@ use crate::{
 
 use super::{
     get_page_iterator, get_schema, page_iter_to_array, read_metadata, Decompressor, FileMetaData,
-    RowGroupMetaData,
+    PageFilter, RowGroupMetaData,
 };
 
 type GroupFilter = Arc<dyn Fn(usize, &RowGroupMetaData) -> bool>;
@@ -25,6 +25,7 @@ pub struct RecordReader<R: Read + Seek> {
     buffer: Vec<u8>,
     decompress_buffer: Vec<u8>,
     groups_filter: GroupFilter,
+    pages_filter: Option<PageFilter>,
     metadata: Rc<FileMetaData>,
     current_group: usize,
     remaining_rows: usize,
@@ -36,6 +37,7 @@ impl<R: Read + Seek> RecordReader<R> {
         projection: Option<Vec<usize>>,
         limit: Option<usize>,
         groups_filter: GroupFilter,
+        pages_filter: Option<PageFilter>,
     ) -> Result<Self> {
         let metadata = read_metadata(&mut reader)?;
 
@@ -75,6 +77,7 @@ impl<R: Read + Seek> RecordReader<R> {
             schema,
             indices: Rc::new(indices),
             groups_filter,
+            pages_filter,
             metadata: Rc::new(metadata),
             current_group: 0,
             buffer: vec![],
@@ -124,7 +127,15 @@ impl<R: Read + Seek> Iterator for RecordReader<R> {
                 // column according to the file's indexing
                 let column = self.indices[column];
                 let column_meta = &columns_meta[column];
-                let pages = get_page_iterator(&metadata, row_group, column, &mut self.reader, b1)?;
+                let pages = get_page_iterator(
+                    &metadata,
+                    row_group,
+                    column,
+                    &mut self.reader,
+                    self.pages_filter.clone(),
+                    b1,
+                )?;
+
                 let mut pages = Decompressor::new(pages, b2);
 
                 let array = page_iter_to_array(&mut pages, column_meta, field.data_type().clone())?;
diff --git a/src/io/parquet/write/utf8/nested.rs b/src/io/parquet/write/utf8/nested.rs
index 65e7d3d13b..d9b7dafbd0 100644
--- a/src/io/parquet/write/utf8/nested.rs
+++ b/src/io/parquet/write/utf8/nested.rs
@@ -1,5 +1,6 @@
-use parquet2::encoding::Encoding;
-use parquet2::{metadata::ColumnDescriptor, page::CompressedDataPage, write::WriteOptions};
+use parquet2::{
+    encoding::Encoding, metadata::ColumnDescriptor, page::CompressedDataPage, write::WriteOptions,
+};
 
 use super::super::{levels, utils};
 use super::basic::{build_statistics, encode_plain};
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());
+    }
+}


Test patch:
diff --git a/arrow-pyarrow-integration-testing/tests/test_sql.py b/arrow-pyarrow-integration-testing/tests/test_sql.py
index e6ec3223d0..5260536342 100644
--- a/arrow-pyarrow-integration-testing/tests/test_sql.py
+++ b/arrow-pyarrow-integration-testing/tests/test_sql.py
@@ -104,7 +104,7 @@ class TestCase(unittest.TestCase):
         data = [
             round(decimal.Decimal(722.82), 2),
             round(decimal.Decimal(-934.11), 2),
-            None
+            None,
         ]
         a = pyarrow.array(data, pyarrow.decimal128(5, 2))
         b = arrow_pyarrow_integration_testing.round_trip(a)
@@ -179,3 +179,17 @@ class TestCase(unittest.TestCase):
             b.validate(full=True)
             assert a.to_pylist() == b.to_pylist()
             assert a.type == b.type
+
+    def test_dict(self):
+        """
+        Python -> Rust -> Python
+        """
+        a = pyarrow.array(
+            ["a", "a", "b", None, "c"],
+            pyarrow.dictionary(pyarrow.int64(), pyarrow.utf8()),
+        )
+        b = arrow_pyarrow_integration_testing.round_trip(a)
+
+        b.validate(full=True)
+        assert a.to_pylist() == b.to_pylist()
+        assert a.type == b.type