Repository: jorgecarleitao/arrow2
Pull request #278 (2021-08-12 06:37:58)

Made IPC integration tests run tests that are not run by arrow-rs
Not sure patching `arrow` is a good idea, since the patch may fail. It should probably be a bit more configurable there.

I will leave this here for some time and re-trigger the job to check that the patch continues to apply for some time.

codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/278?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#278](https://codecov.io/gh/jorgecarleitao/arrow2/pull/278?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (477f25e) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/c7e1a4d6f4e3012543c5888e669ccb09d786f18f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (c7e1a4d) will **not change** coverage.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/jorgecarleitao/arrow2/pull/278/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/278?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     #278   +/-   ##
=======================================
  Coverage   77.15%   77.15%           
=======================================
  Files         254      254           
  Lines       20649    20649           
=======================================
  Hits        15931    15931           
  Misses       4718     4718           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/278?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/278?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Last update [c7e1a4d...477f25e](https://codecov.io/gh/jorgecarleitao/arrow2/pull/278?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/.github/workflows/integration-ipc.yml b/.github/workflows/integration-ipc.yml
index 7e55395b36..acea8b0a40 100644
--- a/.github/workflows/integration-ipc.yml
+++ b/.github/workflows/integration-ipc.yml
@@ -25,6 +25,9 @@ jobs:
       # which is incompatible with this. Let's monkey patch it
       - name: Fix compilation
         run: cp rust/integration-testing/rust_build.sh ci/scripts/rust_build.sh
+      # unskip many of the tests
+      - name: Test more cases
+        run: git apply rust/integration-testing/unskip.patch
       - name: Setup Python
         uses: actions/setup-python@v1
         with:
diff --git a/README.md b/README.md
index c739732b36..9292eb07c5 100644
--- a/README.md
+++ b/README.md
@@ -65,6 +65,7 @@ venv/bin/python parquet_integration/write_parquet.py
 * `MutableArray` API to work in-memory in-place.
 * faster IPC reader (different design that avoids an extra copy of all data)
 * IPC supports 2.0 (compression)
+* Extension type supported
 * All implemented arrow types pass FFI integration tests against pyarrow / C++
 * All implemented arrow types pass IPC integration tests against other implementations
 
@@ -83,7 +84,7 @@ venv/bin/python parquet_integration/write_parquet.py
 ## Features in the original not available in this crate
 
 * Parquet read and write of struct and nested lists.
-* Map types
+* Map type
 
 ## Features in this crate not in pyarrow
 
diff --git a/benches/filter_kernels.rs b/benches/filter_kernels.rs
index 65aaad8675..9d2a59e9b0 100644
--- a/benches/filter_kernels.rs
+++ b/benches/filter_kernels.rs
@@ -40,7 +40,8 @@ fn add_benchmark(c: &mut Criterion) {
         let size = 2usize.pow(log2_size);
 
         let filter_array = create_boolean_array(size, 0.0, 0.9);
-        let filter_array = BooleanArray::from_data(filter_array.values().clone(), None);
+        let filter_array =
+            BooleanArray::from_data(DataType::Boolean, filter_array.values().clone(), None);
 
         let arr_a = create_primitive_array::<f32>(size, DataType::Float32, 0.0);
         c.bench_function(&format!("filter 2^{} f32", log2_size), |b| {
diff --git a/examples/extension.rs b/examples/extension.rs
new file mode 100644
index 0000000000..25210214aa
--- /dev/null
+++ b/examples/extension.rs
@@ -0,0 +1,52 @@
+use std::io::{Cursor, Seek, Write};
+use std::sync::Arc;
+
+use arrow2::array::*;
+use arrow2::datatypes::*;
+use arrow2::error::Result;
+use arrow2::io::ipc::read;
+use arrow2::io::ipc::write;
+use arrow2::record_batch::RecordBatch;
+
+fn main() -> Result<()> {
+    // declare an extension.
+    let extension_type =
+        DataType::Extension("date16".to_string(), Box::new(DataType::UInt16), None);
+
+    // initialize an array with it.
+    let array = UInt16Array::from_slice([1, 2]).to(extension_type.clone());
+
+    // from here on, it works as usual
+    let mut buffer = Cursor::new(vec![]);
+
+    // write to IPC
+    write_ipc(&mut buffer, array)?;
+
+    // read it back
+    let batch = read_ipc(&buffer.into_inner())?;
+
+    // and verify that the datatype is preserved.
+    let array = &batch.columns()[0];
+    assert_eq!(array.data_type(), &extension_type);
+
+    // see https://arrow.apache.org/docs/format/Columnar.html#extension-types
+    // for consuming by other consumers.
+    Ok(())
+}
+
+fn write_ipc<W: Write + Seek>(writer: &mut W, array: impl Array + 'static) -> Result<()> {
+    let schema = Schema::new(vec![Field::new("a", array.data_type().clone(), false)]);
+
+    let mut writer = write::FileWriter::try_new(writer, &schema)?;
+
+    let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(array)])?;
+
+    writer.write(&batch)
+}
+
+fn read_ipc(reader: &[u8]) -> Result<RecordBatch> {
+    let mut reader = Cursor::new(reader);
+    let metadata = read::read_file_metadata(&mut reader)?;
+    let mut reader = read::FileReader::new(&mut reader, metadata, None);
+    reader.next().unwrap()
+}
diff --git a/guide/src/SUMMARY.md b/guide/src/SUMMARY.md
index 6ba760bec7..8158f21788 100644
--- a/guide/src/SUMMARY.md
+++ b/guide/src/SUMMARY.md
@@ -7,6 +7,7 @@
 - [Compute](./compute.md)
 - [Metadata](./metadata.md)
 - [Foreign interfaces](./ffi.md)
+- [Extension](./extension.md)
 - [IO](./io/README.md)
     - [Read CSV](./io/csv_reader.md)
     - [Write CSV](./io/csv_write.md)
diff --git a/guide/src/extension.md b/guide/src/extension.md
new file mode 100644
index 0000000000..fe85860bed
--- /dev/null
+++ b/guide/src/extension.md
@@ -0,0 +1,8 @@
+# Extension types
+
+This crate supports Arrows' ["extension type"](https://arrow.apache.org/docs/format/Columnar.html#extension-types), to declare, use, and share custom logical types.
+The follow example shows how to declare one:
+
+```rust
+{{#include ../../../examples/extension.rs}}
+```
diff --git a/guide/src/high_level.md b/guide/src/high_level.md
index 8d6e93aded..2a9fb52e19 100644
--- a/guide/src/high_level.md
+++ b/guide/src/high_level.md
@@ -4,59 +4,75 @@ The simplest way to think about an arrow `Array` is that it represents
 `Vec<Option<T>>` and has a logical type (see [metadata](../metadata.md))) associated with it.
 
 Probably the simplest array in this crate is `PrimitiveArray<T>`. It can be constructed
-from an iterator as follows:
+from a slice as follows:
+
+```rust
+# use arrow2::array::{Array, PrimitiveArray};
+# fn main() {
+let array = PrimitiveArray::<i32>::from([Some(1), None, Some(123)]);
+assert_eq!(array.len(), 3)
+# }
+```
+
+from a slice of values,
+
+```rust
+# use arrow2::array::{Array, PrimitiveArray};
+# fn main() {
+let array = PrimitiveArray::<f32>::from_slice([1.0, 0.0, 123.0]);
+assert_eq!(array.len(), 3)
+# }
+```
+
+or from an iterator
 
 ```rust
 # use arrow2::array::{Array, PrimitiveArray};
-# use arrow2::datatypes::DataType;
 # fn main() {
-let array = [Some(1), None, Some(123)]
-    .iter()
-    .collect::<PrimitiveArray<i32>>();
+let array: PrimitiveArray<u64> = [Some(1), None, Some(123)].iter().collect();
 assert_eq!(array.len(), 3)
 # }
 ```
 
 A `PrimitiveArray` has 3 components:
 
-1. A physical type (`i32`)
+1. A physical type (e.g. `i32`)
 2. A logical type (e.g. `DataType::Int32`)
 3. Data
 
 The main differences from a `Vec<Option<T>>` are:
 
-* Its data is laid out in memory as a `Buffer<T>` and an `Option<Bitmap>`.
-* It has an associated logical datatype.
+* Its data is laid out in memory as a `Buffer<T>` and an `Option<Bitmap>` (see [../low_level.md])
+* It has an associated logical type (`DataType`).
 
-The first difference allows interoperability with Arrow's ecosystem and efficient SIMD operations (we will re-visit this below); the second difference is that it gives semantic meaning to the array. In the example
+The first allows interoperability with Arrow's ecosystem and efficient SIMD operations
+(we will re-visit this below); the second is that it gives semantic meaning to the array.
+In the example
 
 ```rust
 # use arrow2::array::PrimitiveArray;
 # use arrow2::datatypes::DataType;
 # fn main() {
 let ints = PrimitiveArray::<i32>::from([Some(1), None]);
-let dates = PrimitiveArray::<i32>::from([Some(1), None]);
+let dates = PrimitiveArray::<i32>::from([Some(1), None]).to(DataType::Date32);
 # }
 ```
 
-`ints` and `dates` have the same in-memory representation but different logic representations (e.g. dates are usually represented as a string).
+`ints` and `dates` have the same in-memory representation but different logic
+representations (e.g. dates are usually printed to users as "yyyy-mm-dd").
 
-Some physical types (e.g. `i32`) have a "natural" logical `DataType` (e.g. `DataType::Int32`).
-These types support a more compact notation:
+All physical types (e.g. `i32`) have a "natural" logical `DataType` (e.g. `DataType::Int32`)
+which is assigned when allocating arrays from iterators, slices, etc.
 
 ```rust
 # use arrow2::array::{Array, Int32Array, PrimitiveArray};
 # use arrow2::datatypes::DataType;
 # fn main() {
-/// Int32Array = PrimitiveArray<i32>
-let array = [Some(1), None, Some(123)].iter().collect::<Int32Array>();
-assert_eq!(array.len(), 3);
-let array = Int32Array::from(&[Some(1), None, Some(123)]);
-assert_eq!(array.len(), 3);
-let array = Int32Array::from_slice(&[1, 123]);
-assert_eq!(array.len(), 2);
+let array = PrimitiveArray::<i32>::from_slice([1, 0, 123]);
+assert_eq!(array.data_type(), &DataType::Int32);
 # }
 ```
+they can be cheaply converted to via `.to(DataType)`.
 
 The following arrays are supported:
 
@@ -68,7 +84,8 @@ The following arrays are supported:
 * `FixedSizeBinaryArray` (like `BinaryArray`, but fixed size)
 * `ListArray<i32>` and `ListArray<i64>` (nested arrays)
 * `FixedSizeListArray` (nested arrays of fixed size)
-* `StructArray` (when each row has different logical types)
+* `StructArray` (every row has multiple logical types)
+* `UnionArray` (every row has a different logical type)
 * `DictionaryArray<K>` (nested array with encoded values)
 
 ## Dynamic Array
@@ -78,104 +95,71 @@ implement the trait `Array` and can be cast to `&dyn Array`, i.e. they can be tu
 a trait object. This enables arrays to have types that are dynamic in nature.
 
 ```rust
-# use std::sync::Arc;
 # use arrow2::array::{Array, PrimitiveArray};
-# use arrow2::datatypes::DataType;
 # fn main() {
-let data = vec![
-    Some(vec![Some(1i32), Some(2), Some(3)]),
-    None,
-    Some(vec![Some(4), None, Some(6)]),
-];
-
 let a = PrimitiveArray::<i32>::from(&[Some(1), None]);
 let a: &dyn Array = &a;
 # }
 ```
 
-Note how we have not specified the inner type explicitly in the signature `ListArray<i32>`.
-Instead, `ListArray` has an inner `Array` representing all its values (available via `.values()`).
-
 ### Downcast and `as_any`
 
-Given a trait object `&dyn Array`, we know its logical type via `Array::data_type()` and can use it to downcast the array to its concrete type:
+Given a trait object `array: &dyn Array`, we know its physical type via
+`PhysicalType: array.data_type().to_physical_type()`, which we use to downcast the array
+to its concrete type:
 
 ```rust
 # use arrow2::array::{Array, PrimitiveArray};
-# use arrow2::datatypes::DataType;
+# use arrow2::datatypes::PhysicalType;
 # fn main() {
-let array = [Some(1), None, Some(123)]
-    .iter()
-    .collect::<PrimitiveArray<i32>>();
+let array = PrimitiveArray::<i32>::from(&[Some(1), None]);
 let array = &array as &dyn Array;
-
-let array = array.as_any().downcast_ref::<PrimitiveArray<i32>>().unwrap();
+// ...
+let physical_type: PhysicalType = array.data_type().to_physical_type();
 # }
 ```
 
-There is a many-to-one relationship between `DataType` and an Array (i.e. a physical representation). The relationship is the following:
-
-| `DataType`            | `PhysicalType`            |
-|-----------------------|---------------------------|
-| `UInt8`               | `PrimitiveArray<u8>`      |
-| `UInt16`              | `PrimitiveArray<u16>`     |
-| `UInt32`              | `PrimitiveArray<u32>`     |
-| `UInt64`              | `PrimitiveArray<u64>`     |
-| `Int8`                | `PrimitiveArray<i8>`      |
-| `Int16`               | `PrimitiveArray<i16>`     |
-| `Int32`               | `PrimitiveArray<i32>`     |
-| `Int64`               | `PrimitiveArray<i64>`     |
-| `Float32`             | `PrimitiveArray<f32>`     |
-| `Float64`             | `PrimitiveArray<f64>`     |
-| `Decimal(_,_)`        | `PrimitiveArray<i128>`    |
-| `Date32`              | `PrimitiveArray<i32>`     |
-| `Date64`              | `PrimitiveArray<i64>`     |
-| `Time32(_)`           | `PrimitiveArray<i32>`     |
-| `Time64(_)`           | `PrimitiveArray<i64>`     |
-| `Timestamp(_,_)`      | `PrimitiveArray<i64>`     |
-| `Interval(YearMonth)` | `PrimitiveArray<i32>`     |
-| `Interval(DayTime)`   | `PrimitiveArray<days_ms>` |
-| `Duration(_)`         | `PrimitiveArray<i64>`     |
-| `Binary`              | `BinaryArray<i32>`        |
-| `LargeBinary`         | `BinaryArray<i64>`        |
-| `Utf8`                | `Utf8Array<i32>`          |
-| `LargeUtf8`           | `Utf8Array<i64>`          |
-| `List`                | `ListArray<i32>`          |
-| `LargeList`           | `ListArray<i64>`          |
-| `FixedSizeBinary(_)`  | `FixedSizeBinaryArray`    |
-| `FixedSizeList(_,_)`  | `FixedSizeListArray`      |
-| `Struct(_)`           | `StructArray`             |
-| `Union(_,_,_)`        | `UnionArray`              |
-| `Dictionary(UInt8,_)` | `DictionaryArray<u8>`     |
-| `Dictionary(UInt16,_)`| `DictionaryArray<u16>`    |
-| `Dictionary(UInt32,_)`| `DictionaryArray<u32>`    |
-| `Dictionary(UInt64,_)`| `DictionaryArray<u64>`    |
-| `Dictionary(Int8,_)`  | `DictionaryArray<i8>`     |
-| `Dictionary(Int16,_)` | `DictionaryArray<i16>`    |
-| `Dictionary(Int32,_)` | `DictionaryArray<i32>`    |
-| `Dictionary(Int64,_)` | `DictionaryArray<i64>`    |
-
-In this context, a common pattern to write operators that receive `&dyn Array` is:
+There is a one to one relationship between each variant of `PhysicalType` (an enum) and
+an each implementation of `Array` (a struct):
+
+| `PhysicalType`    | `Array`                |
+|-------------------|------------------------|
+| `Primitive(_)`    | `PrimitiveArray<_>`    |
+| `Binary`          | `BinaryArray<i32>`     |
+| `LargeBinary`     | `BinaryArray<i64>`     |
+| `Utf8`            | `Utf8Array<i32>`       |
+| `LargeUtf8`       | `Utf8Array<i64>`       |
+| `List`            | `ListArray<i32>`       |
+| `LargeList`       | `ListArray<i64>`       |
+| `FixedSizeBinary` | `FixedSizeBinaryArray` |
+| `FixedSizeList`   | `FixedSizeListArray`   |
+| `Struct`          | `StructArray`          |
+| `Union`           | `UnionArray`           |
+| `Dictionary(_)`   | `DictionaryArray<_>`   |
+
+where `_` represents each of the variants (e.g. `PrimitiveType::Int32 <-> i32`).
+
+In this context, a common idiom in using `Array` as a trait object is as follows:
 
 ```rust
-use arrow2::datatypes::DataType;
+use arrow2::datatypes::{PhysicalType, PrimitiveType};
 use arrow2::array::{Array, PrimitiveArray};
 
 fn float_operator(array: &dyn Array) -> Result<Box<dyn Array>, String> {
-    match array.data_type() {
-        DataType::Float32 => {
+    match array.data_type().to_physical_type() {
+        PhysicalType::Primitive(PrimitiveType::Float32) => {
             let array = array.as_any().downcast_ref::<PrimitiveArray<f32>>().unwrap();
             // let array = f32-specific operator
             let array = array.clone();
             Ok(Box::new(array))
         }
-        DataType::Float64 => {
+        PhysicalType::Primitive(PrimitiveType::Float64) => {
             let array = array.as_any().downcast_ref::<PrimitiveArray<f64>>().unwrap();
             // let array = f64-specific operator
             let array = array.clone();
             Ok(Box::new(array))
         }
-        _ => Err("This operator is only valid for float point.".to_string()),
+        _ => Err("This operator is only valid for float point arrays".to_string()),
     }
 }
 ```
@@ -237,7 +221,8 @@ bitwise operations, it is often more performant to operate on chunks of bits ins
 ## Vectorized operations
 
 One of the main advantages of the arrow format and its memory layout is that
-it often enables SIMD. For example, an unary operation `op` on a `PrimitiveArray` is likely auto-vectorized on the following code:
+it often enables SIMD. For example, an unary operation `op` on a `PrimitiveArray`
+likely emits SIMD instructions on the following code:
 
 ```rust
 # use arrow2::buffer::Buffer;
diff --git a/guide/src/metadata.md b/guide/src/metadata.md
index c9009c6767..9ced2cdfeb 100644
--- a/guide/src/metadata.md
+++ b/guide/src/metadata.md
@@ -12,16 +12,18 @@ semantical types defined in Arrow.
 In Arrow2, logical types are declared as variants of the `enum` `arrow2::datatypes::DataType`.
 For example, `DataType::Int32` represents a signed integer of 32 bits.
 
-Each logical type has an associated in-memory physical representation and is associated to specific
-semantics. For example, `Date32` has the same in-memory representation as `Int32`, but the value
-represents the number of days since UNIX epoch.
+Each `DataType` has an associated `enum PhysicalType` (many-to-one) representing the
+particular in-memory representation, and is associated to specific semantics.
+For example, both `DataType::Date32` and `DataType::Int32` have the same `PhysicalType`
+(`PhysicalType::Primitive(PrimitiveType::Int32)`) but `Date32` represents the number of
+days since UNIX epoch.
 
-Logical types are metadata: they annotate arrays with extra information about in-memory data.
+Logical types are metadata: they annotate physical types with extra information about data.
 
 ## `Field` (column metadata)
 
-Besides logical types, the arrow format supports other relevant metadata to the format. All this 
-information is stored in `arrow2::datatypes::Field`.
+Besides logical types, the arrow format supports other relevant metadata to the format.
+All this information is stored in `arrow2::datatypes::Field`.
 
 A `Field` is arrow's metadata associated to a column in the context of a columnar format. 
 It has a name, a logical type `DataType`, whether the column is nullable, etc.
diff --git a/src/array/binary/ffi.rs b/src/array/binary/ffi.rs
index 63d0ea66f3..9a4257cabf 100644
--- a/src/array/binary/ffi.rs
+++ b/src/array/binary/ffi.rs
@@ -44,6 +44,11 @@ unsafe impl<O: Offset, A: ffi::ArrowArrayRef> FromFfi<A> for BinaryArray<O> {
             validity = validity.map(|x| x.slice(offset, length))
         }
 
-        Ok(Self::from_data(offsets, values, validity))
+        Ok(Self::from_data(
+            Self::default_data_type(),
+            offsets,
+            values,
+            validity,
+        ))
     }
 }
diff --git a/src/array/binary/from.rs b/src/array/binary/from.rs
index ef9a287a9c..60fda08413 100644
--- a/src/array/binary/from.rs
+++ b/src/array/binary/from.rs
@@ -31,7 +31,7 @@ impl<O: Offset> BinaryArray<O> {
         // soundness: I is `TrustedLen`
         let (validity, offsets, values) = unsafe { trusted_len_unzip(iterator) };
 
-        Self::from_data(offsets, values, validity)
+        Self::from_data(Self::default_data_type(), offsets, values, validity)
     }
 }
 
diff --git a/src/array/binary/mod.rs b/src/array/binary/mod.rs
index 7c9a2fc3dc..a268d01095 100644
--- a/src/array/binary/mod.rs
+++ b/src/array/binary/mod.rs
@@ -25,14 +25,15 @@ pub struct BinaryArray<O: Offset> {
 // constructors
 impl<O: Offset> BinaryArray<O> {
     /// Creates an empty [`BinaryArray`], i.e. whose `.len` is zero.
-    pub fn new_empty() -> Self {
-        Self::from_data(Buffer::from(&[O::zero()]), Buffer::new(), None)
+    pub fn new_empty(data_type: DataType) -> Self {
+        Self::from_data(data_type, Buffer::from(&[O::zero()]), Buffer::new(), None)
     }
 
     /// Creates an null [`BinaryArray`], i.e. whose `.null_count() == .len()`.
     #[inline]
-    pub fn new_null(length: usize) -> Self {
+    pub fn new_null(data_type: DataType, length: usize) -> Self {
         Self::from_data(
+            data_type,
             Buffer::new_zeroed(length + 1),
             Buffer::new(),
             Some(Bitmap::new_zeroed(length)),
@@ -43,19 +44,24 @@ impl<O: Offset> BinaryArray<O> {
     /// # Panics
     /// * The length of the offset buffer must be larger than 1
     /// * The length of the values must be equal to the last offset value
-    pub fn from_data(offsets: Buffer<O>, values: Buffer<u8>, validity: Option<Bitmap>) -> Self {
+    pub fn from_data(
+        data_type: DataType,
+        offsets: Buffer<O>,
+        values: Buffer<u8>,
+        validity: Option<Bitmap>,
+    ) -> Self {
         check_offsets(&offsets, values.len());
 
         if let Some(validity) = &validity {
             assert_eq!(offsets.len() - 1, validity.len());
         }
 
+        if data_type.to_physical_type() != Self::default_data_type().to_physical_type() {
+            panic!("BinaryArray can only be initialized with DataType::Binary or DataType::LargeBinary")
+        }
+
         Self {
-            data_type: if O::is_large() {
-                DataType::LargeBinary
-            } else {
-                DataType::Binary
-            },
+            data_type,
             offsets,
             values,
             validity,
@@ -63,6 +69,15 @@ impl<O: Offset> BinaryArray<O> {
         }
     }
 
+    /// Returns the default [`DataType`], `DataType::Binary` or `DataType::LargeBinary`
+    pub fn default_data_type() -> DataType {
+        if O::is_large() {
+            DataType::LargeBinary
+        } else {
+            DataType::Binary
+        }
+    }
+
     /// Creates a new [`BinaryArray`] by slicing this [`BinaryArray`].
     /// # Implementation
     /// This function is `O(1)`: all data will be shared between both arrays.
diff --git a/src/array/binary/mutable.rs b/src/array/binary/mutable.rs
index 3ae30d436d..52ae198fa7 100644
--- a/src/array/binary/mutable.rs
+++ b/src/array/binary/mutable.rs
@@ -16,6 +16,7 @@ use super::BinaryArray;
 /// This struct does not allocate a validity until one is required (i.e. push a null to it).
 #[derive(Debug)]
 pub struct MutableBinaryArray<O: Offset> {
+    data_type: DataType,
     offsets: MutableBuffer<O>,
     values: MutableBuffer<u8>,
     validity: Option<MutableBitmap>,
@@ -24,6 +25,7 @@ pub struct MutableBinaryArray<O: Offset> {
 impl<O: Offset> From<MutableBinaryArray<O>> for BinaryArray<O> {
     fn from(other: MutableBinaryArray<O>) -> Self {
         BinaryArray::<O>::from_data(
+            other.data_type,
             other.offsets.into(),
             other.values.into(),
             other.validity.map(|x| x.into()),
@@ -52,6 +54,7 @@ impl<O: Offset> MutableBinaryArray<O> {
         let mut offsets = MutableBuffer::<O>::with_capacity(capacity + 1);
         offsets.push(O::default());
         Self {
+            data_type: BinaryArray::<O>::default_data_type(),
             offsets,
             values: MutableBuffer::<u8>::new(),
             validity: None,
@@ -114,6 +117,7 @@ impl<O: Offset> MutableArray for MutableBinaryArray<O> {
 
     fn as_arc(&mut self) -> Arc<dyn Array> {
         Arc::new(BinaryArray::from_data(
+            self.data_type.clone(),
             std::mem::take(&mut self.offsets).into(),
             std::mem::take(&mut self.values).into(),
             std::mem::take(&mut self.validity).map(|x| x.into()),
@@ -121,11 +125,7 @@ impl<O: Offset> MutableArray for MutableBinaryArray<O> {
     }
 
     fn data_type(&self) -> &DataType {
-        if O::is_large() {
-            &DataType::LargeUtf8
-        } else {
-            &DataType::Utf8
-        }
+        &self.data_type
     }
 
     fn as_any(&self) -> &dyn std::any::Any {
diff --git a/src/array/boolean/ffi.rs b/src/array/boolean/ffi.rs
index 9dedfbc41d..745c27e54c 100644
--- a/src/array/boolean/ffi.rs
+++ b/src/array/boolean/ffi.rs
@@ -34,6 +34,6 @@ unsafe impl<A: ffi::ArrowArrayRef> FromFfi<A> for BooleanArray {
             values = values.slice(offset, length);
             validity = validity.map(|x| x.slice(offset, length))
         }
-        Ok(Self::from_data(values, validity))
+        Ok(Self::from_data(data_type, values, validity))
     }
 }
diff --git a/src/array/boolean/mod.rs b/src/array/boolean/mod.rs
index 3679a361ee..8c1207e390 100644
--- a/src/array/boolean/mod.rs
+++ b/src/array/boolean/mod.rs
@@ -1,4 +1,7 @@
-use crate::{bitmap::Bitmap, datatypes::DataType};
+use crate::{
+    bitmap::Bitmap,
+    datatypes::{DataType, PhysicalType},
+};
 
 use super::{display_fmt, Array};
 
@@ -14,6 +17,7 @@ pub use mutable::*;
 /// Cloning and slicing this struct is `O(1)`.
 #[derive(Debug, Clone)]
 pub struct BooleanArray {
+    data_type: DataType,
     values: Bitmap,
     validity: Option<Bitmap>,
     offset: usize,
@@ -21,28 +25,29 @@ pub struct BooleanArray {
 
 impl BooleanArray {
     /// Returns a new empty [`BooleanArray`].
-    #[inline]
-    pub fn new_empty() -> Self {
-        Self::from_data(Bitmap::new(), None)
+    pub fn new_empty(data_type: DataType) -> Self {
+        Self::from_data(data_type, Bitmap::new(), None)
     }
 
     /// Returns a new [`BooleanArray`] whose all slots are null / `None`.
-    #[inline]
-    pub fn new_null(length: usize) -> Self {
+    pub fn new_null(data_type: DataType, length: usize) -> Self {
         let bitmap = Bitmap::new_zeroed(length);
-        Self::from_data(bitmap.clone(), Some(bitmap))
+        Self::from_data(data_type, bitmap.clone(), Some(bitmap))
     }
 
     /// The canonical method to create a [`BooleanArray`] out of low-end APIs.
     /// # Panics
     /// This function panics iff:
     /// * The validity is not `None` and its length is different from `values`'s length
-    #[inline]
-    pub fn from_data(values: Bitmap, validity: Option<Bitmap>) -> Self {
+    pub fn from_data(data_type: DataType, values: Bitmap, validity: Option<Bitmap>) -> Self {
         if let Some(ref validity) = validity {
             assert_eq!(values.len(), validity.len());
         }
+        if data_type.to_physical_type() != PhysicalType::Boolean {
+            panic!("BooleanArray can only be initialized with DataType::Boolean")
+        }
         Self {
+            data_type,
             values,
             validity,
             offset: 0,
@@ -58,6 +63,7 @@ impl BooleanArray {
     pub fn slice(&self, offset: usize, length: usize) -> Self {
         let validity = self.validity.clone().map(|x| x.slice(offset, length));
         Self {
+            data_type: self.data_type.clone(),
             values: self.values.clone().slice(offset, length),
             validity,
             offset: self.offset + offset,
@@ -100,7 +106,7 @@ impl Array for BooleanArray {
 
     #[inline]
     fn data_type(&self) -> &DataType {
-        &DataType::Boolean
+        &self.data_type
     }
 
     #[inline]
diff --git a/src/array/boolean/mutable.rs b/src/array/boolean/mutable.rs
index c0ccf5938e..8ab6d01faa 100644
--- a/src/array/boolean/mutable.rs
+++ b/src/array/boolean/mutable.rs
@@ -4,7 +4,7 @@ use std::sync::Arc;
 use crate::{
     array::{Array, MutableArray, TryExtend, TryPush},
     bitmap::MutableBitmap,
-    datatypes::DataType,
+    datatypes::{DataType, PhysicalType},
     error::Result,
     trusted_len::TrustedLen,
 };
@@ -17,13 +17,18 @@ use super::BooleanArray;
 /// This struct does not allocate a validity until one is required (i.e. push a null to it).
 #[derive(Debug)]
 pub struct MutableBooleanArray {
+    data_type: DataType,
     values: MutableBitmap,
     validity: Option<MutableBitmap>,
 }
 
 impl From<MutableBooleanArray> for BooleanArray {
     fn from(other: MutableBooleanArray) -> Self {
-        BooleanArray::from_data(other.values.into(), other.validity.map(|x| x.into()))
+        BooleanArray::from_data(
+            other.data_type,
+            other.values.into(),
+            other.validity.map(|x| x.into()),
+        )
     }
 }
 
@@ -49,6 +54,7 @@ impl MutableBooleanArray {
     /// Creates an new [`MutableBooleanArray`] with a capacity of values.
     pub fn with_capacity(capacity: usize) -> Self {
         Self {
+            data_type: DataType::Boolean,
             values: MutableBitmap::with_capacity(capacity),
             validity: None,
         }
@@ -63,8 +69,19 @@ impl MutableBooleanArray {
     }
 
     /// Canonical method to create a new [`MutableBooleanArray`].
-    pub fn from_data(values: MutableBitmap, validity: Option<MutableBitmap>) -> Self {
-        Self { values, validity }
+    pub fn from_data(
+        data_type: DataType,
+        values: MutableBitmap,
+        validity: Option<MutableBitmap>,
+    ) -> Self {
+        if data_type.to_physical_type() != PhysicalType::Boolean {
+            panic!("MutableBooleanArray can only be initialized with DataType::Boolean")
+        }
+        Self {
+            data_type,
+            values,
+            validity,
+        }
     }
 
     /// Pushes a new entry to [`MutableBooleanArray`].
@@ -137,7 +154,11 @@ impl MutableBooleanArray {
     /// Creates a new [`MutableBooleanArray`] from an [`TrustedLen`] of `bool`.
     #[inline]
     pub fn from_trusted_len_values_iter<I: TrustedLen<Item = bool>>(iterator: I) -> Self {
-        Self::from_data(MutableBitmap::from_trusted_len_iter(iterator), None)
+        Self::from_data(
+            DataType::Boolean,
+            MutableBitmap::from_trusted_len_iter(iterator),
+            None,
+        )
     }
 
     /// Creates a new [`MutableBooleanArray`] from a slice of `bool`.
@@ -166,7 +187,7 @@ impl MutableBooleanArray {
             None
         };
 
-        Self::from_data(values, validity)
+        Self::from_data(DataType::Boolean, values, validity)
     }
 
     /// Creates a [`BooleanArray`] from a [`TrustedLen`].
@@ -199,7 +220,7 @@ impl MutableBooleanArray {
             None
         };
 
-        Ok(Self::from_data(values, validity))
+        Ok(Self::from_data(DataType::Boolean, values, validity))
     }
 
     /// Creates a [`BooleanArray`] from a [`TrustedLen`].
@@ -304,7 +325,7 @@ impl<Ptr: std::borrow::Borrow<Option<bool>>> FromIterator<Ptr> for MutableBoolea
             })
             .collect();
 
-        MutableBooleanArray::from_data(values, validity.into())
+        MutableBooleanArray::from_data(DataType::Boolean, values, validity.into())
     }
 }
 
@@ -319,13 +340,14 @@ impl MutableArray for MutableBooleanArray {
 
     fn as_arc(&mut self) -> Arc<dyn Array> {
         Arc::new(BooleanArray::from_data(
+            self.data_type.clone(),
             std::mem::take(&mut self.values).into(),
             std::mem::take(&mut self.validity).map(|x| x.into()),
         ))
     }
 
     fn data_type(&self) -> &DataType {
-        &DataType::Boolean
+        &self.data_type
     }
 
     fn as_any(&self) -> &dyn std::any::Any {
diff --git a/src/array/dictionary/mod.rs b/src/array/dictionary/mod.rs
index beb4960fff..f9d5668d96 100644
--- a/src/array/dictionary/mod.rs
+++ b/src/array/dictionary/mod.rs
@@ -43,16 +43,18 @@ pub struct DictionaryArray<K: DictionaryKey> {
 impl<K: DictionaryKey> DictionaryArray<K> {
     /// Returns a new empty [`DictionaryArray`].
     pub fn new_empty(data_type: DataType) -> Self {
-        let values = new_empty_array(data_type).into();
+        let values = Self::get_child(&data_type);
+        let values = new_empty_array(values.clone()).into();
         Self::from_data(PrimitiveArray::<K>::new_empty(K::DATA_TYPE), values)
     }
 
     /// Returns an [`DictionaryArray`] whose all elements are null
     #[inline]
     pub fn new_null(data_type: DataType, length: usize) -> Self {
+        let values = Self::get_child(&data_type);
         Self::from_data(
             PrimitiveArray::<K>::new_null(K::DATA_TYPE, length),
-            new_empty_array(data_type).into(),
+            new_empty_array(values.clone()).into(),
         )
     }
 
@@ -104,10 +106,10 @@ impl<K: DictionaryKey> DictionaryArray<K> {
 
 impl<K: DictionaryKey> DictionaryArray<K> {
     pub(crate) fn get_child(data_type: &DataType) -> &DataType {
-        if let DataType::Dictionary(_, values) = data_type {
-            values.as_ref()
-        } else {
-            panic!("Wrong DataType")
+        match data_type {
+            DataType::Dictionary(_, values) => values.as_ref(),
+            DataType::Extension(_, inner, _) => Self::get_child(inner),
+            _ => panic!("DictionaryArray must be initialized with DataType::Dictionary"),
         }
     }
 }
diff --git a/src/array/display.rs b/src/array/display.rs
index 7a216993b5..ecf6036e84 100644
--- a/src/array/display.rs
+++ b/src/array/display.rs
@@ -234,6 +234,7 @@ pub fn get_value_display<'a>(array: &'a dyn Array) -> Box<dyn Fn(usize) -> Strin
                 displays[field](index)
             })
         }
+        Extension(_, _, _) => todo!(),
     }
 }
 
diff --git a/src/array/equal/mod.rs b/src/array/equal/mod.rs
index 765907ab4d..32612a1fe0 100644
--- a/src/array/equal/mod.rs
+++ b/src/array/equal/mod.rs
@@ -1,7 +1,4 @@
-use crate::{
-    datatypes::{DataType, IntervalUnit},
-    types::{days_ms, NativeType},
-};
+use crate::types::{days_ms, NativeType};
 
 use super::*;
 
@@ -164,138 +161,76 @@ pub fn equal(lhs: &dyn Array, rhs: &dyn Array) -> bool {
         return false;
     }
 
-    match lhs.data_type() {
-        DataType::Null => {
+    use crate::datatypes::PhysicalType::*;
+    match lhs.data_type().to_physical_type() {
+        Null => {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
             null::equal(lhs, rhs)
         }
-        DataType::Boolean => {
+        Boolean => {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
             boolean::equal(lhs, rhs)
         }
-        DataType::UInt8 => {
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<u8>(lhs, rhs)
-        }
-        DataType::UInt16 => {
-            let lhs = lhs.as_any().downcast_ref().unwrap();
-            let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<u16>(lhs, rhs)
-        }
-        DataType::UInt32 => {
-            let lhs = lhs.as_any().downcast_ref().unwrap();
-            let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<u32>(lhs, rhs)
-        }
-        DataType::UInt64 => {
-            let lhs = lhs.as_any().downcast_ref().unwrap();
-            let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<u64>(lhs, rhs)
-        }
-        DataType::Int8 => {
-            let lhs = lhs.as_any().downcast_ref().unwrap();
-            let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<i8>(lhs, rhs)
-        }
-        DataType::Int16 => {
-            let lhs = lhs.as_any().downcast_ref().unwrap();
-            let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<i16>(lhs, rhs)
-        }
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => {
-            let lhs = lhs.as_any().downcast_ref().unwrap();
-            let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<i32>(lhs, rhs)
-        }
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => {
-            let lhs = lhs.as_any().downcast_ref().unwrap();
-            let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<i64>(lhs, rhs)
-        }
-        DataType::Decimal(_, _) => {
-            let lhs = lhs.as_any().downcast_ref().unwrap();
-            let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<i128>(lhs, rhs)
-        }
-        DataType::Interval(IntervalUnit::DayTime) => {
-            let lhs = lhs.as_any().downcast_ref().unwrap();
-            let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<days_ms>(lhs, rhs)
-        }
-        DataType::Float16 => unreachable!(),
-        DataType::Float32 => {
-            let lhs = lhs.as_any().downcast_ref().unwrap();
-            let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<f32>(lhs, rhs)
-        }
-        DataType::Float64 => {
-            let lhs = lhs.as_any().downcast_ref().unwrap();
-            let rhs = rhs.as_any().downcast_ref().unwrap();
-            primitive::equal::<f64>(lhs, rhs)
-        }
-        DataType::Utf8 => {
+            primitive::equal::<$T>(lhs, rhs)
+        }),
+        Utf8 => {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
             utf8::equal::<i32>(lhs, rhs)
         }
-        DataType::LargeUtf8 => {
+        LargeUtf8 => {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
             utf8::equal::<i64>(lhs, rhs)
         }
-        DataType::Binary => {
+        Binary => {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
             binary::equal::<i32>(lhs, rhs)
         }
-        DataType::LargeBinary => {
+        LargeBinary => {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
             binary::equal::<i64>(lhs, rhs)
         }
-        DataType::List(_) => {
+        List => {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
             list::equal::<i32>(lhs, rhs)
         }
-        DataType::LargeList(_) => {
+        LargeList => {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
             list::equal::<i64>(lhs, rhs)
         }
-        DataType::Struct(_) => {
+        Struct => {
             let lhs = lhs.as_any().downcast_ref::<StructArray>().unwrap();
             let rhs = rhs.as_any().downcast_ref::<StructArray>().unwrap();
             struct_::equal(lhs, rhs)
         }
-        DataType::Dictionary(key_type, _) => {
-            with_match_dictionary_key_type!(key_type.as_ref(), |$T| {
+        Dictionary(key_type) => {
+            with_match_physical_dictionary_key_type!(key_type, |$T| {
                 let lhs = lhs.as_any().downcast_ref().unwrap();
                 let rhs = rhs.as_any().downcast_ref().unwrap();
                 dictionary::equal::<$T>(lhs, rhs)
             })
         }
-        DataType::FixedSizeBinary(_) => {
+        FixedSizeBinary => {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
             fixed_size_binary::equal(lhs, rhs)
         }
-        DataType::FixedSizeList(_, _) => {
+        FixedSizeList => {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
             fixed_size_list::equal(lhs, rhs)
         }
-        DataType::Union(_, _, _) => {
+        Union => {
             let lhs = lhs.as_any().downcast_ref().unwrap();
             let rhs = rhs.as_any().downcast_ref().unwrap();
             union::equal(lhs, rhs)
diff --git a/src/array/ffi.rs b/src/array/ffi.rs
index ea03602336..7368f45420 100644
--- a/src/array/ffi.rs
+++ b/src/array/ffi.rs
@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use crate::datatypes::DataType;
+use crate::datatypes::PhysicalType;
 use crate::{array::*, ffi};
 
 use crate::error::Result;
@@ -41,43 +41,25 @@ type 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),
-        DataType::Int8 => ffi_dyn!(array, PrimitiveArray<i8>),
-        DataType::Int16 => ffi_dyn!(array, PrimitiveArray<i16>),
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => {
-            ffi_dyn!(array, PrimitiveArray<i32>)
-        }
-        DataType::Interval(IntervalUnit::DayTime) => ffi_dyn!(array, PrimitiveArray<days_ms>),
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => ffi_dyn!(array, PrimitiveArray<i64>),
-        DataType::Decimal(_, _) => ffi_dyn!(array, PrimitiveArray<i128>),
-        DataType::UInt8 => ffi_dyn!(array, PrimitiveArray<u8>),
-        DataType::UInt16 => ffi_dyn!(array, PrimitiveArray<u16>),
-        DataType::UInt32 => ffi_dyn!(array, PrimitiveArray<u32>),
-        DataType::UInt64 => ffi_dyn!(array, PrimitiveArray<u64>),
-        DataType::Float16 => unreachable!(),
-        DataType::Float32 => ffi_dyn!(array, PrimitiveArray<f32>),
-        DataType::Float64 => ffi_dyn!(array, PrimitiveArray<f64>),
-        DataType::Binary => ffi_dyn!(array, BinaryArray<i32>),
-        DataType::LargeBinary => ffi_dyn!(array, BinaryArray<i64>),
-        DataType::FixedSizeBinary(_) => ffi_dyn!(array, FixedSizeBinaryArray),
-        DataType::Utf8 => ffi_dyn!(array, Utf8Array::<i32>),
-        DataType::LargeUtf8 => ffi_dyn!(array, Utf8Array::<i64>),
-        DataType::List(_) => ffi_dyn!(array, ListArray::<i32>),
-        DataType::LargeList(_) => ffi_dyn!(array, ListArray::<i64>),
-        DataType::FixedSizeList(_, _) => ffi_dyn!(array, FixedSizeListArray),
-        DataType::Struct(_) => ffi_dyn!(array, StructArray),
-        DataType::Union(_, _, _) => ffi_dyn!(array, UnionArray),
-        DataType::Dictionary(key_type, _) => {
-            with_match_dictionary_key_type!(key_type.as_ref(), |$T| {
+    use PhysicalType::*;
+    match array.data_type().to_physical_type() {
+        Null => ffi_dyn!(array, NullArray),
+        Boolean => ffi_dyn!(array, BooleanArray),
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+            ffi_dyn!(array, PrimitiveArray<$T>)
+        }),
+        Binary => ffi_dyn!(array, BinaryArray<i32>),
+        LargeBinary => ffi_dyn!(array, BinaryArray<i64>),
+        FixedSizeBinary => ffi_dyn!(array, FixedSizeBinaryArray),
+        Utf8 => ffi_dyn!(array, Utf8Array::<i32>),
+        LargeUtf8 => ffi_dyn!(array, Utf8Array::<i64>),
+        List => ffi_dyn!(array, ListArray::<i32>),
+        LargeList => ffi_dyn!(array, ListArray::<i64>),
+        FixedSizeList => ffi_dyn!(array, FixedSizeListArray),
+        Struct => ffi_dyn!(array, StructArray),
+        Union => ffi_dyn!(array, UnionArray),
+        Dictionary(key_type) => {
+            with_match_physical_dictionary_key_type!(key_type, |$T| {
                 let array = array.as_any().downcast_ref::<DictionaryArray<$T>>().unwrap();
                 (
                     array.buffers(),
diff --git a/src/array/fixed_size_binary/mod.rs b/src/array/fixed_size_binary/mod.rs
index 71a9a68751..401b24e8ec 100644
--- a/src/array/fixed_size_binary/mod.rs
+++ b/src/array/fixed_size_binary/mod.rs
@@ -40,7 +40,7 @@ impl FixedSizeBinaryArray {
 
         Self {
             size,
-            data_type: DataType::FixedSizeBinary(size),
+            data_type,
             values,
             validity,
             offset: 0,
@@ -97,10 +97,10 @@ impl FixedSizeBinaryArray {
 
 impl FixedSizeBinaryArray {
     pub(crate) fn get_size(data_type: &DataType) -> &i32 {
-        if let DataType::FixedSizeBinary(size) = data_type {
-            size
-        } else {
-            panic!("Wrong DataType")
+        match data_type {
+            DataType::FixedSizeBinary(size) => size,
+            DataType::Extension(_, child, _) => Self::get_size(child),
+            _ => panic!("Wrong DataType"),
         }
     }
 }
diff --git a/src/array/fixed_size_binary/mutable.rs b/src/array/fixed_size_binary/mutable.rs
index 1376a56386..cfb020d907 100644
--- a/src/array/fixed_size_binary/mutable.rs
+++ b/src/array/fixed_size_binary/mutable.rs
@@ -35,10 +35,11 @@ impl From<MutableFixedSizeBinaryArray> for FixedSizeBinaryArray {
 impl MutableFixedSizeBinaryArray {
     /// Canonical method to create a new [`MutableFixedSizeBinaryArray`].
     pub fn from_data(
-        size: usize,
+        data_type: DataType,
         values: MutableBuffer<u8>,
         validity: Option<MutableBitmap>,
     ) -> Self {
+        let size = *FixedSizeBinaryArray::get_size(&data_type) as usize;
         assert_eq!(
             values.len() % size,
             0,
@@ -52,7 +53,7 @@ impl MutableFixedSizeBinaryArray {
             );
         }
         Self {
-            data_type: DataType::FixedSizeBinary(size as i32),
+            data_type,
             size,
             values,
             validity,
@@ -67,7 +68,7 @@ impl MutableFixedSizeBinaryArray {
     /// Creates a new [`MutableFixedSizeBinaryArray`] with capacity for `capacity` entries.
     pub fn with_capacity(size: usize, capacity: usize) -> Self {
         Self::from_data(
-            size,
+            DataType::FixedSizeBinary(size as i32),
             MutableBuffer::<u8>::with_capacity(capacity * size),
             None,
         )
diff --git a/src/array/fixed_size_list/mod.rs b/src/array/fixed_size_list/mod.rs
index 4147de5b6d..6c8b2f3483 100644
--- a/src/array/fixed_size_list/mod.rs
+++ b/src/array/fixed_size_list/mod.rs
@@ -26,13 +26,18 @@ pub struct FixedSizeListArray {
 impl FixedSizeListArray {
     /// Returns a new empty [`FixedSizeListArray`].
     pub fn new_empty(data_type: DataType) -> Self {
-        let values = new_empty_array(Self::get_child_and_size(&data_type).0.clone()).into();
+        let values =
+            new_empty_array(Self::get_child_and_size(&data_type).0.data_type().clone()).into();
         Self::from_data(data_type, values, None)
     }
 
     /// Returns a new null [`FixedSizeListArray`].
     pub fn new_null(data_type: DataType, length: usize) -> Self {
-        let values = new_null_array(Self::get_child_and_size(&data_type).0.clone(), length).into();
+        let values = new_null_array(
+            Self::get_child_and_size(&data_type).0.data_type().clone(),
+            length,
+        )
+        .into();
         Self::from_data(data_type, values, Some(Bitmap::new_zeroed(length)))
     }
 
@@ -88,16 +93,15 @@ impl FixedSizeListArray {
 }
 
 impl FixedSizeListArray {
-    pub(crate) fn get_child_and_size(data_type: &DataType) -> (&DataType, &i32) {
-        if let DataType::FixedSizeList(field, size) = data_type {
-            (field.data_type(), size)
-        } else {
-            panic!("Wrong DataType")
+    pub(crate) fn get_child_and_size(data_type: &DataType) -> (&Field, &i32) {
+        match data_type {
+            DataType::FixedSizeList(child, size) => (child.as_ref(), size),
+            DataType::Extension(_, child, _) => Self::get_child_and_size(child),
+            _ => panic!("Wrong DataType"),
         }
     }
 
     /// Returns a [`DataType`] consistent with this Array.
-    #[inline]
     pub fn default_datatype(data_type: DataType, size: usize) -> DataType {
         let field = Box::new(Field::new("item", data_type, true));
         DataType::FixedSizeList(field, size as i32)
diff --git a/src/array/growable/binary.rs b/src/array/growable/binary.rs
index f12a3a2a78..f51dd90a5b 100644
--- a/src/array/growable/binary.rs
+++ b/src/array/growable/binary.rs
@@ -4,6 +4,7 @@ use crate::{
     array::{Array, BinaryArray, Offset},
     bitmap::MutableBitmap,
     buffer::MutableBuffer,
+    datatypes::DataType,
 };
 
 use super::{
@@ -14,6 +15,7 @@ use super::{
 /// Concrete [`Growable`] for the [`BinaryArray`].
 pub struct GrowableBinary<'a, O: Offset> {
     arrays: Vec<&'a BinaryArray<O>>,
+    data_type: DataType,
     validity: MutableBitmap,
     values: MutableBuffer<u8>,
     offsets: MutableBuffer<O>,
@@ -25,8 +27,10 @@ pub struct GrowableBinary<'a, O: Offset> {
 
 impl<'a, O: Offset> GrowableBinary<'a, O> {
     /// # Panics
-    /// This function panics if any of the `arrays` is not downcastable to `PrimitiveArray<T>`.
+    /// If `arrays` is empty.
     pub fn new(arrays: Vec<&'a BinaryArray<O>>, mut use_validity: bool, capacity: usize) -> Self {
+        let data_type = arrays[0].data_type().clone();
+
         // if any of the arrays has nulls, insertions from any array requires setting bits
         // as there is at least one array with nulls.
         if !use_validity & arrays.iter().any(|array| array.null_count() > 0) {
@@ -44,6 +48,7 @@ impl<'a, O: Offset> GrowableBinary<'a, O> {
 
         Self {
             arrays,
+            data_type,
             values: MutableBuffer::with_capacity(0),
             offsets,
             length,
@@ -53,11 +58,12 @@ impl<'a, O: Offset> GrowableBinary<'a, O> {
     }
 
     fn to(&mut self) -> BinaryArray<O> {
+        let data_type = self.data_type.clone();
         let validity = std::mem::take(&mut self.validity);
         let offsets = std::mem::take(&mut self.offsets);
         let values = std::mem::take(&mut self.values);
 
-        BinaryArray::<O>::from_data(offsets.into(), values.into(), validity.into())
+        BinaryArray::<O>::from_data(data_type, offsets.into(), values.into(), validity.into())
     }
 }
 
@@ -94,6 +100,6 @@ impl<'a, O: Offset> Growable<'a> for GrowableBinary<'a, O> {
 
 impl<'a, O: Offset> From<GrowableBinary<'a, O>> for BinaryArray<O> {
     fn from(val: GrowableBinary<'a, O>) -> Self {
-        BinaryArray::<O>::from_data(val.offsets.into(), val.values.into(), val.validity.into())
+        BinaryArray::<O>::from_data(val.data_type, val.offsets.into(), val.values.into(), val.validity.into())
     }
 }
diff --git a/src/array/growable/boolean.rs b/src/array/growable/boolean.rs
index 0492465283..1dca6c1b1e 100644
--- a/src/array/growable/boolean.rs
+++ b/src/array/growable/boolean.rs
@@ -3,6 +3,7 @@ use std::sync::Arc;
 use crate::{
     array::{Array, BooleanArray},
     bitmap::MutableBitmap,
+    datatypes::DataType,
 };
 
 use super::{
@@ -13,6 +14,7 @@ use super::{
 /// Concrete [`Growable`] for the [`BooleanArray`].
 pub struct GrowableBoolean<'a> {
     arrays: Vec<&'a BooleanArray>,
+    data_type: DataType,
     validity: MutableBitmap,
     values: MutableBitmap,
     // function used to extend nulls from arrays. This function's lifetime is bound to the array
@@ -22,6 +24,8 @@ pub struct GrowableBoolean<'a> {
 
 impl<'a> GrowableBoolean<'a> {
     pub fn new(arrays: Vec<&'a BooleanArray>, mut use_validity: bool, capacity: usize) -> Self {
+        let data_type = arrays[0].data_type().clone();
+
         // if any of the arrays has nulls, insertions from any array requires setting bits
         // as there is at least one array with nulls.
         if !use_validity & arrays.iter().any(|array| array.null_count() > 0) {
@@ -35,6 +39,7 @@ impl<'a> GrowableBoolean<'a> {
 
         Self {
             arrays,
+            data_type,
             values: MutableBitmap::with_capacity(capacity),
             validity: MutableBitmap::with_capacity(capacity),
             extend_null_bits,
@@ -45,7 +50,7 @@ impl<'a> GrowableBoolean<'a> {
         let validity = std::mem::take(&mut self.validity);
         let values = std::mem::take(&mut self.values);
 
-        BooleanArray::from_data(values.into(), validity.into())
+        BooleanArray::from_data(self.data_type.clone(), values.into(), validity.into())
     }
 }
 
@@ -76,6 +81,6 @@ impl<'a> Growable<'a> for GrowableBoolean<'a> {
 
 impl<'a> From<GrowableBoolean<'a>> for BooleanArray {
     fn from(val: GrowableBoolean<'a>) -> Self {
-        BooleanArray::from_data(val.values.into(), val.validity.into())
+        BooleanArray::from_data(val.data_type, val.values.into(), val.validity.into())
     }
 }
diff --git a/src/array/growable/mod.rs b/src/array/growable/mod.rs
index ecc6503569..7f57b87a6f 100644
--- a/src/array/growable/mod.rs
+++ b/src/array/growable/mod.rs
@@ -96,9 +96,10 @@ pub fn make_growable<'a>(
     let data_type = arrays[0].data_type();
     assert!(arrays.iter().all(|&item| item.data_type() == data_type));
 
-    match data_type {
-        DataType::Null => Box::new(null::GrowableNull::new()),
-        DataType::Boolean => {
+    use PhysicalType::*;
+    match data_type.to_physical_type() {
+        Null => Box::new(null::GrowableNull::new(data_type.clone())),
+        Boolean => {
             let arrays = arrays
                 .iter()
                 .map(|array| array.as_any().downcast_ref().unwrap())
@@ -109,33 +110,10 @@ pub fn make_growable<'a>(
                 capacity,
             ))
         }
-        DataType::Int8 => dyn_growable!(i8, arrays, use_validity, capacity),
-        DataType::Int16 => dyn_growable!(i16, arrays, use_validity, capacity),
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => {
-            dyn_growable!(i32, arrays, use_validity, capacity)
-        }
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => {
-            dyn_growable!(i64, arrays, use_validity, capacity)
-        }
-        DataType::Interval(IntervalUnit::DayTime) => {
-            dyn_growable!(days_ms, arrays, use_validity, capacity)
-        }
-        DataType::Decimal(_, _) => dyn_growable!(i128, arrays, use_validity, capacity),
-        DataType::UInt8 => dyn_growable!(u8, arrays, use_validity, capacity),
-        DataType::UInt16 => dyn_growable!(u16, arrays, use_validity, capacity),
-        DataType::UInt32 => dyn_growable!(u32, arrays, use_validity, capacity),
-        DataType::UInt64 => dyn_growable!(u64, arrays, use_validity, capacity),
-        DataType::Float16 => unreachable!(),
-        DataType::Float32 => dyn_growable!(f32, arrays, use_validity, capacity),
-        DataType::Float64 => dyn_growable!(f64, arrays, use_validity, capacity),
-        DataType::Utf8 => {
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+            dyn_growable!($T, arrays, use_validity, capacity)
+        }),
+        Utf8 => {
             let arrays = arrays
                 .iter()
                 .map(|array| array.as_any().downcast_ref().unwrap())
@@ -146,7 +124,7 @@ pub fn make_growable<'a>(
                 capacity,
             ))
         }
-        DataType::LargeUtf8 => {
+        LargeUtf8 => {
             let arrays = arrays
                 .iter()
                 .map(|array| array.as_any().downcast_ref().unwrap())
@@ -157,7 +135,7 @@ pub fn make_growable<'a>(
                 capacity,
             ))
         }
-        DataType::Binary => {
+        Binary => {
             let arrays = arrays
                 .iter()
                 .map(|array| array.as_any().downcast_ref().unwrap())
@@ -168,7 +146,7 @@ pub fn make_growable<'a>(
                 capacity,
             ))
         }
-        DataType::LargeBinary => {
+        LargeBinary => {
             let arrays = arrays
                 .iter()
                 .map(|array| array.as_any().downcast_ref().unwrap())
@@ -179,7 +157,7 @@ pub fn make_growable<'a>(
                 capacity,
             ))
         }
-        DataType::FixedSizeBinary(_) => {
+        FixedSizeBinary => {
             let arrays = arrays
                 .iter()
                 .map(|array| array.as_any().downcast_ref().unwrap())
@@ -190,8 +168,7 @@ pub fn make_growable<'a>(
                 capacity,
             ))
         }
-
-        DataType::List(_) => {
+        List => {
             let arrays = arrays
                 .iter()
                 .map(|array| array.as_any().downcast_ref().unwrap())
@@ -202,7 +179,7 @@ pub fn make_growable<'a>(
                 capacity,
             ))
         }
-        DataType::LargeList(_) => {
+        LargeList => {
             let arrays = arrays
                 .iter()
                 .map(|array| array.as_any().downcast_ref().unwrap())
@@ -213,7 +190,7 @@ pub fn make_growable<'a>(
                 capacity,
             ))
         }
-        DataType::Struct(_) => {
+        Struct => {
             let arrays = arrays
                 .iter()
                 .map(|array| array.as_any().downcast_ref().unwrap())
@@ -224,10 +201,10 @@ pub fn make_growable<'a>(
                 capacity,
             ))
         }
-        DataType::FixedSizeList(_, _) => todo!(),
-        DataType::Union(_, _, _) => todo!(),
-        DataType::Dictionary(key_type, _) => {
-            with_match_dictionary_key_type!(key_type.as_ref(), |$T| {
+        FixedSizeList => todo!(),
+        Union => todo!(),
+        Dictionary(key_type) => {
+            with_match_physical_dictionary_key_type!(key_type, |$T| {
                 dyn_dict_growable!($T, arrays, use_validity, capacity)
             })
         }
diff --git a/src/array/growable/null.rs b/src/array/growable/null.rs
index 74c4498a89..725968521f 100644
--- a/src/array/growable/null.rs
+++ b/src/array/growable/null.rs
@@ -1,23 +1,30 @@
 use std::sync::Arc;
 
-use crate::array::{Array, NullArray};
+use crate::{
+    array::{Array, NullArray},
+    datatypes::DataType,
+};
 
 use super::Growable;
 
 /// Concrete [`Growable`] for the [`NullArray`].
 pub struct GrowableNull {
+    data_type: DataType,
     length: usize,
 }
 
 impl Default for GrowableNull {
     fn default() -> Self {
-        Self { length: 0 }
+        Self::new(DataType::Null)
     }
 }
 
 impl GrowableNull {
-    pub fn new() -> Self {
-        Self::default()
+    pub fn new(data_type: DataType) -> Self {
+        Self {
+            data_type,
+            length: 0,
+        }
     }
 }
 
@@ -31,16 +38,16 @@ impl<'a> Growable<'a> for GrowableNull {
     }
 
     fn as_arc(&mut self) -> Arc<dyn Array> {
-        Arc::new(NullArray::from_data(self.length))
+        Arc::new(NullArray::from_data(self.data_type.clone(), self.length))
     }
 
     fn as_box(&mut self) -> Box<dyn Array> {
-        Box::new(NullArray::from_data(self.length))
+        Box::new(NullArray::from_data(self.data_type.clone(), self.length))
     }
 }
 
 impl From<GrowableNull> for NullArray {
     fn from(val: GrowableNull) -> Self {
-        NullArray::from_data(val.length)
+        NullArray::from_data(val.data_type, val.length)
     }
 }
diff --git a/src/array/growable/structure.rs b/src/array/growable/structure.rs
index f0f9053e95..620fe19150 100644
--- a/src/array/growable/structure.rs
+++ b/src/array/growable/structure.rs
@@ -3,6 +3,7 @@ use std::sync::Arc;
 use crate::{
     array::{Array, StructArray},
     bitmap::MutableBitmap,
+    datatypes::DataType,
 };
 
 use super::{
@@ -68,7 +69,11 @@ impl<'a> GrowableStruct<'a> {
         let values = std::mem::take(&mut self.values);
         let values = values.into_iter().map(|mut x| x.as_arc()).collect();
 
-        StructArray::from_data(self.arrays[0].fields().to_vec(), values, validity.into())
+        StructArray::from_data(
+            DataType::Struct(self.arrays[0].fields().to_vec()),
+            values,
+            validity.into(),
+        )
     }
 }
 
@@ -116,6 +121,10 @@ impl<'a> From<GrowableStruct<'a>> for StructArray {
     fn from(val: GrowableStruct<'a>) -> Self {
         let values = val.values.into_iter().map(|mut x| x.as_arc()).collect();
 
-        StructArray::from_data(val.arrays[0].fields().to_vec(), values, val.validity.into())
+        StructArray::from_data(
+            DataType::Struct(val.arrays[0].fields().to_vec()),
+            values,
+            val.validity.into(),
+        )
     }
 }
diff --git a/src/array/growable/utf8.rs b/src/array/growable/utf8.rs
index 2d34c64866..355d1e2012 100644
--- a/src/array/growable/utf8.rs
+++ b/src/array/growable/utf8.rs
@@ -56,7 +56,12 @@ impl<'a, O: Offset> GrowableUtf8<'a, O> {
         let values = std::mem::take(&mut self.values);
 
         unsafe {
-            Utf8Array::<O>::from_data_unchecked(offsets.into(), values.into(), validity.into())
+            Utf8Array::<O>::from_data_unchecked(
+                self.arrays[0].data_type().clone(),
+                offsets.into(),
+                values.into(),
+                validity.into(),
+            )
         }
     }
 }
@@ -96,6 +101,7 @@ impl<'a, O: Offset> From<GrowableUtf8<'a, O>> for Utf8Array<O> {
     fn from(val: GrowableUtf8<'a, O>) -> Self {
         unsafe {
             Utf8Array::<O>::from_data_unchecked(
+                val.arrays[0].data_type().clone(),
                 val.offsets.into(),
                 val.values.into(),
                 val.validity.into(),
diff --git a/src/array/list/mod.rs b/src/array/list/mod.rs
index 8e767dda0a..670822b2aa 100644
--- a/src/array/list/mod.rs
+++ b/src/array/list/mod.rs
@@ -120,7 +120,6 @@ impl<O: Offset> ListArray<O> {
 }
 
 impl<O: Offset> ListArray<O> {
-    #[inline]
     pub fn default_datatype(data_type: DataType) -> DataType {
         let field = Box::new(Field::new("item", data_type, true));
         if O::is_large() {
@@ -130,22 +129,22 @@ impl<O: Offset> ListArray<O> {
         }
     }
 
-    #[inline]
     pub fn get_child_field(data_type: &DataType) -> &Field {
         if O::is_large() {
-            if let DataType::LargeList(child) = data_type {
-                child.as_ref()
-            } else {
-                panic!("Wrong DataType")
+            match data_type {
+                DataType::LargeList(child) => child.as_ref(),
+                DataType::Extension(_, child, _) => Self::get_child_field(child),
+                _ => panic!("Wrong DataType"),
             }
-        } else if let DataType::List(child) = data_type {
-            child.as_ref()
         } else {
-            panic!("Wrong DataType")
+            match data_type {
+                DataType::List(child) => child.as_ref(),
+                DataType::Extension(_, child, _) => Self::get_child_field(child),
+                _ => panic!("Wrong DataType"),
+            }
         }
     }
 
-    #[inline]
     pub fn get_child_type(data_type: &DataType) -> &DataType {
         Self::get_child_field(data_type).data_type()
     }
diff --git a/src/array/mod.rs b/src/array/mod.rs
index 46bc3730d8..3b23a377e3 100644
--- a/src/array/mod.rs
+++ b/src/array/mod.rs
@@ -20,7 +20,7 @@ use crate::error::Result;
 use crate::types::days_ms;
 use crate::{
     bitmap::{Bitmap, MutableBitmap},
-    datatypes::{DataType, IntervalUnit},
+    datatypes::DataType,
 };
 
 /// A trait representing an immutable Arrow array. Arrow arrays are trait objects
@@ -164,47 +164,65 @@ macro_rules! with_match_dictionary_key_type {(
     }
 })}
 
+macro_rules! with_match_physical_dictionary_key_type {(
+    $key_type:expr, | $_:tt $T:ident | $($body:tt)*
+) => ({
+    macro_rules! __with_ty__ {( $_ $T:ident ) => ( $($body)* )}
+    use crate::datatypes::DictionaryIndexType::*;
+    match $key_type {
+        Int8 => __with_ty__! { i8 },
+        Int16 => __with_ty__! { i16 },
+        Int32 => __with_ty__! { i32 },
+        Int64 => __with_ty__! { i64 },
+        UInt8 => __with_ty__! { u8 },
+        UInt16 => __with_ty__! { u16 },
+        UInt32 => __with_ty__! { u32 },
+        UInt64 => __with_ty__! { u64 },
+    }
+})}
+
+macro_rules! with_match_primitive_type {(
+    $key_type:expr, | $_:tt $T:ident | $($body:tt)*
+) => ({
+    macro_rules! __with_ty__ {( $_ $T:ident ) => ( $($body)* )}
+    use crate::datatypes::PrimitiveType::*;
+    match $key_type {
+        Int8 => __with_ty__! { i8 },
+        Int16 => __with_ty__! { i16 },
+        Int32 => __with_ty__! { i32 },
+        Int64 => __with_ty__! { i64 },
+        Int128 => __with_ty__! { i128 },
+        DaysMs => __with_ty__! { days_ms },
+        UInt8 => __with_ty__! { u8 },
+        UInt16 => __with_ty__! { u16 },
+        UInt32 => __with_ty__! { u32 },
+        UInt64 => __with_ty__! { u64 },
+        Float32 => __with_ty__! { f32 },
+        Float64 => __with_ty__! { f64 },
+    }
+})}
+
 impl Display for dyn Array {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self.data_type() {
-            DataType::Null => fmt_dyn!(self, NullArray, f),
-            DataType::Boolean => fmt_dyn!(self, BooleanArray, f),
-            DataType::Int8 => fmt_dyn!(self, PrimitiveArray<i8>, f),
-            DataType::Int16 => fmt_dyn!(self, PrimitiveArray<i16>, f),
-            DataType::Int32
-            | DataType::Date32
-            | DataType::Time32(_)
-            | DataType::Interval(IntervalUnit::YearMonth) => {
-                fmt_dyn!(self, PrimitiveArray<i32>, f)
-            }
-            DataType::Interval(IntervalUnit::DayTime) => {
-                fmt_dyn!(self, PrimitiveArray<days_ms>, f)
-            }
-            DataType::Int64
-            | DataType::Date64
-            | DataType::Time64(_)
-            | DataType::Timestamp(_, _)
-            | DataType::Duration(_) => fmt_dyn!(self, PrimitiveArray<i64>, f),
-            DataType::Decimal(_, _) => fmt_dyn!(self, PrimitiveArray<i128>, f),
-            DataType::UInt8 => fmt_dyn!(self, PrimitiveArray<u8>, f),
-            DataType::UInt16 => fmt_dyn!(self, PrimitiveArray<u16>, f),
-            DataType::UInt32 => fmt_dyn!(self, PrimitiveArray<u32>, f),
-            DataType::UInt64 => fmt_dyn!(self, PrimitiveArray<u64>, f),
-            DataType::Float16 => unreachable!(),
-            DataType::Float32 => fmt_dyn!(self, PrimitiveArray<f32>, f),
-            DataType::Float64 => fmt_dyn!(self, PrimitiveArray<f64>, f),
-            DataType::Binary => fmt_dyn!(self, BinaryArray<i32>, f),
-            DataType::LargeBinary => fmt_dyn!(self, BinaryArray<i64>, f),
-            DataType::FixedSizeBinary(_) => fmt_dyn!(self, FixedSizeBinaryArray, f),
-            DataType::Utf8 => fmt_dyn!(self, Utf8Array::<i32>, f),
-            DataType::LargeUtf8 => fmt_dyn!(self, Utf8Array::<i64>, f),
-            DataType::List(_) => fmt_dyn!(self, ListArray::<i32>, f),
-            DataType::LargeList(_) => fmt_dyn!(self, ListArray::<i64>, f),
-            DataType::FixedSizeList(_, _) => fmt_dyn!(self, FixedSizeListArray, f),
-            DataType::Struct(_) => fmt_dyn!(self, StructArray, f),
-            DataType::Union(_, _, _) => fmt_dyn!(self, UnionArray, f),
-            DataType::Dictionary(key_type, _) => {
-                with_match_dictionary_key_type!(key_type.as_ref(), |$T| {
+        use crate::datatypes::PhysicalType::*;
+        match self.data_type().to_physical_type() {
+            Null => fmt_dyn!(self, NullArray, f),
+            Boolean => fmt_dyn!(self, BooleanArray, f),
+            Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+                fmt_dyn!(self, PrimitiveArray<$T>, f)
+            }),
+            Binary => fmt_dyn!(self, BinaryArray<i32>, f),
+            LargeBinary => fmt_dyn!(self, BinaryArray<i64>, f),
+            FixedSizeBinary => fmt_dyn!(self, FixedSizeBinaryArray, f),
+            Utf8 => fmt_dyn!(self, Utf8Array::<i32>, f),
+            LargeUtf8 => fmt_dyn!(self, Utf8Array::<i64>, f),
+            List => fmt_dyn!(self, ListArray::<i32>, f),
+            LargeList => fmt_dyn!(self, ListArray::<i64>, f),
+            FixedSizeList => fmt_dyn!(self, FixedSizeListArray, f),
+            Struct => fmt_dyn!(self, StructArray, f),
+            Union => fmt_dyn!(self, UnionArray, f),
+            Dictionary(key_type) => {
+                with_match_physical_dictionary_key_type!(key_type, |$T| {
                     fmt_dyn!(self, DictionaryArray::<$T>, f)
                 })
             }
@@ -214,46 +232,26 @@ impl Display for dyn Array {
 
 /// Creates a new [`Array`] with a [`Array::len`] of 0.
 pub fn new_empty_array(data_type: DataType) -> Box<dyn Array> {
-    match data_type {
-        DataType::Null => Box::new(NullArray::new_empty()),
-        DataType::Boolean => Box::new(BooleanArray::new_empty()),
-        DataType::Int8 => Box::new(PrimitiveArray::<i8>::new_empty(data_type)),
-        DataType::Int16 => Box::new(PrimitiveArray::<i16>::new_empty(data_type)),
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => {
-            Box::new(PrimitiveArray::<i32>::new_empty(data_type))
-        }
-        DataType::Interval(IntervalUnit::DayTime) => {
-            Box::new(PrimitiveArray::<days_ms>::new_empty(data_type))
-        }
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => Box::new(PrimitiveArray::<i64>::new_empty(data_type)),
-        DataType::Decimal(_, _) => Box::new(PrimitiveArray::<i128>::new_empty(data_type)),
-        DataType::UInt8 => Box::new(PrimitiveArray::<u8>::new_empty(data_type)),
-        DataType::UInt16 => Box::new(PrimitiveArray::<u16>::new_empty(data_type)),
-        DataType::UInt32 => Box::new(PrimitiveArray::<u32>::new_empty(data_type)),
-        DataType::UInt64 => Box::new(PrimitiveArray::<u64>::new_empty(data_type)),
-        DataType::Float16 => unreachable!(),
-        DataType::Float32 => Box::new(PrimitiveArray::<f32>::new_empty(data_type)),
-        DataType::Float64 => Box::new(PrimitiveArray::<f64>::new_empty(data_type)),
-        DataType::Binary => Box::new(BinaryArray::<i32>::new_empty()),
-        DataType::LargeBinary => Box::new(BinaryArray::<i64>::new_empty()),
-        DataType::FixedSizeBinary(_) => Box::new(FixedSizeBinaryArray::new_empty(data_type)),
-        DataType::Utf8 => Box::new(Utf8Array::<i32>::new_empty()),
-        DataType::LargeUtf8 => Box::new(Utf8Array::<i64>::new_empty()),
-        DataType::List(_) => Box::new(ListArray::<i32>::new_empty(data_type)),
-        DataType::LargeList(_) => Box::new(ListArray::<i64>::new_empty(data_type)),
-        DataType::FixedSizeList(_, _) => Box::new(FixedSizeListArray::new_empty(data_type)),
-        DataType::Struct(fields) => Box::new(StructArray::new_empty(&fields)),
-        DataType::Union(_, _, _) => Box::new(UnionArray::new_empty(data_type)),
-        DataType::Dictionary(key_type, value_type) => {
-            with_match_dictionary_key_type!(key_type.as_ref(), |$T| {
-                Box::new(DictionaryArray::<$T>::new_empty(*value_type))
+    use crate::datatypes::PhysicalType::*;
+    match data_type.to_physical_type() {
+        Null => Box::new(NullArray::new_empty(data_type)),
+        Boolean => Box::new(BooleanArray::new_empty(data_type)),
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+            Box::new(PrimitiveArray::<$T>::new_empty(data_type))
+        }),
+        Binary => Box::new(BinaryArray::<i32>::new_empty(data_type)),
+        LargeBinary => Box::new(BinaryArray::<i64>::new_empty(data_type)),
+        FixedSizeBinary => Box::new(FixedSizeBinaryArray::new_empty(data_type)),
+        Utf8 => Box::new(Utf8Array::<i32>::new_empty(data_type)),
+        LargeUtf8 => Box::new(Utf8Array::<i64>::new_empty(data_type)),
+        List => Box::new(ListArray::<i32>::new_empty(data_type)),
+        LargeList => Box::new(ListArray::<i64>::new_empty(data_type)),
+        FixedSizeList => Box::new(FixedSizeListArray::new_empty(data_type)),
+        Struct => Box::new(StructArray::new_empty(data_type)),
+        Union => Box::new(UnionArray::new_empty(data_type)),
+        Dictionary(key_type) => {
+            with_match_physical_dictionary_key_type!(key_type, |$T| {
+                Box::new(DictionaryArray::<$T>::new_empty(data_type))
             })
         }
     }
@@ -263,46 +261,26 @@ pub fn new_empty_array(data_type: DataType) -> Box<dyn Array> {
 /// The array is guaranteed to have [`Array::null_count`] equal to [`Array::len`]
 /// for all types except Union, which does not have a validity.
 pub fn new_null_array(data_type: DataType, length: usize) -> Box<dyn Array> {
-    match data_type {
-        DataType::Null => Box::new(NullArray::new_null(length)),
-        DataType::Boolean => Box::new(BooleanArray::new_null(length)),
-        DataType::Int8 => Box::new(PrimitiveArray::<i8>::new_null(data_type, length)),
-        DataType::Int16 => Box::new(PrimitiveArray::<i16>::new_null(data_type, length)),
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => {
-            Box::new(PrimitiveArray::<i32>::new_null(data_type, length))
-        }
-        DataType::Interval(IntervalUnit::DayTime) => {
-            Box::new(PrimitiveArray::<days_ms>::new_null(data_type, length))
-        }
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => Box::new(PrimitiveArray::<i64>::new_null(data_type, length)),
-        DataType::Decimal(_, _) => Box::new(PrimitiveArray::<i128>::new_null(data_type, length)),
-        DataType::UInt8 => Box::new(PrimitiveArray::<u8>::new_null(data_type, length)),
-        DataType::UInt16 => Box::new(PrimitiveArray::<u16>::new_null(data_type, length)),
-        DataType::UInt32 => Box::new(PrimitiveArray::<u32>::new_null(data_type, length)),
-        DataType::UInt64 => Box::new(PrimitiveArray::<u64>::new_null(data_type, length)),
-        DataType::Float16 => unreachable!(),
-        DataType::Float32 => Box::new(PrimitiveArray::<f32>::new_null(data_type, length)),
-        DataType::Float64 => Box::new(PrimitiveArray::<f64>::new_null(data_type, length)),
-        DataType::Binary => Box::new(BinaryArray::<i32>::new_null(length)),
-        DataType::LargeBinary => Box::new(BinaryArray::<i64>::new_null(length)),
-        DataType::FixedSizeBinary(_) => Box::new(FixedSizeBinaryArray::new_null(data_type, length)),
-        DataType::Utf8 => Box::new(Utf8Array::<i32>::new_null(length)),
-        DataType::LargeUtf8 => Box::new(Utf8Array::<i64>::new_null(length)),
-        DataType::List(_) => Box::new(ListArray::<i32>::new_null(data_type, length)),
-        DataType::LargeList(_) => Box::new(ListArray::<i64>::new_null(data_type, length)),
-        DataType::FixedSizeList(_, _) => Box::new(FixedSizeListArray::new_null(data_type, length)),
-        DataType::Struct(fields) => Box::new(StructArray::new_null(&fields, length)),
-        DataType::Union(_, _, _) => Box::new(UnionArray::new_null(data_type, length)),
-        DataType::Dictionary(key_type, value_type) => {
-            with_match_dictionary_key_type!(key_type.as_ref(), |$T| {
-                Box::new(DictionaryArray::<$T>::new_null(*value_type, length))
+    use crate::datatypes::PhysicalType::*;
+    match data_type.to_physical_type() {
+        Null => Box::new(NullArray::new_null(data_type, length)),
+        Boolean => Box::new(BooleanArray::new_null(data_type, length)),
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+            Box::new(PrimitiveArray::<$T>::new_null(data_type, length))
+        }),
+        Binary => Box::new(BinaryArray::<i32>::new_null(data_type, length)),
+        LargeBinary => Box::new(BinaryArray::<i64>::new_null(data_type, length)),
+        FixedSizeBinary => Box::new(FixedSizeBinaryArray::new_null(data_type, length)),
+        Utf8 => Box::new(Utf8Array::<i32>::new_null(data_type, length)),
+        LargeUtf8 => Box::new(Utf8Array::<i64>::new_null(data_type, length)),
+        List => Box::new(ListArray::<i32>::new_null(data_type, length)),
+        LargeList => Box::new(ListArray::<i64>::new_null(data_type, length)),
+        FixedSizeList => Box::new(FixedSizeListArray::new_null(data_type, length)),
+        Struct => Box::new(StructArray::new_null(data_type, length)),
+        Union => Box::new(UnionArray::new_null(data_type, length)),
+        Dictionary(key_type) => {
+            with_match_physical_dictionary_key_type!(key_type, |$T| {
+                Box::new(DictionaryArray::<$T>::new_null(data_type, length))
             })
         }
     }
@@ -320,43 +298,25 @@ macro_rules! clone_dyn {
 /// This operation is `O(1)` over `len`, as it amounts to increase two ref counts
 /// and moving the concrete struct under a `Box`.
 pub fn clone(array: &dyn Array) -> Box<dyn Array> {
-    match array.data_type() {
-        DataType::Null => clone_dyn!(array, NullArray),
-        DataType::Boolean => clone_dyn!(array, BooleanArray),
-        DataType::Int8 => clone_dyn!(array, PrimitiveArray<i8>),
-        DataType::Int16 => clone_dyn!(array, PrimitiveArray<i16>),
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => {
-            clone_dyn!(array, PrimitiveArray<i32>)
-        }
-        DataType::Interval(IntervalUnit::DayTime) => clone_dyn!(array, PrimitiveArray<days_ms>),
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => clone_dyn!(array, PrimitiveArray<i64>),
-        DataType::Decimal(_, _) => clone_dyn!(array, PrimitiveArray<i128>),
-        DataType::UInt8 => clone_dyn!(array, PrimitiveArray<u8>),
-        DataType::UInt16 => clone_dyn!(array, PrimitiveArray<u16>),
-        DataType::UInt32 => clone_dyn!(array, PrimitiveArray<u32>),
-        DataType::UInt64 => clone_dyn!(array, PrimitiveArray<u64>),
-        DataType::Float16 => unreachable!(),
-        DataType::Float32 => clone_dyn!(array, PrimitiveArray<f32>),
-        DataType::Float64 => clone_dyn!(array, PrimitiveArray<f64>),
-        DataType::Binary => clone_dyn!(array, BinaryArray<i32>),
-        DataType::LargeBinary => clone_dyn!(array, BinaryArray<i64>),
-        DataType::FixedSizeBinary(_) => clone_dyn!(array, FixedSizeBinaryArray),
-        DataType::Utf8 => clone_dyn!(array, Utf8Array::<i32>),
-        DataType::LargeUtf8 => clone_dyn!(array, Utf8Array::<i64>),
-        DataType::List(_) => clone_dyn!(array, ListArray::<i32>),
-        DataType::LargeList(_) => clone_dyn!(array, ListArray::<i64>),
-        DataType::FixedSizeList(_, _) => clone_dyn!(array, FixedSizeListArray),
-        DataType::Struct(_) => clone_dyn!(array, StructArray),
-        DataType::Union(_, _, _) => clone_dyn!(array, UnionArray),
-        DataType::Dictionary(key_type, _) => {
-            with_match_dictionary_key_type!(key_type.as_ref(), |$T| {
+    use crate::datatypes::PhysicalType::*;
+    match array.data_type().to_physical_type() {
+        Null => clone_dyn!(array, NullArray),
+        Boolean => clone_dyn!(array, BooleanArray),
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+            clone_dyn!(array, PrimitiveArray<$T>)
+        }),
+        Binary => clone_dyn!(array, BinaryArray<i32>),
+        LargeBinary => clone_dyn!(array, BinaryArray<i64>),
+        FixedSizeBinary => clone_dyn!(array, FixedSizeBinaryArray),
+        Utf8 => clone_dyn!(array, Utf8Array::<i32>),
+        LargeUtf8 => clone_dyn!(array, Utf8Array::<i64>),
+        List => clone_dyn!(array, ListArray::<i32>),
+        LargeList => clone_dyn!(array, ListArray::<i64>),
+        FixedSizeList => clone_dyn!(array, FixedSizeListArray),
+        Struct => clone_dyn!(array, StructArray),
+        Union => clone_dyn!(array, UnionArray),
+        Dictionary(key_type) => {
+            with_match_physical_dictionary_key_type!(key_type, |$T| {
                 clone_dyn!(array, DictionaryArray::<$T>)
             })
         }
diff --git a/src/array/null.rs b/src/array/null.rs
index 4587e6bb17..dcddf92e51 100644
--- a/src/array/null.rs
+++ b/src/array/null.rs
@@ -12,19 +12,19 @@ pub struct NullArray {
 
 impl NullArray {
     /// Returns a new empty [`NullArray`].
-    pub fn new_empty() -> Self {
-        Self::from_data(0)
+    pub fn new_empty(data_type: DataType) -> Self {
+        Self::from_data(data_type, 0)
     }
 
     /// Returns a new [`NullArray`].
-    pub fn new_null(length: usize) -> Self {
-        Self::from_data(length)
+    pub fn new_null(data_type: DataType, length: usize) -> Self {
+        Self::from_data(data_type, length)
     }
 
     /// Returns a new [`NullArray`].
-    pub fn from_data(length: usize) -> Self {
+    pub fn from_data(data_type: DataType, length: usize) -> Self {
         Self {
-            data_type: DataType::Null,
+            data_type,
             length,
             offset: 0,
         }
diff --git a/src/array/struct_.rs b/src/array/struct_.rs
index 88bb1057b6..d773b51e1d 100644
--- a/src/array/struct_.rs
+++ b/src/array/struct_.rs
@@ -24,7 +24,7 @@ use super::{ffi::ToFfi, new_empty_array, new_null_array, Array, FromFfi};
 ///     Field::new("c", DataType::Int32, false),
 /// ];
 ///
-/// let array = StructArray::from_data(fields, vec![boolean, int], None);
+/// let array = StructArray::from_data(DataType::Struct(fields), vec![boolean, int], None);
 /// ```
 #[derive(Debug, Clone)]
 pub struct StructArray {
@@ -35,21 +35,29 @@ pub struct StructArray {
 
 impl StructArray {
     /// Creates an empty [`StructArray`].
-    pub fn new_empty(fields: &[Field]) -> Self {
-        let values = fields
-            .iter()
-            .map(|field| new_empty_array(field.data_type().clone()).into())
-            .collect();
-        Self::from_data(fields.to_vec(), values, None)
+    pub fn new_empty(data_type: DataType) -> Self {
+        if let DataType::Struct(fields) = &data_type {
+            let values = fields
+                .iter()
+                .map(|field| new_empty_array(field.data_type().clone()).into())
+                .collect();
+            Self::from_data(data_type, values, None)
+        } else {
+            panic!("StructArray must be initialized with DataType::Struct");
+        }
     }
 
     /// Creates a null [`StructArray`] of length `length`.
-    pub fn new_null(fields: &[Field], length: usize) -> Self {
-        let values = fields
-            .iter()
-            .map(|field| new_null_array(field.data_type().clone(), length).into())
-            .collect();
-        Self::from_data(fields.to_vec(), values, Some(Bitmap::new_zeroed(length)))
+    pub fn new_null(data_type: DataType, length: usize) -> Self {
+        if let DataType::Struct(fields) = &data_type {
+            let values = fields
+                .iter()
+                .map(|field| new_null_array(field.data_type().clone(), length).into())
+                .collect();
+            Self::from_data(data_type, values, Some(Bitmap::new_zeroed(length)))
+        } else {
+            panic!("StructArray must be initialized with DataType::Struct");
+        }
     }
 
     /// Canonical method to create a [`StructArray`].
@@ -58,10 +66,11 @@ impl StructArray {
     /// * values's len is different from Fields' length.
     /// * any element of values has a different length than the first element.
     pub fn from_data(
-        fields: Vec<Field>,
+        data_type: DataType,
         values: Vec<Arc<dyn Array>>,
         validity: Option<Bitmap>,
     ) -> Self {
+        let fields = Self::get_fields(&data_type);
         assert!(!fields.is_empty());
         assert_eq!(fields.len(), values.len());
         assert!(values.iter().all(|x| x.len() == values[0].len()));
@@ -69,7 +78,7 @@ impl StructArray {
             assert_eq!(values[0].len(), validity.len());
         }
         Self {
-            data_type: DataType::Struct(fields),
+            data_type,
             values,
             validity,
         }
@@ -122,10 +131,10 @@ impl StructArray {
 impl StructArray {
     /// Returns the fields the `DataType::Struct`.
     pub fn get_fields(data_type: &DataType) -> &[Field] {
-        if let DataType::Struct(fields) = data_type {
-            fields
-        } else {
-            panic!("Wrong datatype passed to Struct.")
+        match data_type {
+            DataType::Struct(fields) => fields,
+            DataType::Extension(_, inner, _) => Self::get_fields(inner),
+            _ => panic!("Wrong datatype passed to Struct."),
         }
     }
 }
@@ -199,6 +208,6 @@ unsafe impl<A: ffi::ArrowArrayRef> FromFfi<A> for StructArray {
         if offset > 0 {
             validity = validity.map(|x| x.slice(offset, length))
         }
-        Ok(Self::from_data(fields, values, validity))
+        Ok(Self::from_data(DataType::Struct(fields), values, validity))
     }
 }
diff --git a/src/array/union/mod.rs b/src/array/union/mod.rs
index f688656e2f..f00dcdff94 100644
--- a/src/array/union/mod.rs
+++ b/src/array/union/mod.rs
@@ -89,33 +89,30 @@ impl UnionArray {
         fields: Vec<Arc<dyn Array>>,
         offsets: Option<Buffer<i32>>,
     ) -> Self {
-        let fields_hash = if let DataType::Union(f, ids, is_sparse) = &data_type {
-            if f.len() != fields.len() {
-                panic!(
-                    "The number of `fields` must equal the number of fields in the Union DataType"
-                )
-            };
-            let same_data_types = f
-                .iter()
-                .zip(fields.iter())
-                .all(|(f, array)| f.data_type() == array.data_type());
-            if !same_data_types {
-                panic!("All fields' datatype in the union must equal the datatypes on the fields.")
-            }
-            if offsets.is_none() != *is_sparse {
-                panic!("Sparsness flag must equal to noness of offsets in UnionArray")
-            }
-            ids.as_ref().map(|ids| {
-                ids.iter()
-                    .map(|x| *x as i8)
-                    .enumerate()
-                    .zip(fields.iter().cloned())
-                    .map(|((i, type_), field)| (type_, (i, field)))
-                    .collect()
-            })
-        } else {
-            panic!("Union struct must be created with the corresponding Union DataType")
+        let (f, ids, is_sparse) = Self::get_all(&data_type);
+
+        if f.len() != fields.len() {
+            panic!("The number of `fields` must equal the number of fields in the Union DataType")
         };
+        let same_data_types = f
+            .iter()
+            .zip(fields.iter())
+            .all(|(f, array)| f.data_type() == array.data_type());
+        if !same_data_types {
+            panic!("All fields' datatype in the union must equal the datatypes on the fields.")
+        }
+        if offsets.is_none() != is_sparse {
+            panic!("Sparsness flag must equal to noness of offsets in UnionArray")
+        }
+        let fields_hash = ids.as_ref().map(|ids| {
+            ids.iter()
+                .map(|x| *x as i8)
+                .enumerate()
+                .zip(fields.iter().cloned())
+                .map(|((i, type_), field)| (type_, (i, field)))
+                .collect()
+        });
+
         // not validated:
         // * `offsets` is valid
         // * max id < fields.len()
@@ -218,20 +215,22 @@ impl Array for UnionArray {
 }
 
 impl UnionArray {
-    pub fn get_fields(data_type: &DataType) -> &[Field] {
-        if let DataType::Union(fields, _, _) = data_type {
-            fields
-        } else {
-            panic!("Wrong datatype passed to UnionArray.")
+    fn get_all(data_type: &DataType) -> (&[Field], Option<&[i32]>, bool) {
+        match data_type {
+            DataType::Union(fields, ids, is_sparse) => {
+                (fields, ids.as_ref().map(|x| x.as_ref()), *is_sparse)
+            }
+            DataType::Extension(_, inner, _) => Self::get_all(inner),
+            _ => panic!("Wrong datatype passed to UnionArray."),
         }
     }
 
+    pub fn get_fields(data_type: &DataType) -> &[Field] {
+        Self::get_all(data_type).0
+    }
+
     pub fn is_sparse(data_type: &DataType) -> bool {
-        if let DataType::Union(_, _, is_sparse) = data_type {
-            *is_sparse
-        } else {
-            panic!("Wrong datatype passed to UnionArray.")
-        }
+        Self::get_all(data_type).2
     }
 }
 
diff --git a/src/array/utf8/ffi.rs b/src/array/utf8/ffi.rs
index adc2a1ea86..1363172591 100644
--- a/src/array/utf8/ffi.rs
+++ b/src/array/utf8/ffi.rs
@@ -32,6 +32,7 @@ unsafe impl<O: Offset, A: ffi::ArrowArrayRef> FromFfi<A> for Utf8Array<O> {
             offsets = offsets.slice(offset, length);
             validity = validity.map(|x| x.slice(offset, length))
         }
-        Ok(Self::from_data(offsets, values, validity))
+        let data_type = Self::default_data_type();
+        Ok(Self::from_data(data_type, offsets, values, validity))
     }
 }
diff --git a/src/array/utf8/mod.rs b/src/array/utf8/mod.rs
index bf627d2464..a5cb62b57a 100644
--- a/src/array/utf8/mod.rs
+++ b/src/array/utf8/mod.rs
@@ -37,14 +37,17 @@ pub struct Utf8Array<O: Offset> {
 impl<O: Offset> Utf8Array<O> {
     /// Returns a new empty [`Utf8Array`].
     #[inline]
-    pub fn new_empty() -> Self {
-        unsafe { Self::from_data_unchecked(Buffer::from(&[O::zero()]), Buffer::new(), None) }
+    pub fn new_empty(data_type: DataType) -> Self {
+        unsafe {
+            Self::from_data_unchecked(data_type, Buffer::from(&[O::zero()]), Buffer::new(), None)
+        }
     }
 
     /// Returns a new [`Utf8Array`] whose all slots are null / `None`.
     #[inline]
-    pub fn new_null(length: usize) -> Self {
+    pub fn new_null(data_type: DataType, length: usize) -> Self {
         Self::from_data(
+            data_type,
             Buffer::new_zeroed(length + 1),
             Buffer::new(),
             Some(Bitmap::new_zeroed(length)),
@@ -54,21 +57,27 @@ impl<O: Offset> Utf8Array<O> {
     /// The canonical method to create a [`Utf8Array`] out of low-end APIs.
     /// # Panics
     /// This function panics iff:
+    /// * The `data_type`'s physical type is not consistent with the offset `O`.
     /// * The `offsets` and `values` are consistent
     /// * The `values` between `offsets` are utf8 encoded
     /// * The validity is not `None` and its length is different from `offsets`'s length minus one.
-    pub fn from_data(offsets: Buffer<O>, values: Buffer<u8>, validity: Option<Bitmap>) -> Self {
+    pub fn from_data(
+        data_type: DataType,
+        offsets: Buffer<O>,
+        values: Buffer<u8>,
+        validity: Option<Bitmap>,
+    ) -> Self {
         check_offsets_and_utf8(&offsets, &values);
         if let Some(ref validity) = validity {
             assert_eq!(offsets.len() - 1, validity.len());
         }
 
+        if data_type.to_physical_type() != Self::default_data_type().to_physical_type() {
+            panic!("Utf8Array can only be initialized with DataType::Utf8 or DataType::LargeUtf8")
+        }
+
         Self {
-            data_type: if O::is_large() {
-                DataType::LargeUtf8
-            } else {
-                DataType::Utf8
-            },
+            data_type,
             offsets,
             values,
             validity,
@@ -76,22 +85,32 @@ impl<O: Offset> Utf8Array<O> {
         }
     }
 
+    /// Returns the default [`DataType`], `DataType::Utf8` or `DataType::LargeUtf8`
+    pub fn default_data_type() -> DataType {
+        if O::is_large() {
+            DataType::LargeUtf8
+        } else {
+            DataType::Utf8
+        }
+    }
+
     /// The same as [`Utf8Array::from_data`] but does not check for utf8.
     /// # Safety
     /// `values` buffer must contain valid utf8 between every `offset`
     pub unsafe fn from_data_unchecked(
+        data_type: DataType,
         offsets: Buffer<O>,
         values: Buffer<u8>,
         validity: Option<Bitmap>,
     ) -> Self {
         check_offsets(&offsets, values.len());
 
+        if data_type.to_physical_type() != Self::default_data_type().to_physical_type() {
+            panic!("Utf8Array can only be initialized with DataType::Utf8 or DataType::LargeUtf8")
+        }
+
         Self {
-            data_type: if O::is_large() {
-                DataType::LargeUtf8
-            } else {
-                DataType::Utf8
-            },
+            data_type,
             offsets,
             values,
             validity,
diff --git a/src/array/utf8/mutable.rs b/src/array/utf8/mutable.rs
index 61aa037aac..80e1e292ae 100644
--- a/src/array/utf8/mutable.rs
+++ b/src/array/utf8/mutable.rs
@@ -17,6 +17,7 @@ use super::Utf8Array;
 /// The mutable version of [`Utf8Array`]. See [`MutableArray`] for more details.
 #[derive(Debug)]
 pub struct MutableUtf8Array<O: Offset> {
+    data_type: DataType,
     offsets: MutableBuffer<O>,
     values: MutableBuffer<u8>,
     validity: Option<MutableBitmap>,
@@ -25,6 +26,7 @@ pub struct MutableUtf8Array<O: Offset> {
 impl<O: Offset> From<MutableUtf8Array<O>> for Utf8Array<O> {
     fn from(other: MutableUtf8Array<O>) -> Self {
         Utf8Array::<O>::from_data(
+            other.data_type,
             other.offsets.into(),
             other.values.into(),
             other.validity.map(|x| x.into()),
@@ -44,6 +46,7 @@ impl<O: Offset> MutableUtf8Array<O> {
         let mut offsets = MutableBuffer::<O>::new();
         offsets.push(O::default());
         Self {
+            data_type: Self::default_data_type(),
             offsets,
             values: MutableBuffer::<u8>::new(),
             validity: None,
@@ -57,6 +60,7 @@ impl<O: Offset> MutableUtf8Array<O> {
     /// * The `values` between `offsets` are not utf8 encoded
     /// * The validity is not `None` and its length is different from `offsets`'s length minus one.
     pub fn from_data(
+        data_type: DataType,
         offsets: MutableBuffer<O>,
         values: MutableBuffer<u8>,
         validity: Option<MutableBitmap>,
@@ -65,7 +69,11 @@ impl<O: Offset> MutableUtf8Array<O> {
         if let Some(ref validity) = validity {
             assert_eq!(offsets.len() - 1, validity.len());
         }
+        if data_type.to_physical_type() != Self::default_data_type().to_physical_type() {
+            panic!("MutableUtf8Array can only be initialized with DataType::Utf8 or DataType::LargeUtf8")
+        }
         Self {
+            data_type,
             offsets,
             values,
             validity,
@@ -80,6 +88,7 @@ impl<O: Offset> MutableUtf8Array<O> {
     /// * The `offsets` and `values` are inconsistent
     /// * The validity is not `None` and its length is different from `offsets`'s length minus one.
     pub unsafe fn from_data_unchecked(
+        data_type: DataType,
         offsets: MutableBuffer<O>,
         values: MutableBuffer<u8>,
         validity: Option<MutableBitmap>,
@@ -88,13 +97,21 @@ impl<O: Offset> MutableUtf8Array<O> {
         if let Some(ref validity) = validity {
             assert_eq!(offsets.len() - 1, validity.len());
         }
+        if data_type.to_physical_type() != Self::default_data_type().to_physical_type() {
+            panic!("MutableUtf8Array can only be initialized with DataType::Utf8 or DataType::LargeUtf8")
+        }
         Self {
+            data_type,
             offsets,
             values,
             validity,
         }
     }
 
+    fn default_data_type() -> DataType {
+        Utf8Array::<O>::default_data_type()
+    }
+
     /// Initializes a new [`MutableUtf8Array`] with a pre-allocated capacity of slots.
     pub fn with_capacity(capacity: usize) -> Self {
         Self::with_capacities(capacity, 0)
@@ -106,6 +123,7 @@ impl<O: Offset> MutableUtf8Array<O> {
         offsets.push(O::default());
 
         Self {
+            data_type: Self::default_data_type(),
             offsets,
             values: MutableBuffer::<u8>::with_capacity(values),
             validity: None,
@@ -171,6 +189,7 @@ impl<O: Offset> MutableArray for MutableUtf8Array<O> {
 
     fn as_arc(&mut self) -> Arc<dyn Array> {
         Arc::new(Utf8Array::from_data(
+            Self::default_data_type(),
             std::mem::take(&mut self.offsets).into(),
             std::mem::take(&mut self.values).into(),
             std::mem::take(&mut self.validity).map(|x| x.into()),
@@ -218,7 +237,7 @@ impl<O: Offset> MutableUtf8Array<O> {
         let (validity, offsets, values) = trusted_len_unzip(iterator);
 
         // soundness: P is `str`
-        Self::from_data_unchecked(offsets, values, validity)
+        Self::from_data_unchecked(Self::default_data_type(), offsets, values, validity)
     }
 
     /// Creates a [`MutableUtf8Array`] from an iterator of trusted length.
@@ -240,7 +259,7 @@ impl<O: Offset> MutableUtf8Array<O> {
         // soundness: I is `TrustedLen`
         let (offsets, values) = unsafe { trusted_len_values_iter(iterator) };
         // soundness: T is AsRef<str>
-        unsafe { Self::from_data_unchecked(offsets, values, None) }
+        unsafe { Self::from_data_unchecked(Self::default_data_type(), offsets, values, None) }
     }
 
     /// Creates a new [`MutableUtf8Array`] from an iterator.
@@ -275,7 +294,12 @@ impl<O: Offset> MutableUtf8Array<O> {
         let (validity, offsets, values) = try_trusted_len_unzip(iterator)?;
 
         // soundness: P is `str`
-        Ok(Self::from_data_unchecked(offsets, values, validity))
+        Ok(Self::from_data_unchecked(
+            Self::default_data_type(),
+            offsets,
+            values,
+            validity,
+        ))
     }
 
     /// Creates a [`MutableUtf8Array`] from an falible iterator of trusted length.
@@ -293,7 +317,7 @@ impl<O: Offset> MutableUtf8Array<O> {
     pub fn from_iter_values<T: AsRef<str>, I: Iterator<Item = T>>(iterator: I) -> Self {
         let (offsets, values) = values_iter(iterator);
         // soundness: T: AsRef<str>
-        unsafe { Self::from_data_unchecked(offsets, values, None) }
+        unsafe { Self::from_data_unchecked(Self::default_data_type(), offsets, values, None) }
     }
 }
 
diff --git a/src/compute/aggregate/memory.rs b/src/compute/aggregate/memory.rs
index 8ec120da5c..1d1cfbe3cf 100644
--- a/src/compute/aggregate/memory.rs
+++ b/src/compute/aggregate/memory.rs
@@ -1,23 +1,12 @@
 use crate::array::*;
 use crate::bitmap::Bitmap;
-use crate::datatypes::{DataType, IntervalUnit};
+use crate::datatypes::PhysicalType;
 use crate::types::days_ms;
 
 fn validity_size(validity: &Option<Bitmap>) -> usize {
     validity.as_ref().map(|b| b.as_slice().0.len()).unwrap_or(0)
 }
 
-macro_rules! dyn_primitive {
-    ($array:expr, $ty:ty) => {{
-        let array = $array
-            .as_any()
-            .downcast_ref::<PrimitiveArray<$ty>>()
-            .unwrap();
-
-        array.values().len() * std::mem::size_of::<$ty>() + validity_size(array.validity())
-    }};
-}
-
 macro_rules! dyn_binary {
     ($array:expr, $ty:ty, $o:ty) => {{
         let array = $array.as_any().downcast_ref::<$ty>().unwrap();
@@ -50,30 +39,23 @@ macro_rules! dyn_dict {
 ///
 /// FFI buffers are included in this estimation.
 pub fn estimated_bytes_size(array: &dyn Array) -> usize {
-    use DataType::*;
-    match array.data_type() {
+    use PhysicalType::*;
+    match array.data_type().to_physical_type() {
         Null => 0,
         Boolean => {
             let array = array.as_any().downcast_ref::<BooleanArray>().unwrap();
             array.values().as_slice().0.len() + validity_size(array.validity())
         }
-        Int8 => dyn_primitive!(array, i8),
-        Int16 => dyn_primitive!(array, i16),
-        Int32 | Date32 | Time32(_) | Interval(IntervalUnit::YearMonth) => {
-            dyn_primitive!(array, i32)
-        }
-        Int64 | Date64 | Timestamp(_, _) | Time64(_) | Duration(_) => dyn_primitive!(array, i64),
-        UInt8 => dyn_primitive!(array, u16),
-        UInt16 => dyn_primitive!(array, u16),
-        UInt32 => dyn_primitive!(array, u32),
-        UInt64 => dyn_primitive!(array, u64),
-        Float16 => unreachable!(),
-        Float32 => dyn_primitive!(array, f32),
-        Float64 => dyn_primitive!(array, f64),
-        Decimal(_, _) => dyn_primitive!(array, i128),
-        Interval(IntervalUnit::DayTime) => dyn_primitive!(array, days_ms),
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+            let array = array
+                .as_any()
+                .downcast_ref::<PrimitiveArray<$T>>()
+                .unwrap();
+
+            array.values().len() * std::mem::size_of::<$T>() + validity_size(array.validity())
+        }),
         Binary => dyn_binary!(array, BinaryArray<i32>, i32),
-        FixedSizeBinary(_) => {
+        FixedSizeBinary => {
             let array = array
                 .as_any()
                 .downcast_ref::<FixedSizeBinaryArray>()
@@ -83,23 +65,23 @@ pub fn estimated_bytes_size(array: &dyn Array) -> usize {
         LargeBinary => dyn_binary!(array, BinaryArray<i64>, i64),
         Utf8 => dyn_binary!(array, Utf8Array<i32>, i32),
         LargeUtf8 => dyn_binary!(array, Utf8Array<i64>, i64),
-        List(_) => {
+        List => {
             let array = array.as_any().downcast_ref::<ListArray<i32>>().unwrap();
             estimated_bytes_size(array.values().as_ref())
                 + array.offsets().len() * std::mem::size_of::<i32>()
                 + validity_size(array.validity())
         }
-        FixedSizeList(_, _) => {
+        FixedSizeList => {
             let array = array.as_any().downcast_ref::<ListArray<i64>>().unwrap();
             estimated_bytes_size(array.values().as_ref()) + validity_size(array.validity())
         }
-        LargeList(_) => {
+        LargeList => {
             let array = array.as_any().downcast_ref::<ListArray<i64>>().unwrap();
             estimated_bytes_size(array.values().as_ref())
                 + array.offsets().len() * std::mem::size_of::<i64>()
                 + validity_size(array.validity())
         }
-        Struct(_) => {
+        Struct => {
             let array = array.as_any().downcast_ref::<StructArray>().unwrap();
             array
                 .values()
@@ -109,7 +91,7 @@ pub fn estimated_bytes_size(array: &dyn Array) -> usize {
                 .sum::<usize>()
                 + validity_size(array.validity())
         }
-        Union(_, _, _) => {
+        Union => {
             let array = array.as_any().downcast_ref::<UnionArray>().unwrap();
             let types = array.types().len() * std::mem::size_of::<i8>();
             let offsets = array
@@ -125,17 +107,9 @@ pub fn estimated_bytes_size(array: &dyn Array) -> usize {
                 .sum::<usize>();
             types + offsets + fields
         }
-        Dictionary(keys, _) => match keys.as_ref() {
-            Int8 => dyn_dict!(array, i8),
-            Int16 => dyn_dict!(array, i16),
-            Int32 => dyn_dict!(array, i32),
-            Int64 => dyn_dict!(array, i64),
-            UInt8 => dyn_dict!(array, u8),
-            UInt16 => dyn_dict!(array, u16),
-            UInt32 => dyn_dict!(array, u32),
-            UInt64 => dyn_dict!(array, u64),
-            _ => unreachable!(),
-        },
+        Dictionary(key_type) => with_match_physical_dictionary_key_type!(key_type, |$T| {
+            dyn_dict!(array, $T)
+        }),
     }
 }
 
diff --git a/src/compute/aggregate/min_max.rs b/src/compute/aggregate/min_max.rs
index 368f2cc6eb..858a10e6ee 100644
--- a/src/compute/aggregate/min_max.rs
+++ b/src/compute/aggregate/min_max.rs
@@ -486,7 +486,7 @@ mod tests {
 
     #[test]
     fn test_boolean_min_max_empty() {
-        let a = BooleanArray::new_empty();
+        let a = BooleanArray::new_empty(DataType::Boolean);
         assert_eq!(None, min_boolean(&a));
         assert_eq!(None, max_boolean(&a));
     }
diff --git a/src/compute/boolean.rs b/src/compute/boolean.rs
index 9fe1d0aac6..03a55f48f2 100644
--- a/src/compute/boolean.rs
+++ b/src/compute/boolean.rs
@@ -17,6 +17,7 @@
 
 use crate::array::{Array, BooleanArray};
 use crate::bitmap::{Bitmap, MutableBitmap};
+use crate::datatypes::DataType;
 use crate::error::{ArrowError, Result};
 
 use super::utils::combine_validities;
@@ -39,7 +40,7 @@ where
 
     let values = op(left_buffer, right_buffer);
 
-    Ok(BooleanArray::from_data(values, validity))
+    Ok(BooleanArray::from_data(DataType::Boolean, values, validity))
 }
 
 /// Performs `AND` operation on two arrays. If either left or right value is null then the
@@ -99,7 +100,7 @@ pub fn or(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
 pub fn not(array: &BooleanArray) -> BooleanArray {
     let values = !array.values();
     let validity = array.validity().clone();
-    BooleanArray::from_data(values, validity)
+    BooleanArray::from_data(DataType::Boolean, values, validity)
 }
 
 /// Returns a non-null [BooleanArray] with whether each value of the array is null.
@@ -123,7 +124,7 @@ pub fn is_null(input: &dyn Array) -> BooleanArray {
         Some(buffer) => !buffer,
     };
 
-    BooleanArray::from_data(values, None)
+    BooleanArray::from_data(DataType::Boolean, values, None)
 }
 
 /// Returns a non-null [BooleanArray] with whether each value of the array is not null.
@@ -142,7 +143,7 @@ pub fn is_not_null(input: &dyn Array) -> BooleanArray {
         None => Bitmap::from_trusted_len_iter(std::iter::repeat(true).take(input.len())),
         Some(buffer) => buffer.clone(),
     };
-    BooleanArray::from_data(values, None)
+    BooleanArray::from_data(DataType::Boolean, values, None)
 }
 
 #[cfg(test)]
diff --git a/src/compute/boolean_kleene.rs b/src/compute/boolean_kleene.rs
index 1cdd2b41c7..4e31f449a5 100644
--- a/src/compute/boolean_kleene.rs
+++ b/src/compute/boolean_kleene.rs
@@ -1,3 +1,4 @@
+use crate::datatypes::DataType;
 use crate::error::{ArrowError, Result};
 use crate::{
     array::{Array, BooleanArray},
@@ -87,7 +88,11 @@ pub fn or(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
         }
         (None, None) => None,
     };
-    Ok(BooleanArray::from_data(lhs_values | rhs_values, validity))
+    Ok(BooleanArray::from_data(
+        DataType::Boolean,
+        lhs_values | rhs_values,
+        validity,
+    ))
 }
 
 /// Logical 'and' with [Kleene logic](https://en.wikipedia.org/wiki/Three-valued_logic#Kleene_and_Priest_logics)
@@ -172,7 +177,11 @@ pub fn and(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
         }
         (None, None) => None,
     };
-    Ok(BooleanArray::from_data(lhs_values & rhs_values, validity))
+    Ok(BooleanArray::from_data(
+        DataType::Boolean,
+        lhs_values & rhs_values,
+        validity,
+    ))
 }
 
 #[cfg(test)]
diff --git a/src/compute/cast/binary_to.rs b/src/compute/cast/binary_to.rs
index 77492ad60c..79edc2af5d 100644
--- a/src/compute/cast/binary_to.rs
+++ b/src/compute/cast/binary_to.rs
@@ -1,16 +1,20 @@
 use std::convert::TryFrom;
 
+use crate::datatypes::DataType;
 use crate::error::{ArrowError, Result};
 use crate::{array::*, buffer::Buffer};
 
-pub fn binary_to_large_binary(from: &BinaryArray<i32>) -> BinaryArray<i64> {
+pub fn binary_to_large_binary(from: &BinaryArray<i32>, to_data_type: DataType) -> BinaryArray<i64> {
     let values = from.values().clone();
     let offsets = from.offsets().iter().map(|x| *x as i64);
     let offsets = Buffer::from_trusted_len_iter(offsets);
-    BinaryArray::<i64>::from_data(offsets, values, from.validity().clone())
+    BinaryArray::<i64>::from_data(to_data_type, offsets, values, from.validity().clone())
 }
 
-pub fn binary_large_to_binary(from: &BinaryArray<i64>) -> Result<BinaryArray<i32>> {
+pub fn binary_large_to_binary(
+    from: &BinaryArray<i64>,
+    to_data_type: DataType,
+) -> Result<BinaryArray<i32>> {
     let values = from.values().clone();
     let _ =
         i32::try_from(*from.offsets().last().unwrap()).map_err(ArrowError::from_external_error)?;
@@ -18,6 +22,7 @@ pub fn binary_large_to_binary(from: &BinaryArray<i64>) -> Result<BinaryArray<i32
     let offsets = from.offsets().iter().map(|x| *x as i32);
     let offsets = Buffer::from_trusted_len_iter(offsets);
     Ok(BinaryArray::<i32>::from_data(
+        to_data_type,
         offsets,
         values,
         from.validity().clone(),
diff --git a/src/compute/cast/mod.rs b/src/compute/cast/mod.rs
index 9dd5ef5f33..451002b7b0 100644
--- a/src/compute/cast/mod.rs
+++ b/src/compute/cast/mod.rs
@@ -426,16 +426,16 @@ fn cast_with_options(
             ))),
         },
         (_, Boolean) => match from_type {
-            UInt8 => primitive_to_boolean_dyn::<u8>(array),
-            UInt16 => primitive_to_boolean_dyn::<u16>(array),
-            UInt32 => primitive_to_boolean_dyn::<u32>(array),
-            UInt64 => primitive_to_boolean_dyn::<u64>(array),
-            Int8 => primitive_to_boolean_dyn::<i8>(array),
-            Int16 => primitive_to_boolean_dyn::<i16>(array),
-            Int32 => primitive_to_boolean_dyn::<i32>(array),
-            Int64 => primitive_to_boolean_dyn::<i64>(array),
-            Float32 => primitive_to_boolean_dyn::<f32>(array),
-            Float64 => primitive_to_boolean_dyn::<f64>(array),
+            UInt8 => primitive_to_boolean_dyn::<u8>(array, to_type.clone()),
+            UInt16 => primitive_to_boolean_dyn::<u16>(array, to_type.clone()),
+            UInt32 => primitive_to_boolean_dyn::<u32>(array, to_type.clone()),
+            UInt64 => primitive_to_boolean_dyn::<u64>(array, to_type.clone()),
+            Int8 => primitive_to_boolean_dyn::<i8>(array, to_type.clone()),
+            Int16 => primitive_to_boolean_dyn::<i16>(array, to_type.clone()),
+            Int32 => primitive_to_boolean_dyn::<i32>(array, to_type.clone()),
+            Int64 => primitive_to_boolean_dyn::<i64>(array, to_type.clone()),
+            Float32 => primitive_to_boolean_dyn::<f32>(array, to_type.clone()),
+            Float64 => primitive_to_boolean_dyn::<f64>(array, to_type.clone()),
             _ => Err(ArrowError::NotYetImplemented(format!(
                 "Casting from {:?} to {:?} not supported",
                 from_type, to_type,
@@ -562,9 +562,12 @@ fn cast_with_options(
 
         (Binary, LargeBinary) => Ok(Box::new(binary_to_large_binary(
             array.as_any().downcast_ref().unwrap(),
+            to_type.clone(),
         ))),
-        (LargeBinary, Binary) => binary_large_to_binary(array.as_any().downcast_ref().unwrap())
-            .map(|x| Box::new(x) as Box<dyn Array>),
+        (LargeBinary, Binary) => {
+            binary_large_to_binary(array.as_any().downcast_ref().unwrap(), to_type.clone())
+                .map(|x| Box::new(x) as Box<dyn Array>)
+        }
 
         // start numeric casts
         (UInt8, UInt16) => primitive_to_primitive_dyn::<u8, u16>(array, to_type, as_options),
diff --git a/src/compute/cast/primitive_to.rs b/src/compute/cast/primitive_to.rs
index dcab568330..ec8d02e850 100644
--- a/src/compute/cast/primitive_to.rs
+++ b/src/compute/cast/primitive_to.rs
@@ -14,19 +14,25 @@ use super::CastOptions;
 
 /// Returns a [`BooleanArray`] where every element is different from zero.
 /// Validity is preserved.
-pub fn primitive_to_boolean<T: NativeType>(from: &PrimitiveArray<T>) -> BooleanArray {
+pub fn primitive_to_boolean<T: NativeType>(
+    from: &PrimitiveArray<T>,
+    to_type: DataType,
+) -> BooleanArray {
     let iter = from.values().iter().map(|v| *v != T::default());
     let values = Bitmap::from_trusted_len_iter(iter);
 
-    BooleanArray::from_data(values, from.validity().clone())
+    BooleanArray::from_data(to_type, values, from.validity().clone())
 }
 
-pub(super) fn primitive_to_boolean_dyn<T>(from: &dyn Array) -> Result<Box<dyn Array>>
+pub(super) fn primitive_to_boolean_dyn<T>(
+    from: &dyn Array,
+    to_type: DataType,
+) -> Result<Box<dyn Array>>
 where
     T: NativeType,
 {
     let from = from.as_any().downcast_ref().unwrap();
-    Ok(Box::new(primitive_to_boolean::<T>(from)))
+    Ok(Box::new(primitive_to_boolean::<T>(from, to_type)))
 }
 
 /// Returns a [`Utf8Array`] where every element is the utf8 representation of the number.
diff --git a/src/compute/cast/utf8_to.rs b/src/compute/cast/utf8_to.rs
index 5a46dd4636..e267199a01 100644
--- a/src/compute/cast/utf8_to.rs
+++ b/src/compute/cast/utf8_to.rs
@@ -107,18 +107,24 @@ pub fn utf8_to_timestamp_ns<O: Offset>(from: &Utf8Array<O>) -> PrimitiveArray<i6
 }
 
 pub fn utf8_to_large_utf8(from: &Utf8Array<i32>) -> Utf8Array<i64> {
+    let data_type = Utf8Array::<i64>::default_data_type();
     let values = from.values().clone();
     let offsets = from.offsets().iter().map(|x| *x as i64);
     let offsets = Buffer::from_trusted_len_iter(offsets);
-    unsafe { Utf8Array::<i64>::from_data_unchecked(offsets, values, from.validity().clone()) }
+    unsafe {
+        Utf8Array::<i64>::from_data_unchecked(data_type, offsets, values, from.validity().clone())
+    }
 }
 
 pub fn utf8_large_to_utf8(from: &Utf8Array<i64>) -> Result<Utf8Array<i32>> {
+    let data_type = Utf8Array::<i32>::default_data_type();
     let values = from.values().clone();
     let _ =
         i32::try_from(*from.offsets().last().unwrap()).map_err(ArrowError::from_external_error)?;
 
     let offsets = from.offsets().iter().map(|x| *x as i32);
     let offsets = Buffer::from_trusted_len_iter(offsets);
-    Ok(unsafe { Utf8Array::<i32>::from_data_unchecked(offsets, values, from.validity().clone()) })
+    Ok(unsafe {
+        Utf8Array::<i32>::from_data_unchecked(data_type, offsets, values, from.validity().clone())
+    })
 }
diff --git a/src/compute/comparison/boolean.rs b/src/compute/comparison/boolean.rs
index 4d0395918e..647e303693 100644
--- a/src/compute/comparison/boolean.rs
+++ b/src/compute/comparison/boolean.rs
@@ -1,6 +1,7 @@
 use crate::array::*;
 use crate::bitmap::Bitmap;
 use crate::buffer::MutableBuffer;
+use crate::datatypes::DataType;
 use crate::scalar::{BooleanScalar, Scalar};
 use crate::{
     bitmap::MutableBitmap,
@@ -46,7 +47,11 @@ where
 
     let values = compare_values_op(lhs.values(), rhs.values(), op);
 
-    Ok(BooleanArray::from_data(values.into(), validity))
+    Ok(BooleanArray::from_data(
+        DataType::Boolean,
+        values.into(),
+        validity,
+    ))
 }
 
 /// Evaluate `op(left, right)` for [`BooleanArray`] and scalar using
@@ -67,7 +72,7 @@ where
         values.push(op(lhs_remainder, rhs))
     };
     let values = MutableBitmap::from_buffer(values, lhs.len()).into();
-    BooleanArray::from_data(values, lhs.validity().clone())
+    BooleanArray::from_data(DataType::Boolean, values, lhs.validity().clone())
 }
 
 /// Perform `lhs == rhs` operation on two arrays.
@@ -148,7 +153,7 @@ pub fn compare(lhs: &BooleanArray, rhs: &BooleanArray, op: Operator) -> Result<B
 
 pub fn compare_scalar(lhs: &BooleanArray, rhs: &BooleanScalar, op: Operator) -> BooleanArray {
     if !rhs.is_valid() {
-        return BooleanArray::new_null(lhs.len());
+        return BooleanArray::new_null(DataType::Boolean, lhs.len());
     }
     compare_scalar_non_null(lhs, rhs.value(), op)
 }
diff --git a/src/compute/comparison/primitive.rs b/src/compute/comparison/primitive.rs
index 65f752426f..561868edea 100644
--- a/src/compute/comparison/primitive.rs
+++ b/src/compute/comparison/primitive.rs
@@ -16,6 +16,7 @@
 // under the License.
 
 use crate::bitmap::Bitmap;
+use crate::datatypes::DataType;
 use crate::scalar::{PrimitiveScalar, Scalar};
 use crate::{array::*, types::NativeType};
 use crate::{
@@ -72,7 +73,11 @@ where
 
     let values = compare_values_op(lhs.values(), rhs.values(), op);
 
-    Ok(BooleanArray::from_data(values.into(), validity))
+    Ok(BooleanArray::from_data(
+        DataType::Boolean,
+        values.into(),
+        validity,
+    ))
 }
 
 /// Evaluate `op(left, right)` for [`PrimitiveArray`] and scalar using
@@ -100,7 +105,11 @@ where
         values.push(op(lhs, rhs))
     };
 
-    BooleanArray::from_data(Bitmap::from_u8_buffer(values, lhs.len()), validity)
+    BooleanArray::from_data(
+        DataType::Boolean,
+        Bitmap::from_u8_buffer(values, lhs.len()),
+        validity,
+    )
 }
 
 /// Perform `lhs == rhs` operation on two arrays.
@@ -225,7 +234,7 @@ pub fn compare_scalar<T: NativeType + Simd8>(
     op: Operator,
 ) -> BooleanArray {
     if !rhs.is_valid() {
-        return BooleanArray::new_null(lhs.len());
+        return BooleanArray::new_null(DataType::Boolean, lhs.len());
     }
     compare_scalar_non_null(lhs, rhs.value(), op)
 }
diff --git a/src/compute/comparison/utf8.rs b/src/compute/comparison/utf8.rs
index e36b89167e..17f365ec23 100644
--- a/src/compute/comparison/utf8.rs
+++ b/src/compute/comparison/utf8.rs
@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use crate::datatypes::DataType;
 use crate::error::{ArrowError, Result};
 use crate::scalar::{Scalar, Utf8Scalar};
 use crate::{array::*, bitmap::Bitmap};
@@ -42,7 +43,7 @@ where
         .map(|(lhs, rhs)| op(lhs, rhs));
     let values = Bitmap::from_trusted_len_iter(values);
 
-    Ok(BooleanArray::from_data(values, validity))
+    Ok(BooleanArray::from_data(DataType::Boolean, values, validity))
 }
 
 /// Evaluate `op(lhs, rhs)` for [`PrimitiveArray`] and scalar using
@@ -57,7 +58,7 @@ where
     let values = lhs.values_iter().map(|lhs| op(lhs, rhs));
     let values = Bitmap::from_trusted_len_iter(values);
 
-    BooleanArray::from_data(values, validity)
+    BooleanArray::from_data(DataType::Boolean, values, validity)
 }
 
 /// Perform `lhs == rhs` operation on [`StringArray`] / [`LargeStringArray`].
@@ -141,7 +142,7 @@ pub fn compare_scalar<O: Offset>(
     op: Operator,
 ) -> BooleanArray {
     if !rhs.is_valid() {
-        return BooleanArray::new_null(lhs.len());
+        return BooleanArray::new_null(DataType::Boolean, lhs.len());
     }
     compare_scalar_non_null(lhs, rhs.value(), op)
 }
diff --git a/src/compute/contains.rs b/src/compute/contains.rs
index d9452da03f..44cc7a8363 100644
--- a/src/compute/contains.rs
+++ b/src/compute/contains.rs
@@ -59,7 +59,7 @@ where
     });
     let values = Bitmap::from_trusted_len_iter(values);
 
-    Ok(BooleanArray::from_data(values, validity))
+    Ok(BooleanArray::from_data(DataType::Boolean, values, validity))
 }
 
 /// Checks if a [`GenericListArray`] contains a value in the [`Utf8Array`]
@@ -93,7 +93,7 @@ where
     });
     let values = Bitmap::from_trusted_len_iter(values);
 
-    Ok(BooleanArray::from_data(values, validity))
+    Ok(BooleanArray::from_data(DataType::Boolean, values, validity))
 }
 
 macro_rules! primitive {
diff --git a/src/compute/filter.rs b/src/compute/filter.rs
index f7132cc472..54102d0e1e 100644
--- a/src/compute/filter.rs
+++ b/src/compute/filter.rs
@@ -15,12 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use crate::array::growable::make_growable;
 use crate::array::growable::Growable;
 use crate::record_batch::RecordBatch;
-use crate::{
-    array::growable::make_growable,
-    datatypes::{DataType, IntervalUnit},
-};
 use crate::{array::*, bitmap::Bitmap, types::NativeType};
 use crate::{
     bitmap::{utils::SlicesIterator, MutableBitmap},
@@ -88,17 +85,6 @@ fn filter_growable<'a>(growable: &mut impl Growable<'a>, chunks: &[(usize, usize
         .for_each(|(start, len)| growable.extend(0, *start, *len));
 }
 
-macro_rules! dyn_build_filter {
-    ($ty:ty, $array:expr, $filter_count:expr, $chunks:expr) => {{
-        let array = $array.as_any().downcast_ref().unwrap();
-        let mut growable =
-            growable::GrowablePrimitive::<$ty>::new(vec![array], false, $filter_count);
-        filter_growable(&mut growable, &$chunks);
-        let array: PrimitiveArray<$ty> = growable.into();
-        Box::new(array)
-    }};
-}
-
 /// Returns a prepared function optimized to filter multiple arrays.
 /// Creating this function requires time, but using it is faster than [filter] when the
 /// same filter needs to be applied to multiple arrays (e.g. a multi-column `RecordBatch`).
@@ -109,60 +95,38 @@ pub fn build_filter(filter: &BooleanArray) -> Result<Filter> {
     let filter_count = iter.slots();
     let chunks = iter.collect::<Vec<_>>();
 
-    Ok(Box::new(move |array: &dyn Array| match array.data_type() {
-        DataType::UInt8 => {
-            dyn_build_filter!(u8, array, filter_count, chunks)
-        }
-        DataType::UInt16 => {
-            dyn_build_filter!(u16, array, filter_count, chunks)
-        }
-        DataType::UInt32 => {
-            dyn_build_filter!(u32, array, filter_count, chunks)
-        }
-        DataType::UInt64 => {
-            dyn_build_filter!(u64, array, filter_count, chunks)
-        }
-        DataType::Int8 => {
-            dyn_build_filter!(i8, array, filter_count, chunks)
-        }
-        DataType::Int16 => {
-            dyn_build_filter!(i16, array, filter_count, chunks)
-        }
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => {
-            dyn_build_filter!(i32, array, filter_count, chunks)
-        }
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => {
-            dyn_build_filter!(i64, array, filter_count, chunks)
-        }
-        DataType::Interval(IntervalUnit::DayTime) => {
-            dyn_build_filter!(days_ms, array, filter_count, chunks)
-        }
-        DataType::Float32 => {
-            dyn_build_filter!(f32, array, filter_count, chunks)
-        }
-        DataType::Float64 => {
-            dyn_build_filter!(f64, array, filter_count, chunks)
-        }
-        DataType::Utf8 => {
-            let array = array.as_any().downcast_ref::<Utf8Array<i32>>().unwrap();
-            let mut growable = growable::GrowableUtf8::<i32>::new(vec![array], false, filter_count);
-            filter_growable(&mut growable, &chunks);
-            let array: Utf8Array<i32> = growable.into();
-            Box::new(array)
-        }
-        _ => {
-            let mut mutable = make_growable(&[array], false, filter_count);
-            chunks
-                .iter()
-                .for_each(|(start, len)| mutable.extend(0, *start, *len));
-            mutable.as_box()
+    use crate::datatypes::PhysicalType::*;
+    Ok(Box::new(move |array: &dyn Array| {
+        match array.data_type().to_physical_type() {
+            Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+                let array = array.as_any().downcast_ref().unwrap();
+                let mut growable =
+                    growable::GrowablePrimitive::<$T>::new(vec![array], false, filter_count);
+                filter_growable(&mut growable, &chunks);
+                let array: PrimitiveArray<$T> = growable.into();
+                Box::new(array)
+            }),
+            Utf8 => {
+                let array = array.as_any().downcast_ref::<Utf8Array<i32>>().unwrap();
+                let mut growable = growable::GrowableUtf8::new(vec![array], false, filter_count);
+                filter_growable(&mut growable, &chunks);
+                let array: Utf8Array<i32> = growable.into();
+                Box::new(array)
+            }
+            LargeUtf8 => {
+                let array = array.as_any().downcast_ref::<Utf8Array<i64>>().unwrap();
+                let mut growable = growable::GrowableUtf8::new(vec![array], false, filter_count);
+                filter_growable(&mut growable, &chunks);
+                let array: Utf8Array<i64> = growable.into();
+                Box::new(array)
+            }
+            _ => {
+                let mut mutable = make_growable(&[array], false, filter_count);
+                chunks
+                    .iter()
+                    .for_each(|(start, len)| mutable.extend(0, *start, *len));
+                mutable.as_box()
+            }
         }
     }))
 }
@@ -185,58 +149,12 @@ pub fn build_filter(filter: &BooleanArray) -> Result<Filter> {
 /// # }
 /// ```
 pub fn filter(array: &dyn Array, filter: &BooleanArray) -> Result<Box<dyn Array>> {
-    match array.data_type() {
-        DataType::UInt8 => {
-            let array = array.as_any().downcast_ref().unwrap();
-            Ok(Box::new(filter_primitive::<u8>(array, filter)))
-        }
-        DataType::UInt16 => {
+    use crate::datatypes::PhysicalType::*;
+    match array.data_type().to_physical_type() {
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
             let array = array.as_any().downcast_ref().unwrap();
-            Ok(Box::new(filter_primitive::<u16>(array, filter)))
-        }
-        DataType::UInt32 => {
-            let array = array.as_any().downcast_ref().unwrap();
-            Ok(Box::new(filter_primitive::<u32>(array, filter)))
-        }
-        DataType::UInt64 => {
-            let array = array.as_any().downcast_ref().unwrap();
-            Ok(Box::new(filter_primitive::<u64>(array, filter)))
-        }
-        DataType::Int8 => {
-            let array = array.as_any().downcast_ref().unwrap();
-            Ok(Box::new(filter_primitive::<i8>(array, filter)))
-        }
-        DataType::Int16 => {
-            let array = array.as_any().downcast_ref().unwrap();
-            Ok(Box::new(filter_primitive::<i16>(array, filter)))
-        }
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => {
-            let array = array.as_any().downcast_ref().unwrap();
-            Ok(Box::new(filter_primitive::<i32>(array, filter)))
-        }
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => {
-            let array = array.as_any().downcast_ref().unwrap();
-            Ok(Box::new(filter_primitive::<i64>(array, filter)))
-        }
-        DataType::Interval(IntervalUnit::DayTime) => {
-            let array = array.as_any().downcast_ref().unwrap();
-            Ok(Box::new(filter_primitive::<days_ms>(array, filter)))
-        }
-        DataType::Float32 => {
-            let array = array.as_any().downcast_ref().unwrap();
-            Ok(Box::new(filter_primitive::<f32>(array, filter)))
-        }
-        DataType::Float64 => {
-            let array = array.as_any().downcast_ref().unwrap();
-            Ok(Box::new(filter_primitive::<f64>(array, filter)))
-        }
+            Ok(Box::new(filter_primitive::<$T>(array, filter)))
+        }),
         _ => {
             let iter = SlicesIterator::new(filter.values());
             let mut mutable = make_growable(&[array], false, iter.slots());
diff --git a/src/compute/like.rs b/src/compute/like.rs
index 4ca328ab3e..abda4f912e 100644
--- a/src/compute/like.rs
+++ b/src/compute/like.rs
@@ -2,6 +2,7 @@ use std::collections::HashMap;
 
 use regex::Regex;
 
+use crate::datatypes::DataType;
 use crate::{array::*, bitmap::Bitmap};
 use crate::{
     compute::utils::combine_validities,
@@ -52,7 +53,7 @@ fn a_like_utf8<O: Offset, F: Fn(bool) -> bool>(
             }
         }))?;
 
-    Ok(BooleanArray::from_data(values, validity))
+    Ok(BooleanArray::from_data(DataType::Boolean, values, validity))
 }
 
 /// Returns `lhs LIKE rhs` operation on two [`Utf8Array`].
@@ -112,7 +113,11 @@ fn a_like_utf8_scalar<O: Offset, F: Fn(bool) -> bool>(
         })?;
         Bitmap::from_trusted_len_iter(lhs.values_iter().map(|x| op(re.is_match(x))))
     };
-    Ok(BooleanArray::from_data(values, validity.clone()))
+    Ok(BooleanArray::from_data(
+        DataType::Boolean,
+        values,
+        validity.clone(),
+    ))
 }
 
 /// Returns `lhs LIKE rhs` operation.
diff --git a/src/compute/regex_match.rs b/src/compute/regex_match.rs
index 3e8d8220b6..1277906736 100644
--- a/src/compute/regex_match.rs
+++ b/src/compute/regex_match.rs
@@ -21,6 +21,7 @@ use regex::Regex;
 
 use super::utils::{combine_validities, unary_utf8_boolean};
 use crate::array::{BooleanArray, Offset, Utf8Array};
+use crate::datatypes::DataType;
 use crate::error::{ArrowError, Result};
 use crate::{array::*, bitmap::Bitmap};
 
@@ -59,7 +60,11 @@ pub fn regex_match<O: Offset>(values: &Utf8Array<O>, regex: &Utf8Array<O>) -> Re
     });
     let new_values = Bitmap::try_from_trusted_len_iter(iterator)?;
 
-    Ok(BooleanArray::from_data(new_values, validity))
+    Ok(BooleanArray::from_data(
+        DataType::Boolean,
+        new_values,
+        validity,
+    ))
 }
 
 /// Regex matches
diff --git a/src/compute/substring.rs b/src/compute/substring.rs
index 2e4fd209b6..c82029b747 100644
--- a/src/compute/substring.rs
+++ b/src/compute/substring.rs
@@ -60,7 +60,12 @@ fn utf8_substring<O: Offset>(array: &Utf8Array<O>, start: O, length: &Option<O>)
         new_values.extend_from_slice(&values[start..start + length]);
     });
 
-    Utf8Array::<O>::from_data(new_offsets.into(), new_values.into(), validity.clone())
+    Utf8Array::<O>::from_data(
+        array.data_type().clone(),
+        new_offsets.into(),
+        new_values.into(),
+        validity.clone(),
+    )
 }
 
 /// Returns an ArrayRef with a substring starting from `start` and with optional length `length` of each of the elements in `array`.
diff --git a/src/compute/take/binary.rs b/src/compute/take/binary.rs
index 3208899467..4bfb46abed 100644
--- a/src/compute/take/binary.rs
+++ b/src/compute/take/binary.rs
@@ -25,6 +25,7 @@ pub fn take<O: Offset, I: Index>(
     values: &BinaryArray<O>,
     indices: &PrimitiveArray<I>,
 ) -> BinaryArray<O> {
+    let data_type = values.data_type().clone();
     let indices_has_validity = indices.null_count() > 0;
     let values_has_validity = values.null_count() > 0;
 
@@ -36,5 +37,5 @@ pub fn take<O: Offset, I: Index>(
         (false, true) => take_indices_validity(values.offsets(), values.values(), indices),
         (true, true) => take_values_indices_validity(values, indices),
     };
-    BinaryArray::<O>::from_data(offsets, values, validity)
+    BinaryArray::<O>::from_data(data_type, offsets, values, validity)
 }
diff --git a/src/compute/take/boolean.rs b/src/compute/take/boolean.rs
index dc1df7820d..2051b14223 100644
--- a/src/compute/take/boolean.rs
+++ b/src/compute/take/boolean.rs
@@ -108,6 +108,7 @@ fn take_values_indices_validity<I: Index>(
 
 /// `take` implementation for boolean arrays
 pub fn take<I: Index>(values: &BooleanArray, indices: &PrimitiveArray<I>) -> BooleanArray {
+    let data_type = values.data_type().clone();
     let indices_has_validity = indices.null_count() > 0;
     let values_has_validity = values.null_count() > 0;
 
@@ -118,7 +119,7 @@ pub fn take<I: Index>(values: &BooleanArray, indices: &PrimitiveArray<I>) -> Boo
         (true, true) => take_values_indices_validity(values, indices),
     };
 
-    BooleanArray::from_data(values, validity)
+    BooleanArray::from_data(data_type, values, validity)
 }
 
 #[cfg(test)]
diff --git a/src/compute/take/mod.rs b/src/compute/take/mod.rs
index 18685b0e38..751ebfc9f3 100644
--- a/src/compute/take/mod.rs
+++ b/src/compute/take/mod.rs
@@ -19,7 +19,7 @@
 
 use crate::{
     array::{new_empty_array, Array, NullArray, PrimitiveArray},
-    datatypes::{DataType, IntervalUnit},
+    datatypes::DataType,
     error::Result,
     types::{days_ms, Index},
 };
@@ -33,87 +33,56 @@ mod primitive;
 mod structure;
 mod utf8;
 
-macro_rules! downcast_take {
-    ($type: ty, $values: expr, $indices: expr) => {{
-        let values = $values
-            .as_any()
-            .downcast_ref()
-            .expect("Unable to downcast to a primitive array");
-        Ok(Box::new(primitive::take::<$type, _>(&values, $indices)))
-    }};
-}
-
-macro_rules! downcast_dict_take {
-    ($type: ty, $values: expr, $indices: expr) => {{
-        let values = $values
-            .as_any()
-            .downcast_ref()
-            .expect("Unable to downcast to a primitive array");
-        Ok(Box::new(dict::take::<$type, _>(&values, $indices)))
-    }};
-}
-
 pub fn take<O: Index>(values: &dyn Array, indices: &PrimitiveArray<O>) -> Result<Box<dyn Array>> {
     if indices.len() == 0 {
         return Ok(new_empty_array(values.data_type().clone()));
     }
 
-    match values.data_type() {
-        DataType::Null => Ok(Box::new(NullArray::from_data(indices.len()))),
-        DataType::Boolean => {
+    use crate::datatypes::PhysicalType::*;
+    match values.data_type().to_physical_type() {
+        Null => Ok(Box::new(NullArray::from_data(
+            values.data_type().clone(),
+            indices.len(),
+        ))),
+        Boolean => {
             let values = values.as_any().downcast_ref().unwrap();
             Ok(Box::new(boolean::take::<O>(values, indices)))
         }
-        DataType::Int8 => downcast_take!(i8, values, indices),
-        DataType::Int16 => downcast_take!(i16, values, indices),
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => downcast_take!(i32, values, indices),
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Duration(_)
-        | DataType::Timestamp(_, _) => downcast_take!(i64, values, indices),
-        DataType::Interval(IntervalUnit::DayTime) => downcast_take!(days_ms, values, indices),
-        DataType::UInt8 => downcast_take!(u8, values, indices),
-        DataType::UInt16 => downcast_take!(u16, values, indices),
-        DataType::UInt32 => downcast_take!(u32, values, indices),
-        DataType::UInt64 => downcast_take!(u64, values, indices),
-        DataType::Float16 => unreachable!(),
-        DataType::Float32 => downcast_take!(f32, values, indices),
-        DataType::Float64 => downcast_take!(f64, values, indices),
-        DataType::Decimal(_, _) => downcast_take!(i128, values, indices),
-        DataType::Utf8 => {
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+            let values = values.as_any().downcast_ref().unwrap();
+            Ok(Box::new(primitive::take::<$T, _>(&values, indices)))
+        }),
+        Utf8 => {
             let values = values.as_any().downcast_ref().unwrap();
             Ok(Box::new(utf8::take::<i32, _>(values, indices)))
         }
-        DataType::LargeUtf8 => {
+        LargeUtf8 => {
             let values = values.as_any().downcast_ref().unwrap();
             Ok(Box::new(utf8::take::<i64, _>(values, indices)))
         }
-        DataType::Binary => {
+        Binary => {
             let values = values.as_any().downcast_ref().unwrap();
             Ok(Box::new(binary::take::<i32, _>(values, indices)))
         }
-        DataType::LargeBinary => {
+        LargeBinary => {
             let values = values.as_any().downcast_ref().unwrap();
             Ok(Box::new(binary::take::<i64, _>(values, indices)))
         }
-        DataType::Dictionary(key_type, _) => {
-            with_match_dictionary_key_type!(key_type.as_ref(), |$T| {
-                downcast_dict_take!($T, values, indices)
+        Dictionary(key_type) => {
+            with_match_physical_dictionary_key_type!(key_type, |$T| {
+                let values = values.as_any().downcast_ref().unwrap();
+                Ok(Box::new(dict::take::<$T, _>(&values, indices)))
             })
         }
-        DataType::Struct(_) => {
+        Struct => {
             let array = values.as_any().downcast_ref().unwrap();
             Ok(Box::new(structure::take::<_>(array, indices)?))
         }
-        DataType::List(_) => {
+        List => {
             let array = values.as_any().downcast_ref().unwrap();
             Ok(Box::new(list::take::<i32, O>(array, indices)))
         }
-        DataType::LargeList(_) => {
+        LargeList => {
             let array = values.as_any().downcast_ref().unwrap();
             Ok(Box::new(list::take::<i64, O>(array, indices)))
         }
@@ -180,7 +149,7 @@ pub fn can_take(data_type: &DataType) -> bool {
 mod tests {
     use std::sync::Arc;
 
-    use crate::datatypes::Field;
+    use crate::datatypes::{Field, IntervalUnit};
     use crate::{array::*, bitmap::MutableBitmap, types::NativeType};
 
     use super::*;
@@ -253,7 +222,7 @@ mod tests {
             Field::new("b", DataType::Int32, true),
         ];
         StructArray::from_data(
-            fields,
+            DataType::Struct(fields),
             vec![
                 Arc::new(boolean) as Arc<dyn Array>,
                 Arc::new(int) as Arc<dyn Array>,
@@ -277,7 +246,7 @@ mod tests {
             .collect::<MutableBitmap>()
             .into();
         let expected = StructArray::from_data(
-            array.fields().to_vec(),
+            array.data_type().clone(),
             vec![
                 Arc::new(boolean) as Arc<dyn Array>,
                 Arc::new(int) as Arc<dyn Array>,
diff --git a/src/compute/take/structure.rs b/src/compute/take/structure.rs
index b724b28759..47aacbae1f 100644
--- a/src/compute/take/structure.rs
+++ b/src/compute/take/structure.rs
@@ -61,7 +61,7 @@ pub fn take<I: Index>(array: &StructArray, indices: &PrimitiveArray<I>) -> Resul
         .collect::<Result<_>>()?;
     let validity = take_validity(array.validity(), indices)?;
     Ok(StructArray::from_data(
-        array.fields().to_vec(),
+        array.data_type().clone(),
         values,
         validity,
     ))
diff --git a/src/compute/take/utf8.rs b/src/compute/take/utf8.rs
index 9d95b86623..99fc091f49 100644
--- a/src/compute/take/utf8.rs
+++ b/src/compute/take/utf8.rs
@@ -25,6 +25,7 @@ pub fn take<O: Offset, I: Index>(
     values: &Utf8Array<O>,
     indices: &PrimitiveArray<I>,
 ) -> Utf8Array<O> {
+    let data_type = values.data_type().clone();
     let indices_has_validity = indices.null_count() > 0;
     let values_has_validity = values.null_count() > 0;
 
@@ -36,7 +37,7 @@ pub fn take<O: Offset, I: Index>(
         (false, true) => take_indices_validity(values.offsets(), values.values(), indices),
         (true, true) => take_values_indices_validity(values, indices),
     };
-    unsafe { Utf8Array::<O>::from_data_unchecked(offsets, values, validity) }
+    unsafe { Utf8Array::<O>::from_data_unchecked(data_type, offsets, values, validity) }
 }
 
 #[cfg(test)]
diff --git a/src/compute/utils.rs b/src/compute/utils.rs
index 5683620b45..6378d9fca3 100644
--- a/src/compute/utils.rs
+++ b/src/compute/utils.rs
@@ -18,6 +18,7 @@
 use crate::{
     array::{Array, BooleanArray, Offset, Utf8Array},
     bitmap::Bitmap,
+    datatypes::DataType,
 };
 
 pub fn combine_validities(lhs: &Option<Bitmap>, rhs: &Option<Bitmap>) -> Option<Bitmap> {
@@ -42,5 +43,5 @@ pub fn unary_utf8_boolean<O: Offset, F: Fn(&str) -> bool>(
         op(value.unwrap())
     });
     let values = Bitmap::from_trusted_len_iter(iterator);
-    BooleanArray::from_data(values, validity)
+    BooleanArray::from_data(DataType::Boolean, values, validity)
 }
diff --git a/src/datatypes/field.rs b/src/datatypes/field.rs
index 3927a878f2..d5a66a5250 100644
--- a/src/datatypes/field.rs
+++ b/src/datatypes/field.rs
@@ -23,7 +23,7 @@ use super::DataType;
 
 /// A logical [`DataType`] and its associated metadata per
 /// [Arrow specification](https://arrow.apache.org/docs/cpp/api/datatype.html)
-#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct Field {
     /// Its name
     pub name: String,
@@ -255,6 +255,7 @@ impl Field {
             | DataType::FixedSizeBinary(_)
             | DataType::Utf8
             | DataType::LargeUtf8
+            | DataType::Extension(_, _, _)
             | DataType::Decimal(_, _) => {
                 if self.data_type != from.data_type {
                     return Err(ArrowError::Schema(
diff --git a/src/datatypes/mod.rs b/src/datatypes/mod.rs
index 7aa4a886ed..9b406e4a20 100644
--- a/src/datatypes/mod.rs
+++ b/src/datatypes/mod.rs
@@ -1,8 +1,10 @@
 //! Metadata declarations such as [`DataType`], [`Field`] and [`Schema`].
 mod field;
+mod physical_type;
 mod schema;
 
 pub use field::Field;
+pub use physical_type::*;
 pub use schema::Schema;
 
 /// The set of datatypes that are supported by this implementation of Apache Arrow.
@@ -20,7 +22,7 @@ pub use schema::Schema;
 /// Nested types can themselves be nested within other arrays.
 /// For more information on these types please see
 /// [the physical memory layout of Apache Arrow](https://arrow.apache.org/docs/format/Columnar.html#physical-memory-layout).
-#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub enum DataType {
     /// Null type, representing an array without values or validity, only a length.
     Null,
@@ -114,6 +116,8 @@ pub enum DataType {
     /// scale is the number of decimal places.
     /// The number 999.99 has a precision of 5 and scale of 2.
     Decimal(usize, usize),
+    /// Extension type.
+    Extension(String, Box<DataType>, Option<String>),
 }
 
 impl std::fmt::Display for DataType {
@@ -169,6 +173,58 @@ impl DataType {
             _ => self == other,
         }
     }
+
+    /// the [`PhysicalType`] of this [`DataType`].
+    pub fn to_physical_type(&self) -> PhysicalType {
+        use DataType::*;
+        match self {
+            Null => PhysicalType::Null,
+            Boolean => PhysicalType::Boolean,
+            Int8 => PhysicalType::Primitive(PrimitiveType::Int8),
+            Int16 => PhysicalType::Primitive(PrimitiveType::Int16),
+            Int32 | Date32 | Time32(_) | Interval(IntervalUnit::YearMonth) => {
+                PhysicalType::Primitive(PrimitiveType::Int32)
+            }
+            Int64 | Date64 | Timestamp(_, _) | Time64(_) | Duration(_) => {
+                PhysicalType::Primitive(PrimitiveType::Int64)
+            }
+            Decimal(_, _) => PhysicalType::Primitive(PrimitiveType::Int128),
+            UInt8 => PhysicalType::Primitive(PrimitiveType::UInt8),
+            UInt16 => PhysicalType::Primitive(PrimitiveType::UInt16),
+            UInt32 => PhysicalType::Primitive(PrimitiveType::UInt32),
+            UInt64 => PhysicalType::Primitive(PrimitiveType::UInt64),
+            Float16 => unreachable!(),
+            Float32 => PhysicalType::Primitive(PrimitiveType::Float32),
+            Float64 => PhysicalType::Primitive(PrimitiveType::Float64),
+            Interval(IntervalUnit::DayTime) => PhysicalType::Primitive(PrimitiveType::DaysMs),
+            Binary => PhysicalType::Binary,
+            FixedSizeBinary(_) => PhysicalType::FixedSizeBinary,
+            LargeBinary => PhysicalType::LargeBinary,
+            Utf8 => PhysicalType::Utf8,
+            LargeUtf8 => PhysicalType::LargeUtf8,
+            List(_) => PhysicalType::List,
+            FixedSizeList(_, _) => PhysicalType::FixedSizeList,
+            LargeList(_) => PhysicalType::LargeList,
+            Struct(_) => PhysicalType::Struct,
+            Union(_, _, _) => PhysicalType::Union,
+            Dictionary(key, _) => PhysicalType::Dictionary(to_dictionary_index_type(key.as_ref())),
+            Extension(_, key, _) => key.to_physical_type(),
+        }
+    }
+}
+
+fn to_dictionary_index_type(data_type: &DataType) -> DictionaryIndexType {
+    match data_type {
+        DataType::Int8 => DictionaryIndexType::Int8,
+        DataType::Int16 => DictionaryIndexType::Int16,
+        DataType::Int32 => DictionaryIndexType::Int32,
+        DataType::Int64 => DictionaryIndexType::Int64,
+        DataType::UInt8 => DictionaryIndexType::UInt8,
+        DataType::UInt16 => DictionaryIndexType::UInt16,
+        DataType::UInt32 => DictionaryIndexType::UInt32,
+        DataType::UInt64 => DictionaryIndexType::UInt64,
+        _ => ::core::unreachable!("A dictionary key type can only be of integer types"),
+    }
 }
 
 // backward compatibility
diff --git a/src/datatypes/physical_type.rs b/src/datatypes/physical_type.rs
new file mode 100644
index 0000000000..aff17e906d
--- /dev/null
+++ b/src/datatypes/physical_type.rs
@@ -0,0 +1,83 @@
+/// the set of valid indices used to index a dictionary-encoded Array.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum DictionaryIndexType {
+    /// A signed 8-bit integer.
+    Int8,
+    /// A signed 16-bit integer.
+    Int16,
+    /// A signed 32-bit integer.
+    Int32,
+    /// A signed 64-bit integer.
+    Int64,
+    /// An unsigned 8-bit integer.
+    UInt8,
+    /// An unsigned 16-bit integer.
+    UInt16,
+    /// An unsigned 32-bit integer.
+    UInt32,
+    /// An unsigned 64-bit integer.
+    UInt64,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum PrimitiveType {
+    /// A signed 8-bit integer.
+    Int8,
+    /// A signed 16-bit integer.
+    Int16,
+    /// A signed 32-bit integer.
+    Int32,
+    /// A signed 64-bit integer.
+    Int64,
+    /// A signed 128-bit integer.
+    Int128,
+    /// An unsigned 8-bit integer.
+    UInt8,
+    /// An unsigned 16-bit integer.
+    UInt16,
+    /// An unsigned 32-bit integer.
+    UInt32,
+    /// An unsigned 64-bit integer.
+    UInt64,
+    /// A 32-bit floating point number.
+    Float32,
+    /// A 64-bit floating point number.
+    Float64,
+    /// Two i32 representing days and ms
+    DaysMs,
+}
+
+/// The set of physical types: unique in-memory representations of an Arrow array.
+/// A physical type has a one-to-many relationship with a [`crate::datatypes::DataType`] and
+/// a one-to-one mapping with each struct in this crate that implements [`crate::array::Array`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum PhysicalType {
+    /// A Null with no allocation.
+    Null,
+    /// A boolean represented as a single bit.
+    Boolean,
+    /// An array where each slot has a known compile-time size.
+    Primitive(PrimitiveType),
+    /// Opaque binary data of variable length.
+    Binary,
+    /// Opaque binary data of fixed size.
+    FixedSizeBinary,
+    /// Opaque binary data of variable length and 64-bit offsets.
+    LargeBinary,
+    /// A variable-length string in Unicode with UTF-8 encoding.
+    Utf8,
+    /// A variable-length string in Unicode with UFT-8 encoding and 64-bit offsets.
+    LargeUtf8,
+    /// A list of some data type with variable length.
+    List,
+    /// A list of some data type with fixed length.
+    FixedSizeList,
+    /// A list of some data type with variable length and 64-bit offsets.
+    LargeList,
+    /// A nested type that contains an arbitrary number of fields.
+    Struct,
+    /// A nested type that represents slots of differing types.
+    Union,
+    /// A dictionary encoded array by `DictionaryIndexType`.
+    Dictionary(DictionaryIndexType),
+}
diff --git a/src/ffi/array.rs b/src/ffi/array.rs
index 611160f525..60af0df0fe 100644
--- a/src/ffi/array.rs
+++ b/src/ffi/array.rs
@@ -21,10 +21,7 @@ use super::ffi::ArrowArrayRef;
 use crate::array::{BooleanArray, FromFfi};
 use crate::error::{ArrowError, Result};
 use crate::types::days_ms;
-use crate::{
-    array::*,
-    datatypes::{DataType, IntervalUnit},
-};
+use crate::{array::*, datatypes::PhysicalType};
 
 /// Reads a valid `ffi` interface into a `Box<dyn Array>`
 /// # Errors
@@ -32,52 +29,30 @@ 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 array: Box<dyn Array> = match array.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)?),
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => {
-            Box::new(PrimitiveArray::<i32>::try_from_ffi(array)?)
-        }
-        DataType::Interval(IntervalUnit::DayTime) => {
-            Box::new(PrimitiveArray::<days_ms>::try_from_ffi(array)?)
-        }
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => Box::new(PrimitiveArray::<i64>::try_from_ffi(array)?),
-        DataType::Decimal(_, _) => Box::new(PrimitiveArray::<i128>::try_from_ffi(array)?),
-        DataType::UInt8 => Box::new(PrimitiveArray::<u8>::try_from_ffi(array)?),
-        DataType::UInt16 => Box::new(PrimitiveArray::<u16>::try_from_ffi(array)?),
-        DataType::UInt32 => Box::new(PrimitiveArray::<u32>::try_from_ffi(array)?),
-        DataType::UInt64 => Box::new(PrimitiveArray::<u64>::try_from_ffi(array)?),
-        DataType::Float16 => unreachable!(),
-        DataType::Float32 => Box::new(PrimitiveArray::<f32>::try_from_ffi(array)?),
-        DataType::Float64 => Box::new(PrimitiveArray::<f64>::try_from_ffi(array)?),
-        DataType::Utf8 => Box::new(Utf8Array::<i32>::try_from_ffi(array)?),
-        DataType::LargeUtf8 => Box::new(Utf8Array::<i64>::try_from_ffi(array)?),
-        DataType::Binary => Box::new(BinaryArray::<i32>::try_from_ffi(array)?),
-        DataType::LargeBinary => Box::new(BinaryArray::<i64>::try_from_ffi(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(key_type, _) => {
-            with_match_dictionary_key_type!(key_type.as_ref(), |$T| {
+    use PhysicalType::*;
+    Ok(match array.field().data_type().to_physical_type() {
+        Boolean => Box::new(BooleanArray::try_from_ffi(array)?),
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+            Box::new(PrimitiveArray::<$T>::try_from_ffi(array)?)
+        }),
+        Utf8 => Box::new(Utf8Array::<i32>::try_from_ffi(array)?),
+        LargeUtf8 => Box::new(Utf8Array::<i64>::try_from_ffi(array)?),
+        Binary => Box::new(BinaryArray::<i32>::try_from_ffi(array)?),
+        LargeBinary => Box::new(BinaryArray::<i64>::try_from_ffi(array)?),
+        List => Box::new(ListArray::<i32>::try_from_ffi(array)?),
+        LargeList => Box::new(ListArray::<i64>::try_from_ffi(array)?),
+        Struct => Box::new(StructArray::try_from_ffi(array)?),
+        Dictionary(key_type) => {
+            with_match_physical_dictionary_key_type!(key_type, |$T| {
                 Box::new(DictionaryArray::<$T>::try_from_ffi(array)?)
             })
         }
-        DataType::Union(_, _, _) => Box::new(UnionArray::try_from_ffi(array)?),
+        Union => Box::new(UnionArray::try_from_ffi(array)?),
         data_type => {
             return Err(ArrowError::NotYetImplemented(format!(
-                "Reading DataType \"{}\" is not yet supported.",
+                "Importing PhysicalType \"{:?}\" is not yet supported.",
                 data_type
             )))
         }
-    };
-
-    Ok(array)
+    })
 }
diff --git a/src/ffi/schema.rs b/src/ffi/schema.rs
index 75a0665000..083e006bee 100644
--- a/src/ffi/schema.rs
+++ b/src/ffi/schema.rs
@@ -356,6 +356,7 @@ fn to_format(data_type: &DataType) -> String {
             r
         }
         DataType::Dictionary(index, _) => to_format(index.as_ref()),
+        DataType::Extension(_, inner, _) => to_format(inner.as_ref()),
     }
 }
 
diff --git a/src/io/ipc/convert.rs b/src/io/ipc/convert.rs
index fc8082fa7b..4f21818996 100644
--- a/src/io/ipc/convert.rs
+++ b/src/io/ipc/convert.rs
@@ -32,6 +32,9 @@ use std::collections::{BTreeMap, HashMap};
 
 use DataType::*;
 
+type Metadata = Option<BTreeMap<String, String>>;
+type Extension = Option<(String, Option<String>)>;
+
 pub fn schema_to_fb_offset<'a>(
     fbb: &mut FlatBufferBuilder<'a>,
     schema: &Schema,
@@ -67,36 +70,54 @@ pub fn schema_to_fb_offset<'a>(
     builder.finish()
 }
 
+fn read_metadata(field: &ipc::Field) -> Metadata {
+    if let Some(list) = field.custom_metadata() {
+        let mut metadata_map = BTreeMap::default();
+        for kv in list {
+            if let (Some(k), Some(v)) = (kv.key(), kv.value()) {
+                metadata_map.insert(k.to_string(), v.to_string());
+            }
+        }
+        Some(metadata_map)
+    } else {
+        None
+    }
+}
+
+pub(crate) fn get_extension(metadata: &Metadata) -> Extension {
+    if let Some(metadata) = metadata {
+        if let Some(name) = metadata.get("ARROW:extension:name") {
+            let metadata = metadata.get("ARROW:extension:metadata").cloned();
+            Some((name.clone(), metadata))
+        } else {
+            None
+        }
+    } else {
+        None
+    }
+}
+
 /// Convert an IPC Field to Arrow Field
 impl<'a> From<ipc::Field<'a>> for Field {
     fn from(field: ipc::Field) -> Field {
+        let metadata = read_metadata(&field);
+
+        let extension = get_extension(&metadata);
+
+        let data_type = get_data_type(field, extension, true);
+
         let mut arrow_field = if let Some(dictionary) = field.dictionary() {
             Field::new_dict(
                 field.name().unwrap(),
-                get_data_type(field, true),
+                data_type,
                 field.nullable(),
                 dictionary.id(),
                 dictionary.isOrdered(),
             )
         } else {
-            Field::new(
-                field.name().unwrap(),
-                get_data_type(field, true),
-                field.nullable(),
-            )
+            Field::new(field.name().unwrap(), data_type, field.nullable())
         };
 
-        let mut metadata = None;
-        if let Some(list) = field.custom_metadata() {
-            let mut metadata_map = BTreeMap::default();
-            for kv in list {
-                if let (Some(k), Some(v)) = (kv.key(), kv.value()) {
-                    metadata_map.insert(k.to_string(), v.to_string());
-                }
-            }
-            metadata = Some(metadata_map);
-        }
-
         arrow_field.set_metadata(metadata);
         arrow_field
     }
@@ -132,7 +153,7 @@ pub fn fb_to_schema(fb: ipc::Schema) -> (Schema, bool) {
 }
 
 /// Get the Arrow data type from the flatbuffer Field table
-pub(crate) fn get_data_type(field: ipc::Field, may_be_dictionary: bool) -> DataType {
+fn get_data_type(field: ipc::Field, extension: Extension, may_be_dictionary: bool) -> DataType {
     if let Some(dictionary) = field.dictionary() {
         if may_be_dictionary {
             let int = dictionary.indexType().unwrap();
@@ -149,11 +170,17 @@ pub(crate) fn get_data_type(field: ipc::Field, may_be_dictionary: bool) -> DataT
             };
             return DataType::Dictionary(
                 Box::new(index_type),
-                Box::new(get_data_type(field, false)),
+                Box::new(get_data_type(field, extension, false)),
             );
         }
     }
 
+    if let Some(extension) = extension {
+        let (name, metadata) = extension;
+        let data_type = get_data_type(field, None, false);
+        return DataType::Extension(name, Box::new(data_type), metadata);
+    }
+
     match field.type_type() {
         ipc::Type::Null => DataType::Null,
         ipc::Type::Bool => DataType::Boolean,
@@ -303,27 +330,58 @@ pub(crate) struct FbFieldType<'b> {
     pub(crate) children: Option<WIPOffset<Vector<'b, ForwardsUOffset<ipc::Field<'b>>>>>,
 }
 
+fn write_metadata<'a>(
+    fbb: &mut FlatBufferBuilder<'a>,
+    metadata: &BTreeMap<String, String>,
+    kv_vec: &mut Vec<WIPOffset<ipc::KeyValue<'a>>>,
+) {
+    for (k, v) in metadata {
+        if k != "ARROW:extension:name" && k != "ARROW:extension:metadata" {
+            let kv_args = ipc::KeyValueArgs {
+                key: Some(fbb.create_string(k.as_str())),
+                value: Some(fbb.create_string(v.as_str())),
+            };
+            kv_vec.push(ipc::KeyValue::create(fbb, &kv_args));
+        }
+    }
+}
+
 /// Create an IPC Field from an Arrow Field
 pub(crate) fn build_field<'a>(
     fbb: &mut FlatBufferBuilder<'a>,
     field: &Field,
 ) -> WIPOffset<ipc::Field<'a>> {
-    // Optional custom metadata.
-    let mut fb_metadata = None;
+    // custom metadata.
+    let mut kv_vec = vec![];
+    if let DataType::Extension(name, _, metadata) = field.data_type() {
+        // append extension information.
+
+        // metadata
+        if let Some(metadata) = metadata {
+            let kv_args = ipc::KeyValueArgs {
+                key: Some(fbb.create_string("ARROW:extension:metadata")),
+                value: Some(fbb.create_string(metadata.as_str())),
+            };
+            kv_vec.push(ipc::KeyValue::create(fbb, &kv_args));
+        }
+
+        // name
+        let kv_args = ipc::KeyValueArgs {
+            key: Some(fbb.create_string("ARROW:extension:name")),
+            value: Some(fbb.create_string(name.as_str())),
+        };
+        kv_vec.push(ipc::KeyValue::create(fbb, &kv_args));
+    }
     if let Some(metadata) = field.metadata() {
         if !metadata.is_empty() {
-            let mut kv_vec = vec![];
-            for (k, v) in metadata {
-                let kv_args = ipc::KeyValueArgs {
-                    key: Some(fbb.create_string(k.as_str())),
-                    value: Some(fbb.create_string(v.as_str())),
-                };
-                let kv_offset = ipc::KeyValue::create(fbb, &kv_args);
-                kv_vec.push(kv_offset);
-            }
-            fb_metadata = Some(fbb.create_vector(&kv_vec));
+            write_metadata(fbb, metadata, &mut kv_vec);
         }
     };
+    let fb_metadata = if !kv_vec.is_empty() {
+        Some(fbb.create_vector(&kv_vec))
+    } else {
+        None
+    };
 
     let fb_field_name = fbb.create_string(field.name().as_str());
     let field_type = get_fb_field_type(field.data_type(), field.is_nullable(), fbb);
@@ -386,6 +444,7 @@ fn type_to_field_type(data_type: &DataType) -> ipc::Type {
         Union(_, _, _) => ipc::Type::Union,
         Struct(_) => ipc::Type::Struct_,
         Dictionary(_, v) => type_to_field_type(v),
+        Extension(_, v, _) => type_to_field_type(v),
     }
 }
 
@@ -625,6 +684,7 @@ pub(crate) fn get_fb_field_type<'a>(
             // type in the DictionaryEncoding metadata in the parent field
             get_fb_field_type(value_type, is_nullable, fbb)
         }
+        Extension(_, value_type, _) => get_fb_field_type(value_type, is_nullable, fbb),
         Decimal(precision, scale) => {
             let mut builder = ipc::DecimalBuilder::new(fbb);
             builder.add_precision(*precision as i32);
diff --git a/src/io/ipc/mod.rs b/src/io/ipc/mod.rs
index f6b347534c..49e22fc18e 100644
--- a/src/io/ipc/mod.rs
+++ b/src/io/ipc/mod.rs
@@ -13,6 +13,7 @@ mod compression;
 mod convert;
 
 pub use convert::fb_to_schema;
+pub(crate) use convert::get_extension;
 pub use gen::Message::root_as_message;
 pub mod read;
 pub mod write;
diff --git a/src/io/ipc/read/array/binary.rs b/src/io/ipc/read/array/binary.rs
index 0e1ed3cd19..a232a16708 100644
--- a/src/io/ipc/read/array/binary.rs
+++ b/src/io/ipc/read/array/binary.rs
@@ -4,6 +4,7 @@ use std::io::{Read, Seek};
 
 use crate::array::{BinaryArray, Offset};
 use crate::buffer::Buffer;
+use crate::datatypes::DataType;
 use crate::error::Result;
 use crate::io::ipc::gen::Message::BodyCompression;
 use crate::types::NativeType;
@@ -14,6 +15,7 @@ use super::super::read_basic::*;
 
 pub fn read_binary<O: Offset, R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
+    data_type: DataType,
     buffers: &mut VecDeque<&gen::Schema::Buffer>,
     reader: &mut R,
     block_offset: u64,
@@ -55,7 +57,9 @@ where
         compression,
     )?;
 
-    Ok(BinaryArray::<O>::from_data(offsets, values, validity))
+    Ok(BinaryArray::<O>::from_data(
+        data_type, offsets, values, validity,
+    ))
 }
 
 pub fn skip_binary(field_nodes: &mut VecDeque<Node>, buffers: &mut VecDeque<&gen::Schema::Buffer>) {
diff --git a/src/io/ipc/read/array/boolean.rs b/src/io/ipc/read/array/boolean.rs
index 866b11e902..b7bd443369 100644
--- a/src/io/ipc/read/array/boolean.rs
+++ b/src/io/ipc/read/array/boolean.rs
@@ -2,6 +2,7 @@ use std::collections::VecDeque;
 use std::io::{Read, Seek};
 
 use crate::array::BooleanArray;
+use crate::datatypes::DataType;
 use crate::error::Result;
 
 use super::super::super::gen;
@@ -10,6 +11,7 @@ use super::super::read_basic::*;
 
 pub fn read_boolean<R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
+    data_type: DataType,
     buffers: &mut VecDeque<&gen::Schema::Buffer>,
     reader: &mut R,
     block_offset: u64,
@@ -35,7 +37,7 @@ pub fn read_boolean<R: Read + Seek>(
         is_little_endian,
         None,
     )?;
-    Ok(BooleanArray::from_data(values, validity))
+    Ok(BooleanArray::from_data(data_type, values, validity))
 }
 
 pub fn skip_boolean(
diff --git a/src/io/ipc/read/array/fixed_size_list.rs b/src/io/ipc/read/array/fixed_size_list.rs
index 8fb9b45cbc..8aca4b7786 100644
--- a/src/io/ipc/read/array/fixed_size_list.rs
+++ b/src/io/ipc/read/array/fixed_size_list.rs
@@ -12,6 +12,7 @@ use super::super::super::gen;
 use super::super::deserialize::{read, skip, Node};
 use super::super::read_basic::*;
 
+#[allow(clippy::too_many_arguments)]
 pub fn read_fixed_size_list<R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
     data_type: DataType,
@@ -37,7 +38,7 @@ pub fn read_fixed_size_list<R: Read + Seek>(
 
     let values = read(
         field_nodes,
-        value_data_type.clone(),
+        value_data_type.data_type().clone(),
         buffers,
         reader,
         block_offset,
@@ -57,7 +58,7 @@ pub fn skip_fixed_size_list(
 
     let _ = buffers.pop_front().unwrap();
 
-    let (data_type, _) = FixedSizeListArray::get_child_and_size(data_type);
+    let (field, _) = FixedSizeListArray::get_child_and_size(data_type);
 
-    skip(field_nodes, data_type, buffers)
+    skip(field_nodes, field.data_type(), buffers)
 }
diff --git a/src/io/ipc/read/array/list.rs b/src/io/ipc/read/array/list.rs
index 61ff8b8612..0f082750b4 100644
--- a/src/io/ipc/read/array/list.rs
+++ b/src/io/ipc/read/array/list.rs
@@ -14,6 +14,7 @@ use super::super::super::gen;
 use super::super::deserialize::{read, skip, Node};
 use super::super::read_basic::*;
 
+#[allow(clippy::too_many_arguments)]
 pub fn read_list<O: Offset, R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
     data_type: DataType,
diff --git a/src/io/ipc/read/array/null.rs b/src/io/ipc/read/array/null.rs
index 7d937cc649..d1cdcb0e49 100644
--- a/src/io/ipc/read/array/null.rs
+++ b/src/io/ipc/read/array/null.rs
@@ -1,11 +1,14 @@
 use std::collections::VecDeque;
 
-use crate::array::NullArray;
+use crate::{array::NullArray, datatypes::DataType};
 
 use super::super::deserialize::Node;
 
-pub fn read_null(field_nodes: &mut VecDeque<Node>) -> NullArray {
-    NullArray::from_data(field_nodes.pop_front().unwrap().0.length() as usize)
+pub fn read_null(field_nodes: &mut VecDeque<Node>, data_type: DataType) -> NullArray {
+    NullArray::from_data(
+        data_type,
+        field_nodes.pop_front().unwrap().0.length() as usize,
+    )
 }
 
 pub fn skip_null(field_nodes: &mut VecDeque<Node>) {
diff --git a/src/io/ipc/read/array/struct_.rs b/src/io/ipc/read/array/struct_.rs
index c259849c37..fd7c3652d8 100644
--- a/src/io/ipc/read/array/struct_.rs
+++ b/src/io/ipc/read/array/struct_.rs
@@ -12,6 +12,7 @@ use super::super::super::gen;
 use super::super::deserialize::{read, skip, Node};
 use super::super::read_basic::*;
 
+#[allow(clippy::too_many_arguments)]
 pub fn read_struct<R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
     data_type: DataType,
@@ -51,7 +52,7 @@ pub fn read_struct<R: Read + Seek>(
         })
         .collect::<Result<Vec<_>>>()?;
 
-    Ok(StructArray::from_data(fields.to_vec(), values, validity))
+    Ok(StructArray::from_data(data_type, values, validity))
 }
 
 pub fn skip_struct(
diff --git a/src/io/ipc/read/array/union.rs b/src/io/ipc/read/array/union.rs
index adaac0f13c..32b0d938fe 100644
--- a/src/io/ipc/read/array/union.rs
+++ b/src/io/ipc/read/array/union.rs
@@ -12,6 +12,7 @@ use super::super::super::gen;
 use super::super::deserialize::{read, skip, Node};
 use super::super::read_basic::*;
 
+#[allow(clippy::too_many_arguments)]
 pub fn read_union<R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
     data_type: DataType,
diff --git a/src/io/ipc/read/array/utf8.rs b/src/io/ipc/read/array/utf8.rs
index 17ffed8da8..788c4c16cd 100644
--- a/src/io/ipc/read/array/utf8.rs
+++ b/src/io/ipc/read/array/utf8.rs
@@ -4,6 +4,7 @@ use std::io::{Read, Seek};
 
 use crate::array::{Offset, Utf8Array};
 use crate::buffer::Buffer;
+use crate::datatypes::DataType;
 use crate::error::Result;
 use crate::io::ipc::gen::Message::BodyCompression;
 use crate::types::NativeType;
@@ -14,6 +15,7 @@ use super::super::read_basic::*;
 
 pub fn read_utf8<O: Offset, R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
+    data_type: DataType,
     buffers: &mut VecDeque<&gen::Schema::Buffer>,
     reader: &mut R,
     block_offset: u64,
@@ -55,7 +57,9 @@ where
         compression,
     )?;
 
-    Ok(Utf8Array::<O>::from_data(offsets, values, validity))
+    Ok(Utf8Array::<O>::from_data(
+        data_type, offsets, values, validity,
+    ))
 }
 
 pub fn skip_utf8(field_nodes: &mut VecDeque<Node>, buffers: &mut VecDeque<&gen::Schema::Buffer>) {
diff --git a/src/io/ipc/read/common.rs b/src/io/ipc/read/common.rs
index e0524c558c..92975c8bd2 100644
--- a/src/io/ipc/read/common.rs
+++ b/src/io/ipc/read/common.rs
@@ -92,6 +92,7 @@ impl<'a, A, I: Iterator<Item = A>> Iterator for ProjectionIter<'a, A, I> {
 /// Creates a record batch from binary data using the `ipc::RecordBatch` indexes and the `Schema`
 /// # Panic
 /// Panics iff the projection is not in increasing order (e.g. `[1, 0]` nor `[0, 1, 1]` are valid)
+#[allow(clippy::too_many_arguments)]
 pub fn read_record_batch<R: Read + Seek>(
     batch: gen::Message::RecordBatch,
     schema: Arc<Schema>,
diff --git a/src/io/ipc/read/deserialize.rs b/src/io/ipc/read/deserialize.rs
index 9866aaf268..663ec06428 100644
--- a/src/io/ipc/read/deserialize.rs
+++ b/src/io/ipc/read/deserialize.rs
@@ -11,7 +11,7 @@ use std::{
 
 use gen::Schema::MetadataVersion;
 
-use crate::datatypes::{DataType, IntervalUnit};
+use crate::datatypes::{DataType, PhysicalType};
 use crate::error::Result;
 use crate::io::ipc::gen::Message::BodyCompression;
 use crate::{array::*, types::days_ms};
@@ -21,6 +21,7 @@ use super::array::*;
 
 pub type Node<'a> = (&'a gen::Message::FieldNode, &'a Option<Arc<dyn Array>>);
 
+#[allow(clippy::too_many_arguments)]
 pub fn read<R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
     data_type: DataType,
@@ -31,146 +32,37 @@ pub fn read<R: Read + Seek>(
     compression: Option<BodyCompression>,
     version: MetadataVersion,
 ) -> Result<Arc<dyn Array>> {
-    match data_type {
-        DataType::Null => {
-            let array = read_null(field_nodes);
+    use PhysicalType::*;
+    match data_type.to_physical_type() {
+        Null => {
+            let array = read_null(field_nodes, data_type);
             Ok(Arc::new(array))
         }
-        DataType::Boolean => {
-            read_boolean(field_nodes, buffers, reader, block_offset, is_little_endian)
-                .map(|x| Arc::new(x) as Arc<dyn Array>)
-        }
-        DataType::Int8 => read_primitive::<i8, _>(
-            field_nodes,
-            data_type,
-            buffers,
-            reader,
-            block_offset,
-            is_little_endian,
-            compression,
-        )
-        .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::Int16 => read_primitive::<i16, _>(
-            field_nodes,
-            data_type,
-            buffers,
-            reader,
-            block_offset,
-            is_little_endian,
-            compression,
-        )
-        .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => read_primitive::<i32, _>(
-            field_nodes,
-            data_type,
-            buffers,
-            reader,
-            block_offset,
-            is_little_endian,
-            compression,
-        )
-        .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => read_primitive::<i64, _>(
+        Boolean => read_boolean(
             field_nodes,
             data_type,
             buffers,
             reader,
             block_offset,
             is_little_endian,
-            compression,
-        )
-        .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::Decimal(_, _) => read_primitive::<i128, _>(
-            field_nodes,
-            data_type,
-            buffers,
-            reader,
-            block_offset,
-            is_little_endian,
-            compression,
-        )
-        .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::Interval(IntervalUnit::DayTime) => read_primitive::<days_ms, _>(
-            field_nodes,
-            data_type,
-            buffers,
-            reader,
-            block_offset,
-            is_little_endian,
-            compression,
-        )
-        .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::UInt8 => read_primitive::<u8, _>(
-            field_nodes,
-            data_type,
-            buffers,
-            reader,
-            block_offset,
-            is_little_endian,
-            compression,
         )
         .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::UInt16 => read_primitive::<u16, _>(
-            field_nodes,
-            data_type,
-            buffers,
-            reader,
-            block_offset,
-            is_little_endian,
-            compression,
-        )
-        .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::UInt32 => read_primitive::<u32, _>(
-            field_nodes,
-            data_type,
-            buffers,
-            reader,
-            block_offset,
-            is_little_endian,
-            compression,
-        )
-        .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::UInt64 => read_primitive::<u64, _>(
-            field_nodes,
-            data_type,
-            buffers,
-            reader,
-            block_offset,
-            is_little_endian,
-            compression,
-        )
-        .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::Float16 => unreachable!(),
-        DataType::Float32 => read_primitive::<f32, _>(
-            field_nodes,
-            data_type,
-            buffers,
-            reader,
-            block_offset,
-            is_little_endian,
-            compression,
-        )
-        .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::Float64 => read_primitive::<f64, _>(
-            field_nodes,
-            data_type,
-            buffers,
-            reader,
-            block_offset,
-            is_little_endian,
-            compression,
-        )
-        .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::Binary => {
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+            read_primitive::<$T, _>(
+                field_nodes,
+                data_type,
+                buffers,
+                reader,
+                block_offset,
+                is_little_endian,
+                compression,
+            )
+            .map(|x| Arc::new(x) as Arc<dyn Array>)
+        }),
+        Binary => {
             let array = read_binary::<i32, _>(
                 field_nodes,
+                data_type,
                 buffers,
                 reader,
                 block_offset,
@@ -179,9 +71,10 @@ pub fn read<R: Read + Seek>(
             )?;
             Ok(Arc::new(array))
         }
-        DataType::LargeBinary => {
+        LargeBinary => {
             let array = read_binary::<i64, _>(
                 field_nodes,
+                data_type,
                 buffers,
                 reader,
                 block_offset,
@@ -190,7 +83,7 @@ pub fn read<R: Read + Seek>(
             )?;
             Ok(Arc::new(array))
         }
-        DataType::FixedSizeBinary(_) => {
+        FixedSizeBinary => {
             let array = read_fixed_size_binary(
                 field_nodes,
                 data_type,
@@ -202,9 +95,10 @@ pub fn read<R: Read + Seek>(
             )?;
             Ok(Arc::new(array))
         }
-        DataType::Utf8 => {
+        Utf8 => {
             let array = read_utf8::<i32, _>(
                 field_nodes,
+                data_type,
                 buffers,
                 reader,
                 block_offset,
@@ -213,9 +107,10 @@ pub fn read<R: Read + Seek>(
             )?;
             Ok(Arc::new(array))
         }
-        DataType::LargeUtf8 => {
+        LargeUtf8 => {
             let array = read_utf8::<i64, _>(
                 field_nodes,
+                data_type,
                 buffers,
                 reader,
                 block_offset,
@@ -224,7 +119,7 @@ pub fn read<R: Read + Seek>(
             )?;
             Ok(Arc::new(array))
         }
-        DataType::List(_) => read_list::<i32, _>(
+        List => read_list::<i32, _>(
             field_nodes,
             data_type,
             buffers,
@@ -235,7 +130,7 @@ pub fn read<R: Read + Seek>(
             version,
         )
         .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::LargeList(_) => read_list::<i64, _>(
+        LargeList => read_list::<i64, _>(
             field_nodes,
             data_type,
             buffers,
@@ -246,7 +141,7 @@ pub fn read<R: Read + Seek>(
             version,
         )
         .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::FixedSizeList(_, _) => read_fixed_size_list(
+        FixedSizeList => read_fixed_size_list(
             field_nodes,
             data_type,
             buffers,
@@ -257,7 +152,7 @@ pub fn read<R: Read + Seek>(
             version,
         )
         .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::Struct(_) => read_struct(
+        Struct => read_struct(
             field_nodes,
             data_type,
             buffers,
@@ -268,8 +163,8 @@ pub fn read<R: Read + Seek>(
             version,
         )
         .map(|x| Arc::new(x) as Arc<dyn Array>),
-        DataType::Dictionary(ref key_type, _) => {
-            with_match_dictionary_key_type!(key_type.as_ref(), |$T| {
+        Dictionary(key_type) => {
+            with_match_physical_dictionary_key_type!(key_type, |$T| {
                 read_dictionary::<$T, _>(
                     field_nodes,
                     buffers,
@@ -280,7 +175,7 @@ pub fn read<R: Read + Seek>(
                 .map(|x| Arc::new(x) as Arc<dyn Array>)
             })
         }
-        DataType::Union(_, _, _) => read_union(
+        Union => read_union(
             field_nodes,
             data_type,
             buffers,
@@ -299,36 +194,19 @@ pub fn skip(
     data_type: &DataType,
     buffers: &mut VecDeque<&gen::Schema::Buffer>,
 ) {
-    match data_type {
-        DataType::Null => skip_null(field_nodes),
-        DataType::Boolean => skip_boolean(field_nodes, buffers),
-        DataType::Int8
-        | DataType::Int16
-        | DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(_)
-        | DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_)
-        | DataType::Decimal(_, _)
-        | DataType::UInt8
-        | DataType::UInt16
-        | DataType::UInt32
-        | DataType::UInt64
-        | DataType::Float32
-        | DataType::Float16
-        | DataType::Float64 => skip_primitive(field_nodes, buffers),
-        DataType::LargeBinary | DataType::Binary => skip_binary(field_nodes, buffers),
-        DataType::LargeUtf8 | DataType::Utf8 => skip_utf8(field_nodes, buffers),
-        DataType::FixedSizeBinary(_) => skip_fixed_size_binary(field_nodes, buffers),
-        DataType::List(_) => skip_list::<i32>(field_nodes, data_type, buffers),
-        DataType::LargeList(_) => skip_list::<i64>(field_nodes, data_type, buffers),
-        DataType::FixedSizeList(_, _) => skip_fixed_size_list(field_nodes, data_type, buffers),
-        DataType::Struct(_) => skip_struct(field_nodes, data_type, buffers),
-        DataType::Dictionary(_, _) => skip_dictionary(field_nodes, buffers),
-        DataType::Union(_, _, _) => skip_union(field_nodes, data_type, buffers),
+    use PhysicalType::*;
+    match data_type.to_physical_type() {
+        Null => skip_null(field_nodes),
+        Boolean => skip_boolean(field_nodes, buffers),
+        Primitive(_) => skip_primitive(field_nodes, buffers),
+        LargeBinary | Binary => skip_binary(field_nodes, buffers),
+        LargeUtf8 | Utf8 => skip_utf8(field_nodes, buffers),
+        FixedSizeBinary => skip_fixed_size_binary(field_nodes, buffers),
+        List => skip_list::<i32>(field_nodes, data_type, buffers),
+        LargeList => skip_list::<i64>(field_nodes, data_type, buffers),
+        FixedSizeList => skip_fixed_size_list(field_nodes, data_type, buffers),
+        Struct => skip_struct(field_nodes, data_type, buffers),
+        Dictionary(_) => skip_dictionary(field_nodes, buffers),
+        Union => skip_union(field_nodes, data_type, buffers),
     }
 }
diff --git a/src/io/ipc/write/serialize.rs b/src/io/ipc/write/serialize.rs
index b238e1b88d..220f7b28fc 100644
--- a/src/io/ipc/write/serialize.rs
+++ b/src/io/ipc/write/serialize.rs
@@ -18,7 +18,7 @@
 use crate::{
     array::*,
     bitmap::Bitmap,
-    datatypes::{DataType, IntervalUnit},
+    datatypes::{DataType, PhysicalType},
     endianess::is_native_little_endian,
     io::ipc::gen::Message,
     trusted_len::TrustedLen,
@@ -346,79 +346,27 @@ pub fn write(
         array.len() as i64,
         array.null_count() as i64,
     ));
-    match array.data_type() {
-        DataType::Null => (),
-        DataType::Boolean => write_boolean(array, buffers, arrow_data, offset, is_little_endian),
-        DataType::Int8 => {
-            write_primitive::<i8>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::Int16 => {
-            write_primitive::<i16>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => {
-            write_primitive::<i32>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => {
-            write_primitive::<i64>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::Decimal(_, _) => {
-            write_primitive::<i128>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::Interval(IntervalUnit::DayTime) => {
-            write_primitive::<days_ms>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::UInt8 => {
-            write_primitive::<u8>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::UInt16 => {
-            write_primitive::<u16>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::UInt32 => {
-            write_primitive::<u32>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::UInt64 => {
-            write_primitive::<u64>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::Float16 => unreachable!(),
-        DataType::Float32 => {
-            write_primitive::<f32>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::Float64 => {
-            write_primitive::<f64>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::Binary => {
-            write_binary::<i32>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::LargeBinary => {
-            write_binary::<i64>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::FixedSizeBinary(_) => {
+    use PhysicalType::*;
+    match array.data_type().to_physical_type() {
+        Null => (),
+        Boolean => write_boolean(array, buffers, arrow_data, offset, is_little_endian),
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+            write_primitive::<$T>(array, buffers, arrow_data, offset, is_little_endian)
+        }),
+        Binary => write_binary::<i32>(array, buffers, arrow_data, offset, is_little_endian),
+        LargeBinary => write_binary::<i64>(array, buffers, arrow_data, offset, is_little_endian),
+        FixedSizeBinary => {
             write_fixed_size_binary(array, buffers, arrow_data, offset, is_little_endian)
         }
-        DataType::Utf8 => write_utf8::<i32>(array, buffers, arrow_data, offset, is_little_endian),
-        DataType::LargeUtf8 => {
-            write_utf8::<i64>(array, buffers, arrow_data, offset, is_little_endian)
-        }
-        DataType::List(_) => {
-            write_list::<i32>(array, buffers, arrow_data, nodes, offset, is_little_endian)
-        }
-        DataType::LargeList(_) => {
-            write_list::<i64>(array, buffers, arrow_data, nodes, offset, is_little_endian)
-        }
-        DataType::FixedSizeList(_, _) => {
+        Utf8 => write_utf8::<i32>(array, buffers, arrow_data, offset, is_little_endian),
+        LargeUtf8 => write_utf8::<i64>(array, buffers, arrow_data, offset, is_little_endian),
+        List => write_list::<i32>(array, buffers, arrow_data, nodes, offset, is_little_endian),
+        LargeList => write_list::<i64>(array, buffers, arrow_data, nodes, offset, is_little_endian),
+        FixedSizeList => {
             write_fixed_size_list(array, buffers, arrow_data, nodes, offset, is_little_endian)
         }
-        DataType::Struct(_) => {
-            write_struct(array, buffers, arrow_data, nodes, offset, is_little_endian)
-        }
-        DataType::Dictionary(_, _) => {
+        Struct => write_struct(array, buffers, arrow_data, nodes, offset, is_little_endian),
+        Dictionary(_) => {
             write_dictionary(
                 array,
                 buffers,
@@ -429,7 +377,7 @@ pub fn write(
                 true,
             );
         }
-        DataType::Union(_, _, _) => {
+        Union => {
             write_union(array, buffers, arrow_data, nodes, offset, is_little_endian);
         }
     }
diff --git a/src/io/json/read/deserialize.rs b/src/io/json/read/deserialize.rs
index fb9e56ef1f..8ac22d6dca 100644
--- a/src/io/json/read/deserialize.rs
+++ b/src/io/json/read/deserialize.rs
@@ -188,7 +188,7 @@ fn read_struct(rows: &[&Value], data_type: DataType) -> StructArray {
         .map(|(_, (data_type, values))| read(&values, data_type.clone()))
         .collect::<Vec<_>>();
 
-    StructArray::from_data(fields.to_vec(), values, None)
+    StructArray::from_data(data_type, values, None)
 }
 
 fn read_dictionary<K: DictionaryKey>(rows: &[&Value], data_type: DataType) -> DictionaryArray<K> {
@@ -224,7 +224,7 @@ fn read_dictionary<K: DictionaryKey>(rows: &[&Value], data_type: DataType) -> Di
 
 pub fn read(rows: &[&Value], data_type: DataType) -> Arc<dyn Array> {
     match &data_type {
-        DataType::Null => Arc::new(NullArray::from_data(rows.len())),
+        DataType::Null => Arc::new(NullArray::from_data(data_type, rows.len())),
         DataType::Boolean => Arc::new(read_boolean(rows)),
         DataType::Int8 => Arc::new(read_int::<i8>(rows, data_type)),
         DataType::Int16 => Arc::new(read_int::<i16>(rows, data_type)),
diff --git a/src/io/json_integration/read.rs b/src/io/json_integration/read.rs
index 188aa6f412..cba5271985 100644
--- a/src/io/json_integration/read.rs
+++ b/src/io/json_integration/read.rs
@@ -24,7 +24,7 @@ use crate::{
     array::*,
     bitmap::{Bitmap, MutableBitmap},
     buffer::Buffer,
-    datatypes::{DataType, Field, IntervalUnit, Schema},
+    datatypes::{DataType, PhysicalType, PrimitiveType, Schema},
     error::{ArrowError, Result},
     record_batch::RecordBatch,
     types::{days_ms, NativeType},
@@ -144,7 +144,7 @@ fn to_primitive<T: NativeType + NumCast>(
     PrimitiveArray::<T>::from_data(data_type, values, validity)
 }
 
-fn to_binary<O: Offset>(json_col: &ArrowJsonColumn) -> Arc<dyn Array> {
+fn to_binary<O: Offset>(json_col: &ArrowJsonColumn, data_type: DataType) -> Arc<dyn Array> {
     let validity = to_validity(&json_col.validity);
     let offsets = to_offsets::<O>(json_col.offset.as_ref());
     let values = json_col
@@ -155,10 +155,10 @@ fn to_binary<O: Offset>(json_col: &ArrowJsonColumn) -> Arc<dyn Array> {
         .map(|value| value.as_str().map(|x| hex::decode(x).unwrap()).unwrap())
         .flatten()
         .collect();
-    Arc::new(BinaryArray::from_data(offsets, values, validity))
+    Arc::new(BinaryArray::from_data(data_type, offsets, values, validity))
 }
 
-fn to_utf8<O: Offset>(json_col: &ArrowJsonColumn) -> Arc<dyn Array> {
+fn to_utf8<O: Offset>(json_col: &ArrowJsonColumn, data_type: DataType) -> Arc<dyn Array> {
     let validity = to_validity(&json_col.validity);
     let offsets = to_offsets::<O>(json_col.offset.as_ref());
     let values = json_col
@@ -169,7 +169,7 @@ fn to_utf8<O: Offset>(json_col: &ArrowJsonColumn) -> Arc<dyn Array> {
         .map(|value| value.as_str().unwrap().as_bytes().to_vec())
         .flatten()
         .collect();
-    Arc::new(Utf8Array::from_data(offsets, values, validity))
+    Arc::new(Utf8Array::from_data(data_type, offsets, values, validity))
 }
 
 fn to_list<O: Offset>(
@@ -181,7 +181,12 @@ fn to_list<O: Offset>(
 
     let child_field = ListArray::<O>::get_child_field(&data_type);
     let children = &json_col.children.as_ref().unwrap()[0];
-    let values = to_array(child_field, children, dictionaries)?;
+    let values = to_array(
+        child_field.data_type().clone(),
+        child_field.dict_id(),
+        children,
+        dictionaries,
+    )?;
     let offsets = to_offsets::<O>(json_col.offset.as_ref());
     Ok(Arc::new(ListArray::<O>::from_data(
         data_type, offsets, values, validity,
@@ -189,13 +194,11 @@ fn to_list<O: Offset>(
 }
 
 fn to_dictionary<K: DictionaryKey>(
-    field: &Field,
+    data_type: DataType,
+    dict_id: i64,
     json_col: &ArrowJsonColumn,
     dictionaries: &HashMap<i64, ArrowJsonDictionaryBatch>,
 ) -> Result<Arc<dyn Array>> {
-    let dict_id = field
-        .dict_id()
-        .ok_or_else(|| ArrowError::Ipc(format!("Unable to find dict_id for field {:?}", field)))?;
     // find dictionary
     let dictionary = dictionaries
         .get(&dict_id)
@@ -204,24 +207,28 @@ fn to_dictionary<K: DictionaryKey>(
     let keys = to_primitive(json_col, K::DATA_TYPE);
 
     // todo: make DataType::Dictionary hold a Field so that it can hold dictionary_id
-    let data_type = DictionaryArray::<K>::get_child(field.data_type());
-    // note: not enough info on nullability of dictionary
-    let value_field = Field::new("value", data_type.clone(), true);
-    let values = to_array(&value_field, &dictionary.data.columns[0], &HashMap::new())?;
+    let inner_data_type = DictionaryArray::<K>::get_child(&data_type);
+    let values = to_array(
+        inner_data_type.clone(),
+        None, // this should not be None: we need to propagate the id as dicts can be nested.
+        &dictionary.data.columns[0],
+        dictionaries,
+    )?;
 
     Ok(Arc::new(DictionaryArray::<K>::from_data(keys, values)))
 }
 
 /// Construct an [`Array`] from the JSON integration format
 pub fn to_array(
-    field: &Field,
+    data_type: DataType,
+    dict_id: Option<i64>,
     json_col: &ArrowJsonColumn,
     dictionaries: &HashMap<i64, ArrowJsonDictionaryBatch>,
 ) -> Result<Arc<dyn Array>> {
-    let data_type = field.data_type();
-    match data_type {
-        DataType::Null => Ok(Arc::new(NullArray::from_data(json_col.count))),
-        DataType::Boolean => {
+    use PhysicalType::*;
+    match data_type.to_physical_type() {
+        Null => Ok(Arc::new(NullArray::from_data(data_type, json_col.count))),
+        Boolean => {
             let validity = to_validity(&json_col.validity);
             let values = json_col
                 .data
@@ -230,36 +237,29 @@ pub fn to_array(
                 .iter()
                 .map(|value| value.as_bool().unwrap())
                 .collect::<Bitmap>();
-            Ok(Arc::new(BooleanArray::from_data(values, validity)))
-        }
-        DataType::Int8 => Ok(Arc::new(to_primitive::<i8>(json_col, data_type.clone()))),
-        DataType::Int16 => Ok(Arc::new(to_primitive::<i16>(json_col, data_type.clone()))),
-        DataType::Int32
-        | DataType::Date32
-        | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth) => {
-            Ok(Arc::new(to_primitive::<i32>(json_col, data_type.clone())))
+            Ok(Arc::new(BooleanArray::from_data(
+                data_type, values, validity,
+            )))
         }
-        DataType::Int64
-        | DataType::Date64
-        | DataType::Time64(_)
-        | DataType::Timestamp(_, _)
-        | DataType::Duration(_) => Ok(Arc::new(to_primitive::<i64>(json_col, data_type.clone()))),
-        DataType::Interval(IntervalUnit::DayTime) => {
-            Ok(Arc::new(to_primitive_interval(json_col, data_type.clone())))
+        Primitive(PrimitiveType::Int8) => Ok(Arc::new(to_primitive::<i8>(json_col, data_type))),
+        Primitive(PrimitiveType::Int16) => Ok(Arc::new(to_primitive::<i16>(json_col, data_type))),
+        Primitive(PrimitiveType::Int32) => Ok(Arc::new(to_primitive::<i32>(json_col, data_type))),
+        Primitive(PrimitiveType::Int64) => Ok(Arc::new(to_primitive::<i64>(json_col, data_type))),
+        Primitive(PrimitiveType::Int128) => Ok(Arc::new(to_decimal(json_col, data_type))),
+        Primitive(PrimitiveType::DaysMs) => {
+            Ok(Arc::new(to_primitive_interval(json_col, data_type)))
         }
-        DataType::Decimal(_, _) => Ok(Arc::new(to_decimal(json_col, data_type.clone()))),
-        DataType::UInt8 => Ok(Arc::new(to_primitive::<u8>(json_col, data_type.clone()))),
-        DataType::UInt16 => Ok(Arc::new(to_primitive::<u16>(json_col, data_type.clone()))),
-        DataType::UInt32 => Ok(Arc::new(to_primitive::<u32>(json_col, data_type.clone()))),
-        DataType::UInt64 => Ok(Arc::new(to_primitive::<u64>(json_col, data_type.clone()))),
-        DataType::Float32 => Ok(Arc::new(to_primitive::<f32>(json_col, data_type.clone()))),
-        DataType::Float64 => Ok(Arc::new(to_primitive::<f64>(json_col, data_type.clone()))),
-        DataType::Binary => Ok(to_binary::<i32>(json_col)),
-        DataType::LargeBinary => Ok(to_binary::<i64>(json_col)),
-        DataType::Utf8 => Ok(to_utf8::<i32>(json_col)),
-        DataType::LargeUtf8 => Ok(to_utf8::<i64>(json_col)),
-        DataType::FixedSizeBinary(_) => {
+        Primitive(PrimitiveType::UInt8) => Ok(Arc::new(to_primitive::<u8>(json_col, data_type))),
+        Primitive(PrimitiveType::UInt16) => Ok(Arc::new(to_primitive::<u16>(json_col, data_type))),
+        Primitive(PrimitiveType::UInt32) => Ok(Arc::new(to_primitive::<u32>(json_col, data_type))),
+        Primitive(PrimitiveType::UInt64) => Ok(Arc::new(to_primitive::<u64>(json_col, data_type))),
+        Primitive(PrimitiveType::Float32) => Ok(Arc::new(to_primitive::<f32>(json_col, data_type))),
+        Primitive(PrimitiveType::Float64) => Ok(Arc::new(to_primitive::<f64>(json_col, data_type))),
+        Binary => Ok(to_binary::<i32>(json_col, data_type)),
+        LargeBinary => Ok(to_binary::<i64>(json_col, data_type)),
+        Utf8 => Ok(to_utf8::<i32>(json_col, data_type)),
+        LargeUtf8 => Ok(to_utf8::<i64>(json_col, data_type)),
+        FixedSizeBinary => {
             let validity = to_validity(&json_col.validity);
 
             let values = json_col
@@ -271,50 +271,60 @@ pub fn to_array(
                 .flatten()
                 .collect();
             Ok(Arc::new(FixedSizeBinaryArray::from_data(
-                data_type.clone(),
-                values,
-                validity,
+                data_type, values, validity,
             )))
         }
-
-        DataType::List(_) => to_list::<i32>(json_col, data_type.clone(), dictionaries),
-        DataType::LargeList(_) => to_list::<i64>(json_col, data_type.clone(), dictionaries),
-
-        DataType::FixedSizeList(child_field, _) => {
+        List => to_list::<i32>(json_col, data_type, dictionaries),
+        LargeList => to_list::<i64>(json_col, data_type, dictionaries),
+        FixedSizeList => {
             let validity = to_validity(&json_col.validity);
 
+            let (child_field, _) = FixedSizeListArray::get_child_and_size(&data_type);
+
             let children = &json_col.children.as_ref().unwrap()[0];
-            let values = to_array(child_field, children, dictionaries)?;
+            let values = to_array(
+                child_field.data_type().clone(),
+                child_field.dict_id(),
+                children,
+                dictionaries,
+            )?;
 
             Ok(Arc::new(FixedSizeListArray::from_data(
-                data_type.clone(),
-                values,
-                validity,
+                data_type, values, validity,
             )))
         }
-        DataType::Struct(fields) => {
+        Struct => {
             let validity = to_validity(&json_col.validity);
 
+            let fields = StructArray::get_fields(&data_type);
+
             let values = fields
                 .iter()
                 .zip(json_col.children.as_ref().unwrap())
-                .map(|(field, col)| to_array(field, col, dictionaries))
+                .map(|(field, col)| to_array(field.data_type().clone(), None, col, dictionaries))
                 .collect::<Result<Vec<_>>>()?;
 
-            let array = StructArray::from_data(fields.clone(), values, validity);
+            let array = StructArray::from_data(data_type, values, validity);
             Ok(Arc::new(array))
         }
-        DataType::Dictionary(key_type, _) => {
-            with_match_dictionary_key_type!(key_type.as_ref(), |$T| {
-                to_dictionary::<$T>(field, json_col, dictionaries)
+        Dictionary(key_type) => {
+            with_match_physical_dictionary_key_type!(key_type, |$T| {
+                to_dictionary::<$T>(data_type, dict_id.unwrap(), json_col, dictionaries)
             })
         }
-        DataType::Float16 => unreachable!(),
-        DataType::Union(fields, _, _) => {
+        Union => {
+            let fields = UnionArray::get_fields(&data_type);
             let fields = fields
                 .iter()
                 .zip(json_col.children.as_ref().unwrap())
-                .map(|(field, col)| to_array(field, col, dictionaries))
+                .map(|(field, col)| {
+                    to_array(
+                        field.data_type().clone(),
+                        field.dict_id(),
+                        col,
+                        dictionaries,
+                    )
+                })
                 .collect::<Result<Vec<_>>>()?;
 
             let types = json_col
@@ -352,7 +362,7 @@ pub fn to_array(
                 })
                 .unwrap_or_default();
 
-            let array = UnionArray::from_data(data_type.clone(), types, fields, offsets);
+            let array = UnionArray::from_data(data_type, types, fields, offsets);
             Ok(Arc::new(array))
         }
     }
@@ -367,7 +377,14 @@ pub fn to_record_batch(
         .fields()
         .iter()
         .zip(&json_batch.columns)
-        .map(|(field, json_col)| to_array(field, json_col, json_dictionaries))
+        .map(|(field, json_col)| {
+            to_array(
+                field.data_type().clone(),
+                field.dict_id(),
+                json_col,
+                json_dictionaries,
+            )
+        })
         .collect::<Result<Vec<_>>>()?;
 
     RecordBatch::try_new(Arc::new(schema.clone()), columns)
diff --git a/src/io/json_integration/schema.rs b/src/io/json_integration/schema.rs
index b6c85bc22d..839e27550c 100644
--- a/src/io/json_integration/schema.rs
+++ b/src/io/json_integration/schema.rs
@@ -26,6 +26,7 @@ use serde_json::{json, Value};
 use crate::error::{ArrowError, Result};
 
 use crate::datatypes::{DataType, Field, IntervalUnit, Schema, TimeUnit};
+use crate::io::ipc::get_extension;
 
 pub trait ToJson {
     /// Generate a JSON representation
@@ -114,6 +115,7 @@ impl ToJson for DataType {
             DataType::Decimal(precision, scale) => {
                 json!({"name": "decimal", "precision": precision, "scale": scale})
             }
+            DataType::Extension(_, inner_data_type, _) => inner_data_type.to_json(),
         }
     }
 }
@@ -176,6 +178,63 @@ fn children(children: Option<&Value>) -> Result<Vec<Field>> {
         .unwrap_or_else(|| Ok(vec![]))
 }
 
+fn read_metadata(metadata: &Value) -> Result<BTreeMap<String, String>> {
+    match metadata {
+        Value::Array(ref values) => {
+            let mut res: BTreeMap<String, String> = BTreeMap::new();
+            for value in values {
+                match value.as_object() {
+                    Some(map) => {
+                        if map.len() != 2 {
+                            return Err(ArrowError::Schema(
+                                "Field 'metadata' must have exact two entries for each key-value map".to_string(),
+                            ));
+                        }
+                        if let (Some(k), Some(v)) = (map.get("key"), map.get("value")) {
+                            if let (Some(k_str), Some(v_str)) = (k.as_str(), v.as_str()) {
+                                res.insert(k_str.to_string().clone(), v_str.to_string().clone());
+                            } else {
+                                return Err(ArrowError::Schema(
+                                    "Field 'metadata' must have map value of string type"
+                                        .to_string(),
+                                ));
+                            }
+                        } else {
+                            return Err(ArrowError::Schema(
+                                "Field 'metadata' lacks map keys named \"key\" or \"value\""
+                                    .to_string(),
+                            ));
+                        }
+                    }
+                    _ => {
+                        return Err(ArrowError::Schema(
+                            "Field 'metadata' contains non-object key-value pair".to_string(),
+                        ));
+                    }
+                }
+            }
+            Ok(res)
+        }
+        Value::Object(ref values) => {
+            let mut res: BTreeMap<String, String> = BTreeMap::new();
+            for (k, v) in values {
+                if let Some(str_value) = v.as_str() {
+                    res.insert(k.clone(), str_value.to_string().clone());
+                } else {
+                    return Err(ArrowError::Schema(format!(
+                        "Field 'metadata' contains non-string value for key {}",
+                        k
+                    )));
+                }
+            }
+            Ok(res)
+        }
+        _ => Err(ArrowError::Schema(
+            "Invalid json value type for field".to_string(),
+        )),
+    }
+}
+
 fn to_data_type(item: &Value, mut children: Vec<Field>) -> Result<DataType> {
     let type_ = item
         .get("name")
@@ -380,104 +439,60 @@ impl TryFrom<&Value> for Field {
 
                 let children = children(map.get("children"))?;
 
-                let data_type = to_data_type(
-                    map.get("type")
-                        .ok_or_else(|| ArrowError::Schema("type missing".to_string()))?,
-                    children,
-                )?;
+                let metadata = if let Some(metadata) = map.get("metadata") {
+                    Some(read_metadata(metadata)?)
+                } else {
+                    None
+                };
 
-                // Referenced example file: testing/data/arrow-ipc-stream/integration/1.0.0-littleendian/generated_custom_metadata.json.gz
-                let metadata = match map.get("metadata") {
-                    Some(&Value::Array(ref values)) => {
-                        let mut res: BTreeMap<String, String> = BTreeMap::new();
-                        for value in values {
-                            match value.as_object() {
-                                Some(map) => {
-                                    if map.len() != 2 {
-                                        return Err(ArrowError::Schema(
-                                            "Field 'metadata' must have exact two entries for each key-value map".to_string(),
-                                        ));
-                                    }
-                                    if let (Some(k), Some(v)) = (map.get("key"), map.get("value")) {
-                                        if let (Some(k_str), Some(v_str)) = (k.as_str(), v.as_str())
-                                        {
-                                            res.insert(
-                                                k_str.to_string().clone(),
-                                                v_str.to_string().clone(),
-                                            );
-                                        } else {
-                                            return Err(ArrowError::Schema("Field 'metadata' must have map value of string type".to_string()));
-                                        }
-                                    } else {
-                                        return Err(ArrowError::Schema("Field 'metadata' lacks map keys named \"key\" or \"value\"".to_string()));
-                                    }
-                                }
-                                _ => {
-                                    return Err(ArrowError::Schema(
-                                        "Field 'metadata' contains non-object key-value pair"
-                                            .to_string(),
-                                    ));
-                                }
-                            }
-                        }
-                        Some(res)
-                    }
-                    // We also support map format, because Schema's metadata supports this.
-                    // See https://github.com/apache/arrow/pull/5907
-                    Some(&Value::Object(ref values)) => {
-                        let mut res: BTreeMap<String, String> = BTreeMap::new();
-                        for (k, v) in values {
-                            if let Some(str_value) = v.as_str() {
-                                res.insert(k.clone(), str_value.to_string().clone());
-                            } else {
-                                return Err(ArrowError::Schema(format!(
-                                    "Field 'metadata' contains non-string value for key {}",
-                                    k
-                                )));
-                            }
-                        }
-                        Some(res)
-                    }
-                    Some(_) => {
-                        return Err(ArrowError::Schema(
-                            "Field `metadata` is not json array".to_string(),
-                        ));
-                    }
-                    _ => None,
+                let extension = get_extension(&metadata);
+
+                let type_ = map
+                    .get("type")
+                    .ok_or_else(|| ArrowError::Schema("type missing".to_string()))?;
+
+                let data_type = to_data_type(type_, children)?;
+
+                let data_type = if let Some((name, metadata)) = extension {
+                    DataType::Extension(name, Box::new(data_type), metadata)
+                } else {
+                    data_type
                 };
 
-                let mut dict_id = 0;
-                let mut dict_is_ordered = false;
+                let data_type = if let Some(dictionary) = map.get("dictionary") {
+                    let index_type = match dictionary.get("indexType") {
+                        Some(t) => to_data_type(t, vec![])?,
+                        _ => {
+                            return Err(ArrowError::Schema(
+                                "Field missing 'indexType' attribute".to_string(),
+                            ));
+                        }
+                    };
+                    DataType::Dictionary(Box::new(index_type), Box::new(data_type))
+                } else {
+                    data_type
+                };
 
-                let data_type = match map.get("dictionary") {
-                    Some(dictionary) => {
-                        let index_type = match dictionary.get("indexType") {
-                            Some(t) => to_data_type(t, vec![])?,
-                            _ => {
-                                return Err(ArrowError::Schema(
-                                    "Field missing 'indexType' attribute".to_string(),
-                                ));
-                            }
-                        };
-                        dict_id = match dictionary.get("id") {
-                            Some(Value::Number(n)) => n.as_i64().unwrap(),
-                            _ => {
-                                return Err(ArrowError::Schema(
-                                    "Field missing 'id' attribute".to_string(),
-                                ));
-                            }
-                        };
-                        dict_is_ordered = match dictionary.get("isOrdered") {
-                            Some(&Value::Bool(n)) => n,
-                            _ => {
-                                return Err(ArrowError::Schema(
-                                    "Field missing 'isOrdered' attribute".to_string(),
-                                ));
-                            }
-                        };
-                        DataType::Dictionary(Box::new(index_type), Box::new(data_type))
-                    }
-                    _ => data_type,
+                let (dict_id, dict_is_ordered) = if let Some(dictionary) = map.get("dictionary") {
+                    let dict_id = match dictionary.get("id") {
+                        Some(Value::Number(n)) => n.as_i64().unwrap(),
+                        _ => {
+                            return Err(ArrowError::Schema(
+                                "Field missing 'id' attribute".to_string(),
+                            ));
+                        }
+                    };
+                    let dict_is_ordered = match dictionary.get("isOrdered") {
+                        Some(&Value::Bool(n)) => n,
+                        _ => {
+                            return Err(ArrowError::Schema(
+                                "Field missing 'isOrdered' attribute".to_string(),
+                            ));
+                        }
+                    };
+                    (dict_id, dict_is_ordered)
+                } else {
+                    (0, false)
                 };
                 let mut f = Field::new_dict(&name, data_type, nullable, dict_id, dict_is_ordered);
                 f.set_metadata(metadata);
diff --git a/src/io/parquet/read/binary/basic.rs b/src/io/parquet/read/binary/basic.rs
index 5171cd87dd..bf07f06e0f 100644
--- a/src/io/parquet/read/binary/basic.rs
+++ b/src/io/parquet/read/binary/basic.rs
@@ -283,11 +283,13 @@ where
 
     Ok(match data_type {
         DataType::LargeBinary | DataType::Binary => Box::new(BinaryArray::from_data(
+            data_type.clone(),
             offsets.into(),
             values.into(),
             validity.into(),
         )),
         DataType::LargeUtf8 | DataType::Utf8 => Box::new(Utf8Array::from_data(
+            data_type.clone(),
             offsets.into(),
             values.into(),
             validity.into(),
@@ -327,11 +329,13 @@ where
 
     Ok(match data_type {
         DataType::LargeBinary | DataType::Binary => Box::new(BinaryArray::from_data(
+            data_type.clone(),
             offsets.into(),
             values.into(),
             validity.into(),
         )),
         DataType::LargeUtf8 | DataType::Utf8 => Box::new(Utf8Array::from_data(
+            data_type.clone(),
             offsets.into(),
             values.into(),
             validity.into(),
diff --git a/src/io/parquet/read/binary/dictionary.rs b/src/io/parquet/read/binary/dictionary.rs
index 011a19dd64..3609fa05a1 100644
--- a/src/io/parquet/read/binary/dictionary.rs
+++ b/src/io/parquet/read/binary/dictionary.rs
@@ -12,9 +12,11 @@ use crate::{
     array::{Array, DictionaryArray, DictionaryKey, Offset, PrimitiveArray, Utf8Array},
     bitmap::{utils::BitmapIter, MutableBitmap},
     buffer::MutableBuffer,
+    datatypes::DataType,
     error::{ArrowError, Result},
 };
 
+#[allow(clippy::too_many_arguments)]
 fn read_dict_optional<K, O>(
     validity_buffer: &[u8],
     indices_buffer: &[u8],
@@ -124,6 +126,7 @@ where
 pub fn iter_to_array<K, O, I, E>(
     mut iter: I,
     metadata: &ColumnChunkMetaData,
+    data_type: DataType,
 ) -> Result<Box<dyn Array>>
 where
     ArrowError: From<E>,
@@ -149,6 +152,12 @@ where
     }
 
     let keys = PrimitiveArray::from_data(K::DATA_TYPE, indices.into(), validity.into());
-    let values = Arc::new(Utf8Array::from_data(offsets.into(), values.into(), None));
+    let data_type = DictionaryArray::<K>::get_child(&data_type).clone();
+    let values = Arc::new(Utf8Array::from_data(
+        data_type,
+        offsets.into(),
+        values.into(),
+        None,
+    ));
     Ok(Box::new(DictionaryArray::<K>::from_data(keys, values)))
 }
diff --git a/src/io/parquet/read/binary/nested.rs b/src/io/parquet/read/binary/nested.rs
index 139e66ae15..211a79ef95 100644
--- a/src/io/parquet/read/binary/nested.rs
+++ b/src/io/parquet/read/binary/nested.rs
@@ -189,11 +189,13 @@ where
 
     let values = match inner_data_type {
         DataType::LargeBinary | DataType::Binary => Arc::new(BinaryArray::from_data(
+            inner_data_type.clone(),
             offsets.into(),
             values.into(),
             validity.into(),
         )) as Arc<dyn Array>,
         DataType::LargeUtf8 | DataType::Utf8 => Arc::new(Utf8Array::from_data(
+            inner_data_type.clone(),
             offsets.into(),
             values.into(),
             validity.into(),
diff --git a/src/io/parquet/read/boolean/basic.rs b/src/io/parquet/read/boolean/basic.rs
index f23f631b2f..914b2921b9 100644
--- a/src/io/parquet/read/boolean/basic.rs
+++ b/src/io/parquet/read/boolean/basic.rs
@@ -1,6 +1,7 @@
 use crate::{
     array::BooleanArray,
     bitmap::{utils::BitmapIter, MutableBitmap},
+    datatypes::DataType,
     error::{ArrowError, Result},
 };
 
@@ -85,7 +86,11 @@ where
         )?
     }
 
-    Ok(BooleanArray::from_data(values.into(), validity.into()))
+    Ok(BooleanArray::from_data(
+        DataType::Boolean,
+        values.into(),
+        validity.into(),
+    ))
 }
 
 pub async fn stream_to_array<I, E>(pages: I, metadata: &ColumnChunkMetaData) -> Result<BooleanArray>
@@ -109,7 +114,11 @@ where
         )?
     }
 
-    Ok(BooleanArray::from_data(values.into(), validity.into()))
+    Ok(BooleanArray::from_data(
+        DataType::Boolean,
+        values.into(),
+        validity.into(),
+    ))
 }
 
 fn extend_from_page(
diff --git a/src/io/parquet/read/boolean/nested.rs b/src/io/parquet/read/boolean/nested.rs
index 1cb0ffc7c5..fc48477ea8 100644
--- a/src/io/parquet/read/boolean/nested.rs
+++ b/src/io/parquet/read/boolean/nested.rs
@@ -157,7 +157,22 @@ where
         )?
     }
 
-    let values = Arc::new(BooleanArray::from_data(values.into(), validity.into()));
+    let inner_data_type = match data_type {
+        DataType::List(ref inner) => inner.data_type(),
+        DataType::LargeList(ref inner) => inner.data_type(),
+        _ => {
+            return Err(ArrowError::NotYetImplemented(format!(
+                "Read nested datatype {:?}",
+                data_type
+            )))
+        }
+    };
+
+    let values = Arc::new(BooleanArray::from_data(
+        inner_data_type.clone(),
+        values.into(),
+        validity.into(),
+    ));
 
     create_list(data_type, &mut nested, values)
 }
diff --git a/src/io/parquet/read/mod.rs b/src/io/parquet/read/mod.rs
index 1ade626d39..212677e0f5 100644
--- a/src/io/parquet/read/mod.rs
+++ b/src/io/parquet/read/mod.rs
@@ -152,8 +152,8 @@ fn dict_read<
         Int64 | Date64 | Time64(_) | Duration(_) | Timestamp(_, _) => {
             primitive::iter_to_dict_array::<K, _, _, _, _, _>(iter, metadata, data_type, |x: i64| x)
         }
-        Utf8 => binary::iter_to_dict_array::<K, i32, _, _>(iter, metadata),
-        LargeUtf8 => binary::iter_to_dict_array::<K, i64, _, _>(iter, metadata),
+        Utf8 => binary::iter_to_dict_array::<K, i32, _, _>(iter, metadata, data_type),
+        LargeUtf8 => binary::iter_to_dict_array::<K, i64, _, _>(iter, metadata, data_type),
         other => Err(ArrowError::NotYetImplemented(format!(
             "Reading dictionaries of type {:?}",
             other
diff --git a/src/io/parquet/read/primitive/dictionary.rs b/src/io/parquet/read/primitive/dictionary.rs
index 8d7c7de4bc..6cf2a5245c 100644
--- a/src/io/parquet/read/primitive/dictionary.rs
+++ b/src/io/parquet/read/primitive/dictionary.rs
@@ -18,6 +18,7 @@ use crate::{
     types::NativeType as ArrowNativeType,
 };
 
+#[allow(clippy::too_many_arguments)]
 fn read_dict_optional<K, T, A, F>(
     validity_buffer: &[u8],
     indices_buffer: &[u8],
diff --git a/src/record_batch.rs b/src/record_batch.rs
index 6168693623..40164d59eb 100644
--- a/src/record_batch.rs
+++ b/src/record_batch.rs
@@ -360,7 +360,7 @@ impl From<RecordBatch> for StructArray {
             .zip(batch.columns.iter())
             .map(|t| (t.0.clone(), t.1.clone()))
             .unzip();
-        StructArray::from_data(fields, values, None)
+        StructArray::from_data(DataType::Struct(fields), values, None)
     }
 }
 
diff --git a/src/scalar/mod.rs b/src/scalar/mod.rs
index 39bb9a858c..da3f2241af 100644
--- a/src/scalar/mod.rs
+++ b/src/scalar/mod.rs
@@ -29,21 +29,6 @@ pub trait Scalar: std::fmt::Debug {
     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();
@@ -85,8 +70,8 @@ macro_rules! dyn_new_list {
 
 /// 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() {
+    use PhysicalType::*;
+    match array.data_type().to_physical_type() {
         Null => Box::new(NullScalar::new()),
         Boolean => {
             let array = array.as_any().downcast_ref::<BooleanArray>().unwrap();
@@ -97,28 +82,25 @@ pub fn new_scalar(array: &dyn Array, index: usize) -> Box<dyn Scalar> {
             };
             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),
+        Primitive(primitive) => with_match_primitive_type!(primitive, |$T| {
+            let array = array
+                .as_any()
+                .downcast_ref::<PrimitiveArray<$T>>()
+                .unwrap();
+            let value = if array.is_valid(index) {
+                Some(array.value(index))
+            } else {
+                None
+            };
+            Box::new(PrimitiveScalar::new(array.data_type().clone(), value))
+        }),
         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(_) => {
+        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
@@ -131,9 +113,9 @@ pub fn new_scalar(array: &dyn Array, index: usize) -> Box<dyn Scalar> {
                 Box::new(StructScalar::new(array.data_type().clone(), None))
             }
         }
-        FixedSizeBinary(_) => todo!(),
-        FixedSizeList(_, _) => todo!(),
-        Union(_, _, _) => todo!(),
-        Dictionary(_, _) => todo!(),
+        FixedSizeBinary => todo!(),
+        FixedSizeList => todo!(),
+        Union => todo!(),
+        Dictionary(_) => todo!(),
     }
 }
diff --git a/src/types/mod.rs b/src/types/mod.rs
index 3a30ee7801..61125a6282 100644
--- a/src/types/mod.rs
+++ b/src/types/mod.rs
@@ -15,7 +15,7 @@ mod index;
 pub mod simd;
 pub use index::*;
 
-use crate::datatypes::{DataType, IntervalUnit, TimeUnit};
+use crate::datatypes::{DataType, IntervalUnit, PhysicalType, PrimitiveType};
 
 /// Trait denoting anything that has a natural logical [`DataType`].
 /// For example, [`DataType::Int32`] for `i32`.
@@ -31,11 +31,11 @@ pub unsafe trait Relation {
 }
 
 macro_rules! create_relation {
-    ($native_ty:ty, $($impl_pattern:pat)|+) => {
+    ($native_ty:ty, $physical_ty:expr) => {
         unsafe impl Relation for $native_ty {
             #[inline]
             fn is_valid(data_type: &DataType) -> bool {
-                matches!(data_type, $($impl_pattern)|+)
+                data_type.to_physical_type() == $physical_ty
             }
         }
     };
@@ -127,34 +127,17 @@ natural_type!(f64, DataType::Float64);
 natural_type!(days_ms, DataType::Interval(IntervalUnit::DayTime));
 natural_type!(i128, DataType::Decimal(32, 32)); // users should set the decimal when creating an array
 
-create_relation!(u8, &DataType::UInt8);
-create_relation!(u16, &DataType::UInt16);
-create_relation!(u32, &DataType::UInt32);
-create_relation!(u64, &DataType::UInt64);
-create_relation!(i8, &DataType::Int8);
-create_relation!(i16, &DataType::Int16);
-create_relation!(
-    i32,
-    &DataType::Int32
-        | &DataType::Date32
-        | &DataType::Time32(TimeUnit::Millisecond)
-        | &DataType::Time32(TimeUnit::Second)
-        | &DataType::Interval(IntervalUnit::YearMonth)
-);
-
-create_relation!(
-    i64,
-    &DataType::Int64
-        | &DataType::Date64
-        | &DataType::Time64(TimeUnit::Microsecond)
-        | &DataType::Time64(TimeUnit::Nanosecond)
-        | &DataType::Timestamp(_, _)
-        | &DataType::Duration(_)
-);
-
-create_relation!(i128, &DataType::Decimal(_, _));
-create_relation!(f32, &DataType::Float32);
-create_relation!(f64, &DataType::Float64);
+create_relation!(u8, PhysicalType::Primitive(PrimitiveType::UInt8));
+create_relation!(u16, PhysicalType::Primitive(PrimitiveType::UInt16));
+create_relation!(u32, PhysicalType::Primitive(PrimitiveType::UInt32));
+create_relation!(u64, PhysicalType::Primitive(PrimitiveType::UInt64));
+create_relation!(i8, PhysicalType::Primitive(PrimitiveType::Int8));
+create_relation!(i16, PhysicalType::Primitive(PrimitiveType::Int16));
+create_relation!(i32, PhysicalType::Primitive(PrimitiveType::Int32));
+create_relation!(i64, PhysicalType::Primitive(PrimitiveType::Int64));
+create_relation!(i128, PhysicalType::Primitive(PrimitiveType::Int128));
+create_relation!(f32, PhysicalType::Primitive(PrimitiveType::Float32));
+create_relation!(f64, PhysicalType::Primitive(PrimitiveType::Float64));
 
 /// The in-memory representation of the DayMillisecond variant of arrow's "Interval" logical type.
 #[derive(Debug, Copy, Clone, Default, PartialEq, Eq, Hash)]
@@ -217,7 +200,7 @@ unsafe impl NativeType for days_ms {
     }
 }
 
-create_relation!(days_ms, &DataType::Interval(IntervalUnit::DayTime));
+create_relation!(days_ms, PhysicalType::Primitive(PrimitiveType::DaysMs));
 
 impl days_ms {
     /// A new [`days_ms`].


Test patch:
diff --git a/integration-testing/Cargo.toml b/integration-testing/Cargo.toml
index f2c6e31d22..e07ae2c141 100644
--- a/integration-testing/Cargo.toml
+++ b/integration-testing/Cargo.toml
@@ -30,7 +30,7 @@ publish = false
 logging = ["tracing-subscriber"]
 
 [dependencies]
-arrow2 = { path = "../", features = ["io_ipc", "io_json_integration"], default-features = false }
+arrow2 = { path = "../", features = ["io_ipc", "io_ipc_compression", "io_json_integration"], default-features = false }
 arrow-flight = { path = "../arrow-flight" }
 async-trait = "0.1.41"
 clap = "2.33"
diff --git a/integration-testing/unskip.patch b/integration-testing/unskip.patch
new file mode 100644
index 0000000000..fb84d7d94d
--- /dev/null
+++ b/integration-testing/unskip.patch
@@ -0,0 +1,40 @@
+diff --git a/dev/archery/archery/integration/datagen.py b/dev/archery/archery/integration/datagen.py
+index d0c4b3d6c..bc4b83e68 100644
+--- a/dev/archery/archery/integration/datagen.py
++++ b/dev/archery/archery/integration/datagen.py
+@@ -1568,8 +1568,7 @@ def get_generated_json_files(tempdir=None):
+         .skip_category('Go')    # TODO(ARROW-7901)
+         .skip_category('JS'),   # TODO(ARROW-7900)
+ 
+-        generate_decimal128_case()
+-        .skip_category('Rust'),
++        generate_decimal128_case(),
+ 
+         generate_decimal256_case()
+         .skip_category('Go')  # TODO(ARROW-7948): Decimal + Go
+@@ -1579,8 +1578,7 @@ def get_generated_json_files(tempdir=None):
+         generate_datetime_case(),
+ 
+         generate_interval_case()
+-        .skip_category('JS')  # TODO(ARROW-5239): Intervals + JS
+-        .skip_category('Rust'),
++        .skip_category('JS'),  # TODO(ARROW-5239): Intervals + JS
+ 
+         generate_month_day_nano_interval_case()
+         .skip_category('Go')
+@@ -1603,13 +1601,11 @@ def get_generated_json_files(tempdir=None):
+ 
+         generate_nested_large_offsets_case()
+         .skip_category('Go')
+-        .skip_category('JS')
+-        .skip_category('Rust'),
++        .skip_category('JS'),
+ 
+         generate_unions_case()
+         .skip_category('Go')
+-        .skip_category('JS')
+-        .skip_category('Rust'),
++        .skip_category('JS'),
+ 
+         generate_custom_metadata_case()
+         .skip_category('JS'),
diff --git a/tests/it/array/binary/mod.rs b/tests/it/array/binary/mod.rs
index 1f48730acd..a7c6401034 100644
--- a/tests/it/array/binary/mod.rs
+++ b/tests/it/array/binary/mod.rs
@@ -1,6 +1,7 @@
 use arrow2::{
     array::{Array, BinaryArray},
     bitmap::Bitmap,
+    datatypes::DataType,
 };
 
 #[test]
@@ -24,6 +25,7 @@ fn basics() {
     assert!(array.is_valid(2));
 
     let array2 = BinaryArray::<i32>::from_data(
+        DataType::Binary,
         array.offsets().clone(),
         array.values().clone(),
         array.validity().clone(),
@@ -40,7 +42,7 @@ fn basics() {
 
 #[test]
 fn empty() {
-    let array = BinaryArray::<i32>::new_empty();
+    let array = BinaryArray::<i32>::new_empty(DataType::Binary);
     assert_eq!(array.values().as_slice(), b"");
     assert_eq!(array.offsets().as_slice(), &[0]);
     assert_eq!(array.validity(), &None);
diff --git a/tests/it/array/boolean/mod.rs b/tests/it/array/boolean/mod.rs
index ce2cb815d1..f237687812 100644
--- a/tests/it/array/boolean/mod.rs
+++ b/tests/it/array/boolean/mod.rs
@@ -1,6 +1,7 @@
 use arrow2::{
     array::{Array, BooleanArray},
     bitmap::Bitmap,
+    datatypes::DataType,
 };
 
 mod mutable;
@@ -23,7 +24,11 @@ fn basics() {
     assert!(!array.is_valid(1));
     assert!(array.is_valid(2));
 
-    let array2 = BooleanArray::from_data(array.values().clone(), array.validity().clone());
+    let array2 = BooleanArray::from_data(
+        DataType::Boolean,
+        array.values().clone(),
+        array.validity().clone(),
+    );
     assert_eq!(array, array2);
 
     let array = array.slice(1, 2);
@@ -33,7 +38,7 @@ fn basics() {
 
 #[test]
 fn empty() {
-    let array = BooleanArray::new_empty();
+    let array = BooleanArray::new_empty(DataType::Boolean);
     assert_eq!(array.values().len(), 0);
     assert_eq!(array.validity(), &None);
 }
diff --git a/tests/it/array/boolean/mutable.rs b/tests/it/array/boolean/mutable.rs
index 70ca6845f2..d3d8a3228e 100644
--- a/tests/it/array/boolean/mutable.rs
+++ b/tests/it/array/boolean/mutable.rs
@@ -1,5 +1,6 @@
 use arrow2::array::{MutableArray, MutableBooleanArray};
 use arrow2::bitmap::MutableBitmap;
+use arrow2::datatypes::DataType;
 use arrow2::error::Result;
 
 #[test]
@@ -53,7 +54,11 @@ fn try_from_trusted_len_iter() {
 
 #[test]
 fn reserve() {
-    let mut a = MutableBooleanArray::from_data(MutableBitmap::new(), Some(MutableBitmap::new()));
+    let mut a = MutableBooleanArray::from_data(
+        DataType::Boolean,
+        MutableBitmap::new(),
+        Some(MutableBitmap::new()),
+    );
 
     a.reserve(10);
     assert!(a.validity().as_ref().unwrap().capacity() > 0);
diff --git a/tests/it/array/fixed_size_binary/mutable.rs b/tests/it/array/fixed_size_binary/mutable.rs
index d8c27ccee5..7373182084 100644
--- a/tests/it/array/fixed_size_binary/mutable.rs
+++ b/tests/it/array/fixed_size_binary/mutable.rs
@@ -5,7 +5,11 @@ use arrow2::datatypes::DataType;
 
 #[test]
 fn basic() {
-    let a = MutableFixedSizeBinaryArray::from_data(2, MutableBuffer::from([1, 2, 3, 4]), None);
+    let a = MutableFixedSizeBinaryArray::from_data(
+        DataType::FixedSizeBinary(2),
+        MutableBuffer::from([1, 2, 3, 4]),
+        None,
+    );
     assert_eq!(a.len(), 2);
     assert_eq!(a.data_type(), &DataType::FixedSizeBinary(2));
     assert_eq!(a.values(), &MutableBuffer::from([1, 2, 3, 4]));
@@ -17,18 +21,26 @@ fn basic() {
 #[allow(clippy::eq_op)]
 #[test]
 fn equal() {
-    let a = MutableFixedSizeBinaryArray::from_data(2, MutableBuffer::from([1, 2, 3, 4]), None);
+    let a = MutableFixedSizeBinaryArray::from_data(
+        DataType::FixedSizeBinary(2),
+        MutableBuffer::from([1, 2, 3, 4]),
+        None,
+    );
     assert_eq!(a, a);
-    let b = MutableFixedSizeBinaryArray::from_data(2, MutableBuffer::from([1, 2]), None);
+    let b = MutableFixedSizeBinaryArray::from_data(
+        DataType::FixedSizeBinary(2),
+        MutableBuffer::from([1, 2]),
+        None,
+    );
     assert_eq!(b, b);
     assert!(a != b);
     let a = MutableFixedSizeBinaryArray::from_data(
-        2,
+        DataType::FixedSizeBinary(2),
         MutableBuffer::from([1, 2, 3, 4]),
         Some(MutableBitmap::from([true, false])),
     );
     let b = MutableFixedSizeBinaryArray::from_data(
-        2,
+        DataType::FixedSizeBinary(2),
         MutableBuffer::from([1, 2, 3, 4]),
         Some(MutableBitmap::from([false, true])),
     );
diff --git a/tests/it/array/growable/null.rs b/tests/it/array/growable/null.rs
index 934f6a24c9..47e25ed3b9 100644
--- a/tests/it/array/growable/null.rs
+++ b/tests/it/array/growable/null.rs
@@ -1,15 +1,20 @@
-use arrow2::array::growable::{Growable, GrowableNull};
-use arrow2::array::*;
+use arrow2::{
+    array::{
+        growable::{Growable, GrowableNull},
+        NullArray,
+    },
+    datatypes::DataType,
+};
 
 #[test]
 fn null() {
-    let mut mutable = GrowableNull::new();
+    let mut mutable = GrowableNull::default();
 
     mutable.extend(0, 1, 2);
     mutable.extend(1, 0, 1);
 
     let result: NullArray = mutable.into();
 
-    let expected = NullArray::from_data(3);
+    let expected = NullArray::from_data(DataType::Null, 3);
     assert_eq!(result, expected);
 }
diff --git a/tests/it/array/growable/struct_.rs b/tests/it/array/growable/struct_.rs
index f26e4c08ae..0dd85e0dcd 100644
--- a/tests/it/array/growable/struct_.rs
+++ b/tests/it/array/growable/struct_.rs
@@ -7,7 +7,7 @@ use arrow2::array::{
 use arrow2::bitmap::Bitmap;
 use arrow2::datatypes::{DataType, Field};
 
-fn some_values() -> (Vec<Field>, Vec<Arc<dyn Array>>) {
+fn some_values() -> (DataType, Vec<Arc<dyn Array>>) {
     let strings: Arc<dyn Array> = Arc::new(Utf8Array::<i32>::from(&[
         Some("a"),
         Some("aa"),
@@ -26,7 +26,7 @@ fn some_values() -> (Vec<Field>, Vec<Arc<dyn Array>>) {
         Field::new("f1", DataType::Utf8, true),
         Field::new("f2", DataType::Int32, true),
     ];
-    (fields, vec![strings, ints])
+    (DataType::Struct(fields), vec![strings, ints])
 }
 
 #[test]
diff --git a/tests/it/array/utf8/mod.rs b/tests/it/array/utf8/mod.rs
index c9cc823a74..49c993d575 100644
--- a/tests/it/array/utf8/mod.rs
+++ b/tests/it/array/utf8/mod.rs
@@ -1,4 +1,4 @@
-use arrow2::{array::*, bitmap::Bitmap, buffer::Buffer, error::Result};
+use arrow2::{array::*, bitmap::Bitmap, buffer::Buffer, datatypes::DataType, error::Result};
 
 mod mutable;
 
@@ -23,6 +23,7 @@ fn basics() {
     assert!(array.is_valid(2));
 
     let array2 = Utf8Array::<i32>::from_data(
+        DataType::Utf8,
         array.offsets().clone(),
         array.values().clone(),
         array.validity().clone(),
@@ -39,7 +40,7 @@ fn basics() {
 
 #[test]
 fn empty() {
-    let array = Utf8Array::<i32>::new_empty();
+    let array = Utf8Array::<i32>::new_empty(DataType::Utf8);
     assert_eq!(array.values().as_slice(), b"");
     assert_eq!(array.offsets().as_slice(), &[0]);
     assert_eq!(array.validity(), &None);
@@ -59,7 +60,10 @@ fn from_slice() {
 
     let offsets = Buffer::from(&[0, 1, 2, 4]);
     let values = Buffer::from("abcc".as_bytes());
-    assert_eq!(b, Utf8Array::<i32>::from_data(offsets, values, None));
+    assert_eq!(
+        b,
+        Utf8Array::<i32>::from_data(DataType::Utf8, offsets, values, None)
+    );
 }
 
 #[test]
@@ -68,7 +72,10 @@ fn from_iter_values() {
 
     let offsets = Buffer::from(&[0, 1, 2, 4]);
     let values = Buffer::from("abcc".as_bytes());
-    assert_eq!(b, Utf8Array::<i32>::from_data(offsets, values, None));
+    assert_eq!(
+        b,
+        Utf8Array::<i32>::from_data(DataType::Utf8, offsets, values, None)
+    );
 }
 
 #[test]
@@ -78,7 +85,10 @@ fn from_trusted_len_iter() {
 
     let offsets = Buffer::from(&[0, 1, 2, 4]);
     let values = Buffer::from("abcc".as_bytes());
-    assert_eq!(b, Utf8Array::<i32>::from_data(offsets, values, None));
+    assert_eq!(
+        b,
+        Utf8Array::<i32>::from_data(DataType::Utf8, offsets, values, None)
+    );
 }
 
 #[test]
@@ -92,5 +102,8 @@ fn try_from_trusted_len_iter() {
 
     let offsets = Buffer::from(&[0, 1, 2, 4]);
     let values = Buffer::from("abcc".as_bytes());
-    assert_eq!(b, Utf8Array::<i32>::from_data(offsets, values, None));
+    assert_eq!(
+        b,
+        Utf8Array::<i32>::from_data(DataType::Utf8, offsets, values, None)
+    );
 }
diff --git a/tests/it/io/ipc/common.rs b/tests/it/io/ipc/common.rs
index 6a46c03393..8eff523789 100644
--- a/tests/it/io/ipc/common.rs
+++ b/tests/it/io/ipc/common.rs
@@ -12,7 +12,7 @@ use arrow2::{
 use flate2::read::GzDecoder;
 
 /// Read gzipped JSON file
-pub fn read_gzip_json(version: &str, file_name: &str) -> (Schema, Vec<RecordBatch>) {
+pub fn read_gzip_json(version: &str, file_name: &str) -> Result<(Schema, Vec<RecordBatch>)> {
     let testdata = crate::test_util::arrow_test_data();
     let file = File::open(format!(
         "{}/arrow-ipc-stream/integration/{}/{}.json.gz",
@@ -23,10 +23,11 @@ pub fn read_gzip_json(version: &str, file_name: &str) -> (Schema, Vec<RecordBatc
     let mut s = String::new();
     gz.read_to_string(&mut s).unwrap();
     // convert to Arrow JSON
-    let arrow_json: ArrowJson = serde_json::from_str(&s).unwrap();
+    let arrow_json: ArrowJson = serde_json::from_str(&s)?;
 
     let schema = serde_json::to_value(arrow_json.schema).unwrap();
-    let schema = Schema::try_from(&schema).unwrap();
+
+    let schema = Schema::try_from(&schema)?;
 
     // read dictionaries
     let mut dictionaries = HashMap::new();
@@ -41,10 +42,9 @@ pub fn read_gzip_json(version: &str, file_name: &str) -> (Schema, Vec<RecordBatc
         .batches
         .iter()
         .map(|batch| to_record_batch(&schema, batch, &dictionaries))
-        .collect::<Result<Vec<_>>>()
-        .unwrap();
+        .collect::<Result<Vec<_>>>()?;
 
-    (schema, batches)
+    Ok((schema, batches))
 }
 
 pub fn read_arrow_stream(version: &str, file_name: &str) -> (Schema, Vec<RecordBatch>) {
diff --git a/tests/it/io/ipc/read/file.rs b/tests/it/io/ipc/read/file.rs
index 08e45b68c2..cf1920ab25 100644
--- a/tests/it/io/ipc/read/file.rs
+++ b/tests/it/io/ipc/read/file.rs
@@ -16,7 +16,7 @@ fn test_file(version: &str, file_name: &str) -> Result<()> {
     let reader = FileReader::new(&mut file, metadata, None);
 
     // read expected JSON output
-    let (schema, batches) = read_gzip_json(version, file_name);
+    let (schema, batches) = read_gzip_json(version, file_name)?;
 
     assert_eq!(&schema, reader.schema().as_ref());
 
@@ -117,6 +117,11 @@ fn read_generated_100_union() -> Result<()> {
     test_file("1.0.0-bigendian", "generated_union")
 }
 
+#[test]
+fn read_generated_100_extension() -> Result<()> {
+    test_file("1.0.0-littleendian", "generated_extension")
+}
+
 #[test]
 fn read_generated_017_union() -> Result<()> {
     test_file("0.17.1", "generated_union")
diff --git a/tests/it/io/ipc/read/stream.rs b/tests/it/io/ipc/read/stream.rs
index d6ad005e78..7a69e19178 100644
--- a/tests/it/io/ipc/read/stream.rs
+++ b/tests/it/io/ipc/read/stream.rs
@@ -16,7 +16,7 @@ fn test_file(version: &str, file_name: &str) -> Result<()> {
     let reader = StreamReader::new(file, metadata);
 
     // read expected JSON output
-    let (schema, batches) = read_gzip_json(version, file_name);
+    let (schema, batches) = read_gzip_json(version, file_name)?;
 
     assert_eq!(&schema, reader.schema().as_ref());
 
diff --git a/tests/it/io/ipc/write/file.rs b/tests/it/io/ipc/write/file.rs
index 7a46393804..cc7d292b22 100644
--- a/tests/it/io/ipc/write/file.rs
+++ b/tests/it/io/ipc/write/file.rs
@@ -36,7 +36,7 @@ fn round_trip(batch: RecordBatch) -> Result<()> {
 }
 
 fn test_file(version: &str, file_name: &str) -> Result<()> {
-    let (schema, batches) = read_gzip_json(version, file_name);
+    let (schema, batches) = read_gzip_json(version, file_name)?;
 
     let mut result = Vec::<u8>::new();
 
@@ -56,7 +56,7 @@ fn test_file(version: &str, file_name: &str) -> Result<()> {
     let reader = FileReader::new(&mut reader, metadata, None);
 
     // read expected JSON output
-    let (expected_schema, expected_batches) = read_gzip_json(version, file_name);
+    let (expected_schema, expected_batches) = read_gzip_json(version, file_name)?;
 
     assert_eq!(schema.as_ref(), &expected_schema);
 
diff --git a/tests/it/io/ipc/write/stream.rs b/tests/it/io/ipc/write/stream.rs
index e9019d01f9..16a4153071 100644
--- a/tests/it/io/ipc/write/stream.rs
+++ b/tests/it/io/ipc/write/stream.rs
@@ -30,7 +30,7 @@ fn test_file(version: &str, file_name: &str) {
     let schema = reader.schema().clone();
 
     // read expected JSON output
-    let (expected_schema, expected_batches) = read_gzip_json(version, file_name);
+    let (expected_schema, expected_batches) = read_gzip_json(version, file_name).unwrap();
 
     assert_eq!(schema.as_ref(), &expected_schema);
 
diff --git a/tests/it/io/json/mod.rs b/tests/it/io/json/mod.rs
index b8bc9a5f7b..1bf3d0856c 100644
--- a/tests/it/io/json/mod.rs
+++ b/tests/it/io/json/mod.rs
@@ -212,11 +212,11 @@ fn case_struct() -> (String, Schema, Vec<Box<dyn Array>>) {
 
     // build expected output
     let d = Utf8Array::<i32>::from(&vec![Some("text"), None, Some("text"), None]);
-    let c = StructArray::from_data(vec![d_field], vec![Arc::new(d)], None);
+    let c = StructArray::from_data(DataType::Struct(vec![d_field]), vec![Arc::new(d)], None);
 
     let b = BooleanArray::from(vec![Some(true), Some(false), Some(true), None]);
     let expected = StructArray::from_data(
-        vec![Field::new("b", DataType::Boolean, true), c_field],
+        DataType::Struct(vec![Field::new("b", DataType::Boolean, true), c_field]),
         vec![Arc::new(b), Arc::new(c)],
         None,
     );
diff --git a/tests/it/io/json/read.rs b/tests/it/io/json/read.rs
index 89dc10e39b..4a7c0b0d3d 100644
--- a/tests/it/io/json/read.rs
+++ b/tests/it/io/json/read.rs
@@ -154,7 +154,7 @@ fn nested_list_arrays() {
         None,
     ]);
 
-    let c = StructArray::from_data(vec![d_field], vec![Arc::new(d)], None);
+    let c = StructArray::from_data(DataType::Struct(vec![d_field]), vec![Arc::new(d)], None);
 
     let b = BooleanArray::from(vec![
         Some(true),
@@ -165,7 +165,7 @@ fn nested_list_arrays() {
         Some(true),
     ]);
     let a_struct = StructArray::from_data(
-        vec![b_field, c_field],
+        DataType::Struct(vec![b_field, c_field]),
         vec![Arc::new(b) as Arc<dyn Array>, Arc::new(c) as Arc<dyn Array>],
         None,
     );
diff --git a/tests/it/io/json/write.rs b/tests/it/io/json/write.rs
index 13337ec7bd..93c6fb27f8 100644
--- a/tests/it/io/json/write.rs
+++ b/tests/it/io/json/write.rs
@@ -51,11 +51,11 @@ fn write_nested_structs() {
     ]);
 
     let c1 = StructArray::from_data(
-        fields,
+        DataType::Struct(fields),
         vec![
             Arc::new(Int32Array::from(&[Some(1), None, Some(5)])),
             Arc::new(StructArray::from_data(
-                vec![c121],
+                DataType::Struct(vec![c121]),
                 vec![Arc::new(Utf8Array::<i32>::from(&vec![
                     Some("e"),
                     Some("f"),
@@ -192,11 +192,11 @@ fn write_list_of_struct() {
     let schema = Schema::new(vec![field_c1, field_c2]);
 
     let s = StructArray::from_data(
-        fields,
+        DataType::Struct(fields),
         vec![
             Arc::new(Int32Array::from(&[Some(1), None, Some(5)])),
             Arc::new(StructArray::from_data(
-                inner,
+               DataType::Struct(inner),
                 vec![Arc::new(Utf8Array::<i32>::from(&vec![
                     Some("e"),
                     Some("f"),
diff --git a/tests/it/io/parquet/mod.rs b/tests/it/io/parquet/mod.rs
index 2810aaf467..31964d6d4a 100644
--- a/tests/it/io/parquet/mod.rs
+++ b/tests/it/io/parquet/mod.rs
@@ -440,7 +440,7 @@ fn integration_read(data: &[u8]) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
 }
 
 fn test_file(version: &str, file_name: &str) -> Result<()> {
-    let (schema, batches) = read_gzip_json(version, file_name);
+    let (schema, batches) = read_gzip_json(version, file_name)?;
 
     let data = integration_write(&schema, &batches)?;
 
diff --git a/tests/it/io/print.rs b/tests/it/io/print.rs
index a2ee1b97ef..040e7c4b55 100644
--- a/tests/it/io/print.rs
+++ b/tests/it/io/print.rs
@@ -360,7 +360,7 @@ fn write_struct() -> Result<()> {
 
     let validity = Some(Bitmap::from(&[true, false, true]));
 
-    let array = StructArray::from_data(fields, values, validity);
+    let array = StructArray::from_data(DataType::Struct(fields), values, validity);
 
     let schema = Schema::new(vec![Field::new("a", array.data_type().clone(), true)]);