Repository: jorgecarleitao/arrow2
Pull request #1208 (2022-08-04 19:30:22)

Simpler IPC API
This PR removes from the public API internal methods and exposes them with more intuitive APIs.

This reduces the mental load of the IPC API public surface to users, thereby allowing them to more easily use the crate.

It also makes it easier to read from `flight`, as all utilities to do so are now available under `io::flight`.


codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#1208](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (931c7ac) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/497d4310ede92cc8acd223fcedce0ef2763e09eb?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (497d431) will **decrease** coverage by `0.16%`.
> The diff coverage is `71.85%`.

> :exclamation: Current head 931c7ac differs from pull request most recent head ffcc0c2. Consider uploading reports for the commit ffcc0c2 to get more accurate results

```diff
@@            Coverage Diff             @@
##             main    #1208      +/-   ##
==========================================
- Coverage   83.49%   83.32%   -0.17%     
==========================================
  Files         356      357       +1     
  Lines       37081    37159      +78     
==========================================
+ Hits        30961    30964       +3     
- Misses       6120     6195      +75     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/io/flight/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2ZsaWdodC9tb2QucnM=) | `37.64% <0.00%> (-29.36%)` | :arrow_down: |
| [src/io/ipc/read/file\_async.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL2ZpbGVfYXN5bmMucnM=) | `61.19% <ø> (ø)` | |
| [src/io/ipc/read/reader.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL3JlYWRlci5ycw==) | `94.66% <ø> (-1.32%)` | :arrow_down: |
| [src/mmap/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL21tYXAvbW9kLnJz) | `84.84% <ø> (ø)` | |
| [src/io/ipc/read/file.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL2ZpbGUucnM=) | `96.87% <96.87%> (ø)` | |
| [src/io/ipc/read/array/boolean.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL2FycmF5L2Jvb2xlYW4ucnM=) | `90.74% <0.00%> (-7.41%)` | :arrow_down: |
| [src/io/ipc/read/schema.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL3NjaGVtYS5ycw==) | `95.81% <0.00%> (+0.32%)` | :arrow_up: |
| [src/bitmap/immutable.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2JpdG1hcC9pbW11dGFibGUucnM=) | `86.41% <0.00%> (+0.61%)` | :arrow_up: |
| [src/bitmap/utils/slice\_iterator.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2JpdG1hcC91dGlscy9zbGljZV9pdGVyYXRvci5ycw==) | `98.78% <0.00%> (+1.21%)` | :arrow_up: |
| [src/array/binary/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1208/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2JpbmFyeS9tb2QucnM=) | `90.12% <0.00%> (+1.23%)` | :arrow_up: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)


Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 791d758ea2..7b620163ac 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -47,7 +47,7 @@ regex-syntax = { version = "^0.6", optional = true }
 streaming-iterator = { version = "0.1", optional = true }
 fallible-streaming-iterator = { version = "0.1", optional = true }
 
-json-deserializer = { version = "0.3", optional = true }
+json-deserializer = { version = "0.4", optional = true, features = ["preserve_order"] }
 indexmap = { version = "^1.6", optional = true }
 
 # used to print columns in a nice columnar format
@@ -155,13 +155,23 @@ io_flight = ["io_ipc", "arrow-format/flight-data"]
 
 # base64 + io_ipc because arrow schemas are stored as base64-encoded ipc format.
 io_parquet = ["parquet2", "io_ipc", "base64", "futures", "streaming-iterator", "fallible-streaming-iterator"]
+
 io_parquet_compression = [
-    "parquet2/zstd",
-    "parquet2/snappy",
-    "parquet2/gzip",
-    "parquet2/lz4",
-    "parquet2/brotli",
+    "io_parquet_zstd",
+    "io_parquet_gzip",
+    "io_parquet_snappy",
+    "io_parquet_lz4",
+    "io_parquet_brotli"
 ]
+
+# compression backends
+io_parquet_zstd = ["parquet2/zstd"]
+io_parquet_snappy = ["parquet2/snappy"]
+io_parquet_gzip = ["parquet2/gzip"]
+io_parquet_lz4_flex = ["parquet2/lz4_flex"]
+io_parquet_lz4 = ["parquet2/lz4"]
+io_parquet_brotli = ["parquet2/brotli"]
+
 io_avro = ["avro-schema", "streaming-iterator"]
 io_avro_compression = [
     "avro-schema/compression",
diff --git a/src/array/binary/mod.rs b/src/array/binary/mod.rs
index af06d19501..b1efa2f6f2 100644
--- a/src/array/binary/mod.rs
+++ b/src/array/binary/mod.rs
@@ -157,13 +157,13 @@ impl<O: Offset> BinaryArray<O> {
         &self.data_type
     }
 
-    /// Returns the values of this [`Utf8Array`].
+    /// Returns the values of this [`BinaryArray`].
     #[inline]
     pub fn values(&self) -> &Buffer<u8> {
         &self.values
     }
 
-    /// Returns the offsets of this [`Utf8Array`].
+    /// Returns the offsets of this [`BinaryArray`].
     #[inline]
     pub fn offsets(&self) -> &Buffer<O> {
         &self.offsets
diff --git a/src/array/fixed_size_list/mod.rs b/src/array/fixed_size_list/mod.rs
index d61a64c258..dac5f19c70 100644
--- a/src/array/fixed_size_list/mod.rs
+++ b/src/array/fixed_size_list/mod.rs
@@ -3,6 +3,7 @@ use crate::{
     datatypes::{DataType, Field},
     error::Error,
 };
+use std::sync::Arc;
 
 use super::{new_empty_array, new_null_array, Array};
 
@@ -118,8 +119,8 @@ impl FixedSizeListArray {
     }
 
     /// Boxes self into a [`Arc<dyn Array>`].
-    pub fn arced(self) -> std::sync::Arc<dyn Array> {
-        std::sync::Arc::new(self)
+    pub fn arced(self) -> Arc<dyn Array> {
+        Arc::new(self)
     }
 }
 
diff --git a/src/array/growable/mod.rs b/src/array/growable/mod.rs
index 398c429731..eb79f7fed9 100644
--- a/src/array/growable/mod.rs
+++ b/src/array/growable/mod.rs
@@ -3,6 +3,7 @@
 
 use crate::array::*;
 use crate::datatypes::*;
+use std::sync::Arc;
 
 mod binary;
 pub use binary::GrowableBinary;
@@ -44,7 +45,7 @@ pub trait Growable<'a> {
 
     /// Converts this [`Growable`] to an [`Arc<dyn Array>`], thereby finishing the mutation.
     /// Self will be empty after such operation.
-    fn as_arc(&mut self) -> std::sync::Arc<dyn Array> {
+    fn as_arc(&mut self) -> Arc<dyn Array> {
         self.as_box().into()
     }
 
diff --git a/src/array/list/mod.rs b/src/array/list/mod.rs
index 13f781f680..819f6b1dd2 100644
--- a/src/array/list/mod.rs
+++ b/src/array/list/mod.rs
@@ -4,6 +4,7 @@ use crate::{
     datatypes::{DataType, Field},
     error::Error,
 };
+use std::sync::Arc;
 
 use super::{
     new_empty_array,
@@ -126,8 +127,8 @@ impl<O: Offset> ListArray<O> {
     }
 
     /// Boxes self into a [`Arc<dyn Array>`].
-    pub fn arced(self) -> std::sync::Arc<dyn Array> {
-        std::sync::Arc::new(self)
+    pub fn arced(self) -> Arc<dyn Array> {
+        Arc::new(self)
     }
 }
 
diff --git a/src/array/list/mutable.rs b/src/array/list/mutable.rs
index 8d0dc3b05e..71283104fb 100644
--- a/src/array/list/mutable.rs
+++ b/src/array/list/mutable.rs
@@ -157,7 +157,7 @@ impl<O: Offset, M: MutableArray> MutableListArray<O, M> {
         &mut self.values
     }
 
-    /// The offseta
+    /// The offsets
     pub fn offsets(&self) -> &Vec<O> {
         &self.offsets
     }
diff --git a/src/array/mod.rs b/src/array/mod.rs
index 8010b5cf92..951e14d71a 100644
--- a/src/array/mod.rs
+++ b/src/array/mod.rs
@@ -132,7 +132,7 @@ pub trait MutableArray: std::fmt::Debug + Send + Sync {
     /// The optional validity of the array.
     fn validity(&self) -> Option<&MutableBitmap>;
 
-    /// Convert itself to an (immutable) ['Array'].
+    /// Convert itself to an (immutable) [`Array`].
     fn as_box(&mut self) -> Box<dyn Array>;
 
     /// Convert itself to an (immutable) atomically reference counted [`Array`].
@@ -391,7 +391,7 @@ pub use list::{ListArray, ListValuesIter, MutableListArray};
 pub use map::MapArray;
 pub use null::NullArray;
 pub use primitive::*;
-pub use struct_::StructArray;
+pub use struct_::{MutableStructArray, StructArray};
 pub use union::UnionArray;
 pub use utf8::{MutableUtf8Array, Utf8Array, Utf8ValuesIter};
 
diff --git a/src/array/struct_/mod.rs b/src/array/struct_/mod.rs
index a7c88b6fb1..3d25d8c188 100644
--- a/src/array/struct_/mod.rs
+++ b/src/array/struct_/mod.rs
@@ -9,6 +9,8 @@ use super::{new_empty_array, new_null_array, Array};
 mod ffi;
 pub(super) mod fmt;
 mod iterator;
+mod mutable;
+pub use mutable::*;
 
 /// A [`StructArray`] is a nested [`Array`] with an optional validity representing
 /// multiple [`Array`] with the same number of rows.
diff --git a/src/array/struct_/mutable.rs b/src/array/struct_/mutable.rs
new file mode 100644
index 0000000000..cc02146a77
--- /dev/null
+++ b/src/array/struct_/mutable.rs
@@ -0,0 +1,216 @@
+use std::sync::Arc;
+
+use crate::{
+    array::{Array, MutableArray},
+    bitmap::MutableBitmap,
+    datatypes::DataType,
+};
+
+use super::StructArray;
+
+/// Converting a [`MutableStructArray`] into a [`StructArray`] is `O(1)`.
+#[derive(Debug)]
+pub struct MutableStructArray {
+    data_type: DataType,
+    values: Vec<Box<dyn MutableArray>>,
+    validity: Option<MutableBitmap>,
+}
+
+impl From<MutableStructArray> for StructArray {
+    fn from(other: MutableStructArray) -> Self {
+        let validity = if other.validity.as_ref().map(|x| x.unset_bits()).unwrap_or(0) > 0 {
+            other.validity.map(|x| x.into())
+        } else {
+            None
+        };
+
+        StructArray::from_data(
+            other.data_type,
+            other.values.into_iter().map(|mut v| v.as_box()).collect(),
+            validity,
+        )
+    }
+}
+
+impl MutableStructArray {
+    /// Creates a new [`MutableStructArray`].
+    pub fn new(data_type: DataType, values: Vec<Box<dyn MutableArray>>) -> Self {
+        Self::from_data(data_type, values, None)
+    }
+
+    /// Create a [`MutableStructArray`] out of low-end APIs.
+    /// # Panics
+    /// This function panics iff:
+    /// * `data_type` is not [`DataType::Struct`]
+    /// * The inner types of `data_type` are not equal to those of `values`
+    /// * `validity` is not `None` and its length is different from the `values`'s length
+    pub fn from_data(
+        data_type: DataType,
+        values: Vec<Box<dyn MutableArray>>,
+        validity: Option<MutableBitmap>,
+    ) -> Self {
+        match data_type.to_logical_type() {
+            DataType::Struct(ref fields) => assert!(fields
+                .iter()
+                .map(|f| f.data_type())
+                .eq(values.iter().map(|f| f.data_type()))),
+            _ => panic!("StructArray must be initialized with DataType::Struct"),
+        };
+        let self_ = Self {
+            data_type,
+            values,
+            validity,
+        };
+        self_.assert_lengths();
+        self_
+    }
+
+    fn assert_lengths(&self) {
+        let first_len = self.values.first().map(|v| v.len());
+        if let Some(len) = first_len {
+            if !self.values.iter().all(|x| x.len() == len) {
+                let lengths: Vec<_> = self.values.iter().map(|v| v.len()).collect();
+                panic!("StructArray child lengths differ: {:?}", lengths);
+            }
+        }
+        if let Some(validity) = &self.validity {
+            assert_eq!(first_len.unwrap_or(0), validity.len());
+        }
+    }
+
+    /// Extract the low-end APIs from the [`MutableStructArray`].
+    pub fn into_data(self) -> (DataType, Vec<Box<dyn MutableArray>>, Option<MutableBitmap>) {
+        (self.data_type, self.values, self.validity)
+    }
+
+    /// The mutable values
+    pub fn mut_values(&mut self) -> &mut Vec<Box<dyn MutableArray>> {
+        &mut self.values
+    }
+
+    /// The values
+    pub fn values(&self) -> &Vec<Box<dyn MutableArray>> {
+        &self.values
+    }
+
+    /// Return the `i`th child array.
+    pub fn value<A: MutableArray + 'static>(&mut self, i: usize) -> Option<&mut A> {
+        self.values[i].as_mut_any().downcast_mut::<A>()
+    }
+}
+
+impl MutableStructArray {
+    /// Reserves `additional` entries.
+    pub fn reserve(&mut self, additional: usize) {
+        for v in &mut self.values {
+            v.reserve(additional);
+        }
+        if let Some(x) = self.validity.as_mut() {
+            x.reserve(additional)
+        }
+    }
+
+    /// Call this after pushing into each child array.
+    /// # Panics
+    /// This function panics if any of the children does not have exactly one more
+    /// element than before the last call of `push`.
+    pub fn push(&mut self, valid: bool) {
+        match &mut self.validity {
+            Some(validity) => validity.push(valid),
+            None => match valid {
+                true => (),
+                false => self.init_validity(),
+            },
+        };
+        self.assert_lengths();
+    }
+
+    fn push_null(&mut self) {
+        for v in &mut self.values {
+            v.push_null();
+        }
+        self.push(false);
+    }
+
+    fn init_validity(&mut self) {
+        let mut validity = MutableBitmap::with_capacity(self.values.capacity());
+        let len = self.len();
+        if len > 0 {
+            validity.extend_constant(len, true);
+            validity.set(len - 1, false);
+        }
+        self.validity = Some(validity)
+    }
+
+    /// Converts itself into an [`Array`].
+    pub fn into_arc(self) -> Arc<dyn Array> {
+        let a: StructArray = self.into();
+        Arc::new(a)
+    }
+
+    /// Shrinks the capacity of the [`MutableStructArray`] to fit its current length.
+    pub fn shrink_to_fit(&mut self) {
+        for v in &mut self.values {
+            v.shrink_to_fit();
+        }
+        if let Some(validity) = self.validity.as_mut() {
+            validity.shrink_to_fit()
+        }
+    }
+}
+
+impl MutableArray for MutableStructArray {
+    fn len(&self) -> usize {
+        self.values.first().map(|v| v.len()).unwrap_or(0)
+    }
+
+    fn validity(&self) -> Option<&MutableBitmap> {
+        self.validity.as_ref()
+    }
+
+    fn as_box(&mut self) -> Box<dyn Array> {
+        Box::new(StructArray::from_data(
+            self.data_type.clone(),
+            std::mem::take(&mut self.values)
+                .into_iter()
+                .map(|mut v| v.as_box())
+                .collect(),
+            std::mem::take(&mut self.validity).map(|x| x.into()),
+        ))
+    }
+
+    fn as_arc(&mut self) -> Arc<dyn Array> {
+        Arc::new(StructArray::from_data(
+            self.data_type.clone(),
+            std::mem::take(&mut self.values)
+                .into_iter()
+                .map(|mut v| v.as_box())
+                .collect(),
+            std::mem::take(&mut self.validity).map(|x| x.into()),
+        ))
+    }
+
+    fn data_type(&self) -> &DataType {
+        &self.data_type
+    }
+
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn as_mut_any(&mut self) -> &mut dyn std::any::Any {
+        self
+    }
+
+    fn push_null(&mut self) {
+        self.push_null()
+    }
+
+    fn shrink_to_fit(&mut self) {
+        self.shrink_to_fit()
+    }
+
+    fn reserve(&mut self, additional: usize) {
+        self.reserve(additional)
+    }
+}
diff --git a/src/io/flight/mod.rs b/src/io/flight/mod.rs
index 4072c0d0d0..17f7c20d88 100644
--- a/src/io/flight/mod.rs
+++ b/src/io/flight/mod.rs
@@ -14,6 +14,7 @@ use crate::{
     io::ipc::write::common::{encode_chunk, DictionaryTracker, EncodedData},
 };
 
+use super::ipc::read::Dictionaries;
 use super::ipc::{IpcField, IpcSchema};
 
 pub use super::ipc::write::default_ipc_fields;
@@ -114,7 +115,7 @@ pub fn deserialize_schemas(bytes: &[u8]) -> Result<(Schema, IpcSchema)> {
     read::deserialize_schema(bytes)
 }
 
-/// Deserializes [`FlightData`] to [`Chunk`].
+/// Deserializes [`FlightData`] representing a record batch message to [`Chunk`].
 pub fn deserialize_batch(
     data: &FlightData,
     fields: &[Field],
@@ -149,3 +150,102 @@ pub fn deserialize_batch(
         )),
     }
 }
+
+/// Deserializes [`FlightData`], assuming it to be a dictionary message, into `dictionaries`.
+pub fn deserialize_dictionary(
+    data: &FlightData,
+    fields: &[Field],
+    ipc_schema: &IpcSchema,
+    dictionaries: &mut read::Dictionaries,
+) -> Result<()> {
+    let message = ipc::MessageRef::read_as_root(&data.data_header)?;
+
+    let chunk = if let ipc::MessageHeaderRef::DictionaryBatch(chunk) = message
+        .header()?
+        .ok_or_else(|| Error::oos("Header is required"))?
+    {
+        chunk
+    } else {
+        return Ok(());
+    };
+
+    let length = data.data_body.len();
+    let mut reader = std::io::Cursor::new(&data.data_body);
+    read::read_dictionary(
+        chunk,
+        fields,
+        ipc_schema,
+        dictionaries,
+        &mut reader,
+        0,
+        length as u64,
+        &mut Default::default(),
+    )?;
+
+    Ok(())
+}
+
+/// Deserializes [`FlightData`] into either a [`Chunk`] (when the message is a record batch)
+/// or by upserting into `dictionaries` (when the message is a dictionary)
+pub fn deserialize_message(
+    data: &FlightData,
+    fields: &[Field],
+    ipc_schema: &IpcSchema,
+    dictionaries: &mut Dictionaries,
+) -> Result<Option<Chunk<Box<dyn Array>>>> {
+    let FlightData {
+        data_header,
+        data_body,
+        ..
+    } = data;
+
+    let message = arrow_format::ipc::MessageRef::read_as_root(data_header)?;
+    let header = message
+        .header()?
+        .ok_or_else(|| Error::oos("IPC Message must contain a header"))?;
+
+    match header {
+        ipc::MessageHeaderRef::RecordBatch(batch) => {
+            let length = data_body.len();
+            let mut reader = std::io::Cursor::new(data_body);
+
+            let chunk = read::read_record_batch(
+                batch,
+                fields,
+                ipc_schema,
+                None,
+                None,
+                dictionaries,
+                arrow_format::ipc::MetadataVersion::V5,
+                &mut reader,
+                0,
+                length as u64,
+                &mut Default::default(),
+            )?;
+
+            Ok(chunk.into())
+        }
+        ipc::MessageHeaderRef::DictionaryBatch(dict_batch) => {
+            let length = data_body.len();
+            let mut reader = std::io::Cursor::new(data_body);
+
+            read::read_dictionary(
+                dict_batch,
+                fields,
+                ipc_schema,
+                dictionaries,
+                &mut reader,
+                0,
+                length as u64,
+                &mut Default::default(),
+            )?;
+            Ok(None)
+        }
+        t => {
+            return Err(Error::nyi(format!(
+                "Reading types other than record batches not yet supported, unable to read {:?}",
+                t
+            )));
+        }
+    }
+}
diff --git a/src/io/ipc/read/file.rs b/src/io/ipc/read/file.rs
new file mode 100644
index 0000000000..ad8dc34811
--- /dev/null
+++ b/src/io/ipc/read/file.rs
@@ -0,0 +1,319 @@
+use ahash::AHashMap;
+use std::convert::TryInto;
+use std::io::{Read, Seek, SeekFrom};
+
+use crate::array::Array;
+use crate::chunk::Chunk;
+use crate::datatypes::Schema;
+use crate::error::{Error, Result};
+use crate::io::ipc::IpcSchema;
+
+use super::super::{ARROW_MAGIC, CONTINUATION_MARKER};
+use super::common::*;
+use super::schema::fb_to_schema;
+use super::Dictionaries;
+use super::OutOfSpecKind;
+use arrow_format::ipc::planus::ReadAsRoot;
+
+/// Metadata of an Arrow IPC file, written in the footer of the file.
+#[derive(Debug, Clone)]
+pub struct FileMetadata {
+    /// The schema that is read from the file footer
+    pub schema: Schema,
+
+    /// The files' [`IpcSchema`]
+    pub ipc_schema: IpcSchema,
+
+    /// The blocks in the file
+    ///
+    /// A block indicates the regions in the file to read to get data
+    pub blocks: Vec<arrow_format::ipc::Block>,
+
+    /// Dictionaries associated to each dict_id
+    pub(crate) dictionaries: Option<Vec<arrow_format::ipc::Block>>,
+
+    /// The total size of the file in bytes
+    pub size: u64,
+}
+
+fn read_dictionary_message<R: Read + Seek>(
+    reader: &mut R,
+    offset: u64,
+    data: &mut Vec<u8>,
+) -> Result<()> {
+    let mut message_size: [u8; 4] = [0; 4];
+    reader.seek(SeekFrom::Start(offset))?;
+    reader.read_exact(&mut message_size)?;
+    if message_size == CONTINUATION_MARKER {
+        reader.read_exact(&mut message_size)?;
+    };
+    let message_length = i32::from_le_bytes(message_size);
+
+    let message_length: usize = message_length
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    data.clear();
+    data.try_reserve(message_length)?;
+    reader
+        .by_ref()
+        .take(message_length as u64)
+        .read_to_end(data)?;
+
+    Ok(())
+}
+
+pub(crate) fn get_dictionary_batch<'a>(
+    message: &'a arrow_format::ipc::MessageRef,
+) -> Result<arrow_format::ipc::DictionaryBatchRef<'a>> {
+    let header = message
+        .header()
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferHeader(err)))?
+        .ok_or_else(|| Error::from(OutOfSpecKind::MissingMessageHeader))?;
+    match header {
+        arrow_format::ipc::MessageHeaderRef::DictionaryBatch(batch) => Ok(batch),
+        _ => Err(Error::from(OutOfSpecKind::UnexpectedMessageType)),
+    }
+}
+
+fn read_dictionary_block<R: Read + Seek>(
+    reader: &mut R,
+    metadata: &FileMetadata,
+    block: &arrow_format::ipc::Block,
+    dictionaries: &mut Dictionaries,
+    message_scratch: &mut Vec<u8>,
+    dictionary_scratch: &mut Vec<u8>,
+) -> Result<()> {
+    let offset: u64 = block
+        .offset
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
+    let length: u64 = block
+        .meta_data_length
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
+    read_dictionary_message(reader, offset, message_scratch)?;
+
+    let message = arrow_format::ipc::MessageRef::read_as_root(message_scratch.as_ref())
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
+
+    let batch = get_dictionary_batch(&message)?;
+
+    read_dictionary(
+        batch,
+        &metadata.schema.fields,
+        &metadata.ipc_schema,
+        dictionaries,
+        reader,
+        offset + length,
+        metadata.size,
+        dictionary_scratch,
+    )
+}
+
+/// Reads all file's dictionaries, if any
+/// This function is IO-bounded
+pub fn read_file_dictionaries<R: Read + Seek>(
+    reader: &mut R,
+    metadata: &FileMetadata,
+    scratch: &mut Vec<u8>,
+) -> Result<Dictionaries> {
+    let mut dictionaries = Default::default();
+
+    let blocks = if let Some(blocks) = &metadata.dictionaries {
+        blocks
+    } else {
+        return Ok(AHashMap::new());
+    };
+    // use a temporary smaller scratch for the messages
+    let mut message_scratch = Default::default();
+
+    for block in blocks {
+        read_dictionary_block(
+            reader,
+            metadata,
+            block,
+            &mut dictionaries,
+            &mut message_scratch,
+            scratch,
+        )?;
+    }
+    Ok(dictionaries)
+}
+
+/// Reads the footer's length and magic number in footer
+fn read_footer_len<R: Read + Seek>(reader: &mut R) -> Result<(u64, usize)> {
+    // read footer length and magic number in footer
+    let end = reader.seek(SeekFrom::End(-10))? + 10;
+
+    let mut footer: [u8; 10] = [0; 10];
+
+    reader.read_exact(&mut footer)?;
+    let footer_len = i32::from_le_bytes(footer[..4].try_into().unwrap());
+
+    if footer[4..] != ARROW_MAGIC {
+        return Err(Error::from(OutOfSpecKind::InvalidFooter));
+    }
+    let footer_len = footer_len
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    Ok((end, footer_len))
+}
+
+pub(super) fn deserialize_footer(footer_data: &[u8], size: u64) -> Result<FileMetadata> {
+    let footer = arrow_format::ipc::FooterRef::read_as_root(footer_data)
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferFooter(err)))?;
+
+    let blocks = footer
+        .record_batches()
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferRecordBatches(err)))?
+        .ok_or_else(|| Error::from(OutOfSpecKind::MissingRecordBatches))?;
+
+    let blocks = blocks
+        .iter()
+        .map(|block| {
+            block
+                .try_into()
+                .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferRecordBatches(err)))
+        })
+        .collect::<Result<Vec<_>>>()?;
+
+    let ipc_schema = footer
+        .schema()
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferSchema(err)))?
+        .ok_or_else(|| Error::from(OutOfSpecKind::MissingSchema))?;
+    let (schema, ipc_schema) = fb_to_schema(ipc_schema)?;
+
+    let dictionaries = footer
+        .dictionaries()
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferDictionaries(err)))?
+        .map(|dictionaries| {
+            dictionaries
+                .into_iter()
+                .map(|block| {
+                    block.try_into().map_err(|err| {
+                        Error::from(OutOfSpecKind::InvalidFlatbufferRecordBatches(err))
+                    })
+                })
+                .collect::<Result<Vec<_>>>()
+        })
+        .transpose()?;
+
+    Ok(FileMetadata {
+        schema,
+        ipc_schema,
+        blocks,
+        dictionaries,
+        size,
+    })
+}
+
+/// Read the Arrow IPC file's metadata
+pub fn read_file_metadata<R: Read + Seek>(reader: &mut R) -> Result<FileMetadata> {
+    // check if header contain the correct magic bytes
+    let mut magic_buffer: [u8; 6] = [0; 6];
+    let start = reader.seek(SeekFrom::Current(0))?;
+    reader.read_exact(&mut magic_buffer)?;
+    if magic_buffer != ARROW_MAGIC {
+        return Err(Error::from(OutOfSpecKind::InvalidHeader));
+    }
+
+    let (end, footer_len) = read_footer_len(reader)?;
+
+    // read footer
+    reader.seek(SeekFrom::End(-10 - footer_len as i64))?;
+
+    let mut serialized_footer = vec![];
+    serialized_footer.try_reserve(footer_len)?;
+    reader
+        .by_ref()
+        .take(footer_len as u64)
+        .read_to_end(&mut serialized_footer)?;
+
+    deserialize_footer(&serialized_footer, end - start)
+}
+
+pub(crate) fn get_record_batch(
+    message: arrow_format::ipc::MessageRef,
+) -> Result<arrow_format::ipc::RecordBatchRef> {
+    let header = message
+        .header()
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferHeader(err)))?
+        .ok_or_else(|| Error::from(OutOfSpecKind::MissingMessageHeader))?;
+    match header {
+        arrow_format::ipc::MessageHeaderRef::RecordBatch(batch) => Ok(batch),
+        _ => Err(Error::from(OutOfSpecKind::UnexpectedMessageType)),
+    }
+}
+
+/// Reads the record batch at position `index` from the reader.
+///
+/// This function is useful for random access to the file. For example, if
+/// you have indexed the file somewhere else, this allows pruning
+/// certain parts of the file.
+/// # Panics
+/// This function panics iff `index >= metadata.blocks.len()`
+#[allow(clippy::too_many_arguments)]
+pub fn read_batch<R: Read + Seek>(
+    reader: &mut R,
+    dictionaries: &Dictionaries,
+    metadata: &FileMetadata,
+    projection: Option<&[usize]>,
+    limit: Option<usize>,
+    index: usize,
+    message_scratch: &mut Vec<u8>,
+    data_scratch: &mut Vec<u8>,
+) -> Result<Chunk<Box<dyn Array>>> {
+    let block = metadata.blocks[index];
+
+    let offset: u64 = block
+        .offset
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let length: u64 = block
+        .meta_data_length
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    // read length
+    reader.seek(SeekFrom::Start(offset))?;
+    let mut meta_buf = [0; 4];
+    reader.read_exact(&mut meta_buf)?;
+    if meta_buf == CONTINUATION_MARKER {
+        // continuation marker encountered, read message next
+        reader.read_exact(&mut meta_buf)?;
+    }
+    let meta_len = i32::from_le_bytes(meta_buf)
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
+
+    message_scratch.clear();
+    message_scratch.try_reserve(meta_len)?;
+    reader
+        .by_ref()
+        .take(meta_len as u64)
+        .read_to_end(message_scratch)?;
+
+    let message = arrow_format::ipc::MessageRef::read_as_root(message_scratch.as_ref())
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
+
+    let batch = get_record_batch(message)?;
+
+    read_record_batch(
+        batch,
+        &metadata.schema.fields,
+        &metadata.ipc_schema,
+        projection,
+        limit,
+        dictionaries,
+        message
+            .version()
+            .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferVersion(err)))?,
+        reader,
+        offset + length,
+        metadata.size,
+        data_scratch,
+    )
+}
diff --git a/src/io/ipc/read/file_async.rs b/src/io/ipc/read/file_async.rs
index 82999c28ea..bc74890cc2 100644
--- a/src/io/ipc/read/file_async.rs
+++ b/src/io/ipc/read/file_async.rs
@@ -14,7 +14,7 @@ use crate::error::{Error, Result};
 use crate::io::ipc::{IpcSchema, ARROW_MAGIC, CONTINUATION_MARKER};
 
 use super::common::{apply_projection, prepare_projection, read_dictionary, read_record_batch};
-use super::reader::{deserialize_footer, get_record_batch};
+use super::file::{deserialize_footer, get_record_batch};
 use super::Dictionaries;
 use super::FileMetadata;
 use super::OutOfSpecKind;
diff --git a/src/io/ipc/read/mod.rs b/src/io/ipc/read/mod.rs
index 2645d91440..88f1deaf3a 100644
--- a/src/io/ipc/read/mod.rs
+++ b/src/io/ipc/read/mod.rs
@@ -12,8 +12,9 @@ mod array;
 mod common;
 mod deserialize;
 mod error;
+pub(crate) mod file;
 mod read_basic;
-pub(crate) mod reader;
+mod reader;
 mod schema;
 mod stream;
 
@@ -28,11 +29,10 @@ pub mod stream_async;
 pub mod file_async;
 
 pub(crate) use common::first_dict_field;
-pub use common::{read_dictionary, read_record_batch};
-pub use reader::{
-    read_batch, read_file_dictionaries, read_file_metadata, FileMetadata, FileReader,
-};
-pub use schema::deserialize_schema;
+pub(crate) use common::{read_dictionary, read_record_batch};
+pub use file::{read_batch, read_file_dictionaries, read_file_metadata, FileMetadata};
+pub use reader::FileReader;
+pub(crate) use schema::deserialize_schema;
 pub use stream::{read_stream_metadata, StreamMetadata, StreamReader, StreamState};
 
 /// how dictionaries are tracked in this crate
diff --git a/src/io/ipc/read/reader.rs b/src/io/ipc/read/reader.rs
index 37b8437c27..da584fadee 100644
--- a/src/io/ipc/read/reader.rs
+++ b/src/io/ipc/read/reader.rs
@@ -1,322 +1,14 @@
 use ahash::AHashMap;
-use std::convert::TryInto;
-use std::io::{Read, Seek, SeekFrom};
+use std::io::{Read, Seek};
 
 use crate::array::Array;
 use crate::chunk::Chunk;
 use crate::datatypes::Schema;
-use crate::error::{Error, Result};
-use crate::io::ipc::IpcSchema;
+use crate::error::Result;
 
-use super::super::{ARROW_MAGIC, CONTINUATION_MARKER};
 use super::common::*;
-use super::schema::fb_to_schema;
 use super::Dictionaries;
-use super::OutOfSpecKind;
-use arrow_format::ipc::planus::ReadAsRoot;
-
-/// Metadata of an Arrow IPC file, written in the footer of the file.
-#[derive(Debug, Clone)]
-pub struct FileMetadata {
-    /// The schema that is read from the file footer
-    pub schema: Schema,
-
-    /// The files' [`IpcSchema`]
-    pub ipc_schema: IpcSchema,
-
-    /// The blocks in the file
-    ///
-    /// A block indicates the regions in the file to read to get data
-    pub blocks: Vec<arrow_format::ipc::Block>,
-
-    /// Dictionaries associated to each dict_id
-    pub(crate) dictionaries: Option<Vec<arrow_format::ipc::Block>>,
-
-    /// The total size of the file in bytes
-    pub size: u64,
-}
-
-fn read_dictionary_message<R: Read + Seek>(
-    reader: &mut R,
-    offset: u64,
-    data: &mut Vec<u8>,
-) -> Result<()> {
-    let mut message_size: [u8; 4] = [0; 4];
-    reader.seek(SeekFrom::Start(offset))?;
-    reader.read_exact(&mut message_size)?;
-    if message_size == CONTINUATION_MARKER {
-        reader.read_exact(&mut message_size)?;
-    };
-    let message_length = i32::from_le_bytes(message_size);
-
-    let message_length: usize = message_length
-        .try_into()
-        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
-
-    data.clear();
-    data.try_reserve(message_length)?;
-    reader
-        .by_ref()
-        .take(message_length as u64)
-        .read_to_end(data)?;
-
-    Ok(())
-}
-
-pub(crate) fn get_dictionary_batch<'a>(
-    message: &'a arrow_format::ipc::MessageRef,
-) -> Result<arrow_format::ipc::DictionaryBatchRef<'a>> {
-    let header = message
-        .header()
-        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferHeader(err)))?
-        .ok_or_else(|| Error::from(OutOfSpecKind::MissingMessageHeader))?;
-    match header {
-        arrow_format::ipc::MessageHeaderRef::DictionaryBatch(batch) => Ok(batch),
-        _ => Err(Error::from(OutOfSpecKind::UnexpectedMessageType)),
-    }
-}
-
-fn read_dictionary_block<R: Read + Seek>(
-    reader: &mut R,
-    metadata: &FileMetadata,
-    block: &arrow_format::ipc::Block,
-    dictionaries: &mut Dictionaries,
-    message_scratch: &mut Vec<u8>,
-    dictionary_scratch: &mut Vec<u8>,
-) -> Result<()> {
-    let offset: u64 = block
-        .offset
-        .try_into()
-        .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
-    let length: u64 = block
-        .meta_data_length
-        .try_into()
-        .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
-    read_dictionary_message(reader, offset, message_scratch)?;
-
-    let message = arrow_format::ipc::MessageRef::read_as_root(message_scratch.as_ref())
-        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
-
-    let batch = get_dictionary_batch(&message)?;
-
-    read_dictionary(
-        batch,
-        &metadata.schema.fields,
-        &metadata.ipc_schema,
-        dictionaries,
-        reader,
-        offset + length,
-        metadata.size,
-        dictionary_scratch,
-    )
-}
-
-/// Reads all file's dictionaries, if any
-/// This function is IO-bounded
-pub fn read_file_dictionaries<R: Read + Seek>(
-    reader: &mut R,
-    metadata: &FileMetadata,
-    scratch: &mut Vec<u8>,
-) -> Result<Dictionaries> {
-    let mut dictionaries = Default::default();
-
-    let blocks = if let Some(blocks) = &metadata.dictionaries {
-        blocks
-    } else {
-        return Ok(AHashMap::new());
-    };
-    // use a temporary smaller scratch for the messages
-    let mut message_scratch = Default::default();
-
-    for block in blocks {
-        read_dictionary_block(
-            reader,
-            metadata,
-            block,
-            &mut dictionaries,
-            &mut message_scratch,
-            scratch,
-        )?;
-    }
-    Ok(dictionaries)
-}
-
-/// Reads the footer's length and magic number in footer
-fn read_footer_len<R: Read + Seek>(reader: &mut R) -> Result<(u64, usize)> {
-    // read footer length and magic number in footer
-    let end = reader.seek(SeekFrom::End(-10))? + 10;
-
-    let mut footer: [u8; 10] = [0; 10];
-
-    reader.read_exact(&mut footer)?;
-    let footer_len = i32::from_le_bytes(footer[..4].try_into().unwrap());
-
-    if footer[4..] != ARROW_MAGIC {
-        return Err(Error::from(OutOfSpecKind::InvalidFooter));
-    }
-    let footer_len = footer_len
-        .try_into()
-        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
-
-    Ok((end, footer_len))
-}
-
-pub(super) fn deserialize_footer(footer_data: &[u8], size: u64) -> Result<FileMetadata> {
-    let footer = arrow_format::ipc::FooterRef::read_as_root(footer_data)
-        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferFooter(err)))?;
-
-    let blocks = footer
-        .record_batches()
-        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferRecordBatches(err)))?
-        .ok_or_else(|| Error::from(OutOfSpecKind::MissingRecordBatches))?;
-
-    let blocks = blocks
-        .iter()
-        .map(|block| {
-            block
-                .try_into()
-                .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferRecordBatches(err)))
-        })
-        .collect::<Result<Vec<_>>>()?;
-
-    let ipc_schema = footer
-        .schema()
-        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferSchema(err)))?
-        .ok_or_else(|| Error::from(OutOfSpecKind::MissingSchema))?;
-    let (schema, ipc_schema) = fb_to_schema(ipc_schema)?;
-
-    let dictionaries = footer
-        .dictionaries()
-        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferDictionaries(err)))?
-        .map(|dictionaries| {
-            dictionaries
-                .into_iter()
-                .map(|block| {
-                    block.try_into().map_err(|err| {
-                        Error::from(OutOfSpecKind::InvalidFlatbufferRecordBatches(err))
-                    })
-                })
-                .collect::<Result<Vec<_>>>()
-        })
-        .transpose()?;
-
-    Ok(FileMetadata {
-        schema,
-        ipc_schema,
-        blocks,
-        dictionaries,
-        size,
-    })
-}
-
-/// Read the Arrow IPC file's metadata
-pub fn read_file_metadata<R: Read + Seek>(reader: &mut R) -> Result<FileMetadata> {
-    // check if header contain the correct magic bytes
-    let mut magic_buffer: [u8; 6] = [0; 6];
-    let start = reader.seek(SeekFrom::Current(0))?;
-    reader.read_exact(&mut magic_buffer)?;
-    if magic_buffer != ARROW_MAGIC {
-        return Err(Error::from(OutOfSpecKind::InvalidHeader));
-    }
-
-    let (end, footer_len) = read_footer_len(reader)?;
-
-    // read footer
-    reader.seek(SeekFrom::End(-10 - footer_len as i64))?;
-
-    let mut serialized_footer = vec![];
-    serialized_footer.try_reserve(footer_len)?;
-    reader
-        .by_ref()
-        .take(footer_len as u64)
-        .read_to_end(&mut serialized_footer)?;
-
-    deserialize_footer(&serialized_footer, end - start)
-}
-
-pub(crate) fn get_record_batch(
-    message: arrow_format::ipc::MessageRef,
-) -> Result<arrow_format::ipc::RecordBatchRef> {
-    let header = message
-        .header()
-        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferHeader(err)))?
-        .ok_or_else(|| Error::from(OutOfSpecKind::MissingMessageHeader))?;
-    match header {
-        arrow_format::ipc::MessageHeaderRef::RecordBatch(batch) => Ok(batch),
-        _ => Err(Error::from(OutOfSpecKind::UnexpectedMessageType)),
-    }
-}
-
-/// Reads the record batch at position `index` from the reader.
-///
-/// This function is useful for random access to the file. For example, if
-/// you have indexed the file somewhere else, this allows pruning
-/// certain parts of the file.
-/// # Panics
-/// This function panics iff `index >= metadata.blocks.len()`
-#[allow(clippy::too_many_arguments)]
-pub fn read_batch<R: Read + Seek>(
-    reader: &mut R,
-    dictionaries: &Dictionaries,
-    metadata: &FileMetadata,
-    projection: Option<&[usize]>,
-    limit: Option<usize>,
-    index: usize,
-    message_scratch: &mut Vec<u8>,
-    data_scratch: &mut Vec<u8>,
-) -> Result<Chunk<Box<dyn Array>>> {
-    let block = metadata.blocks[index];
-
-    let offset: u64 = block
-        .offset
-        .try_into()
-        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
-
-    let length: u64 = block
-        .meta_data_length
-        .try_into()
-        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
-
-    // read length
-    reader.seek(SeekFrom::Start(offset))?;
-    let mut meta_buf = [0; 4];
-    reader.read_exact(&mut meta_buf)?;
-    if meta_buf == CONTINUATION_MARKER {
-        // continuation marker encountered, read message next
-        reader.read_exact(&mut meta_buf)?;
-    }
-    let meta_len = i32::from_le_bytes(meta_buf)
-        .try_into()
-        .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
-
-    message_scratch.clear();
-    message_scratch.try_reserve(meta_len)?;
-    reader
-        .by_ref()
-        .take(meta_len as u64)
-        .read_to_end(message_scratch)?;
-
-    let message = arrow_format::ipc::MessageRef::read_as_root(message_scratch.as_ref())
-        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
-
-    let batch = get_record_batch(message)?;
-
-    read_record_batch(
-        batch,
-        &metadata.schema.fields,
-        &metadata.ipc_schema,
-        projection,
-        limit,
-        dictionaries,
-        message
-            .version()
-            .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferVersion(err)))?,
-        reader,
-        offset + length,
-        metadata.size,
-        data_scratch,
-    )
-}
+use super::{read_batch, read_file_dictionaries, FileMetadata};
 
 /// An iterator of [`Chunk`]s from an Arrow IPC file.
 pub struct FileReader<R: Read + Seek> {
diff --git a/src/io/json/read/infer_schema.rs b/src/io/json/read/infer_schema.rs
index 2ea4f84058..38c147728a 100644
--- a/src/io/json/read/infer_schema.rs
+++ b/src/io/json/read/infer_schema.rs
@@ -1,5 +1,4 @@
 use std::borrow::Borrow;
-use std::collections::BTreeMap;
 
 use indexmap::map::IndexMap as HashMap;
 use indexmap::set::IndexSet as HashSet;
@@ -30,7 +29,7 @@ fn filter_map_nulls(dt: DataType) -> Option<DataType> {
     }
 }
 
-fn infer_object(inner: &BTreeMap<String, Value>) -> Result<DataType> {
+fn infer_object(inner: &HashMap<String, Value>) -> Result<DataType> {
     let fields = inner
         .iter()
         .filter_map(|(key, value)| {
diff --git a/src/io/parquet/write/sink.rs b/src/io/parquet/write/sink.rs
index 9ba2d4aea9..5086b8e09e 100644
--- a/src/io/parquet/write/sink.rs
+++ b/src/io/parquet/write/sink.rs
@@ -1,7 +1,7 @@
 use ahash::AHashMap;
 use std::{pin::Pin, task::Poll};
 
-use futures::{future::BoxFuture, AsyncWrite, FutureExt, Sink, TryFutureExt};
+use futures::{future::BoxFuture, AsyncWrite, AsyncWriteExt, FutureExt, Sink, TryFutureExt};
 use parquet2::metadata::KeyValue;
 use parquet2::write::FileStreamer;
 use parquet2::write::WriteOptions as ParquetWriteOptions;
@@ -218,6 +218,7 @@ where
 
                     this.task = Some(Box::pin(async move {
                         writer.end(kv_meta).map_err(Error::from).await?;
+                        writer.into_inner().close().map_err(Error::from).await?;
                         Ok(None)
                     }));
                     this.poll_complete(cx)
diff --git a/src/mmap/mod.rs b/src/mmap/mod.rs
index 55bd7d01f8..d8bd90c3a2 100644
--- a/src/mmap/mod.rs
+++ b/src/mmap/mod.rs
@@ -8,7 +8,7 @@ use crate::datatypes::{DataType, Field};
 use crate::error::Error;
 use crate::ffi::mmap;
 
-use crate::io::ipc::read::reader::{get_dictionary_batch, get_record_batch};
+use crate::io::ipc::read::file::{get_dictionary_batch, get_record_batch};
 use crate::io::ipc::read::{first_dict_field, Dictionaries, FileMetadata};
 use crate::io::ipc::read::{IpcBuffer, Node, OutOfSpecKind};
 use crate::io::ipc::{IpcField, CONTINUATION_MARKER};


Test patch:
diff --git a/integration-testing/Cargo.toml b/integration-testing/Cargo.toml
index 8cc7d9a0e2..87b41ffc99 100644
--- a/integration-testing/Cargo.toml
+++ b/integration-testing/Cargo.toml
@@ -29,7 +29,7 @@ logging = ["tracing-subscriber"]
 
 [dependencies]
 arrow2 = { path = "../", features = ["io_ipc", "io_ipc_compression", "io_flight", "io_json_integration"] }
-arrow-format = { version = "0.7", features = ["full"] }
+arrow-format = { version = "0.7", features = ["flight-data", "flight-service"] }
 async-trait = "0.1.41"
 clap = { version = "^3", features = ["derive"] }
 futures = "0.3"
diff --git a/integration-testing/src/flight_client_scenarios/integration_test.rs b/integration-testing/src/flight_client_scenarios/integration_test.rs
index 3cdcd17f2e..8955a3e121 100644
--- a/integration-testing/src/flight_client_scenarios/integration_test.rs
+++ b/integration-testing/src/flight_client_scenarios/integration_test.rs
@@ -21,23 +21,15 @@ use arrow2::{
     array::Array,
     chunk::Chunk,
     datatypes::*,
-    io::ipc::{
-        read::{self, Dictionaries},
-        write, IpcSchema,
-    },
     io::{
         flight::{self, deserialize_batch, serialize_batch},
-        ipc::IpcField,
+        ipc::{read::Dictionaries, write, IpcField, IpcSchema},
     },
 };
-use arrow_format::flight::service::flight_service_client::FlightServiceClient;
-use arrow_format::ipc;
-use arrow_format::{
-    flight::data::{
-        flight_descriptor::DescriptorType, FlightData, FlightDescriptor, Location, Ticket,
-    },
-    ipc::planus::ReadAsRoot,
+use arrow_format::flight::data::{
+    flight_descriptor::DescriptorType, FlightData, FlightDescriptor, Location, Ticket,
 };
+use arrow_format::flight::service::flight_service_client::FlightServiceClient;
 use futures::{stream::BoxStream, StreamExt, TryStreamExt};
 use tonic::{Request, Streaming};
 
@@ -233,30 +225,14 @@ async fn read_dictionaries(
     dictionaries: &mut Dictionaries,
 ) -> Option<FlightData> {
     let mut data = stream.next().await?.ok()?;
-    let mut message =
-        ipc::MessageRef::read_as_root(&data.data_header).expect("Error parsing first message");
-
-    while let ipc::MessageHeaderRef::DictionaryBatch(chunk) = message
-        .header()
-        .expect("Header to be valid flatbuffers")
-        .expect("Header to be present")
-    {
-        let length = data.data_body.len();
-        let mut reader = std::io::Cursor::new(&data.data_body);
-        read::read_dictionary(
-            chunk,
-            fields,
-            ipc_schema,
-            dictionaries,
-            &mut reader,
-            0,
-            length as u64,
-            &mut Default::default(),
-        )
-        .expect("Error reading dictionary");
 
+    let existing = dictionaries.len();
+    loop {
+        flight::deserialize_dictionary(&data, fields, ipc_schema, dictionaries).ok()?;
+        if dictionaries.len() == existing {
+            break;
+        }
         data = stream.next().await?.ok()?;
-        message = ipc::MessageRef::read_as_root(&data.data_header).expect("Error parsing message");
     }
 
     Some(data)
diff --git a/integration-testing/src/flight_server_scenarios/integration_test.rs b/integration-testing/src/flight_server_scenarios/integration_test.rs
index 9d21cf0c97..89df0a041b 100644
--- a/integration-testing/src/flight_server_scenarios/integration_test.rs
+++ b/integration-testing/src/flight_server_scenarios/integration_test.rs
@@ -26,14 +26,13 @@ use tonic::{transport::Server, Request, Response, Status, Streaming};
 use arrow_format::flight::data::flight_descriptor::*;
 use arrow_format::flight::data::*;
 use arrow_format::flight::service::flight_service_server::*;
-use arrow_format::ipc::planus::ReadAsRoot;
-use arrow_format::ipc::MessageHeaderRef;
 
 use arrow2::array::Array;
 use arrow2::chunk::Chunk;
-use arrow2::datatypes::{Field, Schema};
+use arrow2::datatypes::Schema;
 use arrow2::io::flight::{
-    deserialize_schemas, serialize_batch, serialize_schema, serialize_schema_to_info,
+    deserialize_message, deserialize_schemas, serialize_batch, serialize_schema,
+    serialize_schema_to_info,
 };
 use arrow2::io::ipc;
 use arrow2::io::ipc::read::Dictionaries;
@@ -228,14 +227,14 @@ impl FlightService for Service {
         let stream = try_stream! {
             pin_mut!(input_stream);
             for await item in input_stream {
-                let FlightData {data_header, data_body, app_metadata, ..} = item.map_err(|_| Status::invalid_argument(format!("Invalid")))?;
-                save_message(&data_header,
-                    &data_body,
-                    &schema,
+                let data = item.map_err(|_| Status::invalid_argument(format!("Invalid")))?;
+                let maybe_chunk = deserialize_message(&data, &schema.fields,
                     &ipc_schema,
-                    &mut dictionaries,
-                    &mut chunks)?;
-                yield PutResult {app_metadata}
+                    &mut dictionaries).map_err(|_| Status::invalid_argument(format!("Invalid")))?;
+                if let Some(chunk) = maybe_chunk {
+                    chunks.push(chunk)
+                }
+                yield PutResult {app_metadata: data.app_metadata}
             }
             let dataset = IntegrationDataset {
                 schema,
@@ -269,86 +268,3 @@ impl FlightService for Service {
         Err(Status::unimplemented("do_exchange"))
     }
 }
-
-fn chunk_from_message(
-    batch: arrow_format::ipc::RecordBatchRef<'_>,
-    data_body: &[u8],
-    fields: &[Field],
-    ipc_schema: &ipc::IpcSchema,
-    dictionaries: &mut Dictionaries,
-) -> Result<Chunk<Box<dyn Array>>, Status> {
-    let length = data_body.len();
-    let mut reader = std::io::Cursor::new(data_body);
-
-    ipc::read::read_record_batch(
-        batch,
-        fields,
-        ipc_schema,
-        None,
-        None,
-        dictionaries,
-        arrow_format::ipc::MetadataVersion::V5,
-        &mut reader,
-        0,
-        length as u64,
-        &mut Default::default(),
-    )
-    .map_err(|e| Status::internal(format!("Could not convert to Chunk: {:?}", e)))
-}
-
-fn dictionary_from_message(
-    dict_batch: arrow_format::ipc::DictionaryBatchRef<'_>,
-    data_body: &[u8],
-    fields: &[Field],
-    ipc_schema: &ipc::IpcSchema,
-    dictionaries: &mut Dictionaries,
-) -> Result<(), Status> {
-    let length = data_body.len();
-    let mut reader = std::io::Cursor::new(data_body);
-
-    ipc::read::read_dictionary(
-        dict_batch,
-        fields,
-        ipc_schema,
-        dictionaries,
-        &mut reader,
-        0,
-        length as u64,
-        &mut Default::default(),
-    )
-    .map_err(|e| Status::internal(format!("Could not convert to Dictionary: {:?}", e)))
-}
-
-fn save_message(
-    header: &[u8],
-    body: &[u8],
-    schema: &Schema,
-    ipc_schema: &ipc::IpcSchema,
-    dictionaries: &mut Dictionaries,
-    chunks: &mut Vec<Chunk<Box<dyn Array>>>,
-) -> Result<(), Status> {
-    let message = arrow_format::ipc::MessageRef::read_as_root(header)
-        .map_err(|e| Status::internal(format!("Could not parse message: {:?}", e)))?;
-    let header = message
-        .header()
-        .map_err(|x| Status::internal(x.to_string()))?
-        .ok_or_else(|| Status::internal("Message must contain a header".to_string()))?;
-
-    match header {
-        MessageHeaderRef::RecordBatch(batch) => {
-            let chunk = chunk_from_message(batch, body, &schema.fields, ipc_schema, dictionaries)?;
-
-            chunks.push(chunk);
-        }
-        MessageHeaderRef::DictionaryBatch(dict_batch) => {
-            dictionary_from_message(dict_batch, body, &schema.fields, ipc_schema, dictionaries)?;
-        }
-        t => {
-            return Err(Status::internal(format!(
-                "Reading types other than record batches not yet supported, unable to read {:?}",
-                t
-            )));
-        }
-    }
-    Ok(())
-}
diff --git a/tests/it/array/struct_/mod.rs b/tests/it/array/struct_/mod.rs
index 9ec34eca0e..8869c215e5 100644
--- a/tests/it/array/struct_/mod.rs
+++ b/tests/it/array/struct_/mod.rs
@@ -1,4 +1,5 @@
 mod iterator;
+mod mutable;
 
 use arrow2::array::*;
 use arrow2::bitmap::Bitmap;
diff --git a/tests/it/array/struct_/mutable.rs b/tests/it/array/struct_/mutable.rs
new file mode 100644
index 0000000000..19f2f12f15
--- /dev/null
+++ b/tests/it/array/struct_/mutable.rs
@@ -0,0 +1,33 @@
+use arrow2::{
+    array::*,
+    datatypes::{DataType, Field},
+};
+
+#[test]
+fn push() {
+    let c1 = Box::new(MutablePrimitiveArray::<i32>::new()) as Box<dyn MutableArray>;
+    let values = vec![c1];
+    let data_type = DataType::Struct(vec![Field::new("f1", DataType::Int32, true)]);
+    let mut a = MutableStructArray::new(data_type, values);
+
+    a.value::<MutablePrimitiveArray<i32>>(0)
+        .unwrap()
+        .push(Some(1));
+    a.push(true);
+    a.value::<MutablePrimitiveArray<i32>>(0).unwrap().push(None);
+    a.push(false);
+    a.value::<MutablePrimitiveArray<i32>>(0)
+        .unwrap()
+        .push(Some(2));
+    a.push(true);
+
+    assert_eq!(a.len(), 3);
+    assert!(a.is_valid(0));
+    assert!(!a.is_valid(1));
+    assert!(a.is_valid(2));
+
+    assert_eq!(
+        a.value::<MutablePrimitiveArray<i32>>(0).unwrap().values(),
+        &Vec::from([1, 0, 2])
+    );
+}
diff --git a/tests/it/io/ndjson/mod.rs b/tests/it/io/ndjson/mod.rs
index 14e8d6378d..0f90da3c14 100644
--- a/tests/it/io/ndjson/mod.rs
+++ b/tests/it/io/ndjson/mod.rs
@@ -311,3 +311,22 @@ fn case(case: &str) -> (String, Box<dyn Array>) {
         _ => todo!(),
     }
 }
+
+#[test]
+fn infer_object() -> Result<()> {
+    let data = r#"{"i64": 1, "f64": 0.1, "utf8": "foo1", "bools": true}
+    {"i64": 2, "f64": 0.2, "utf8": "foo2", "bools": false}
+    {"i64": 3, "f64": 0.3, "utf8": "foo3"}
+    {"i64": 4, "f64": 0.4, "utf8": "foo4", "bools": false}
+    "#;
+    let u64_fld = Field::new("i64", DataType::Int64, true);
+    let f64_fld = Field::new("f64", DataType::Float64, true);
+    let utf8_fld = Field::new("utf8", DataType::Utf8, true);
+    let bools_fld = Field::new("bools", DataType::Boolean, true);
+
+    let expected = DataType::Struct(vec![u64_fld, f64_fld, utf8_fld, bools_fld]);
+    let actual = infer(data)?;
+
+    assert_eq!(expected, actual);
+    Ok(())
+}