Repository: jorgecarleitao/arrow2
Pull request #256 (2021-08-05 20:27:13)

Added support to filter parquet pages.
This is the corresponding of https://github.com/jorgecarleitao/parquet2/pull/34, adding support to skip parquet pages based on its statistics.

# Backwards incompatible changes:

* `RecordReader` now requires an optional argument, `page_filter`. Pass `None` to recover previous behavior
* `get_page_iterator` now requires an optional argument, `page_filter`. Pass `None` to recover previous behavior

The page filter's signature is `Arc<dyn Fn(&ColumnDescriptor, &DataPageHeader) -> bool>`: use the first argument to tell which column is being considered and the second argument for information about the page.

In particular, `DataPageHeader::statistics()` contains the page statistics.

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

[![Impacted file tree graph](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/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/256?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     #256      +/-   ##
==========================================
+ Coverage   76.76%   77.44%   +0.68%     
==========================================
  Files         232      232              
  Lines       19820    20076     +256     
==========================================
+ Hits        15214    15548     +334     
+ Misses       4606     4528      -78     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/io/parquet/read/record\_batch.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvcmVhZC9yZWNvcmRfYmF0Y2gucnM=) | `88.15% <87.50%> (-0.25%)` | :arrow_down: |
| [src/io/parquet/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvbW9kLnJz) | `94.06% <100.00%> (+0.11%)` | :arrow_up: |
| [src/io/parquet/read/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvcmVhZC9tb2QucnM=) | `86.08% <100.00%> (ø)` | |
| [src/io/print.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3ByaW50LnJz) | `97.63% <0.00%> (+0.77%)` | :arrow_up: |
| [src/array/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/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% <0.00%> (+0.90%)` | :arrow_up: |
| [src/array/primitive/mutable.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L3ByaW1pdGl2ZS9tdXRhYmxlLnJz) | `66.81% <0.00%> (+1.32%)` | :arrow_up: |
| [src/array/dictionary/mutable.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2RpY3Rpb25hcnkvbXV0YWJsZS5ycw==) | `87.50% <0.00%> (+5.00%)` | :arrow_up: |
| [src/compute/aggregate/min\_max.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2NvbXB1dGUvYWdncmVnYXRlL21pbl9tYXgucnM=) | `94.50% <0.00%> (+8.24%)` | :arrow_up: |
| [src/compute/cast/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2NvbXB1dGUvY2FzdC9tb2QucnM=) | `38.07% <0.00%> (+10.97%)` | :arrow_up: |
| [src/compute/cast/utf8\_to.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2NvbXB1dGUvY2FzdC91dGY4X3RvLnJz) | `27.45% <0.00%> (+13.72%)` | :arrow_up: |
| ... and [2 more](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256/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/256?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/256?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Last update [d545253...faa39dc](https://codecov.io/gh/jorgecarleitao/arrow2/pull/256?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/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/compute/cast/mod.rs b/src/compute/cast/mod.rs
index 2a99c899c3..9d09e6bc75 100644
--- a/src/compute/cast/mod.rs
+++ b/src/compute/cast/mod.rs
@@ -53,7 +53,7 @@ impl Default for CastOptions {
 
 impl CastOptions {
     fn with_wrapped(&self, v: bool) -> Self {
-        let mut option = self.clone();
+        let mut option = *self;
         option.wrapped = v;
         option
     }
@@ -1234,6 +1234,7 @@ mod tests {
         assert_eq!(expected, result.as_ref());
     }
 
+    #[test]
     fn list_to_list() {
         let data = vec![
             Some(vec![Some(1i32), Some(2), Some(3)]),
diff --git a/src/compute/cast/primitive_to.rs b/src/compute/cast/primitive_to.rs
index 862af498e4..6b6bb90f3f 100644
--- a/src/compute/cast/primitive_to.rs
+++ b/src/compute/cast/primitive_to.rs
@@ -89,11 +89,7 @@ where
     I: NativeType + num::traits::AsPrimitive<O>,
     O: NativeType,
 {
-    unary(
-        from,
-        |x| num::traits::AsPrimitive::<O>::as_(x),
-        to_type.clone(),
-    )
+    unary(from, num::traits::AsPrimitive::<O>::as_, to_type.clone())
 }
 
 /// Cast [`PrimitiveArray`] to a [`PrimitiveArray`] of the same physical type.
diff --git a/src/io/ipc/write/mod.rs b/src/io/ipc/write/mod.rs
index e2be9700e0..7bdf9089fe 100644
--- a/src/io/ipc/write/mod.rs
+++ b/src/io/ipc/write/mod.rs
@@ -22,6 +22,7 @@ mod stream;
 mod writer;
 
 pub use super::gen::Schema::MetadataVersion;
+pub use common::IpcWriteOptions;
 pub use schema::schema_to_bytes;
 pub use serialize::{write, write_dictionary};
 pub use stream::StreamWriter;
diff --git a/src/io/parquet/mod.rs b/src/io/parquet/mod.rs
index 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/mod.rs b/src/io/parquet/read/mod.rs
index 8cddeb220b..6c53a92211 100644
--- a/src/io/parquet/read/mod.rs
+++ b/src/io/parquet/read/mod.rs
@@ -25,7 +25,7 @@ pub use parquet2::{
     page::{CompressedDataPage, DataPage, DataPageHeader},
     read::{
         decompress, get_page_iterator as _get_page_iterator, read_metadata as _read_metadata,
-        streaming_iterator, Decompressor, PageIterator, StreamingIterator,
+        streaming_iterator, Decompressor, PageFilter, PageIterator, StreamingIterator,
     },
     schema::types::{
         LogicalType, ParquetType, PhysicalType, PrimitiveConvertedType,
@@ -40,10 +40,16 @@ pub fn get_page_iterator<'b, RR: Read + Seek>(
     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,
     )?)
 }
 
@@ -394,7 +400,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/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/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/integration-testing/src/flight_client_scenarios/integration_test.rs b/integration-testing/src/flight_client_scenarios/integration_test.rs
index 9c4179e7fd..7cb128cdf6 100644
--- a/integration-testing/src/flight_client_scenarios/integration_test.rs
+++ b/integration-testing/src/flight_client_scenarios/integration_test.rs
@@ -76,7 +76,7 @@ async fn upload_data(
 ) -> Result {
     let (mut upload_tx, upload_rx) = mpsc::channel(10);
 
-    let options = write::common::IpcWriteOptions::default();
+    let options = write::IpcWriteOptions::default();
     let mut schema_flight_data =
         arrow_flight::utils::flight_data_from_arrow_schema(&schema, &options);
     schema_flight_data.flight_descriptor = Some(descriptor.clone());
@@ -128,7 +128,7 @@ async fn send_batch(
     upload_tx: &mut mpsc::Sender<FlightData>,
     metadata: &[u8],
     batch: &RecordBatch,
-    options: &write::common::IpcWriteOptions,
+    options: &write::IpcWriteOptions,
 ) -> Result {
     let (dictionary_flight_data, mut batch_flight_data) =
         arrow_flight::utils::flight_data_from_arrow_batch(batch, &options);
diff --git a/integration-testing/src/flight_server_scenarios/integration_test.rs b/integration-testing/src/flight_server_scenarios/integration_test.rs
index b10df3dc9b..118564cf8a 100644
--- a/integration-testing/src/flight_server_scenarios/integration_test.rs
+++ b/integration-testing/src/flight_server_scenarios/integration_test.rs
@@ -106,7 +106,7 @@ impl FlightService for FlightServiceImpl {
             .get(&key)
             .ok_or_else(|| Status::not_found(format!("Could not find flight. {}", key)))?;
 
-        let options = ipc::write::common::IpcWriteOptions::default();
+        let options = ipc::write::IpcWriteOptions::default();
 
         let schema = std::iter::once({
             Ok(arrow_flight::utils::flight_data_from_arrow_schema(
@@ -174,7 +174,7 @@ impl FlightService for FlightServiceImpl {
 
                 let total_records: usize = flight.chunks.iter().map(|chunk| chunk.num_rows()).sum();
 
-                let options = ipc::write::common::IpcWriteOptions::default();
+                let options = ipc::write::IpcWriteOptions::default();
                 let schema =
                     arrow_flight::utils::ipc_message_from_arrow_schema(&flight.schema, &options)
                         .expect(