Repository: jorgecarleitao/arrow2
Pull request #1199 (2022-08-01 05:27:37)

Migrated Avro code to avro-schema repo
Companion to https://github.com/DataEngineeringLabs/avro-schema/pull/1

codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#1199](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (89190c7) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/3f3febfb17575c4212220be8893a5838debb1907?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (3f3febf) will **decrease** coverage by `0.06%`.
> The diff coverage is `87.87%`.

```diff
@@            Coverage Diff             @@
##             main    #1199      +/-   ##
==========================================
- Coverage   83.44%   83.38%   -0.07%     
==========================================
  Files         366      354      -12     
  Lines       36697    36330     -367     
==========================================
- Hits        30621    30292     -329     
+ Misses       6076     6038      -38     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/io/avro/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2F2cm8vbW9kLnJz) | `0.00% <0.00%> (-86.67%)` | :arrow_down: |
| [src/io/avro/read/util.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2F2cm8vcmVhZC91dGlsLnJz) | `88.88% <ø> (-7.41%)` | :arrow_down: |
| [src/io/avro/write/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2F2cm8vd3JpdGUvbW9kLnJz) | `100.00% <ø> (ø)` | |
| [src/io/avro/read/schema.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2F2cm8vcmVhZC9zY2hlbWEucnM=) | `74.50% <83.33%> (+2.39%)` | :arrow_up: |
| [src/io/avro/write/serialize.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2F2cm8vd3JpdGUvc2VyaWFsaXplLnJz) | `87.43% <85.71%> (ø)` | |
| [src/io/avro/read/deserialize.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2F2cm8vcmVhZC9kZXNlcmlhbGl6ZS5ycw==) | `81.11% <100.00%> (+0.10%)` | :arrow_up: |
| [src/io/avro/read/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2F2cm8vcmVhZC9tb2QucnM=) | `89.28% <100.00%> (+0.09%)` | :arrow_up: |
| [src/io/avro/write/schema.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2F2cm8vd3JpdGUvc2NoZW1hLnJz) | `81.35% <100.00%> (+4.27%)` | :arrow_up: |
| [src/io/ipc/read/reader.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL3JlYWRlci5ycw==) | `96.32% <0.00%> (-0.34%)` | :arrow_down: |
| [src/array/binary/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199/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: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199?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/1199?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Last update [3f3febf...89190c7](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1199?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/Cargo.toml b/Cargo.toml
index 3fe3cb06c7..791d758ea2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -21,14 +21,19 @@ num-traits = "0.2"
 dyn-clone = "1"
 bytemuck = { version = "1", features = ["derive"] }
 chrono = { version = "0.4", default_features = false, features = ["std"] }
-chrono-tz = { version = "0.6", optional = true }
-# To efficiently cast numbers to strings
-lexical-core = { version = "0.8", optional = true }
+
 # We need to Hash values before sending them to an hasher. This
 # crate provides HashMap that assumes pre-hashed values.
 hash_hasher = "^2.0.3"
 # For SIMD utf8 validation
 simdutf8 = "0.1.3"
+# faster hashing
+ahash = { version = "0.7" }
+
+# for timezone support
+chrono-tz = { version = "0.6", optional = true }
+# To efficiently cast numbers to strings
+lexical-core = { version = "0.8", optional = true }
 
 # for csv io
 csv = { version = "^1.1", optional = true }
@@ -38,6 +43,7 @@ csv-core = { version = "0.1", optional = true }
 csv-async = { version = "^1.1", optional = true }
 
 regex = { version = "^1.3", optional = true }
+regex-syntax = { version = "^0.6", optional = true }
 streaming-iterator = { version = "0.1", optional = true }
 fallible-streaming-iterator = { version = "0.1", optional = true }
 
@@ -64,20 +70,14 @@ base64 = { version = "0.13.0", optional = true }
 # to write to parquet as a stream
 futures = { version = "0.3", optional = true }
 
-# for faster hashing
-ahash = { version = "0.7" }
+# to read IPC as a stream
+async-stream = { version = "0.3.2", optional = true }
 
 # parquet support
 parquet2 = { version = "0.14.0", optional = true, default_features = false }
 
 # avro support
-avro-schema = { version = "0.2", optional = true }
-# compression of avro
-libflate = { version = "1.1.1", optional = true }
-snap = { version = "1", optional = true }
-crc = { version = "2", optional = true }
-# async avro
-async-stream = { version = "0.3.2", optional = true }
+avro-schema = { version = "0.3", optional = true }
 
 # ORC support
 orc-format = { version = "0.3.0", optional = true }
@@ -135,6 +135,7 @@ full = [
     "io_avro_compression",
     "io_avro_async",
     "regex",
+    "regex-syntax",
     "compute",
     # parses timezones used in timestamp conversions
     "chrono-tz",
@@ -161,14 +162,11 @@ io_parquet_compression = [
     "parquet2/lz4",
     "parquet2/brotli",
 ]
-
-io_avro = ["avro-schema", "streaming-iterator", "fallible-streaming-iterator", "serde_json"]
+io_avro = ["avro-schema", "streaming-iterator"]
 io_avro_compression = [
-    "libflate",
-    "snap",
-    "crc",
+    "avro-schema/compression",
 ]
-io_avro_async = ["io_avro", "futures", "async-stream"]
+io_avro_async = ["avro-schema/async"]
 
 io_orc = [ "orc-format" ]
 
@@ -190,7 +188,7 @@ compute_filter = []
 compute_hash = ["multiversion"]
 compute_if_then_else = []
 compute_length = []
-compute_like = ["regex"]
+compute_like = ["regex", "regex-syntax"]
 compute_limit = []
 compute_merge_sort = ["itertools", "compute_sort"]
 compute_nullif = ["compute_comparison"]
diff --git a/benches/avro_read.rs b/benches/avro_read.rs
index 9143db9e7e..37088492df 100644
--- a/benches/avro_read.rs
+++ b/benches/avro_read.rs
@@ -4,6 +4,7 @@ use avro_rs::types::Record;
 use criterion::*;
 
 use arrow2::error::Result;
+use arrow2::io::avro::avro_schema::read::read_metadata;
 use arrow2::io::avro::read;
 use avro_rs::*;
 use avro_rs::{Codec, Schema as AvroSchema};
@@ -42,17 +43,10 @@ fn write(size: usize, has_codec: bool) -> Result<Vec<u8>> {
 fn read_batch(buffer: &[u8], size: usize) -> Result<()> {
     let mut file = Cursor::new(buffer);
 
-    let (avro_schema, schema, codec, file_marker) = read::read_metadata(&mut file)?;
+    let metadata = read_metadata(&mut file)?;
+    let schema = read::infer_schema(&metadata.record)?;
 
-    let reader = read::Reader::new(
-        read::Decompressor::new(
-            read::BlockStreamIterator::new(&mut file, file_marker),
-            codec,
-        ),
-        avro_schema,
-        schema.fields,
-        None,
-    );
+    let reader = read::Reader::new(file, metadata, schema.fields, None);
 
     let mut rows = 0;
     for maybe_batch in reader {
diff --git a/examples/avro_kafka.rs b/examples/avro_kafka.rs
index 5fb392fd93..7645024939 100644
--- a/examples/avro_kafka.rs
+++ b/examples/avro_kafka.rs
@@ -2,6 +2,8 @@ use arrow2::{
     datatypes::{DataType, Field},
     error::Error,
     io::avro,
+    io::avro::avro_schema::file::Block,
+    io::avro::avro_schema::schema::{Field as AvroField, Schema},
 };
 
 fn read_schema_id<R: std::io::Read>(reader: &mut R) -> Result<u32, Error> {
@@ -16,7 +18,7 @@ fn read_schema_id<R: std::io::Read>(reader: &mut R) -> Result<u32, Error> {
     Ok(u32::from_be_bytes(header[1..].try_into().unwrap()))
 }
 
-fn read_block<R: std::io::Read>(reader: &mut R, block: &mut avro::Block) -> Result<(), Error> {
+fn read_block<R: std::io::Read>(reader: &mut R, block: &mut Block) -> Result<(), Error> {
     // (we could lump multiple records together by extending the block instead of clearing)
     block.data.clear();
     reader.read_to_end(&mut block.data)?;
@@ -34,8 +36,8 @@ fn main() -> Result<(), Error> {
 
     // say that from the registry we concluded that this schema has fields
     let avro_fields = vec![
-        avro_schema::Schema::String(None),
-        avro_schema::Schema::String(None),
+        AvroField::new("first_name", Schema::String(None)),
+        AvroField::new("last_name", Schema::String(None)),
     ];
     // (which we map to arrow fields as)
     let fields = vec![
@@ -44,7 +46,7 @@ fn main() -> Result<(), Error> {
     ];
 
     // the below allow us to read it to arrow (a chunk of a single element)
-    let mut block = avro::Block::default();
+    let mut block = Block::default();
     read_block(&mut stream, &mut block)?;
 
     let chunk = avro::read::deserialize(&block, &fields, &avro_fields, &[true, true])?;
diff --git a/examples/avro_read.rs b/examples/avro_read.rs
index 2d0148cf9d..749cf3c50d 100644
--- a/examples/avro_read.rs
+++ b/examples/avro_read.rs
@@ -2,6 +2,7 @@ use std::fs::File;
 use std::io::BufReader;
 
 use arrow2::error::Result;
+use arrow2::io::avro::avro_schema;
 use arrow2::io::avro::read;
 
 fn main() -> Result<()> {
@@ -12,16 +13,13 @@ fn main() -> Result<()> {
 
     let file = &mut BufReader::new(File::open(path)?);
 
-    let (avro_schema, schema, codec, file_marker) = read::read_metadata(file)?;
+    let metadata = avro_schema::read::read_metadata(file)?;
 
-    println!("{:#?}", avro_schema);
+    let schema = read::infer_schema(&metadata.record)?;
 
-    let reader = read::Reader::new(
-        read::Decompressor::new(read::BlockStreamIterator::new(file, file_marker), codec),
-        avro_schema,
-        schema.fields,
-        None,
-    );
+    println!("{:#?}", metadata);
+
+    let reader = read::Reader::new(file, metadata, schema.fields, None);
 
     for maybe_chunk in reader {
         let columns = maybe_chunk?;
diff --git a/examples/avro_read_async.rs b/examples/avro_read_async.rs
index 9c9c3649da..ac7ad0b645 100644
--- a/examples/avro_read_async.rs
+++ b/examples/avro_read_async.rs
@@ -1,11 +1,14 @@
+use std::sync::Arc;
+
 use futures::pin_mut;
 use futures::StreamExt;
 use tokio::fs::File;
 use tokio_util::compat::*;
 
 use arrow2::error::Result;
-use arrow2::io::avro::read::{decompress_block, deserialize};
-use arrow2::io::avro::read_async::*;
+use arrow2::io::avro::avro_schema::file::Block;
+use arrow2::io::avro::avro_schema::read_async::{block_stream, decompress_block, read_metadata};
+use arrow2::io::avro::read::{deserialize, infer_schema};
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<()> {
@@ -16,22 +19,28 @@ async fn main() -> Result<()> {
 
     let mut reader = File::open(file_path).await?.compat();
 
-    let (avro_schemas, schema, compression, marker) = read_metadata(&mut reader).await?;
-    let avro_schemas = Box::new(avro_schemas);
-    let projection = Box::new(schema.fields.iter().map(|_| true).collect::<Vec<_>>());
+    let metadata = read_metadata(&mut reader).await?;
+    let schema = infer_schema(&metadata.record)?;
+    let metadata = Arc::new(metadata);
+    let projection = Arc::new(schema.fields.iter().map(|_| true).collect::<Vec<_>>());
 
-    let blocks = block_stream(&mut reader, marker).await;
+    let blocks = block_stream(&mut reader, metadata.marker).await;
 
     pin_mut!(blocks);
     while let Some(mut block) = blocks.next().await.transpose()? {
         let schema = schema.clone();
-        let avro_schemas = avro_schemas.clone();
+        let metadata = metadata.clone();
         let projection = projection.clone();
         // the content here is CPU-bounded. It should run on a dedicated thread pool
         let handle = tokio::task::spawn_blocking(move || {
             let mut decompressed = Block::new(0, vec![]);
-            decompress_block(&mut block, &mut decompressed, compression)?;
-            deserialize(&decompressed, &schema.fields, &avro_schemas, &projection)
+            decompress_block(&mut block, &mut decompressed, metadata.compression)?;
+            deserialize(
+                &decompressed,
+                &schema.fields,
+                &metadata.record.fields,
+                &projection,
+            )
         });
         let chunk = handle.await.unwrap()?;
         assert!(!chunk.is_empty());
diff --git a/examples/avro_write.rs b/examples/avro_write.rs
index 5e8224c12c..6042172913 100644
--- a/examples/avro_write.rs
+++ b/examples/avro_write.rs
@@ -4,6 +4,7 @@ use arrow2::{
     array::{Array, Int32Array},
     datatypes::{Field, Schema},
     error::Result,
+    io::avro::avro_schema,
     io::avro::write,
 };
 
@@ -11,26 +12,27 @@ fn write_avro<W: std::io::Write>(
     file: &mut W,
     arrays: &[&dyn Array],
     schema: &Schema,
-    compression: Option<write::Compression>,
+    compression: Option<avro_schema::file::Compression>,
 ) -> Result<()> {
-    let avro_fields = write::to_avro_schema(schema)?;
+    let record = write::to_record(schema)?;
 
     let mut serializers = arrays
         .iter()
-        .zip(avro_fields.iter())
+        .zip(record.fields.iter())
         .map(|(array, field)| write::new_serializer(*array, &field.schema))
         .collect::<Vec<_>>();
-    let mut block = write::Block::new(arrays[0].len(), vec![]);
+    let mut block = avro_schema::file::Block::new(arrays[0].len(), vec![]);
 
     write::serialize(&mut serializers, &mut block);
 
-    let mut compressed_block = write::CompressedBlock::default();
+    let mut compressed_block = avro_schema::file::CompressedBlock::default();
 
-    let _was_compressed = write::compress(&mut block, &mut compressed_block, compression)?;
+    let _was_compressed =
+        avro_schema::write::compress(&mut block, &mut compressed_block, compression)?;
 
-    write::write_metadata(file, avro_fields.clone(), compression)?;
+    avro_schema::write::write_metadata(file, record, compression)?;
 
-    write::write_block(file, &compressed_block)?;
+    avro_schema::write::write_block(file, &compressed_block)?;
 
     Ok(())
 }
diff --git a/examples/ipc_file_mmap.rs b/examples/ipc_file_mmap.rs
new file mode 100644
index 0000000000..51dd2a2747
--- /dev/null
+++ b/examples/ipc_file_mmap.rs
@@ -0,0 +1,35 @@
+//! Example showing how to memory map an Arrow IPC file into a [`Chunk`].
+use std::sync::Arc;
+
+use arrow2::error::Result;
+use arrow2::io::ipc::read;
+use arrow2::mmap::{mmap_dictionaries_unchecked, mmap_unchecked};
+
+// Arrow2 requires a struct that implements `Clone + AsRef<[u8]>`, which
+// usually `Arc<Mmap>` supports. Here we mock it
+#[derive(Clone)]
+struct Mmap(Vec<u8>);
+
+impl AsRef<[u8]> for Mmap {
+    #[inline]
+    fn as_ref(&self) -> &[u8] {
+        self.0.as_ref()
+    }
+}
+
+fn main() -> Result<()> {
+    // given a mmap
+    let mmap = Arc::new(Mmap(vec![]));
+
+    // read the metadata
+    let metadata = read::read_file_metadata(&mut std::io::Cursor::new(mmap.as_ref()))?;
+
+    // mmap the dictionaries
+    let dictionaries = unsafe { mmap_dictionaries_unchecked(&metadata, mmap.clone())? };
+
+    // and finally mmap a chunk (0 in this case).
+    let chunk = unsafe { mmap_unchecked(&metadata, &dictionaries, mmap, 0) }?;
+
+    println!("{chunk:?}");
+    Ok(())
+}
diff --git a/guide/src/SUMMARY.md b/guide/src/SUMMARY.md
index 85ffbce583..f79cced198 100644
--- a/guide/src/SUMMARY.md
+++ b/guide/src/SUMMARY.md
@@ -14,6 +14,7 @@
     - [Read Parquet](./io/parquet_read.md)
     - [Write Parquet](./io/parquet_write.md)
     - [Read Arrow](./io/ipc_read.md)
+    - [Memory map Arrow](./io/ipc_mmap.md)
     - [Read Arrow stream](./io/ipc_stream_read.md)
     - [Write Arrow](./io/ipc_write.md)
     - [Read Avro](./io/avro_read.md)
diff --git a/guide/src/io/ipc_mmap.md b/guide/src/io/ipc_mmap.md
new file mode 100644
index 0000000000..be5bcbe7c6
--- /dev/null
+++ b/guide/src/io/ipc_mmap.md
@@ -0,0 +1,10 @@
+# Read Arrow
+
+When compiled with feature `io_ipc`, this crate can be used to memory map IPC Arrow files
+into arrays.
+
+The example below shows how to memory map an IPC Arrow file into `Chunk`es:
+
+```rust
+{{#include ../../../examples/ipc_file_mmap.rs}}
+```
diff --git a/src/array/binary/mutable.rs b/src/array/binary/mutable.rs
index 7c44a5ff44..9990a542c2 100644
--- a/src/array/binary/mutable.rs
+++ b/src/array/binary/mutable.rs
@@ -230,6 +230,10 @@ impl<O: Offset> MutableArray for MutableBinaryArray<O> {
         self.push::<&[u8]>(None)
     }
 
+    fn reserve(&mut self, additional: usize) {
+        self.reserve(additional)
+    }
+
     fn shrink_to_fit(&mut self) {
         self.shrink_to_fit()
     }
diff --git a/src/array/boolean/mutable.rs b/src/array/boolean/mutable.rs
index 39acab03e0..83fac4b7ef 100644
--- a/src/array/boolean/mutable.rs
+++ b/src/array/boolean/mutable.rs
@@ -495,6 +495,10 @@ impl MutableArray for MutableBooleanArray {
         self.push(None)
     }
 
+    fn reserve(&mut self, additional: usize) {
+        self.reserve(additional)
+    }
+
     fn shrink_to_fit(&mut self) {
         self.shrink_to_fit()
     }
diff --git a/src/array/dictionary/mutable.rs b/src/array/dictionary/mutable.rs
index c5ca7efb31..665d6fc923 100644
--- a/src/array/dictionary/mutable.rs
+++ b/src/array/dictionary/mutable.rs
@@ -135,6 +135,11 @@ impl<K: DictionaryKey, M: MutableArray> MutableDictionaryArray<K, M> {
         Box::new(a)
     }
 
+    /// Reserves `additional` slots.
+    pub fn reserve(&mut self, additional: usize) {
+        self.keys.reserve(additional);
+    }
+
     /// Shrinks the capacity of the [`MutableDictionaryArray`] to fit its current length.
     pub fn shrink_to_fit(&mut self) {
         self.values.shrink_to_fit();
@@ -196,6 +201,11 @@ impl<K: DictionaryKey, M: 'static + MutableArray> MutableArray for MutableDictio
     fn push_null(&mut self) {
         self.keys.push(None)
     }
+
+    fn reserve(&mut self, additional: usize) {
+        self.reserve(additional)
+    }
+
     fn shrink_to_fit(&mut self) {
         self.shrink_to_fit()
     }
diff --git a/src/array/fixed_size_binary/mutable.rs b/src/array/fixed_size_binary/mutable.rs
index 528689ae4b..dc70e49e4a 100644
--- a/src/array/fixed_size_binary/mutable.rs
+++ b/src/array/fixed_size_binary/mutable.rs
@@ -175,6 +175,14 @@ impl MutableFixedSizeBinaryArray {
         std::slice::from_raw_parts(self.values.as_ptr().add(i * self.size), self.size)
     }
 
+    /// Reserves `additional` slots.
+    pub fn reserve(&mut self, additional: usize) {
+        self.values.reserve(additional * self.size);
+        if let Some(x) = self.validity.as_mut() {
+            x.reserve(additional)
+        }
+    }
+
     /// Shrinks the capacity of the [`MutableFixedSizeBinaryArray`] to fit its current length.
     pub fn shrink_to_fit(&mut self) {
         self.values.shrink_to_fit();
@@ -238,6 +246,10 @@ impl MutableArray for MutableFixedSizeBinaryArray {
         self.push::<&[u8]>(None);
     }
 
+    fn reserve(&mut self, additional: usize) {
+        self.reserve(additional)
+    }
+
     fn shrink_to_fit(&mut self) {
         self.shrink_to_fit()
     }
diff --git a/src/array/fixed_size_list/mutable.rs b/src/array/fixed_size_list/mutable.rs
index 5798e4472e..c638e84033 100644
--- a/src/array/fixed_size_list/mutable.rs
+++ b/src/array/fixed_size_list/mutable.rs
@@ -99,6 +99,15 @@ impl<M: MutableArray> MutableFixedSizeListArray<M> {
             None => self.init_validity(),
         }
     }
+
+    /// Reserves `additional` slots.
+    pub fn reserve(&mut self, additional: usize) {
+        self.values.reserve(additional);
+        if let Some(x) = self.validity.as_mut() {
+            x.reserve(additional)
+        }
+    }
+
     /// Shrinks the capacity of the [`MutableFixedSizeListArray`] to fit its current length.
     pub fn shrink_to_fit(&mut self) {
         self.values.shrink_to_fit();
@@ -157,6 +166,10 @@ impl<M: MutableArray + 'static> MutableArray for MutableFixedSizeListArray<M> {
         }
     }
 
+    fn reserve(&mut self, additional: usize) {
+        self.reserve(additional)
+    }
+
     fn shrink_to_fit(&mut self) {
         self.shrink_to_fit()
     }
diff --git a/src/array/list/mod.rs b/src/array/list/mod.rs
index 88ab651323..13f781f680 100644
--- a/src/array/list/mod.rs
+++ b/src/array/list/mod.rs
@@ -324,7 +324,7 @@ impl<O: Offset> ListArray<O> {
     /// Returns a the inner [`Field`]
     /// # Errors
     /// Panics iff the logical type is not consistent with this struct.
-    fn try_get_child(data_type: &DataType) -> Result<&Field, Error> {
+    pub fn try_get_child(data_type: &DataType) -> Result<&Field, Error> {
         if O::IS_LARGE {
             match data_type.to_logical_type() {
                 DataType::LargeList(child) => Ok(child.as_ref()),
diff --git a/src/array/list/mutable.rs b/src/array/list/mutable.rs
index 2f6d8d1ebf..8d0dc3b05e 100644
--- a/src/array/list/mutable.rs
+++ b/src/array/list/mutable.rs
@@ -193,6 +193,14 @@ impl<O: Offset, M: MutableArray> MutableListArray<O, M> {
         Box::new(a)
     }
 
+    /// Reserves `additional` slots.
+    pub fn reserve(&mut self, additional: usize) {
+        self.offsets.reserve(additional);
+        if let Some(x) = self.validity.as_mut() {
+            x.reserve(additional)
+        }
+    }
+
     /// Shrinks the capacity of the [`MutableListArray`] to fit its current length.
     pub fn shrink_to_fit(&mut self) {
         self.values.shrink_to_fit();
@@ -254,6 +262,11 @@ impl<O: Offset, M: MutableArray + 'static> MutableArray for MutableListArray<O,
     fn push_null(&mut self) {
         self.push_null()
     }
+
+    fn reserve(&mut self, additional: usize) {
+        self.reserve(additional)
+    }
+
     fn shrink_to_fit(&mut self) {
         self.shrink_to_fit();
     }
diff --git a/src/array/mod.rs b/src/array/mod.rs
index cc5038fa0f..8010b5cf92 100644
--- a/src/array/mod.rs
+++ b/src/array/mod.rs
@@ -163,6 +163,9 @@ pub trait MutableArray: std::fmt::Debug + Send + Sync {
             .unwrap_or(true)
     }
 
+    /// Reserves additional slots to its capacity.
+    fn reserve(&mut self, additional: usize);
+
     /// Shrink the array to fit its length.
     fn shrink_to_fit(&mut self);
 }
diff --git a/src/array/primitive/mutable.rs b/src/array/primitive/mutable.rs
index c390690d3b..5f2014db31 100644
--- a/src/array/primitive/mutable.rs
+++ b/src/array/primitive/mutable.rs
@@ -419,6 +419,10 @@ impl<T: NativeType> MutableArray for MutablePrimitiveArray<T> {
         self.push(None)
     }
 
+    fn reserve(&mut self, additional: usize) {
+        self.reserve(additional)
+    }
+
     fn shrink_to_fit(&mut self) {
         self.shrink_to_fit()
     }
diff --git a/src/array/utf8/mutable.rs b/src/array/utf8/mutable.rs
index 5215094237..0b87b97c25 100644
--- a/src/array/utf8/mutable.rs
+++ b/src/array/utf8/mutable.rs
@@ -318,6 +318,10 @@ impl<O: Offset> MutableArray for MutableUtf8Array<O> {
         self.push::<&str>(None)
     }
 
+    fn reserve(&mut self, additional: usize) {
+        self.reserve(additional, 0)
+    }
+
     fn shrink_to_fit(&mut self) {
         self.shrink_to_fit()
     }
diff --git a/src/compute/like.rs b/src/compute/like.rs
index cc7517e823..bf363972a1 100644
--- a/src/compute/like.rs
+++ b/src/compute/like.rs
@@ -17,8 +17,41 @@ fn is_like_pattern(c: char) -> bool {
     c == '%' || c == '_'
 }
 
+/// Transforms a like `pattern` to a regex compatible pattern. To achieve that, it does:
+///
+/// 1. Replace like wildcards for regex expressions as the pattern will be evaluated using regex match: `%` => `.*` and `_` => `.`
+/// 2. Escape regex meta characters to match them and not be evaluated as regex special chars. For example: `.` => `\\.`
+/// 3. Replace escaped like wildcards removing the escape characters to be able to match it as a regex. For example: `\\%` => `%`
 fn replace_pattern(pattern: &str) -> String {
-    pattern.replace('%', ".*").replace('_', ".")
+    let mut result = String::new();
+    let text = String::from(pattern);
+    let mut chars_iter = text.chars().peekable();
+    while let Some(c) = chars_iter.next() {
+        if c == '\\' {
+            let next = chars_iter.peek();
+            match next {
+                Some(next) if is_like_pattern(*next) => {
+                    result.push(*next);
+                    // Skipping the next char as it is already appended
+                    chars_iter.next();
+                }
+                _ => {
+                    result.push('\\');
+                    result.push('\\');
+                }
+            }
+        } else if regex_syntax::is_meta_character(c) {
+            result.push('\\');
+            result.push(c);
+        } else if c == '%' {
+            result.push_str(".*");
+        } else if c == '_' {
+            result.push('.');
+        } else {
+            result.push(c);
+        }
+    }
+    result
 }
 
 #[inline]
@@ -108,7 +141,10 @@ fn a_like_utf8_scalar<O: Offset, F: Fn(bool) -> bool>(
 
     let values = if !rhs.contains(is_like_pattern) {
         Bitmap::from_trusted_len_iter(lhs.values_iter().map(|x| op(x == rhs)))
-    } else if rhs.ends_with('%') && !rhs[..rhs.len() - 1].contains(is_like_pattern) {
+    } else if rhs.ends_with('%')
+        && !rhs.ends_with("\\%")
+        && !rhs[..rhs.len() - 1].contains(is_like_pattern)
+    {
         // fast path, can use starts_with
         let starts_with = &rhs[..rhs.len() - 1];
         Bitmap::from_trusted_len_iter(lhs.values_iter().map(|x| op(x.starts_with(starts_with))))
@@ -260,7 +296,10 @@ fn a_like_binary_scalar<O: Offset, F: Fn(bool) -> bool>(
 
     let values = if !pattern.contains(is_like_pattern) {
         Bitmap::from_trusted_len_iter(lhs.values_iter().map(|x| op(x == rhs)))
-    } else if pattern.ends_with('%') && !pattern[..pattern.len() - 1].contains(is_like_pattern) {
+    } else if pattern.ends_with('%')
+        && !pattern.ends_with("\\%")
+        && !pattern[..pattern.len() - 1].contains(is_like_pattern)
+    {
         // fast path, can use starts_with
         let starts_with = &rhs[..rhs.len() - 1];
         Bitmap::from_trusted_len_iter(lhs.values_iter().map(|x| op(x.starts_with(starts_with))))
diff --git a/src/ffi/array.rs b/src/ffi/array.rs
index 46858deff7..b50fdaa725 100644
--- a/src/ffi/array.rs
+++ b/src/ffi/array.rs
@@ -493,7 +493,7 @@ impl InternalArrowArray {
     }
 }
 
-impl ArrowArrayRef for Box<InternalArrowArray> {
+impl ArrowArrayRef for InternalArrowArray {
     /// the data_type as declared in the schema
     fn data_type(&self) -> &DataType {
         &self.data_type
diff --git a/src/ffi/mmap.rs b/src/ffi/mmap.rs
new file mode 100644
index 0000000000..4bd006393f
--- /dev/null
+++ b/src/ffi/mmap.rs
@@ -0,0 +1,612 @@
+use std::collections::VecDeque;
+use std::sync::Arc;
+
+use crate::array::{Array, DictionaryKey, FixedSizeListArray, ListArray, Offset, StructArray};
+use crate::datatypes::DataType;
+use crate::error::Error;
+
+use crate::io::ipc::read::{Dictionaries, OutOfSpecKind};
+use crate::io::ipc::read::{IpcBuffer, Node};
+use crate::io::ipc::IpcField;
+use crate::types::NativeType;
+
+use super::{export_array_to_c, try_from, ArrowArray, InternalArrowArray};
+
+#[allow(dead_code)]
+struct PrivateData<T> {
+    // the owner of the pointers' regions
+    data: T,
+    buffers_ptr: Box<[*const std::os::raw::c_void]>,
+    children_ptr: Box<[*mut ArrowArray]>,
+    dictionary_ptr: Option<*mut ArrowArray>,
+}
+
+fn get_buffer_bounds(buffers: &mut VecDeque<IpcBuffer>) -> Result<(usize, usize), Error> {
+    let buffer = buffers
+        .pop_front()
+        .ok_or_else(|| Error::from(OutOfSpecKind::ExpectedBuffer))?;
+
+    let offset: usize = buffer
+        .offset()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let length: usize = buffer
+        .length()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    Ok((offset, length))
+}
+
+fn get_buffer<'a, T: NativeType>(
+    data: &'a [u8],
+    block_offset: usize,
+    buffers: &mut VecDeque<IpcBuffer>,
+    num_rows: usize,
+) -> Result<&'a [u8], Error> {
+    let (offset, length) = get_buffer_bounds(buffers)?;
+
+    // verify that they are in-bounds
+    let values = data
+        .get(block_offset + offset..block_offset + offset + length)
+        .ok_or_else(|| Error::OutOfSpec("buffer out of bounds".to_string()))?;
+
+    // validate alignment
+    let v: &[T] = bytemuck::try_cast_slice(values)
+        .map_err(|_| Error::OutOfSpec("buffer not aligned for mmap".to_string()))?;
+
+    if v.len() < num_rows {
+        return Err(Error::OutOfSpec(
+            "buffer's length is too small in mmap".to_string(),
+        ));
+    }
+
+    Ok(values)
+}
+
+fn get_validity<'a>(
+    data: &'a [u8],
+    block_offset: usize,
+    buffers: &mut VecDeque<IpcBuffer>,
+    null_count: usize,
+) -> Result<Option<&'a [u8]>, Error> {
+    let validity = get_buffer_bounds(buffers)?;
+    let (offset, length) = validity;
+
+    Ok(if null_count > 0 {
+        // verify that they are in-bounds and get its pointer
+        Some(
+            data.get(block_offset + offset..block_offset + offset + length)
+                .ok_or_else(|| Error::OutOfSpec("buffer out of bounds".to_string()))?,
+        )
+    } else {
+        None
+    })
+}
+
+fn create_array<
+    T: AsRef<[u8]>,
+    I: Iterator<Item = Option<*const u8>>,
+    II: Iterator<Item = ArrowArray>,
+>(
+    data: Arc<T>,
+    num_rows: usize,
+    null_count: usize,
+    buffers: I,
+    children: II,
+    dictionary: Option<ArrowArray>,
+) -> ArrowArray {
+    let buffers_ptr = buffers
+        .map(|maybe_buffer| match maybe_buffer {
+            Some(b) => b as *const std::os::raw::c_void,
+            None => std::ptr::null(),
+        })
+        .collect::<Box<[_]>>();
+    let n_buffers = buffers_ptr.len() as i64;
+
+    let children_ptr = children
+        .map(|child| Box::into_raw(Box::new(child)))
+        .collect::<Box<_>>();
+    let n_children = children_ptr.len() as i64;
+
+    let dictionary_ptr = dictionary.map(|array| Box::into_raw(Box::new(array)));
+
+    let mut private_data = Box::new(PrivateData::<Arc<T>> {
+        data,
+        buffers_ptr,
+        children_ptr,
+        dictionary_ptr,
+    });
+
+    ArrowArray {
+        length: num_rows as i64,
+        null_count: null_count as i64,
+        offset: 0, // IPC files are by definition not offset
+        n_buffers,
+        n_children,
+        buffers: private_data.buffers_ptr.as_mut_ptr(),
+        children: private_data.children_ptr.as_mut_ptr(),
+        dictionary: private_data.dictionary_ptr.unwrap_or(std::ptr::null_mut()),
+        release: Some(release::<Arc<T>>),
+        private_data: Box::into_raw(private_data) as *mut ::std::os::raw::c_void,
+    }
+}
+
+/// callback used to drop [`ArrowArray`] when it is exported
+unsafe extern "C" fn release<T>(array: *mut ArrowArray) {
+    if array.is_null() {
+        return;
+    }
+    let array = &mut *array;
+
+    // take ownership of `private_data`, therefore dropping it
+    let private = Box::from_raw(array.private_data as *mut PrivateData<T>);
+    for child in private.children_ptr.iter() {
+        let _ = Box::from_raw(*child);
+    }
+
+    if let Some(ptr) = private.dictionary_ptr {
+        let _ = Box::from_raw(ptr);
+    }
+
+    array.release = None;
+}
+
+fn mmap_binary<O: Offset, T: AsRef<[u8]>>(
+    data: Arc<T>,
+    node: &Node,
+    block_offset: usize,
+    buffers: &mut VecDeque<IpcBuffer>,
+) -> Result<ArrowArray, Error> {
+    let num_rows: usize = node
+        .length()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let null_count: usize = node
+        .null_count()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let data_ref = data.as_ref().as_ref();
+
+    let validity = get_validity(data_ref, block_offset, buffers, null_count)?.map(|x| x.as_ptr());
+
+    let offsets = get_buffer::<O>(data_ref, block_offset, buffers, num_rows + 1)?.as_ptr();
+    let values = get_buffer::<u8>(data_ref, block_offset, buffers, 0)?.as_ptr();
+
+    // NOTE: offsets and values invariants are _not_ validated
+    Ok(create_array(
+        data,
+        num_rows,
+        null_count,
+        [validity, Some(offsets), Some(values)].into_iter(),
+        [].into_iter(),
+        None,
+    ))
+}
+
+fn mmap_fixed_size_binary<T: AsRef<[u8]>>(
+    data: Arc<T>,
+    node: &Node,
+    block_offset: usize,
+    buffers: &mut VecDeque<IpcBuffer>,
+) -> Result<ArrowArray, Error> {
+    let num_rows: usize = node
+        .length()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let null_count: usize = node
+        .null_count()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let data_ref = data.as_ref().as_ref();
+
+    let validity = get_validity(data_ref, block_offset, buffers, null_count)?.map(|x| x.as_ptr());
+
+    let values = get_buffer::<u8>(data_ref, block_offset, buffers, num_rows + 1)?.as_ptr();
+
+    Ok(create_array(
+        data,
+        num_rows,
+        null_count,
+        [validity, Some(values)].into_iter(),
+        [].into_iter(),
+        None,
+    ))
+}
+
+fn mmap_null<T: AsRef<[u8]>>(
+    data: Arc<T>,
+    node: &Node,
+    _block_offset: usize,
+    _buffers: &mut VecDeque<IpcBuffer>,
+) -> Result<ArrowArray, Error> {
+    let num_rows: usize = node
+        .length()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let null_count: usize = node
+        .null_count()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    Ok(create_array(
+        data,
+        num_rows,
+        null_count,
+        [].into_iter(),
+        [].into_iter(),
+        None,
+    ))
+}
+
+fn mmap_boolean<T: AsRef<[u8]>>(
+    data: Arc<T>,
+    node: &Node,
+    block_offset: usize,
+    buffers: &mut VecDeque<IpcBuffer>,
+) -> Result<ArrowArray, Error> {
+    let num_rows: usize = node
+        .length()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let null_count: usize = node
+        .null_count()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let data_ref = data.as_ref().as_ref();
+
+    let validity = get_validity(data_ref, block_offset, buffers, null_count)?.map(|x| x.as_ptr());
+
+    let values = get_buffer_bounds(buffers)?;
+    let (offset, length) = values;
+
+    // verify that they are in-bounds and get its pointer
+    let values = data_ref[block_offset + offset..block_offset + offset + length].as_ptr();
+
+    Ok(create_array(
+        data,
+        num_rows,
+        null_count,
+        [validity, Some(values)].into_iter(),
+        [].into_iter(),
+        None,
+    ))
+}
+
+fn mmap_primitive<P: NativeType, T: AsRef<[u8]>>(
+    data: Arc<T>,
+    node: &Node,
+    block_offset: usize,
+    buffers: &mut VecDeque<IpcBuffer>,
+) -> Result<ArrowArray, Error> {
+    let data_ref = data.as_ref().as_ref();
+
+    let num_rows: usize = node
+        .length()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let null_count: usize = node
+        .null_count()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let validity = get_validity(data_ref, block_offset, buffers, null_count)?.map(|x| x.as_ptr());
+
+    let values = get_buffer::<P>(data_ref, block_offset, buffers, num_rows)?.as_ptr();
+
+    Ok(create_array(
+        data,
+        num_rows,
+        null_count,
+        [validity, Some(values)].into_iter(),
+        [].into_iter(),
+        None,
+    ))
+}
+
+#[allow(clippy::too_many_arguments)]
+fn mmap_list<O: Offset, T: AsRef<[u8]>>(
+    data: Arc<T>,
+    node: &Node,
+    block_offset: usize,
+    data_type: &DataType,
+    ipc_field: &IpcField,
+    dictionaries: &Dictionaries,
+    field_nodes: &mut VecDeque<Node>,
+    buffers: &mut VecDeque<IpcBuffer>,
+) -> Result<ArrowArray, Error> {
+    let child = ListArray::<O>::try_get_child(data_type)?.data_type();
+
+    let num_rows: usize = node
+        .length()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let null_count: usize = node
+        .null_count()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let data_ref = data.as_ref().as_ref();
+
+    let validity = get_validity(data_ref, block_offset, buffers, null_count)?.map(|x| x.as_ptr());
+
+    let offsets = get_buffer::<O>(data_ref, block_offset, buffers, num_rows + 1)?.as_ptr();
+
+    let values = get_array(
+        data.clone(),
+        block_offset,
+        child,
+        &ipc_field.fields[0],
+        dictionaries,
+        field_nodes,
+        buffers,
+    )?;
+
+    // NOTE: offsets and values invariants are _not_ validated
+    Ok(create_array(
+        data,
+        num_rows,
+        null_count,
+        [validity, Some(offsets)].into_iter(),
+        [values].into_iter(),
+        None,
+    ))
+}
+
+#[allow(clippy::too_many_arguments)]
+fn mmap_fixed_size_list<T: AsRef<[u8]>>(
+    data: Arc<T>,
+    node: &Node,
+    block_offset: usize,
+    data_type: &DataType,
+    ipc_field: &IpcField,
+    dictionaries: &Dictionaries,
+    field_nodes: &mut VecDeque<Node>,
+    buffers: &mut VecDeque<IpcBuffer>,
+) -> Result<ArrowArray, Error> {
+    let child = FixedSizeListArray::try_child_and_size(data_type)?
+        .0
+        .data_type();
+
+    let num_rows: usize = node
+        .length()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let null_count: usize = node
+        .null_count()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let data_ref = data.as_ref().as_ref();
+
+    let validity = get_validity(data_ref, block_offset, buffers, null_count)?.map(|x| x.as_ptr());
+
+    let values = get_array(
+        data.clone(),
+        block_offset,
+        child,
+        &ipc_field.fields[0],
+        dictionaries,
+        field_nodes,
+        buffers,
+    )?;
+
+    Ok(create_array(
+        data,
+        num_rows,
+        null_count,
+        [validity].into_iter(),
+        [values].into_iter(),
+        None,
+    ))
+}
+
+#[allow(clippy::too_many_arguments)]
+fn mmap_struct<T: AsRef<[u8]>>(
+    data: Arc<T>,
+    node: &Node,
+    block_offset: usize,
+    data_type: &DataType,
+    ipc_field: &IpcField,
+    dictionaries: &Dictionaries,
+    field_nodes: &mut VecDeque<Node>,
+    buffers: &mut VecDeque<IpcBuffer>,
+) -> Result<ArrowArray, Error> {
+    let children = StructArray::try_get_fields(data_type)?;
+
+    let num_rows: usize = node
+        .length()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let null_count: usize = node
+        .null_count()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let data_ref = data.as_ref().as_ref();
+
+    let validity = get_validity(data_ref, block_offset, buffers, null_count)?.map(|x| x.as_ptr());
+
+    let values = children
+        .iter()
+        .map(|f| &f.data_type)
+        .zip(ipc_field.fields.iter())
+        .map(|(child, ipc)| {
+            get_array(
+                data.clone(),
+                block_offset,
+                child,
+                ipc,
+                dictionaries,
+                field_nodes,
+                buffers,
+            )
+        })
+        .collect::<Result<Vec<_>, Error>>()?;
+
+    Ok(create_array(
+        data,
+        num_rows,
+        null_count,
+        [validity].into_iter(),
+        values.into_iter(),
+        None,
+    ))
+}
+
+#[allow(clippy::too_many_arguments)]
+fn mmap_dict<K: DictionaryKey, T: AsRef<[u8]>>(
+    data: Arc<T>,
+    node: &Node,
+    block_offset: usize,
+    _: &DataType,
+    ipc_field: &IpcField,
+    dictionaries: &Dictionaries,
+    _: &mut VecDeque<Node>,
+    buffers: &mut VecDeque<IpcBuffer>,
+) -> Result<ArrowArray, Error> {
+    let num_rows: usize = node
+        .length()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let null_count: usize = node
+        .null_count()
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let data_ref = data.as_ref().as_ref();
+
+    let dictionary = dictionaries
+        .get(&ipc_field.dictionary_id.unwrap())
+        .ok_or_else(|| Error::oos("Missing dictionary"))?
+        .clone();
+
+    let validity = get_validity(data_ref, block_offset, buffers, null_count)?.map(|x| x.as_ptr());
+
+    let values = get_buffer::<K>(data_ref, block_offset, buffers, num_rows)?.as_ptr();
+
+    Ok(create_array(
+        data,
+        num_rows,
+        null_count,
+        [validity, Some(values)].into_iter(),
+        [].into_iter(),
+        Some(export_array_to_c(dictionary)),
+    ))
+}
+
+fn get_array<T: AsRef<[u8]>>(
+    data: Arc<T>,
+    block_offset: usize,
+    data_type: &DataType,
+    ipc_field: &IpcField,
+    dictionaries: &Dictionaries,
+    field_nodes: &mut VecDeque<Node>,
+    buffers: &mut VecDeque<IpcBuffer>,
+) -> Result<ArrowArray, Error> {
+    use crate::datatypes::PhysicalType::*;
+    let node = field_nodes
+        .pop_front()
+        .ok_or_else(|| Error::from(OutOfSpecKind::ExpectedBuffer))?;
+
+    match data_type.to_physical_type() {
+        Null => mmap_null(data, &node, block_offset, buffers),
+        Boolean => mmap_boolean(data, &node, block_offset, buffers),
+        Primitive(p) => with_match_primitive_type!(p, |$T| {
+            mmap_primitive::<$T, _>(data, &node, block_offset, buffers)
+        }),
+        Utf8 | Binary => mmap_binary::<i32, _>(data, &node, block_offset, buffers),
+        FixedSizeBinary => mmap_fixed_size_binary(data, &node, block_offset, buffers),
+        LargeBinary | LargeUtf8 => mmap_binary::<i64, _>(data, &node, block_offset, buffers),
+        List => mmap_list::<i32, _>(
+            data,
+            &node,
+            block_offset,
+            data_type,
+            ipc_field,
+            dictionaries,
+            field_nodes,
+            buffers,
+        ),
+        LargeList => mmap_list::<i64, _>(
+            data,
+            &node,
+            block_offset,
+            data_type,
+            ipc_field,
+            dictionaries,
+            field_nodes,
+            buffers,
+        ),
+        FixedSizeList => mmap_fixed_size_list(
+            data,
+            &node,
+            block_offset,
+            data_type,
+            ipc_field,
+            dictionaries,
+            field_nodes,
+            buffers,
+        ),
+        Struct => mmap_struct(
+            data,
+            &node,
+            block_offset,
+            data_type,
+            ipc_field,
+            dictionaries,
+            field_nodes,
+            buffers,
+        ),
+        Dictionary(key_type) => match_integer_type!(key_type, |$T| {
+            mmap_dict::<$T, _>(
+                data,
+                &node,
+                block_offset,
+                data_type,
+                ipc_field,
+                dictionaries,
+                field_nodes,
+                buffers,
+            )
+        }),
+        _ => todo!(),
+    }
+}
+
+/// Maps a memory region to an [`Array`].
+pub(crate) unsafe fn mmap<T: AsRef<[u8]>>(
+    data: Arc<T>,
+    block_offset: usize,
+    data_type: DataType,
+    ipc_field: &IpcField,
+    dictionaries: &Dictionaries,
+    field_nodes: &mut VecDeque<Node>,
+    buffers: &mut VecDeque<IpcBuffer>,
+) -> Result<Box<dyn Array>, Error> {
+    let array = get_array(
+        data,
+        block_offset,
+        &data_type,
+        ipc_field,
+        dictionaries,
+        field_nodes,
+        buffers,
+    )?;
+    // The unsafety comes from the fact that `array` is not necessarily valid -
+    // the IPC file may be corrupted (e.g. invalid offsets or non-utf8 data)
+    unsafe { try_from(InternalArrowArray::new(array, data_type)) }
+}
diff --git a/src/ffi/mod.rs b/src/ffi/mod.rs
index 10fc2fb994..b416d1c464 100644
--- a/src/ffi/mod.rs
+++ b/src/ffi/mod.rs
@@ -3,6 +3,9 @@
 mod array;
 mod bridge;
 mod generated;
+#[cfg(feature = "io_ipc")]
+#[cfg_attr(docsrs, doc(cfg(feature = "io_ipc")))]
+pub(crate) mod mmap;
 mod schema;
 mod stream;
 
@@ -44,5 +47,5 @@ pub unsafe fn import_array_from_c(
     array: ArrowArray,
     data_type: DataType,
 ) -> Result<Box<dyn Array>> {
-    try_from(Box::new(InternalArrowArray::new(array, data_type)))
+    try_from(InternalArrowArray::new(array, data_type))
 }
diff --git a/src/io/avro/mod.rs b/src/io/avro/mod.rs
index 5addae95e4..bf7bda85f1 100644
--- a/src/io/avro/mod.rs
+++ b/src/io/avro/mod.rs
@@ -1,23 +1,16 @@
 //! Read and write from and to Apache Avro
 
-pub mod read;
-#[cfg(feature = "io_avro_async")]
-#[cfg_attr(docsrs, doc(cfg(feature = "io_avro_async")))]
-pub mod read_async;
-pub mod write;
-#[cfg(feature = "io_avro_async")]
-#[cfg_attr(docsrs, doc(cfg(feature = "io_avro_async")))]
-pub mod write_async;
+pub use avro_schema;
 
-/// Valid compressions
-#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
-pub enum Compression {
-    /// Deflate
-    Deflate,
-    /// Snappy
-    Snappy,
+impl From<avro_schema::error::Error> for crate::error::Error {
+    fn from(error: avro_schema::error::Error) -> Self {
+        Self::ExternalFormat(error.to_string())
+    }
 }
 
+pub mod read;
+pub mod write;
+
 // macros that can operate in sync and async code.
 macro_rules! avro_decode {
     ($reader:ident $($_await:tt)*) => {
@@ -46,86 +39,4 @@ macro_rules! avro_decode {
     }
 }
 
-macro_rules! read_header {
-    ($reader:ident $($_await:tt)*) => {{
-        let mut items = ahash::AHashMap::new();
-
-        loop {
-            let len = zigzag_i64($reader)$($_await)*? as usize;
-            if len == 0 {
-                break Ok(items);
-            }
-
-            items.reserve(len);
-            for _ in 0..len {
-                let key = _read_binary($reader)$($_await)*?;
-                let key = String::from_utf8(key)
-                    .map_err(|_| Error::ExternalFormat("Invalid Avro header".to_string()))?;
-                let value = _read_binary($reader)$($_await)*?;
-                items.insert(key, value);
-            }
-        }
-    }};
-}
-
-macro_rules! read_metadata {
-    ($reader:ident $($_await:tt)*) => {{
-        let mut magic_number = [0u8; 4];
-        $reader.read_exact(&mut magic_number)$($_await)*?;
-
-        // see https://avro.apache.org/docs/current/spec.html#Object+Container+Files
-        if magic_number != [b'O', b'b', b'j', 1u8] {
-            return Err(Error::ExternalFormat(
-                "Avro header does not contain a valid magic number".to_string(),
-            ));
-        }
-
-        let header = read_header($reader)$($_await)*?;
-
-        let (schema, compression) = deserialize_header(header)?;
-
-        let marker = read_file_marker($reader)$($_await)*?;
-
-        Ok((schema, compression, marker))
-    }};
-}
-
-pub(crate) use {avro_decode, read_header, read_metadata};
-
-/// A compressed Avro block.
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct CompressedBlock {
-    /// The number of rows
-    pub number_of_rows: usize,
-    /// The compressed data
-    pub data: Vec<u8>,
-}
-
-impl CompressedBlock {
-    /// Creates a new CompressedBlock
-    pub fn new(number_of_rows: usize, data: Vec<u8>) -> Self {
-        Self {
-            number_of_rows,
-            data,
-        }
-    }
-}
-
-/// An uncompressed Avro block.
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct Block {
-    /// The number of rows
-    pub number_of_rows: usize,
-    /// The uncompressed data
-    pub data: Vec<u8>,
-}
-
-impl Block {
-    /// Creates a new Block
-    pub fn new(number_of_rows: usize, data: Vec<u8>) -> Self {
-        Self {
-            number_of_rows,
-            data,
-        }
-    }
-}
+pub(crate) use avro_decode;
diff --git a/src/io/avro/read/block.rs b/src/io/avro/read/block.rs
deleted file mode 100644
index 62cc5a88ef..0000000000
--- a/src/io/avro/read/block.rs
+++ /dev/null
@@ -1,101 +0,0 @@
-//! APIs to read from Avro format to arrow.
-use std::io::Read;
-
-use fallible_streaming_iterator::FallibleStreamingIterator;
-
-use crate::error::{Error, Result};
-
-use super::super::CompressedBlock;
-use super::util;
-
-fn read_size<R: Read>(reader: &mut R) -> Result<(usize, usize)> {
-    let rows = match util::zigzag_i64(reader) {
-        Ok(a) => a,
-        Err(Error::Io(io_err)) => {
-            if let std::io::ErrorKind::UnexpectedEof = io_err.kind() {
-                // end
-                return Ok((0, 0));
-            } else {
-                return Err(Error::Io(io_err));
-            }
-        }
-        Err(other) => return Err(other),
-    };
-    let bytes = util::zigzag_i64(reader)?;
-    Ok((rows as usize, bytes as usize))
-}
-
-/// Reads a [`CompressedBlock`] from the `reader`.
-/// # Error
-/// This function errors iff either the block cannot be read or the sync marker does not match
-fn read_block<R: Read>(
-    reader: &mut R,
-    block: &mut CompressedBlock,
-    file_marker: [u8; 16],
-) -> Result<()> {
-    let (rows, bytes) = read_size(reader)?;
-    block.number_of_rows = rows;
-    if rows == 0 {
-        return Ok(());
-    };
-
-    block.data.clear();
-    block.data.try_reserve(bytes)?;
-    reader.take(bytes as u64).read_to_end(&mut block.data)?;
-
-    let mut marker = [0u8; 16];
-    reader.read_exact(&mut marker)?;
-
-    if marker != file_marker {
-        return Err(Error::ExternalFormat(
-            "Avro: the sync marker in the block does not correspond to the file marker".to_string(),
-        ));
-    }
-    Ok(())
-}
-
-/// [`FallibleStreamingIterator`] of compressed avro blocks
-pub struct BlockStreamIterator<R: Read> {
-    buf: CompressedBlock,
-    reader: R,
-    file_marker: [u8; 16],
-}
-
-impl<R: Read> BlockStreamIterator<R> {
-    /// Creates a new [`BlockStreamIterator`].
-    pub fn new(reader: R, file_marker: [u8; 16]) -> Self {
-        Self {
-            reader,
-            file_marker,
-            buf: CompressedBlock::new(0, vec![]),
-        }
-    }
-
-    /// The buffer of [`BlockStreamIterator`].
-    pub fn buffer(&mut self) -> &mut CompressedBlock {
-        &mut self.buf
-    }
-
-    /// Deconstructs itself
-    pub fn into_inner(self) -> (R, Vec<u8>) {
-        (self.reader, self.buf.data)
-    }
-}
-
-impl<R: Read> FallibleStreamingIterator for BlockStreamIterator<R> {
-    type Error = Error;
-    type Item = CompressedBlock;
-
-    fn advance(&mut self) -> Result<()> {
-        read_block(&mut self.reader, &mut self.buf, self.file_marker)?;
-        Ok(())
-    }
-
-    fn get(&self) -> Option<&Self::Item> {
-        if self.buf.number_of_rows > 0 {
-            Some(&self.buf)
-        } else {
-            None
-        }
-    }
-}
diff --git a/src/io/avro/read/decompress.rs b/src/io/avro/read/decompress.rs
deleted file mode 100644
index bc5fcbd582..0000000000
--- a/src/io/avro/read/decompress.rs
+++ /dev/null
@@ -1,119 +0,0 @@
-//! APIs to read from Avro format to arrow.
-use std::io::Read;
-
-use fallible_streaming_iterator::FallibleStreamingIterator;
-
-use crate::error::{Error, Result};
-
-use super::super::{Block, CompressedBlock};
-use super::BlockStreamIterator;
-use super::Compression;
-
-#[cfg(feature = "io_avro_compression")]
-const CRC_TABLE: crc::Crc<u32> = crc::Crc::<u32>::new(&crc::CRC_32_ISO_HDLC);
-
-/// Decompresses an Avro block.
-/// Returns whether the buffers where swapped.
-pub fn decompress_block(
-    block: &mut CompressedBlock,
-    decompressed: &mut Block,
-    compression: Option<Compression>,
-) -> Result<bool> {
-    decompressed.number_of_rows = block.number_of_rows;
-    let block = &mut block.data;
-    let decompressed = &mut decompressed.data;
-
-    match compression {
-        None => {
-            std::mem::swap(block, decompressed);
-            Ok(true)
-        }
-        #[cfg(feature = "io_avro_compression")]
-        Some(Compression::Deflate) => {
-            decompressed.clear();
-            let mut decoder = libflate::deflate::Decoder::new(&block[..]);
-            decoder.read_to_end(decompressed)?;
-            Ok(false)
-        }
-        #[cfg(feature = "io_avro_compression")]
-        Some(Compression::Snappy) => {
-            let crc = &block[block.len() - 4..];
-            let block = &block[..block.len() - 4];
-
-            let len = snap::raw::decompress_len(block)
-                .map_err(|e| Error::ExternalFormat(e.to_string()))?;
-            decompressed.clear();
-            decompressed.resize(len, 0);
-            snap::raw::Decoder::new()
-                .decompress(block, decompressed)
-                .map_err(|e| Error::ExternalFormat(e.to_string()))?;
-
-            let expected_crc = u32::from_be_bytes([crc[0], crc[1], crc[2], crc[3]]);
-
-            let actual_crc = CRC_TABLE.checksum(decompressed);
-            if expected_crc != actual_crc {
-                return Err(Error::ExternalFormat(
-                    "The crc of snap-compressed block does not match".to_string(),
-                ));
-            }
-            Ok(false)
-        }
-        #[cfg(not(feature = "io_avro_compression"))]
-        Some(Compression::Deflate) => Err(Error::InvalidArgumentError(
-            "The avro file is deflate-encoded but feature 'io_avro_compression' is not active."
-                .to_string(),
-        )),
-        #[cfg(not(feature = "io_avro_compression"))]
-        Some(Compression::Snappy) => Err(Error::InvalidArgumentError(
-            "The avro file is snappy-encoded but feature 'io_avro_compression' is not active."
-                .to_string(),
-        )),
-    }
-}
-
-/// [`FallibleStreamingIterator`] of decompressed Avro blocks
-pub struct Decompressor<R: Read> {
-    blocks: BlockStreamIterator<R>,
-    codec: Option<Compression>,
-    buf: Block,
-    was_swapped: bool,
-}
-
-impl<R: Read> Decompressor<R> {
-    /// Creates a new [`Decompressor`].
-    pub fn new(blocks: BlockStreamIterator<R>, codec: Option<Compression>) -> Self {
-        Self {
-            blocks,
-            codec,
-            buf: Block::new(0, vec![]),
-            was_swapped: false,
-        }
-    }
-
-    /// Deconstructs itself into its internal reader
-    pub fn into_inner(self) -> R {
-        self.blocks.into_inner().0
-    }
-}
-
-impl<R: Read> FallibleStreamingIterator for Decompressor<R> {
-    type Error = Error;
-    type Item = Block;
-
-    fn advance(&mut self) -> Result<()> {
-        if self.was_swapped {
-            std::mem::swap(&mut self.blocks.buffer().data, &mut self.buf.data);
-        }
-        self.blocks.advance()?;
-        self.was_swapped = decompress_block(self.blocks.buffer(), &mut self.buf, self.codec)?;
-        Ok(())
-    }
-
-    fn get(&self) -> Option<&Self::Item> {
-        if self.buf.number_of_rows > 0 {
-            Some(&self.buf)
-        } else {
-            None
-        }
-    }
-}
diff --git a/src/io/avro/read/deserialize.rs b/src/io/avro/read/deserialize.rs
index 1bce0fb52e..9fa9f15254 100644
--- a/src/io/avro/read/deserialize.rs
+++ b/src/io/avro/read/deserialize.rs
@@ -1,7 +1,8 @@
 use std::convert::TryInto;
 
-use avro_schema::Record;
-use avro_schema::{Enum, Schema as AvroSchema};
+use avro_schema::file::Block;
+use avro_schema::schema::Record;
+use avro_schema::schema::{Enum, Field as AvroField, Schema as AvroSchema};
 
 use crate::array::*;
 use crate::chunk::Chunk;
@@ -10,13 +11,12 @@ use crate::error::Error;
 use crate::error::Result;
 use crate::types::months_days_ns;
 
-use super::super::Block;
 use super::nested::*;
 use super::util;
 
 fn make_mutable(
     data_type: &DataType,
-    avro_schema: Option<&AvroSchema>,
+    avro_field: Option<&AvroSchema>,
     capacity: usize,
 ) -> Result<Box<dyn MutableArray>> {
     Ok(match data_type.to_physical_type() {
@@ -34,7 +34,7 @@ fn make_mutable(
             Box::new(MutableUtf8Array::<i32>::with_capacity(capacity)) as Box<dyn MutableArray>
         }
         PhysicalType::Dictionary(_) => {
-            if let Some(AvroSchema::Enum(Enum { symbols, .. })) = avro_schema {
+            if let Some(AvroSchema::Enum(Enum { symbols, .. })) = avro_field {
                 let values = Utf8Array::<i32>::from_slice(symbols);
                 Box::new(FixedItemsUtf8Dictionary::with_capacity(values, capacity))
                     as Box<dyn MutableArray>
@@ -437,24 +437,30 @@ fn skip_item<'a>(field: &Field, avro_field: &AvroSchema, mut block: &'a [u8]) ->
     Ok(block)
 }
 
-/// Deserializes a [`Block`] into [`Chunk`], projected
+/// Deserializes a [`Block`] assumed to be encoded according to [`AvroField`] into [`Chunk`],
+/// using `projection` to ignore `avro_fields`.
+/// # Panics
+/// `fields`, `avro_fields` and `projection` must have the same length.
 pub fn deserialize(
     block: &Block,
     fields: &[Field],
-    avro_schemas: &[AvroSchema],
+    avro_fields: &[AvroField],
     projection: &[bool],
 ) -> Result<Chunk<Box<dyn Array>>> {
+    assert_eq!(fields.len(), avro_fields.len());
+    assert_eq!(fields.len(), projection.len());
+
     let rows = block.number_of_rows;
     let mut block = block.data.as_ref();
 
     // create mutables, one per field
     let mut arrays: Vec<Box<dyn MutableArray>> = fields
         .iter()
-        .zip(avro_schemas.iter())
+        .zip(avro_fields.iter())
         .zip(projection.iter())
-        .map(|((field, avro_schema), projection)| {
+        .map(|((field, avro_field), projection)| {
             if *projection {
-                make_mutable(&field.data_type, Some(avro_schema), rows)
+                make_mutable(&field.data_type, Some(&avro_field.schema), rows)
             } else {
                 // just something; we are not going to use it
                 make_mutable(&DataType::Int32, None, 0)
@@ -467,14 +473,14 @@ pub fn deserialize(
         let iter = arrays
             .iter_mut()
             .zip(fields.iter())
-            .zip(avro_schemas.iter())
+            .zip(avro_fields.iter())
             .zip(projection.iter());
 
         for (((array, field), avro_field), projection) in iter {
             block = if *projection {
-                deserialize_item(array.as_mut(), field.is_nullable, avro_field, block)
+                deserialize_item(array.as_mut(), field.is_nullable, &avro_field.schema, block)
             } else {
-                skip_item(field, avro_field, block)
+                skip_item(field, &avro_field.schema, block)
             }?
         }
     }
diff --git a/src/io/avro/read/header.rs b/src/io/avro/read/header.rs
deleted file mode 100644
index c985ce2508..0000000000
--- a/src/io/avro/read/header.rs
+++ /dev/null
@@ -1,29 +0,0 @@
-use ahash::AHashMap;
-use avro_schema::Schema;
-use serde_json;
-
-use crate::error::{Error, Result};
-
-use super::Compression;
-
-/// Deserializes the Avro header into an Avro [`Schema`] and optional [`Compression`].
-pub(crate) fn deserialize_header(
-    header: AHashMap<String, Vec<u8>>,
-) -> Result<(Schema, Option<Compression>)> {
-    let schema = header
-        .get("avro.schema")
-        .ok_or_else(|| Error::ExternalFormat("Avro schema must be present".to_string()))
-        .and_then(|bytes| {
-            serde_json::from_slice(bytes.as_ref()).map_err(|e| Error::ExternalFormat(e.to_string()))
-        })?;
-
-    let compression = header.get("avro.codec").and_then(|bytes| {
-        let bytes: &[u8] = bytes.as_ref();
-        match bytes {
-            b"snappy" => Some(Compression::Snappy),
-            b"deflate" => Some(Compression::Deflate),
-            _ => None,
-        }
-    });
-    Ok((schema, compression))
-}
diff --git a/src/io/avro/read/mod.rs b/src/io/avro/read/mod.rs
index f02af4052e..5014499c12 100644
--- a/src/io/avro/read/mod.rs
+++ b/src/io/avro/read/mod.rs
@@ -1,51 +1,28 @@
 //! APIs to read from Avro format to arrow.
 use std::io::Read;
 
-use avro_schema::{Record, Schema as AvroSchema};
-use fallible_streaming_iterator::FallibleStreamingIterator;
+use avro_schema::file::FileMetadata;
+use avro_schema::read::fallible_streaming_iterator::FallibleStreamingIterator;
+use avro_schema::read::{block_iterator, BlockStreamingIterator};
+use avro_schema::schema::Field as AvroField;
 
-mod block;
-mod decompress;
-pub use block::BlockStreamIterator;
-pub use decompress::{decompress_block, Decompressor};
 mod deserialize;
 pub use deserialize::deserialize;
-mod header;
 mod nested;
 mod schema;
 mod util;
 
-pub(super) use header::deserialize_header;
-pub(super) use schema::infer_schema;
+pub use schema::infer_schema;
 
 use crate::array::Array;
 use crate::chunk::Chunk;
-use crate::datatypes::{Field, Schema};
+use crate::datatypes::Field;
 use crate::error::Result;
 
-use super::Compression;
-
-/// Reads the avro metadata from `reader` into a [`Schema`], [`Compression`] and magic marker.
-#[allow(clippy::type_complexity)]
-pub fn read_metadata<R: std::io::Read>(
-    reader: &mut R,
-) -> Result<(Vec<AvroSchema>, Schema, Option<Compression>, [u8; 16])> {
-    let (avro_schema, codec, marker) = util::read_schema(reader)?;
-    let schema = infer_schema(&avro_schema)?;
-
-    let avro_schema = if let AvroSchema::Record(Record { fields, .. }) = avro_schema {
-        fields.into_iter().map(|x| x.schema).collect()
-    } else {
-        panic!()
-    };
-
-    Ok((avro_schema, schema, codec, marker))
-}
-
 /// Single threaded, blocking reader of Avro; [`Iterator`] of [`Chunk`].
 pub struct Reader<R: Read> {
-    iter: Decompressor<R>,
-    avro_schemas: Vec<AvroSchema>,
+    iter: BlockStreamingIterator<R>,
+    avro_fields: Vec<AvroField>,
     fields: Vec<Field>,
     projection: Vec<bool>,
 }
@@ -53,15 +30,16 @@ pub struct Reader<R: Read> {
 impl<R: Read> Reader<R> {
     /// Creates a new [`Reader`].
     pub fn new(
-        iter: Decompressor<R>,
-        avro_schemas: Vec<AvroSchema>,
+        reader: R,
+        metadata: FileMetadata,
         fields: Vec<Field>,
         projection: Option<Vec<bool>>,
     ) -> Self {
         let projection = projection.unwrap_or_else(|| fields.iter().map(|_| true).collect());
+
         Self {
-            iter,
-            avro_schemas,
+            iter: block_iterator(reader, metadata.compression, metadata.marker),
+            avro_fields: metadata.record.fields,
             fields,
             projection,
         }
@@ -78,12 +56,12 @@ impl<R: Read> Iterator for Reader<R> {
 
     fn next(&mut self) -> Option<Self::Item> {
         let fields = &self.fields[..];
-        let avro_schemas = &self.avro_schemas;
+        let avro_fields = &self.avro_fields;
         let projection = &self.projection;
 
         self.iter
             .next()
             .transpose()
-            .map(|maybe_block| deserialize(maybe_block?, fields, avro_schemas, projection))
+            .map(|maybe_block| deserialize(maybe_block?, fields, avro_fields, projection))
     }
 }
diff --git a/src/io/avro/read/nested.rs b/src/io/avro/read/nested.rs
index 450930cad7..981b6894f9 100644
--- a/src/io/avro/read/nested.rs
+++ b/src/io/avro/read/nested.rs
@@ -112,6 +112,10 @@ impl<O: Offset> MutableArray for DynMutableListArray<O> {
         self.push_null()
     }
 
+    fn reserve(&mut self, _: usize) {
+        todo!();
+    }
+
     fn shrink_to_fit(&mut self) {
         todo!();
     }
@@ -195,6 +199,10 @@ impl MutableArray for FixedItemsUtf8Dictionary {
         self.push_null()
     }
 
+    fn reserve(&mut self, _: usize) {
+        todo!();
+    }
+
     fn shrink_to_fit(&mut self) {
         todo!();
     }
@@ -287,6 +295,10 @@ impl MutableArray for DynMutableStructArray {
         self.push_null()
     }
 
+    fn reserve(&mut self, _: usize) {
+        todo!();
+    }
+
     fn shrink_to_fit(&mut self) {
         todo!();
     }
diff --git a/src/io/avro/read/schema.rs b/src/io/avro/read/schema.rs
index 13408efd9c..b45fc95c0e 100644
--- a/src/io/avro/read/schema.rs
+++ b/src/io/avro/read/schema.rs
@@ -1,4 +1,4 @@
-use avro_schema::{Enum, Fixed, Record, Schema as AvroSchema};
+use avro_schema::schema::{Enum, Fixed, Record, Schema as AvroSchema};
 
 use crate::datatypes::*;
 use crate::error::{Error, Result};
@@ -19,26 +19,21 @@ fn external_props(schema: &AvroSchema) -> Metadata {
     props
 }
 
-/// Infers an [`Schema`] from the root [`AvroSchema`].
+/// Infers an [`Schema`] from the root [`Record`].
 /// This
-pub fn infer_schema(schema: &AvroSchema) -> Result<Schema> {
-    if let AvroSchema::Record(Record { fields, .. }) = schema {
-        Ok(fields
-            .iter()
-            .map(|field| {
-                schema_to_field(
-                    &field.schema,
-                    Some(&field.name),
-                    external_props(&field.schema),
-                )
-            })
-            .collect::<Result<Vec<_>>>()?
-            .into())
-    } else {
-        Err(Error::OutOfSpec(
-            "The root AvroSchema must be of type Record".to_string(),
-        ))
-    }
+pub fn infer_schema(record: &Record) -> Result<Schema> {
+    Ok(record
+        .fields
+        .iter()
+        .map(|field| {
+            schema_to_field(
+                &field.schema,
+                Some(&field.name),
+                external_props(&field.schema),
+            )
+        })
+        .collect::<Result<Vec<_>>>()?
+        .into())
 }
 
 fn schema_to_field(schema: &AvroSchema, name: Option<&str>, props: Metadata) -> Result<Field> {
@@ -48,24 +43,24 @@ fn schema_to_field(schema: &AvroSchema, name: Option<&str>, props: Metadata) ->
         AvroSchema::Boolean => DataType::Boolean,
         AvroSchema::Int(logical) => match logical {
             Some(logical) => match logical {
-                avro_schema::IntLogical::Date => DataType::Date32,
-                avro_schema::IntLogical::Time => DataType::Time32(TimeUnit::Millisecond),
+                avro_schema::schema::IntLogical::Date => DataType::Date32,
+                avro_schema::schema::IntLogical::Time => DataType::Time32(TimeUnit::Millisecond),
             },
             None => DataType::Int32,
         },
         AvroSchema::Long(logical) => match logical {
             Some(logical) => match logical {
-                avro_schema::LongLogical::Time => DataType::Time64(TimeUnit::Microsecond),
-                avro_schema::LongLogical::TimestampMillis => {
+                avro_schema::schema::LongLogical::Time => DataType::Time64(TimeUnit::Microsecond),
+                avro_schema::schema::LongLogical::TimestampMillis => {
                     DataType::Timestamp(TimeUnit::Millisecond, Some("00:00".to_string()))
                 }
-                avro_schema::LongLogical::TimestampMicros => {
+                avro_schema::schema::LongLogical::TimestampMicros => {
                     DataType::Timestamp(TimeUnit::Microsecond, Some("00:00".to_string()))
                 }
-                avro_schema::LongLogical::LocalTimestampMillis => {
+                avro_schema::schema::LongLogical::LocalTimestampMillis => {
                     DataType::Timestamp(TimeUnit::Millisecond, None)
                 }
-                avro_schema::LongLogical::LocalTimestampMicros => {
+                avro_schema::schema::LongLogical::LocalTimestampMicros => {
                     DataType::Timestamp(TimeUnit::Microsecond, None)
                 }
             },
@@ -75,7 +70,7 @@ fn schema_to_field(schema: &AvroSchema, name: Option<&str>, props: Metadata) ->
         AvroSchema::Double => DataType::Float64,
         AvroSchema::Bytes(logical) => match logical {
             Some(logical) => match logical {
-                avro_schema::BytesLogical::Decimal(precision, scale) => {
+                avro_schema::schema::BytesLogical::Decimal(precision, scale) => {
                     DataType::Decimal(*precision, *scale)
                 }
             },
@@ -134,10 +129,10 @@ fn schema_to_field(schema: &AvroSchema, name: Option<&str>, props: Metadata) ->
         }
         AvroSchema::Fixed(Fixed { size, logical, .. }) => match logical {
             Some(logical) => match logical {
-                avro_schema::FixedLogical::Decimal(precision, scale) => {
+                avro_schema::schema::FixedLogical::Decimal(precision, scale) => {
                     DataType::Decimal(*precision, *scale)
                 }
-                avro_schema::FixedLogical::Duration => {
+                avro_schema::schema::FixedLogical::Duration => {
                     DataType::Interval(IntervalUnit::MonthDayNano)
                 }
             },
diff --git a/src/io/avro/read/util.rs b/src/io/avro/read/util.rs
index b14870a390..f3cb07c6b0 100644
--- a/src/io/avro/read/util.rs
+++ b/src/io/avro/read/util.rs
@@ -1,12 +1,8 @@
-use ahash::AHashMap;
 use std::io::Read;
 
-use avro_schema::Schema;
-
 use crate::error::{Error, Result};
 
-use super::super::{avro_decode, read_header, read_metadata};
-use super::{deserialize_header, Compression};
+use super::super::avro_decode;
 
 pub fn zigzag_i64<R: Read>(reader: &mut R) -> Result<i64> {
     let z = decode_variable(reader)?;
@@ -20,28 +16,3 @@ pub fn zigzag_i64<R: Read>(reader: &mut R) -> Result<i64> {
 fn decode_variable<R: Read>(reader: &mut R) -> Result<u64> {
     avro_decode!(reader)
 }
-
-fn _read_binary<R: Read>(reader: &mut R) -> Result<Vec<u8>> {
-    let len: usize = zigzag_i64(reader)? as usize;
-    let mut buf = vec![];
-    buf.try_reserve(len)?;
-    reader.take(len as u64).read_to_end(&mut buf)?;
-    Ok(buf)
-}
-
-fn read_header<R: Read>(reader: &mut R) -> Result<AHashMap<String, Vec<u8>>> {
-    read_header!(reader)
-}
-
-fn read_file_marker<R: Read>(reader: &mut R) -> Result<[u8; 16]> {
-    let mut marker = [0u8; 16];
-    reader.read_exact(&mut marker)?;
-    Ok(marker)
-}
-
-/// Reads the schema from `reader`, returning the file's [`Schema`] and [`Compression`].
-/// # Error
-/// This function errors iff the header is not a valid avro file header.
-pub fn read_schema<R: Read>(reader: &mut R) -> Result<(Schema, Option<Compression>, [u8; 16])> {
-    read_metadata!(reader)
-}
diff --git a/src/io/avro/read_async/block.rs b/src/io/avro/read_async/block.rs
deleted file mode 100644
index ff92d6a97e..0000000000
--- a/src/io/avro/read_async/block.rs
+++ /dev/null
@@ -1,77 +0,0 @@
-//! APIs to read from Avro format to arrow.
-use async_stream::try_stream;
-use futures::AsyncRead;
-use futures::AsyncReadExt;
-use futures::Stream;
-
-use crate::error::{Error, Result};
-
-use super::CompressedBlock;
-
-use super::utils::zigzag_i64;
-
-async fn read_size<R: AsyncRead + Unpin + Send>(reader: &mut R) -> Result<(usize, usize)> {
-    let rows = match zigzag_i64(reader).await {
-        Ok(a) => a,
-        Err(Error::Io(io_err)) => {
-            if let std::io::ErrorKind::UnexpectedEof = io_err.kind() {
-                // end
-                return Ok((0, 0));
-            } else {
-                return Err(Error::Io(io_err));
-            }
-        }
-        Err(other) => return Err(other),
-    };
-    let bytes = zigzag_i64(reader).await?;
-    Ok((rows as usize, bytes as usize))
-}
-
-/// Reads a [`CompressedBlock`] from the `reader`.
-/// # Error
-/// This function errors iff either the block cannot be read or the sync marker does not match
-async fn read_block<R: AsyncRead + Unpin + Send>(
-    reader: &mut R,
-    block: &mut CompressedBlock,
-    file_marker: [u8; 16],
-) -> Result<()> {
-    let (rows, bytes) = read_size(reader).await?;
-    block.number_of_rows = rows;
-    if rows == 0 {
-        return Ok(());
-    };
-
-    block.data.clear();
-    block.data.try_reserve(bytes)?;
-    reader
-        .take(bytes as u64)
-        .read_to_end(&mut block.data)
-        .await?;
-
-    let mut marker = [0u8; 16];
-    reader.read_exact(&mut marker).await?;
-
-    if marker != file_marker {
-        return Err(Error::ExternalFormat(
-            "Avro: the sync marker in the block does not correspond to the file marker".to_string(),
-        ));
-    }
-    Ok(())
-}
-
-/// Returns a fallible [`Stream`] of Avro blocks bound to `reader`
-pub async fn block_stream<R: AsyncRead + Unpin + Send>(
-    reader: &mut R,
-    file_marker: [u8; 16],
-) -> impl Stream<Item = Result<CompressedBlock>> + '_ {
-    try_stream! {
-        loop {
-            let mut block = CompressedBlock::new(0, vec![]);
-            read_block(reader, &mut block, file_marker).await?;
-            if block.number_of_rows == 0 {
-                break
-            }
-            yield block
-        }
-    }
-}
diff --git a/src/io/avro/read_async/metadata.rs b/src/io/avro/read_async/metadata.rs
deleted file mode 100644
index 213428e0f0..0000000000
--- a/src/io/avro/read_async/metadata.rs
+++ /dev/null
@@ -1,60 +0,0 @@
-//! Async Avro
-use ahash::AHashMap;
-use avro_schema::{Record, Schema as AvroSchema};
-use futures::AsyncRead;
-use futures::AsyncReadExt;
-
-use crate::datatypes::Schema;
-use crate::error::{Error, Result};
-
-use super::super::read::deserialize_header;
-use super::super::read::infer_schema;
-use super::super::Compression;
-use super::super::{read_header, read_metadata};
-use super::utils::zigzag_i64;
-
-/// Reads Avro's metadata from `reader` into a [`AvroSchema`], [`Compression`] and magic marker.
-#[allow(clippy::type_complexity)]
-async fn read_metadata_async<R: AsyncRead + Unpin + Send>(
-    reader: &mut R,
-) -> Result<(AvroSchema, Option<Compression>, [u8; 16])> {
-    read_metadata!(reader.await)
-}
-
-/// Reads the avro metadata from `reader` into a [`AvroSchema`], [`Compression`] and magic marker.
-#[allow(clippy::type_complexity)]
-pub async fn read_metadata<R: AsyncRead + Unpin + Send>(
-    reader: &mut R,
-) -> Result<(Vec<AvroSchema>, Schema, Option<Compression>, [u8; 16])> {
-    let (avro_schema, codec, marker) = read_metadata_async(reader).await?;
-    let schema = infer_schema(&avro_schema)?;
-
-    let avro_schema = if let AvroSchema::Record(Record { fields, .. }) = avro_schema {
-        fields.into_iter().map(|x| x.schema).collect()
-    } else {
-        panic!()
-    };
-
-    Ok((avro_schema, schema, codec, marker))
-}
-
-/// Reads the file marker asynchronously
-async fn read_file_marker<R: AsyncRead + Unpin + Send>(reader: &mut R) -> Result<[u8; 16]> {
-    let mut marker = [0u8; 16];
-    reader.read_exact(&mut marker).await?;
-    Ok(marker)
-}
-
-async fn _read_binary<R: AsyncRead + Unpin + Send>(reader: &mut R) -> Result<Vec<u8>> {
-    let len: usize = zigzag_i64(reader).await? as usize;
-    let mut buf = vec![];
-    buf.try_reserve(len)?;
-    reader.take(len as u64).read_to_end(&mut buf).await?;
-    Ok(buf)
-}
-
-async fn read_header<R: AsyncRead + Unpin + Send>(
-    reader: &mut R,
-) -> Result<AHashMap<String, Vec<u8>>> {
-    read_header!(reader.await)
-}
diff --git a/src/io/avro/read_async/mod.rs b/src/io/avro/read_async/mod.rs
deleted file mode 100644
index dc3d7c276b..0000000000
--- a/src/io/avro/read_async/mod.rs
+++ /dev/null
@@ -1,9 +0,0 @@
-//! Async read Avro
-
-mod block;
-mod metadata;
-pub(self) mod utils;
-
-pub use super::{Block, CompressedBlock};
-pub use block::block_stream;
-pub use metadata::read_metadata;
diff --git a/src/io/avro/read_async/utils.rs b/src/io/avro/read_async/utils.rs
deleted file mode 100644
index 25ed898d09..0000000000
--- a/src/io/avro/read_async/utils.rs
+++ /dev/null
@@ -1,19 +0,0 @@
-use futures::AsyncRead;
-use futures::AsyncReadExt;
-
-use crate::error::{Error, Result};
-
-use super::super::avro_decode;
-
-pub async fn zigzag_i64<R: AsyncRead + Unpin + Send>(reader: &mut R) -> Result<i64> {
-    let z = decode_variable(reader).await?;
-    Ok(if z & 0x1 == 0 {
-        (z >> 1) as i64
-    } else {
-        !(z >> 1) as i64
-    })
-}
-
-async fn decode_variable<R: AsyncRead + Unpin + Send>(reader: &mut R) -> Result<u64> {
-    avro_decode!(reader.await)
-}
diff --git a/src/io/avro/write/block.rs b/src/io/avro/write/block.rs
deleted file mode 100644
index 0758c95c4a..0000000000
--- a/src/io/avro/write/block.rs
+++ /dev/null
@@ -1,19 +0,0 @@
-use std::io::Write;
-
-use crate::error::Result;
-
-use super::super::CompressedBlock;
-use super::{util::zigzag_encode, SYNC_NUMBER};
-
-/// Writes a [`CompressedBlock`] to `writer`
-pub fn write_block<W: Write>(writer: &mut W, compressed_block: &CompressedBlock) -> Result<()> {
-    // write size and rows
-    zigzag_encode(compressed_block.number_of_rows as i64, writer)?;
-    zigzag_encode(compressed_block.data.len() as i64, writer)?;
-
-    writer.write_all(&compressed_block.data)?;
-
-    writer.write_all(&SYNC_NUMBER)?;
-
-    Ok(())
-}
diff --git a/src/io/avro/write/compress.rs b/src/io/avro/write/compress.rs
deleted file mode 100644
index 5121003a2b..0000000000
--- a/src/io/avro/write/compress.rs
+++ /dev/null
@@ -1,62 +0,0 @@
-//! APIs to read from Avro format to arrow.
-
-use crate::error::Result;
-
-use super::Compression;
-use super::{Block, CompressedBlock};
-
-#[cfg(feature = "io_avro_compression")]
-const CRC_TABLE: crc::Crc<u32> = crc::Crc::<u32>::new(&crc::CRC_32_ISO_HDLC);
-
-/// Compresses a [`Block`] to a [`CompressedBlock`].
-pub fn compress(
-    block: &mut Block,
-    compressed: &mut CompressedBlock,
-    compression: Option<Compression>,
-) -> Result<bool> {
-    compressed.number_of_rows = block.number_of_rows;
-    let block = &mut block.data;
-    let compressed = &mut compressed.data;
-
-    match compression {
-        None => {
-            std::mem::swap(block, compressed);
-            Ok(true)
-        }
-        #[cfg(feature = "io_avro_compression")]
-        Some(Compression::Deflate) => {
-            use std::io::Write;
-            compressed.clear();
-            let mut encoder = libflate::deflate::Encoder::new(compressed);
-            encoder.write_all(block)?;
-            encoder.finish();
-            Ok(false)
-        }
-        #[cfg(feature = "io_avro_compression")]
-        Some(Compression::Snappy) => {
-            use snap::raw::{max_compress_len, Encoder};
-
-            compressed.clear();
-
-            let required_len = max_compress_len(block.len());
-            compressed.resize(required_len, 0);
-            let compressed_bytes = Encoder::new()
-                .compress(block, compressed)
-                .map_err(|e| crate::error::Error::ExternalFormat(e.to_string()))?;
-            compressed.truncate(compressed_bytes);
-
-            compressed.extend(CRC_TABLE.checksum(block).to_be_bytes());
-            Ok(false)
-        }
-        #[cfg(not(feature = "io_avro_compression"))]
-        Some(Compression::Deflate) => Err(crate::error::Error::InvalidArgumentError(
-            "Trying to compress Avro with deflate but feature 'io_avro_compression' is not active."
-                .to_string(),
-        )),
-        #[cfg(not(feature = "io_avro_compression"))]
-        Some(Compression::Snappy) => Err(crate::error::Error::InvalidArgumentError(
-            "Trying to compress Avro with snappy but feature 'io_avro_compression' is not active."
-                .to_string(),
-        )),
-    }
-}
diff --git a/src/io/avro/write/header.rs b/src/io/avro/write/header.rs
deleted file mode 100644
index 370a0532b5..0000000000
--- a/src/io/avro/write/header.rs
+++ /dev/null
@@ -1,28 +0,0 @@
-use ahash::AHashMap;
-use avro_schema::Schema;
-use serde_json;
-
-use crate::error::{Error, Result};
-
-use super::Compression;
-
-/// Serializes an [`Schema`] and optional [`Compression`] into an avro header.
-pub(crate) fn serialize_header(
-    schema: &Schema,
-    compression: Option<Compression>,
-) -> Result<AHashMap<String, Vec<u8>>> {
-    let schema = serde_json::to_string(schema).map_err(|e| Error::ExternalFormat(e.to_string()))?;
-
-    let mut header = AHashMap::<String, Vec<u8>>::default();
-
-    header.insert("avro.schema".to_string(), schema.into_bytes());
-    if let Some(compression) = compression {
-        let value = match compression {
-            Compression::Snappy => b"snappy".to_vec(),
-            Compression::Deflate => b"deflate".to_vec(),
-        };
-        header.insert("avro.codec".to_string(), value);
-    };
-
-    Ok(header)
-}
diff --git a/src/io/avro/write/mod.rs b/src/io/avro/write/mod.rs
index 158ec571a8..f821f0dd1e 100644
--- a/src/io/avro/write/mod.rs
+++ b/src/io/avro/write/mod.rs
@@ -1,46 +1,10 @@
 //! APIs to write to Avro format.
-use avro_schema::{Field as AvroField, Record, Schema as AvroSchema};
+use avro_schema::file::Block;
 
-use crate::error::Result;
-
-pub use super::Compression;
-
-mod header;
-pub(super) use header::serialize_header;
 mod schema;
-pub use schema::to_avro_schema;
+pub use schema::to_record;
 mod serialize;
 pub use serialize::{can_serialize, new_serializer, BoxSerializer};
-mod block;
-pub use block::*;
-mod compress;
-pub(super) mod util;
-pub use compress::compress;
-
-pub use super::{Block, CompressedBlock};
-
-pub(super) const SYNC_NUMBER: [u8; 16] = [1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4];
-// * Four bytes, ASCII 'O', 'b', 'j', followed by 1.
-pub(super) const AVRO_MAGIC: [u8; 4] = [b'O', b'b', b'j', 1u8];
-
-/// Writes Avro's metadata to `writer`.
-pub fn write_metadata<W: std::io::Write>(
-    writer: &mut W,
-    fields: Vec<AvroField>,
-    compression: Option<Compression>,
-) -> Result<()> {
-    writer.write_all(&AVRO_MAGIC)?;
-
-    // * file metadata, including the schema.
-    let schema = AvroSchema::Record(Record::new("", fields));
-
-    write_schema(writer, &schema, compression)?;
-
-    // The 16-byte, randomly-generated sync marker for this file.
-    writer.write_all(&SYNC_NUMBER)?;
-
-    Ok(())
-}
 
 /// consumes a set of [`BoxSerializer`] into an [`Block`].
 /// # Panics
@@ -62,19 +26,3 @@ pub fn serialize<'a>(serializers: &mut [BoxSerializer<'a>], block: &mut Block) {
         }
     }
 }
-
-pub(super) fn write_schema<W: std::io::Write>(
-    writer: &mut W,
-    schema: &AvroSchema,
-    compression: Option<Compression>,
-) -> Result<()> {
-    let header = serialize_header(schema, compression)?;
-
-    util::zigzag_encode(header.len() as i64, writer)?;
-    for (name, item) in header {
-        util::write_binary(name.as_bytes(), writer)?;
-        util::write_binary(&item, writer)?;
-    }
-    writer.write_all(&[0])?;
-    Ok(())
-}
diff --git a/src/io/avro/write/schema.rs b/src/io/avro/write/schema.rs
index 26a14db8b3..94673f45a4 100644
--- a/src/io/avro/write/schema.rs
+++ b/src/io/avro/write/schema.rs
@@ -1,4 +1,4 @@
-use avro_schema::{
+use avro_schema::schema::{
     BytesLogical, Field as AvroField, Fixed, FixedLogical, IntLogical, LongLogical, Record,
     Schema as AvroSchema,
 };
@@ -6,9 +6,20 @@ use avro_schema::{
 use crate::datatypes::*;
 use crate::error::{Error, Result};
 
-/// Converts a [`Schema`] to a vector of [`AvroField`] with it.
-pub fn to_avro_schema(schema: &Schema) -> Result<Vec<AvroField>> {
-    schema.fields.iter().map(field_to_field).collect()
+/// Converts a [`Schema`] to an Avro [`Record`].
+pub fn to_record(schema: &Schema) -> Result<Record> {
+    let fields = schema
+        .fields
+        .iter()
+        .map(field_to_field)
+        .collect::<Result<_>>()?;
+    Ok(Record {
+        name: "".to_string(),
+        namespace: None,
+        doc: None,
+        aliases: vec![],
+        fields,
+    })
 }
 
 fn field_to_field(field: &Field) -> Result<AvroField> {
diff --git a/src/io/avro/write/serialize.rs b/src/io/avro/write/serialize.rs
index 443848ab61..7fbcfdb776 100644
--- a/src/io/avro/write/serialize.rs
+++ b/src/io/avro/write/serialize.rs
@@ -1,4 +1,5 @@
-use avro_schema::{Record, Schema as AvroSchema};
+use avro_schema::schema::{Record, Schema as AvroSchema};
+use avro_schema::write::encode;
 
 use crate::bitmap::utils::zip_validity;
 use crate::datatypes::{IntervalUnit, PhysicalType, PrimitiveType};
@@ -6,7 +7,6 @@ use crate::types::months_days_ns;
 use crate::{array::*, datatypes::DataType};
 
 use super::super::super::iterator::*;
-use super::util;
 
 // Zigzag representation of false and true respectively.
 const IS_NULL: u8 = 0;
@@ -20,7 +20,7 @@ fn utf8_required<O: Offset>(array: &Utf8Array<O>) -> BoxSerializer {
     Box::new(BufStreamingIterator::new(
         array.values_iter(),
         |x, buf| {
-            util::zigzag_encode(x.len() as i64, buf).unwrap();
+            encode::zigzag_encode(x.len() as i64, buf).unwrap();
             buf.extend_from_slice(x.as_bytes());
         },
         vec![],
@@ -33,7 +33,7 @@ fn utf8_optional<O: Offset>(array: &Utf8Array<O>) -> BoxSerializer {
         |x, buf| {
             if let Some(x) = x {
                 buf.push(IS_VALID);
-                util::zigzag_encode(x.len() as i64, buf).unwrap();
+                encode::zigzag_encode(x.len() as i64, buf).unwrap();
                 buf.extend_from_slice(x.as_bytes());
             } else {
                 buf.push(IS_NULL);
@@ -47,7 +47,7 @@ fn binary_required<O: Offset>(array: &BinaryArray<O>) -> BoxSerializer {
     Box::new(BufStreamingIterator::new(
         array.values_iter(),
         |x, buf| {
-            util::zigzag_encode(x.len() as i64, buf).unwrap();
+            encode::zigzag_encode(x.len() as i64, buf).unwrap();
             buf.extend_from_slice(x);
         },
         vec![],
@@ -60,7 +60,7 @@ fn binary_optional<O: Offset>(array: &BinaryArray<O>) -> BoxSerializer {
         |x, buf| {
             if let Some(x) = x {
                 buf.push(IS_VALID);
-                util::zigzag_encode(x.len() as i64, buf).unwrap();
+                encode::zigzag_encode(x.len() as i64, buf).unwrap();
                 buf.extend_from_slice(x);
             } else {
                 buf.push(IS_NULL);
@@ -105,13 +105,13 @@ fn list_required<'a, O: Offset>(array: &'a ListArray<O>, schema: &AvroSchema) ->
     Box::new(BufStreamingIterator::new(
         lengths,
         move |length, buf| {
-            util::zigzag_encode(length, buf).unwrap();
+            encode::zigzag_encode(length, buf).unwrap();
             let mut rows = 0;
             while let Some(item) = inner.next() {
                 buf.extend_from_slice(item);
                 rows += 1;
                 if rows == length {
-                    util::zigzag_encode(0, buf).unwrap();
+                    encode::zigzag_encode(0, buf).unwrap();
                     break;
                 }
             }
@@ -133,13 +133,13 @@ fn list_optional<'a, O: Offset>(array: &'a ListArray<O>, schema: &AvroSchema) ->
         move |length, buf| {
             if let Some(length) = length {
                 buf.push(IS_VALID);
-                util::zigzag_encode(length, buf).unwrap();
+                encode::zigzag_encode(length, buf).unwrap();
                 let mut rows = 0;
                 while let Some(item) = inner.next() {
                     buf.extend_from_slice(item);
                     rows += 1;
                     if rows == length {
-                        util::zigzag_encode(0, buf).unwrap();
+                        encode::zigzag_encode(0, buf).unwrap();
                         break;
                     }
                 }
@@ -278,7 +278,7 @@ pub fn new_serializer<'a>(array: &'a dyn Array, schema: &AvroSchema) -> BoxSeria
                 |x, buf| {
                     if let Some(x) = x {
                         buf.push(IS_VALID);
-                        util::zigzag_encode(*x as i64, buf).unwrap();
+                        encode::zigzag_encode(*x as i64, buf).unwrap();
                     } else {
                         buf.push(IS_NULL);
                     }
@@ -294,7 +294,7 @@ pub fn new_serializer<'a>(array: &'a dyn Array, schema: &AvroSchema) -> BoxSeria
             Box::new(BufStreamingIterator::new(
                 values.values().iter(),
                 |x, buf| {
-                    util::zigzag_encode(*x as i64, buf).unwrap();
+                    encode::zigzag_encode(*x as i64, buf).unwrap();
                 },
                 vec![],
             ))
@@ -309,7 +309,7 @@ pub fn new_serializer<'a>(array: &'a dyn Array, schema: &AvroSchema) -> BoxSeria
                 |x, buf| {
                     if let Some(x) = x {
                         buf.push(IS_VALID);
-                        util::zigzag_encode(*x, buf).unwrap();
+                        encode::zigzag_encode(*x, buf).unwrap();
                     } else {
                         buf.push(IS_NULL);
                     }
@@ -325,7 +325,7 @@ pub fn new_serializer<'a>(array: &'a dyn Array, schema: &AvroSchema) -> BoxSeria
             Box::new(BufStreamingIterator::new(
                 values.values().iter(),
                 |x, buf| {
-                    util::zigzag_encode(*x, buf).unwrap();
+                    encode::zigzag_encode(*x, buf).unwrap();
                 },
                 vec![],
             ))
@@ -401,7 +401,7 @@ pub fn new_serializer<'a>(array: &'a dyn Array, schema: &AvroSchema) -> BoxSeria
                 values.values().iter(),
                 |x, buf| {
                     let len = ((x.leading_zeros() / 8) - ((x.leading_zeros() / 8) % 2)) as usize;
-                    util::zigzag_encode((16 - len) as i64, buf).unwrap();
+                    encode::zigzag_encode((16 - len) as i64, buf).unwrap();
                     buf.extend_from_slice(&x.to_be_bytes()[len..]);
                 },
                 vec![],
@@ -419,7 +419,7 @@ pub fn new_serializer<'a>(array: &'a dyn Array, schema: &AvroSchema) -> BoxSeria
                         buf.push(IS_VALID);
                         let len =
                             ((x.leading_zeros() / 8) - ((x.leading_zeros() / 8) % 2)) as usize;
-                        util::zigzag_encode((16 - len) as i64, buf).unwrap();
+                        encode::zigzag_encode((16 - len) as i64, buf).unwrap();
                         buf.extend_from_slice(&x.to_be_bytes()[len..]);
                     } else {
                         buf.push(IS_NULL);
diff --git a/src/io/avro/write/util.rs b/src/io/avro/write/util.rs
deleted file mode 100644
index 52d13c026e..0000000000
--- a/src/io/avro/write/util.rs
+++ /dev/null
@@ -1,26 +0,0 @@
-use crate::error::Result;
-
-#[inline]
-pub fn zigzag_encode<W: std::io::Write>(n: i64, writer: &mut W) -> Result<()> {
-    _zigzag_encode(((n << 1) ^ (n >> 63)) as u64, writer)
-}
-
-#[inline]
-fn _zigzag_encode<W: std::io::Write>(mut z: u64, writer: &mut W) -> Result<()> {
-    loop {
-        if z <= 0x7F {
-            writer.write_all(&[(z & 0x7F) as u8])?;
-            break;
-        } else {
-            writer.write_all(&[(0x80 | (z & 0x7F)) as u8])?;
-            z >>= 7;
-        }
-    }
-    Ok(())
-}
-
-pub(crate) fn write_binary<W: std::io::Write>(bytes: &[u8], writer: &mut W) -> Result<()> {
-    zigzag_encode(bytes.len() as i64, writer)?;
-    writer.write_all(bytes)?;
-    Ok(())
-}
diff --git a/src/io/avro/write_async/block.rs b/src/io/avro/write_async/block.rs
deleted file mode 100644
index 6eba0a7fda..0000000000
--- a/src/io/avro/write_async/block.rs
+++ /dev/null
@@ -1,26 +0,0 @@
-use futures::{AsyncWrite, AsyncWriteExt};
-
-use crate::error::Result;
-
-use super::super::write::{util::zigzag_encode, SYNC_NUMBER};
-use super::super::CompressedBlock;
-
-/// Writes a [`CompressedBlock`] to `writer`
-pub async fn write_block<W>(writer: &mut W, compressed_block: &CompressedBlock) -> Result<()>
-where
-    W: AsyncWrite + Unpin,
-{
-    // write size and rows
-    let mut scratch = Vec::with_capacity(10);
-    zigzag_encode(compressed_block.number_of_rows as i64, &mut scratch)?;
-    writer.write_all(&scratch).await?;
-    scratch.clear();
-    zigzag_encode(compressed_block.data.len() as i64, &mut scratch)?;
-    writer.write_all(&scratch).await?;
-
-    writer.write_all(&compressed_block.data).await?;
-
-    writer.write_all(&SYNC_NUMBER).await?;
-
-    Ok(())
-}
diff --git a/src/io/avro/write_async/mod.rs b/src/io/avro/write_async/mod.rs
deleted file mode 100644
index 81340125d4..0000000000
--- a/src/io/avro/write_async/mod.rs
+++ /dev/null
@@ -1,38 +0,0 @@
-//! Async write Avro
-mod block;
-pub use block::write_block;
-
-use avro_schema::{Field as AvroField, Record, Schema as AvroSchema};
-use futures::{AsyncWrite, AsyncWriteExt};
-
-use crate::error::Result;
-
-use super::{
-    write::{write_schema, AVRO_MAGIC, SYNC_NUMBER},
-    Compression,
-};
-
-/// Writes Avro's metadata to `writer`.
-pub async fn write_metadata<W>(
-    writer: &mut W,
-    fields: Vec<AvroField>,
-    compression: Option<Compression>,
-) -> Result<()>
-where
-    W: AsyncWrite + Unpin,
-{
-    writer.write_all(&AVRO_MAGIC).await?;
-
-    // * file metadata, including the schema.
-    let schema = AvroSchema::Record(Record::new("", fields));
-
-    let mut scratch = vec![];
-    write_schema(&mut scratch, &schema, compression)?;
-
-    writer.write_all(&scratch).await?;
-
-    // The 16-byte, randomly-generated sync marker for this file.
-    writer.write_all(&SYNC_NUMBER).await?;
-
-    Ok(())
-}
diff --git a/src/io/ipc/mod.rs b/src/io/ipc/mod.rs
index 3d8a8ccb99..1f8476cef7 100644
--- a/src/io/ipc/mod.rs
+++ b/src/io/ipc/mod.rs
@@ -81,7 +81,7 @@ pub mod read;
 pub mod write;
 
 const ARROW_MAGIC: [u8; 6] = [b'A', b'R', b'R', b'O', b'W', b'1'];
-const CONTINUATION_MARKER: [u8; 4] = [0xff; 4];
+pub(crate) const CONTINUATION_MARKER: [u8; 4] = [0xff; 4];
 
 /// Struct containing `dictionary_id` and nested `IpcField`, allowing users
 /// to specify the dictionary ids of the IPC fields when writing to IPC.
diff --git a/src/io/ipc/read/common.rs b/src/io/ipc/read/common.rs
index 306df0061c..9a1ea3ce1c 100644
--- a/src/io/ipc/read/common.rs
+++ b/src/io/ipc/read/common.rs
@@ -212,7 +212,7 @@ fn find_first_dict_field<'a>(
     find_first_dict_field_d(id, &field.data_type, ipc_field)
 }
 
-fn first_dict_field<'a>(
+pub(crate) fn first_dict_field<'a>(
     id: i64,
     fields: &'a [Field],
     ipc_fields: &'a [IpcField],
@@ -253,45 +253,41 @@ pub fn read_dictionary<R: Read + Seek>(
         .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferId(err)))?;
     let (first_field, first_ipc_field) = first_dict_field(id, fields, &ipc_schema.fields)?;
 
-    // As the dictionary batch does not contain the type of the
-    // values array, we need to retrieve this from the schema.
-    // Get an array representing this dictionary's values.
-    let dictionary_values: Box<dyn Array> = match first_field.data_type.to_logical_type() {
-        DataType::Dictionary(_, ref value_type, _) => {
-            let batch = batch
-                .data()
-                .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferData(err)))?
-                .ok_or_else(|| Error::from(OutOfSpecKind::MissingData))?;
-
-            // Make a fake schema for the dictionary batch.
-            let fields = vec![Field::new("", value_type.as_ref().clone(), false)];
-            let ipc_schema = IpcSchema {
-                fields: vec![first_ipc_field.clone()],
-                is_little_endian: ipc_schema.is_little_endian,
-            };
-            let chunk = read_record_batch(
-                batch,
-                &fields,
-                &ipc_schema,
-                None,
-                None, // we must read the whole dictionary
-                dictionaries,
-                arrow_format::ipc::MetadataVersion::V5,
-                reader,
-                block_offset,
-                file_size,
-                scratch,
-            )?;
-            chunk.into_arrays().pop().unwrap()
-        }
-        _ => {
+    let batch = batch
+        .data()
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferData(err)))?
+        .ok_or_else(|| Error::from(OutOfSpecKind::MissingData))?;
+
+    let value_type =
+        if let DataType::Dictionary(_, value_type, _) = first_field.data_type.to_logical_type() {
+            value_type.as_ref()
+        } else {
             return Err(Error::from(OutOfSpecKind::InvalidIdDataType {
                 requested_id: id,
-            }))
-        }
+            }));
+        };
+
+    // Make a fake schema for the dictionary batch.
+    let fields = vec![Field::new("", value_type.clone(), false)];
+    let ipc_schema = IpcSchema {
+        fields: vec![first_ipc_field.clone()],
+        is_little_endian: ipc_schema.is_little_endian,
     };
-
-    dictionaries.insert(id, dictionary_values);
+    let chunk = read_record_batch(
+        batch,
+        &fields,
+        &ipc_schema,
+        None,
+        None, // we must read the whole dictionary
+        dictionaries,
+        arrow_format::ipc::MetadataVersion::V5,
+        reader,
+        block_offset,
+        file_size,
+        scratch,
+    )?;
+
+    dictionaries.insert(id, chunk.into_arrays().pop().unwrap());
 
     Ok(())
 }
diff --git a/src/io/ipc/read/file_async.rs b/src/io/ipc/read/file_async.rs
index 4b36a44026..82999c28ea 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_serialized_batch};
+use super::reader::{deserialize_footer, get_record_batch};
 use super::Dictionaries;
 use super::FileMetadata;
 use super::OutOfSpecKind;
@@ -205,7 +205,7 @@ where
     let message = arrow_format::ipc::MessageRef::read_as_root(meta_buffer)
         .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
 
-    let batch = get_serialized_batch(&message)?;
+    let batch = get_record_batch(message)?;
 
     let block_length: usize = message
         .body_length()
diff --git a/src/io/ipc/read/mod.rs b/src/io/ipc/read/mod.rs
index f623e0aef5..2645d91440 100644
--- a/src/io/ipc/read/mod.rs
+++ b/src/io/ipc/read/mod.rs
@@ -27,6 +27,7 @@ pub mod stream_async;
 #[cfg_attr(docsrs, doc(cfg(feature = "io_ipc_read_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,
diff --git a/src/io/ipc/read/reader.rs b/src/io/ipc/read/reader.rs
index ab72adec96..37b8437c27 100644
--- a/src/io/ipc/read/reader.rs
+++ b/src/io/ipc/read/reader.rs
@@ -27,13 +27,13 @@ pub struct FileMetadata {
     /// The blocks in the file
     ///
     /// A block indicates the regions in the file to read to get data
-    pub(crate) blocks: Vec<arrow_format::ipc::Block>,
+    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(crate) size: u64,
+    pub size: u64,
 }
 
 fn read_dictionary_message<R: Read + Seek>(
@@ -63,6 +63,19 @@ fn read_dictionary_message<R: Read + Seek>(
     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,
@@ -84,27 +97,18 @@ fn read_dictionary_block<R: Read + Seek>(
     let message = arrow_format::ipc::MessageRef::read_as_root(message_scratch.as_ref())
         .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
 
-    let header = message
-        .header()
-        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferHeader(err)))?
-        .ok_or_else(|| Error::from(OutOfSpecKind::MissingMessageHeader))?;
+    let batch = get_dictionary_batch(&message)?;
 
-    match header {
-        arrow_format::ipc::MessageHeaderRef::DictionaryBatch(batch) => {
-            let block_offset = offset + length;
-            read_dictionary(
-                batch,
-                &metadata.schema.fields,
-                &metadata.ipc_schema,
-                dictionaries,
-                reader,
-                block_offset,
-                metadata.size,
-                dictionary_scratch,
-            )
-        }
-        _ => Err(Error::from(OutOfSpecKind::UnexpectedMessageType)),
-    }
+    read_dictionary(
+        batch,
+        &metadata.schema.fields,
+        &metadata.ipc_schema,
+        dictionaries,
+        reader,
+        offset + length,
+        metadata.size,
+        dictionary_scratch,
+    )
 }
 
 /// Reads all file's dictionaries, if any
@@ -116,7 +120,7 @@ pub fn read_file_dictionaries<R: Read + Seek>(
 ) -> Result<Dictionaries> {
     let mut dictionaries = Default::default();
 
-    let blocks = if let Some(blocks) = metadata.dictionaries.as_deref() {
+    let blocks = if let Some(blocks) = &metadata.dictionaries {
         blocks
     } else {
         return Ok(AHashMap::new());
@@ -230,9 +234,9 @@ pub fn read_file_metadata<R: Read + Seek>(reader: &mut R) -> Result<FileMetadata
     deserialize_footer(&serialized_footer, end - start)
 }
 
-pub(super) fn get_serialized_batch<'a>(
-    message: &'a arrow_format::ipc::MessageRef,
-) -> Result<arrow_format::ipc::RecordBatchRef<'a>> {
+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)))?
@@ -268,6 +272,11 @@ pub fn read_batch<R: Read + Seek>(
         .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];
@@ -290,17 +299,7 @@ pub fn read_batch<R: Read + Seek>(
     let message = arrow_format::ipc::MessageRef::read_as_root(message_scratch.as_ref())
         .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
 
-    let batch = get_serialized_batch(&message)?;
-
-    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))?;
+    let batch = get_record_batch(message)?;
 
     read_record_batch(
         batch,
diff --git a/src/io/ipc/write/common.rs b/src/io/ipc/write/common.rs
index 02ddc1eca8..8b1a2a03c5 100644
--- a/src/io/ipc/write/common.rs
+++ b/src/io/ipc/write/common.rs
@@ -370,8 +370,8 @@ pub struct EncodedData {
 
 /// Calculate an 8-byte boundary and return the number of bytes needed to pad to 8 bytes
 #[inline]
-pub(crate) fn pad_to_8(len: usize) -> usize {
-    (((len + 7) & !7) - len) as usize
+pub(crate) fn pad_to_64(len: usize) -> usize {
+    (((len + 63) & !63) - len) as usize
 }
 
 /// An array [`Chunk`] with optional accompanying IPC fields.
diff --git a/src/io/ipc/write/common_async.rs b/src/io/ipc/write/common_async.rs
index 8792a9bd0b..21c04603d8 100644
--- a/src/io/ipc/write/common_async.rs
+++ b/src/io/ipc/write/common_async.rs
@@ -4,7 +4,7 @@ use futures::AsyncWriteExt;
 use crate::error::Result;
 
 use super::super::CONTINUATION_MARKER;
-use super::common::pad_to_8;
+use super::common::pad_to_64;
 use super::common::EncodedData;
 
 /// Write a message's IPC data and buffers, returning metadata and buffer data lengths written
@@ -14,10 +14,10 @@ pub async fn write_message<W: AsyncWrite + Unpin + Send>(
 ) -> Result<(usize, usize)> {
     let arrow_data_len = encoded.arrow_data.len();
 
-    let a = 8 - 1;
+    let a = 64 - 1;
     let buffer = encoded.ipc_message;
     let flatbuf_size = buffer.len();
-    let prefix_size = 8;
+    let prefix_size = 8; // the message length
     let aligned_size = (flatbuf_size + prefix_size + a) & !a;
     let padding_bytes = aligned_size - flatbuf_size - prefix_size;
 
@@ -57,7 +57,7 @@ async fn write_body_buffers<W: AsyncWrite + Unpin + Send>(
     data: &[u8],
 ) -> Result<usize> {
     let len = data.len();
-    let pad_len = pad_to_8(data.len());
+    let pad_len = pad_to_64(data.len());
     let total_len = len + pad_len;
 
     // write body buffer
diff --git a/src/io/ipc/write/common_sync.rs b/src/io/ipc/write/common_sync.rs
index bafde8664c..9f281e2206 100644
--- a/src/io/ipc/write/common_sync.rs
+++ b/src/io/ipc/write/common_sync.rs
@@ -3,7 +3,7 @@ use std::io::Write;
 use crate::error::Result;
 
 use super::super::CONTINUATION_MARKER;
-use super::common::pad_to_8;
+use super::common::pad_to_64;
 use super::common::EncodedData;
 
 /// Write a message's IPC data and buffers, returning metadata and buffer data lengths written
@@ -38,7 +38,7 @@ pub fn write_message<W: Write>(writer: &mut W, encoded: EncodedData) -> Result<(
 
 fn write_body_buffers<W: Write>(mut writer: W, data: &[u8]) -> Result<usize> {
     let len = data.len();
-    let pad_len = pad_to_8(data.len());
+    let pad_len = pad_to_64(data.len());
     let total_len = len + pad_len;
 
     // write body buffer
diff --git a/src/io/ipc/write/serialize.rs b/src/io/ipc/write/serialize.rs
index 40eb13a00b..28647bd911 100644
--- a/src/io/ipc/write/serialize.rs
+++ b/src/io/ipc/write/serialize.rs
@@ -7,7 +7,7 @@ use crate::{
 
 use super::super::compression;
 use super::super::endianess::is_native_little_endian;
-use super::common::{pad_to_8, Compression};
+use super::common::{pad_to_64, Compression};
 
 fn write_primitive<T: NativeType>(
     array: &PrimitiveArray<T>,
@@ -564,8 +564,8 @@ pub fn write(
 }
 
 #[inline]
-fn pad_buffer_to_8(buffer: &mut Vec<u8>, length: usize) {
-    let pad_len = pad_to_8(length);
+fn pad_buffer_to_64(buffer: &mut Vec<u8>, length: usize) {
+    let pad_len = pad_to_64(length);
     buffer.extend_from_slice(&vec![0u8; pad_len]);
 }
 
@@ -748,7 +748,7 @@ fn write_buffer_from_iter<T: NativeType, I: TrustedLen<Item = T>>(
 fn finish_buffer(arrow_data: &mut Vec<u8>, start: usize, offset: &mut i64) -> ipc::Buffer {
     let buffer_len = (arrow_data.len() - start) as i64;
 
-    pad_buffer_to_8(arrow_data, arrow_data.len() - start);
+    pad_buffer_to_64(arrow_data, arrow_data.len() - start);
     let total_len = (arrow_data.len() - start) as i64;
 
     let buffer = ipc::Buffer {
diff --git a/src/io/orc/read/mod.rs b/src/io/orc/read/mod.rs
index 123addf3f9..597ecbc9fb 100644
--- a/src/io/orc/read/mod.rs
+++ b/src/io/orc/read/mod.rs
@@ -256,7 +256,6 @@ fn try_extend<O: Offset + TryFrom<u64>, I: Iterator<Item = u64>>(
     iter: I,
 ) -> Result<(), orc_format::error::Error> {
     for item in iter {
-        println!("{item}");
         let item: O = item
             .try_into()
             .map_err(|_| orc_format::error::Error::OutOfSpec)?;
diff --git a/src/io/parquet/read/statistics/dictionary.rs b/src/io/parquet/read/statistics/dictionary.rs
index b2553c62f1..a4bc349741 100644
--- a/src/io/parquet/read/statistics/dictionary.rs
+++ b/src/io/parquet/read/statistics/dictionary.rs
@@ -59,6 +59,10 @@ impl MutableArray for DynMutableDictionary {
         todo!()
     }
 
+    fn reserve(&mut self, _: usize) {
+        todo!();
+    }
+
     fn shrink_to_fit(&mut self) {
         todo!()
     }
diff --git a/src/io/parquet/read/statistics/list.rs b/src/io/parquet/read/statistics/list.rs
index 6b4bb2d7d3..496c7deb9d 100644
--- a/src/io/parquet/read/statistics/list.rs
+++ b/src/io/parquet/read/statistics/list.rs
@@ -73,6 +73,10 @@ impl MutableArray for DynMutableListArray {
         todo!()
     }
 
+    fn reserve(&mut self, _: usize) {
+        todo!();
+    }
+
     fn shrink_to_fit(&mut self) {
         todo!()
     }
diff --git a/src/io/parquet/read/statistics/map.rs b/src/io/parquet/read/statistics/map.rs
index 71ef8154dd..c31dfb9d9e 100644
--- a/src/io/parquet/read/statistics/map.rs
+++ b/src/io/parquet/read/statistics/map.rs
@@ -58,6 +58,10 @@ impl MutableArray for DynMutableMapArray {
         todo!()
     }
 
+    fn reserve(&mut self, _: usize) {
+        todo!();
+    }
+
     fn shrink_to_fit(&mut self) {
         todo!()
     }
diff --git a/src/io/parquet/read/statistics/struct_.rs b/src/io/parquet/read/statistics/struct_.rs
index 7ae80af3e7..c19a22b1db 100644
--- a/src/io/parquet/read/statistics/struct_.rs
+++ b/src/io/parquet/read/statistics/struct_.rs
@@ -55,6 +55,10 @@ impl MutableArray for DynMutableStructArray {
         todo!()
     }
 
+    fn reserve(&mut self, _: usize) {
+        todo!();
+    }
+
     fn shrink_to_fit(&mut self) {
         todo!()
     }
diff --git a/src/lib.rs b/src/lib.rs
index 01fe444d5c..16bd56ed04 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -17,13 +17,16 @@ pub mod bitmap;
 pub mod buffer;
 pub mod chunk;
 pub mod error;
+#[cfg(feature = "io_ipc")]
+#[cfg_attr(docsrs, doc(cfg(feature = "io_ipc")))]
+pub mod mmap;
+
 pub mod scalar;
 pub mod trusted_len;
 pub mod types;
 
 pub mod compute;
 pub mod io;
-//pub mod record_batch;
 pub mod temporal_conversions;
 
 pub mod datatypes;
diff --git a/src/mmap/mod.rs b/src/mmap/mod.rs
new file mode 100644
index 0000000000..55bd7d01f8
--- /dev/null
+++ b/src/mmap/mod.rs
@@ -0,0 +1,226 @@
+//! Memory maps regions defined on the IPC format into [`Array`].
+use std::collections::VecDeque;
+use std::sync::Arc;
+
+use crate::array::Array;
+use crate::chunk::Chunk;
+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::{first_dict_field, Dictionaries, FileMetadata};
+use crate::io::ipc::read::{IpcBuffer, Node, OutOfSpecKind};
+use crate::io::ipc::{IpcField, CONTINUATION_MARKER};
+
+use arrow_format::ipc::planus::ReadAsRoot;
+use arrow_format::ipc::{Block, MessageRef, RecordBatchRef};
+
+fn read_message(
+    mut bytes: &[u8],
+    block: arrow_format::ipc::Block,
+) -> Result<(MessageRef, usize), Error> {
+    let offset: usize = block
+        .offset
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let block_length: usize = block
+        .meta_data_length
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    bytes = &bytes[offset..];
+    let mut message_length = bytes[..4].try_into().unwrap();
+    bytes = &bytes[4..];
+
+    if message_length == CONTINUATION_MARKER {
+        // continuation marker encountered, read message next
+        message_length = bytes[..4].try_into().unwrap();
+        bytes = &bytes[4..];
+    };
+
+    let message_length: usize = i32::from_le_bytes(message_length)
+        .try_into()
+        .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
+
+    let message = arrow_format::ipc::MessageRef::read_as_root(&bytes[..message_length])
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
+
+    Ok((message, offset + block_length))
+}
+
+fn get_buffers_nodes(
+    batch: RecordBatchRef,
+) -> Result<(VecDeque<IpcBuffer>, VecDeque<Node>), Error> {
+    let compression = batch.compression()?;
+    if compression.is_some() {
+        return Err(Error::nyi(
+            "mmap can only be done on uncompressed IPC files",
+        ));
+    }
+
+    let buffers = batch
+        .buffers()
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferBuffers(err)))?
+        .ok_or_else(|| Error::from(OutOfSpecKind::MissingMessageBuffers))?;
+    let buffers = buffers.iter().collect::<VecDeque<_>>();
+
+    let field_nodes = batch
+        .nodes()
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferNodes(err)))?
+        .ok_or_else(|| Error::from(OutOfSpecKind::MissingMessageNodes))?;
+    let field_nodes = field_nodes.iter().collect::<VecDeque<_>>();
+
+    Ok((buffers, field_nodes))
+}
+
+unsafe fn _mmap_record<T: AsRef<[u8]>>(
+    fields: &[Field],
+    ipc_fields: &[IpcField],
+    data: Arc<T>,
+    batch: RecordBatchRef,
+    offset: usize,
+    dictionaries: &Dictionaries,
+) -> Result<Chunk<Box<dyn Array>>, Error> {
+    let (mut buffers, mut field_nodes) = get_buffers_nodes(batch)?;
+
+    fields
+        .iter()
+        .map(|f| &f.data_type)
+        .cloned()
+        .zip(ipc_fields)
+        .map(|(data_type, ipc_field)| {
+            mmap::mmap(
+                data.clone(),
+                offset,
+                data_type,
+                ipc_field,
+                dictionaries,
+                &mut field_nodes,
+                &mut buffers,
+            )
+        })
+        .collect::<Result<_, Error>>()
+        .and_then(Chunk::try_new)
+}
+
+unsafe fn _mmap_unchecked<T: AsRef<[u8]>>(
+    fields: &[Field],
+    ipc_fields: &[IpcField],
+    data: Arc<T>,
+    block: Block,
+    dictionaries: &Dictionaries,
+) -> Result<Chunk<Box<dyn Array>>, Error> {
+    let (message, offset) = read_message(data.as_ref().as_ref(), block)?;
+    let batch = get_record_batch(message)?;
+    _mmap_record(
+        fields,
+        ipc_fields,
+        data.clone(),
+        batch,
+        offset,
+        dictionaries,
+    )
+}
+
+/// Memory maps an record batch from an IPC file into a [`Chunk`].
+/// # Errors
+/// This function errors when:
+/// * The IPC file is not valid
+/// * the buffers on the file are un-aligned with their corresponding data. This can happen when:
+///     * the file was written with 8-bit alignment
+///     * the file contains type decimal 128 or 256
+/// # Safety
+/// The caller must ensure that `data` contains a valid buffers, for example:
+/// * Offsets in variable-sized containers must be in-bounds and increasing
+/// * Utf8 data is valid
+pub unsafe fn mmap_unchecked<T: AsRef<[u8]>>(
+    metadata: &FileMetadata,
+    dictionaries: &Dictionaries,
+    data: Arc<T>,
+    chunk: usize,
+) -> Result<Chunk<Box<dyn Array>>, Error> {
+    let block = metadata.blocks[chunk];
+
+    let (message, offset) = read_message(data.as_ref().as_ref(), block)?;
+    let batch = get_record_batch(message)?;
+    _mmap_record(
+        &metadata.schema.fields,
+        &metadata.ipc_schema.fields,
+        data.clone(),
+        batch,
+        offset,
+        dictionaries,
+    )
+}
+
+unsafe fn mmap_dictionary<T: AsRef<[u8]>>(
+    metadata: &FileMetadata,
+    data: Arc<T>,
+    block: Block,
+    dictionaries: &mut Dictionaries,
+) -> Result<(), Error> {
+    let (message, offset) = read_message(data.as_ref().as_ref(), block)?;
+    let batch = get_dictionary_batch(&message)?;
+
+    let id = batch
+        .id()
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferId(err)))?;
+    let (first_field, first_ipc_field) =
+        first_dict_field(id, &metadata.schema.fields, &metadata.ipc_schema.fields)?;
+
+    let batch = batch
+        .data()
+        .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferData(err)))?
+        .ok_or_else(|| Error::from(OutOfSpecKind::MissingData))?;
+
+    let value_type =
+        if let DataType::Dictionary(_, value_type, _) = first_field.data_type.to_logical_type() {
+            value_type.as_ref()
+        } else {
+            return Err(Error::from(OutOfSpecKind::InvalidIdDataType {
+                requested_id: id,
+            }));
+        };
+
+    // Make a fake schema for the dictionary batch.
+    let field = Field::new("", value_type.clone(), false);
+
+    let chunk = _mmap_record(
+        &[field],
+        &[first_ipc_field.clone()],
+        data.clone(),
+        batch,
+        offset,
+        dictionaries,
+    )?;
+
+    dictionaries.insert(id, chunk.into_arrays().pop().unwrap());
+
+    Ok(())
+}
+
+/// Memory maps dictionaries from an IPC file into
+/// # Safety
+/// The caller must ensure that `data` contains a valid buffers, for example:
+/// * Offsets in variable-sized containers must be in-bounds and increasing
+/// * Utf8 data is valid
+pub unsafe fn mmap_dictionaries_unchecked<T: AsRef<[u8]>>(
+    metadata: &FileMetadata,
+    data: Arc<T>,
+) -> Result<Dictionaries, Error> {
+    let blocks = if let Some(blocks) = &metadata.dictionaries {
+        blocks
+    } else {
+        return Ok(Default::default());
+    };
+
+    let mut dictionaries = Default::default();
+
+    blocks
+        .iter()
+        .cloned()
+        .try_for_each(|block| mmap_dictionary(metadata, data.clone(), block, &mut dictionaries))?;
+    Ok(dictionaries)
+}


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index c9b25238f8..74ebe22eed 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -114,6 +114,25 @@ jobs:
       - name: Run
         run: MIRIFLAGS="-Zmiri-disable-isolation" cargo miri test --tests --features io_ipc,io_json_integration io::ipc::write::write_sliced_list
 
+  miri-checks-mmap:
+    name: MIRI on IO IPC mmaping
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions-rs/toolchain@v1
+        with:
+          toolchain: nightly-2022-07-12
+          override: true
+      - uses: Swatinem/rust-cache@v1
+        with:
+          key: key1
+      - name: Install Miri
+        run: |
+          rustup component add miri
+          cargo miri setup
+      - name: Run
+        run: cargo miri test --tests --features io_ipc io::ipc::mmap
+
   feature-compilation:
     name: Feature coverage
     runs-on: ubuntu-latest
diff --git a/tests/it/compute/like.rs b/tests/it/compute/like.rs
index 61cb01cc13..3bd4c6aaf3 100644
--- a/tests/it/compute/like.rs
+++ b/tests/it/compute/like.rs
@@ -39,6 +39,28 @@ fn test_like_binary_scalar() -> Result<()> {
     Ok(())
 }
 
+#[test]
+fn test_like_utf8_scalar() -> Result<()> {
+    let array = Utf8Array::<i32>::from_slice(&["Arrow", "Arrow", "Arrow", "BA"]);
+
+    let result = like_utf8_scalar(&array, "A%").unwrap();
+    assert_eq!(result, BooleanArray::from_slice(&[true, true, true, false]));
+
+    let result = like_utf8_scalar(&array, "Arrow").unwrap();
+    assert_eq!(result, BooleanArray::from_slice(&[true, true, true, false]));
+
+    let array = Utf8Array::<i32>::from_slice(&["A%", "Arrow"]);
+
+    let result = like_utf8_scalar(&array, "A\\%").unwrap();
+    assert_eq!(result, BooleanArray::from_slice(&[true, false]));
+
+    let array = Utf8Array::<i32>::from_slice(&["A_row", "Arrow"]);
+    let result = like_utf8_scalar(&array, "A\\_row").unwrap();
+    assert_eq!(result, BooleanArray::from_slice(&[true, false]));
+
+    Ok(())
+}
+
 #[test]
 fn test_nlike_binary_scalar() -> Result<()> {
     let array = BinaryArray::<i32>::from_slice(&["Arrow", "Arrow", "Arrow", "BA"]);
diff --git a/tests/it/io/avro/read.rs b/tests/it/io/avro/read.rs
index deb8ffd802..5c301f7d1b 100644
--- a/tests/it/io/avro/read.rs
+++ b/tests/it/io/avro/read.rs
@@ -6,6 +6,7 @@ use avro_rs::{Days, Decimal, Duration, Millis, Months, Schema as AvroSchema};
 use arrow2::array::*;
 use arrow2::datatypes::*;
 use arrow2::error::Result;
+use arrow2::io::avro::avro_schema::read::read_metadata;
 use arrow2::io::avro::read;
 
 pub(super) fn schema() -> (AvroSchema, Schema) {
@@ -196,14 +197,10 @@ pub(super) fn read_avro(
 ) -> Result<(Chunk<Box<dyn Array>>, Schema)> {
     let file = &mut avro;
 
-    let (avro_schema, schema, codec, file_marker) = read::read_metadata(file)?;
+    let metadata = read_metadata(file)?;
+    let schema = read::infer_schema(&metadata.record)?;
 
-    let mut reader = read::Reader::new(
-        read::Decompressor::new(read::BlockStreamIterator::new(file, file_marker), codec),
-        avro_schema,
-        schema.fields.clone(),
-        projection.clone(),
-    );
+    let mut reader = read::Reader::new(file, metadata, schema.fields.clone(), projection.clone());
 
     let schema = if let Some(projection) = projection {
         let fields = schema
diff --git a/tests/it/io/avro/read_async.rs b/tests/it/io/avro/read_async.rs
index 3dd04cdb47..6a6b09ac9f 100644
--- a/tests/it/io/avro/read_async.rs
+++ b/tests/it/io/avro/read_async.rs
@@ -4,7 +4,8 @@ use futures::pin_mut;
 use futures::StreamExt;
 
 use arrow2::error::Result;
-use arrow2::io::avro::read_async::*;
+use arrow2::io::avro::avro_schema::read_async::{block_stream, read_metadata};
+use arrow2::io::avro::read;
 
 use super::read::{schema, write_avro};
 
@@ -14,11 +15,12 @@ async fn test(codec: Codec) -> Result<()> {
 
     let mut reader = &mut &avro_data[..];
 
-    let (_, schema, _, marker) = read_metadata(&mut reader).await?;
+    let metadata = read_metadata(&mut reader).await?;
+    let schema = read::infer_schema(&metadata.record)?;
 
     assert_eq!(schema, expected_schema);
 
-    let blocks = block_stream(&mut reader, marker).await;
+    let blocks = block_stream(&mut reader, metadata.marker).await;
 
     pin_mut!(blocks);
     while let Some(block) = blocks.next().await.transpose()? {
diff --git a/tests/it/io/avro/write.rs b/tests/it/io/avro/write.rs
index d4b7e410c1..091da09c75 100644
--- a/tests/it/io/avro/write.rs
+++ b/tests/it/io/avro/write.rs
@@ -2,7 +2,9 @@ use arrow2::array::*;
 use arrow2::chunk::Chunk;
 use arrow2::datatypes::*;
 use arrow2::error::Result;
-use arrow2::io::avro::{write, CompressedBlock};
+use arrow2::io::avro::avro_schema::file::{Block, CompressedBlock, Compression};
+use arrow2::io::avro::avro_schema::write::{compress, write_block, write_metadata};
+use arrow2::io::avro::write;
 use arrow2::types::months_days_ns;
 
 use super::read::read_avro;
@@ -107,24 +109,24 @@ pub(super) fn data() -> Chunk<Box<dyn Array>> {
 pub(super) fn serialize_to_block<R: AsRef<dyn Array>>(
     columns: &Chunk<R>,
     schema: &Schema,
-    compression: Option<write::Compression>,
+    compression: Option<Compression>,
 ) -> Result<CompressedBlock> {
-    let avro_fields = write::to_avro_schema(schema)?;
+    let record = write::to_record(schema)?;
 
     let mut serializers = columns
         .arrays()
         .iter()
         .map(|x| x.as_ref())
-        .zip(avro_fields.iter())
+        .zip(record.fields.iter())
         .map(|(array, field)| write::new_serializer(array, &field.schema))
         .collect::<Vec<_>>();
-    let mut block = write::Block::new(columns.len(), vec![]);
+    let mut block = Block::new(columns.len(), vec![]);
 
     write::serialize(&mut serializers, &mut block);
 
-    let mut compressed_block = write::CompressedBlock::default();
+    let mut compressed_block = CompressedBlock::default();
 
-    write::compress(&mut block, &mut compressed_block, compression)?;
+    compress(&mut block, &mut compressed_block, compression)?;
 
     Ok(compressed_block)
 }
@@ -132,21 +134,21 @@ pub(super) fn serialize_to_block<R: AsRef<dyn Array>>(
 fn write_avro<R: AsRef<dyn Array>>(
     columns: &Chunk<R>,
     schema: &Schema,
-    compression: Option<write::Compression>,
+    compression: Option<Compression>,
 ) -> Result<Vec<u8>> {
     let compressed_block = serialize_to_block(columns, schema, compression)?;
 
-    let avro_fields = write::to_avro_schema(schema)?;
+    let avro_fields = write::to_record(schema)?;
     let mut file = vec![];
 
-    write::write_metadata(&mut file, avro_fields, compression)?;
+    write_metadata(&mut file, avro_fields, compression)?;
 
-    write::write_block(&mut file, &compressed_block)?;
+    write_block(&mut file, &compressed_block)?;
 
     Ok(file)
 }
 
-fn roundtrip(compression: Option<write::Compression>) -> Result<()> {
+fn roundtrip(compression: Option<Compression>) -> Result<()> {
     let expected = data();
     let expected_schema = schema();
 
@@ -169,13 +171,13 @@ fn no_compression() -> Result<()> {
 #[cfg(feature = "io_avro_compression")]
 #[test]
 fn snappy() -> Result<()> {
-    roundtrip(Some(write::Compression::Snappy))
+    roundtrip(Some(Compression::Snappy))
 }
 
 #[cfg(feature = "io_avro_compression")]
 #[test]
 fn deflate() -> Result<()> {
-    roundtrip(Some(write::Compression::Deflate))
+    roundtrip(Some(Compression::Deflate))
 }
 
 fn large_format_schema() -> Schema {
diff --git a/tests/it/io/avro/write_async.rs b/tests/it/io/avro/write_async.rs
index 1c679c564c..b98071accb 100644
--- a/tests/it/io/avro/write_async.rs
+++ b/tests/it/io/avro/write_async.rs
@@ -2,8 +2,9 @@ use arrow2::array::*;
 use arrow2::chunk::Chunk;
 use arrow2::datatypes::*;
 use arrow2::error::Result;
+use arrow2::io::avro::avro_schema::file::Compression;
+use arrow2::io::avro::avro_schema::write_async::{write_block, write_metadata};
 use arrow2::io::avro::write;
-use arrow2::io::avro::write_async;
 
 use super::read::read_avro;
 use super::write::{data, schema, serialize_to_block};
@@ -11,22 +12,22 @@ use super::write::{data, schema, serialize_to_block};
 async fn write_avro<R: AsRef<dyn Array>>(
     columns: &Chunk<R>,
     schema: &Schema,
-    compression: Option<write::Compression>,
+    compression: Option<Compression>,
 ) -> Result<Vec<u8>> {
     // usually done on a different thread pool
     let compressed_block = serialize_to_block(columns, schema, compression)?;
 
-    let avro_fields = write::to_avro_schema(schema)?;
+    let record = write::to_record(schema)?;
     let mut file = vec![];
 
-    write_async::write_metadata(&mut file, avro_fields.clone(), compression).await?;
+    write_metadata(&mut file, record, compression).await?;
 
-    write_async::write_block(&mut file, &compressed_block).await?;
+    write_block(&mut file, &compressed_block).await?;
 
     Ok(file)
 }
 
-async fn roundtrip(compression: Option<write::Compression>) -> Result<()> {
+async fn roundtrip(compression: Option<Compression>) -> Result<()> {
     let expected = data();
     let expected_schema = schema();
 
diff --git a/tests/it/io/ipc/mmap.rs b/tests/it/io/ipc/mmap.rs
new file mode 100644
index 0000000000..ef317b9a69
--- /dev/null
+++ b/tests/it/io/ipc/mmap.rs
@@ -0,0 +1,119 @@
+use arrow2::array::*;
+use arrow2::chunk::Chunk;
+use arrow2::datatypes::{DataType, Field, Schema};
+use arrow2::error::Result;
+use arrow2::io::ipc::read::read_file_metadata;
+use std::sync::Arc;
+
+use super::write::file::write;
+
+fn round_trip(array: Box<dyn Array>) -> Result<()> {
+    let schema = Schema::from(vec![Field::new("a", array.data_type().clone(), true)]);
+    let columns = Chunk::try_new(vec![array.clone()])?;
+
+    let data = Arc::new(write(&[columns], &schema, None, None)?);
+
+    let metadata = read_file_metadata(&mut std::io::Cursor::new(data.as_ref()))?;
+
+    let dictionaries =
+        unsafe { arrow2::mmap::mmap_dictionaries_unchecked(&metadata, data.clone())? };
+
+    let new_array = unsafe { arrow2::mmap::mmap_unchecked(&metadata, &dictionaries, data, 0)? };
+    assert_eq!(new_array.into_arrays()[0], array);
+    Ok(())
+}
+
+#[test]
+fn utf8() -> Result<()> {
+    let array = Utf8Array::<i32>::from([None, None, Some("bb")])
+        .slice(1, 2)
+        .boxed();
+    round_trip(array)
+}
+
+#[test]
+fn fixed_size_binary() -> Result<()> {
+    let array = FixedSizeBinaryArray::from([None, None, Some([1, 2])])
+        .slice(1, 2)
+        .boxed();
+    round_trip(array)
+}
+
+#[test]
+fn primitive() -> Result<()> {
+    let array = PrimitiveArray::<i32>::from([None, None, Some(3)])
+        .slice(1, 2)
+        .boxed();
+    round_trip(array)
+}
+
+#[test]
+fn boolean() -> Result<()> {
+    let array = BooleanArray::from([None, None, Some(true)])
+        .slice(1, 2)
+        .boxed();
+    round_trip(array)
+}
+
+#[test]
+fn null() -> Result<()> {
+    let array = NullArray::new(DataType::Null, 10).boxed();
+    round_trip(array)
+}
+
+#[test]
+fn fixed_size_list() -> Result<()> {
+    let data = vec![
+        Some(vec![Some(1i32), Some(2), Some(3)]),
+        None,
+        Some(vec![Some(4), None, Some(6)]),
+    ];
+
+    let mut array = MutableFixedSizeListArray::new(MutablePrimitiveArray::<i32>::new(), 3);
+    array.try_extend(data)?;
+
+    let array: FixedSizeListArray = array.into();
+    round_trip(array.slice(1, 2).boxed())
+}
+
+#[test]
+fn list() -> Result<()> {
+    let data = vec![
+        Some(vec![Some(1i32), Some(2), Some(3)]),
+        None,
+        Some(vec![Some(4), None, Some(6)]),
+    ];
+
+    let mut array = MutableListArray::<i32, MutablePrimitiveArray<i32>>::new();
+    array.try_extend(data).unwrap();
+    let array = array.into_box().slice(1, 2);
+    round_trip(array)
+}
+
+#[test]
+fn struct_() -> Result<()> {
+    let array = PrimitiveArray::<i32>::from([None, None, None, Some(3), Some(4)]).boxed();
+
+    let array = StructArray::new(
+        DataType::Struct(vec![Field::new("f1", array.data_type().clone(), true)]),
+        vec![array],
+        Some([true, true, false, true, false].into()),
+    )
+    .slice(1, 4)
+    .boxed();
+
+    round_trip(array)
+}
+
+#[test]
+fn dict() -> Result<()> {
+    let keys = PrimitiveArray::<i32>::from([None, None, None, Some(3), Some(4)]);
+
+    let values = PrimitiveArray::<i32>::from_slice([0, 1, 2, 3, 4, 5]).boxed();
+
+    let array = DictionaryArray::try_from_keys(keys, values)?
+        .slice(1, 4)
+        .boxed();
+
+    round_trip(array)
+}
diff --git a/tests/it/io/ipc/mod.rs b/tests/it/io/ipc/mod.rs
index 63cce5b2e8..6d3e71c5db 100644
--- a/tests/it/io/ipc/mod.rs
+++ b/tests/it/io/ipc/mod.rs
@@ -15,3 +15,5 @@ mod read_stream_async;
 
 #[cfg(feature = "io_ipc_read_async")]
 mod read_file_async;
+
+mod mmap;
diff --git a/tests/it/io/ipc/write/mod.rs b/tests/it/io/ipc/write/mod.rs
index f3a4fbfe57..48ca5ea6cd 100644
--- a/tests/it/io/ipc/write/mod.rs
+++ b/tests/it/io/ipc/write/mod.rs
@@ -1,3 +1,3 @@
-mod file;
+pub mod file;
 mod file_append;
 mod stream;