Repository: jorgecarleitao/arrow2
Pull request #1110 (2022-06-27 09:33:26)

Made `row_group::get_field_columns` public
This function can be used when reading synchronously multiple columns from a parquet file in parallel.


Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 5d355c6a0c..4bc21d5948 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -67,7 +67,7 @@ futures = { version = "0.3", optional = true }
 ahash = { version = "0.7", optional = true }
 
 # parquet support
-parquet2 = { version = "0.13.1", optional = true, default_features = false }
+parquet2 = { version = "0.14.0", optional = true, default_features = false }
 
 # avro support
 avro-schema = { version = "0.2", optional = true }
diff --git a/examples/avro_read_async.rs b/examples/avro_read_async.rs
index d3c5172049..9c9c3649da 100644
--- a/examples/avro_read_async.rs
+++ b/examples/avro_read_async.rs
@@ -33,8 +33,8 @@ async fn main() -> Result<()> {
             decompress_block(&mut block, &mut decompressed, compression)?;
             deserialize(&decompressed, &schema.fields, &avro_schemas, &projection)
         });
-        let batch = handle.await.unwrap()?;
-        assert!(!batch.is_empty());
+        let chunk = handle.await.unwrap()?;
+        assert!(!chunk.is_empty());
     }
 
     Ok(())
diff --git a/examples/ipc_file_read.rs b/examples/ipc_file_read.rs
index 23d179adb4..6478b25d48 100644
--- a/examples/ipc_file_read.rs
+++ b/examples/ipc_file_read.rs
@@ -33,7 +33,7 @@ fn read_batch(path: &str) -> Result<(Schema, Chunk<Box<dyn Array>>)> {
     let schema = metadata.schema.clone();
 
     // advanced way: read the dictionary
-    let dictionaries = read::read_file_dictionaries(&mut file, &metadata)?;
+    let dictionaries = read::read_file_dictionaries(&mut file, &metadata, &mut Default::default())?;
 
     let chunk_index = 0;
 
@@ -43,7 +43,8 @@ fn read_batch(path: &str) -> Result<(Schema, Chunk<Box<dyn Array>>)> {
         &metadata,
         None,
         chunk_index,
-        &mut vec![],
+        &mut Default::default(),
+        &mut Default::default(),
     )?;
 
     Ok((schema, chunk))
diff --git a/examples/parquet_read.rs b/examples/parquet_read.rs
index a29555e6ec..92fa83c521 100644
--- a/examples/parquet_read.rs
+++ b/examples/parquet_read.rs
@@ -11,7 +11,7 @@ fn main() -> Result<()> {
     let file_path = &args[1];
 
     let reader = File::open(file_path)?;
-    let reader = read::FileReader::try_new(reader, None, None, None, None)?;
+    let reader = read::FileReader::try_new(reader, None, Some(1024 * 8 * 8), None, None)?;
 
     println!("{:#?}", reader.schema());
 
diff --git a/examples/parquet_read_parallel/src/main.rs b/examples/parquet_read_parallel/src/main.rs
index 313205bbfd..6ab26d61f7 100644
--- a/examples/parquet_read_parallel/src/main.rs
+++ b/examples/parquet_read_parallel/src/main.rs
@@ -15,32 +15,37 @@ use arrow2::{
 
 mod logger;
 
+/// Advances each iterator in parallel
 /// # Panic
 /// If the iterators are empty
-fn deserialize_parallel(columns: &mut [ArrayIter<'static>]) -> Result<Chunk<Box<dyn Array>>> {
+fn deserialize_parallel(iters: &mut [ArrayIter<'static>]) -> Result<Chunk<Box<dyn Array>>> {
     // CPU-bounded
-    let columns = columns
+    let arrays = iters
         .par_iter_mut()
         .map(|iter| iter.next().transpose())
         .collect::<Result<Vec<_>>>()?;
 
-    Chunk::try_new(columns.into_iter().map(|x| x.unwrap()).collect())
+    Chunk::try_new(arrays.into_iter().map(|x| x.unwrap()).collect())
 }
 
 fn parallel_read(path: &str, row_group: usize) -> Result<()> {
+    // open the file
     let mut file = BufReader::new(File::open(path)?);
+
+    // read Parquet's metadata and infer Arrow schema
     let metadata = read::read_metadata(&mut file)?;
     let schema = read::infer_schema(&metadata)?;
 
+    // select the row group from the metadata
     let row_group = &metadata.row_groups[row_group];
 
-    let chunk_size = 1024 * 8;
+    let chunk_size = 1024 * 8 * 8;
 
     // read (IO-bounded) all columns into memory (use a subset of the fields to project)
     let mut columns =
         read::read_columns_many(&mut file, row_group, schema.fields, Some(chunk_size))?;
 
-    // deserialize (CPU-bounded) to arrow
+    // deserialize (CPU-bounded) to Arrow in chunks
     let mut num_rows = row_group.num_rows();
     while num_rows > 0 {
         num_rows = num_rows.saturating_sub(chunk_size);
diff --git a/examples/parquet_write_async.rs b/examples/parquet_write_async.rs
new file mode 100644
index 0000000000..b6205c8fb0
--- /dev/null
+++ b/examples/parquet_write_async.rs
@@ -0,0 +1,56 @@
+use futures::SinkExt;
+use tokio::fs::File;
+
+use arrow2::{
+    array::{Array, Int32Array},
+    chunk::Chunk,
+    datatypes::{Field, Schema},
+    error::Result,
+    io::parquet::write::{
+        transverse, CompressionOptions, Encoding, FileSink, Version, WriteOptions,
+    },
+};
+use tokio_util::compat::TokioAsyncReadCompatExt;
+
+async fn write_batch(path: &str, schema: Schema, columns: Chunk<Box<dyn Array>>) -> Result<()> {
+    let options = WriteOptions {
+        write_statistics: true,
+        compression: CompressionOptions::Uncompressed,
+        version: Version::V2,
+    };
+
+    let mut stream = futures::stream::iter(vec![Ok(columns)].into_iter());
+
+    // Create a new empty file
+    let file = File::create(path).await?.compat();
+
+    let encodings = schema
+        .fields
+        .iter()
+        .map(|f| transverse(&f.data_type, |_| Encoding::Plain))
+        .collect();
+
+    let mut writer = FileSink::try_new(file, schema, encodings, options)?;
+
+    writer.send_all(&mut stream).await?;
+    writer.close().await?;
+    Ok(())
+}
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<()> {
+    let array = Int32Array::from(&[
+        Some(0),
+        Some(1),
+        Some(2),
+        Some(3),
+        Some(4),
+        Some(5),
+        Some(6),
+    ]);
+    let field = Field::new("c1", array.data_type().clone(), true);
+    let schema = Schema::from(vec![field]);
+    let columns = Chunk::new(vec![array.boxed()]);
+
+    write_batch("test.parquet", schema, columns).await
+}
diff --git a/examples/parquet_write_parallel/src/main.rs b/examples/parquet_write_parallel/src/main.rs
index ac1d216f90..980c817e42 100644
--- a/examples/parquet_write_parallel/src/main.rs
+++ b/examples/parquet_write_parallel/src/main.rs
@@ -41,7 +41,7 @@ impl FallibleStreamingIterator for Bla {
     }
 }
 
-fn parallel_write(path: &str, schema: Schema, batches: &[Chunk]) -> Result<()> {
+fn parallel_write(path: &str, schema: Schema, chunks: &[Chunk]) -> Result<()> {
     // declare the options
     let options = WriteOptions {
         write_statistics: true,
@@ -70,9 +70,9 @@ fn parallel_write(path: &str, schema: Schema, batches: &[Chunk]) -> Result<()> {
     // derive the parquet schema (physical types) from arrow's schema.
     let parquet_schema = to_parquet_schema(&schema)?;
 
-    let row_groups = batches.iter().map(|batch| {
+    let row_groups = chunks.iter().map(|batch| {
         // write batch to pages; parallelized by rayon
-        let columns = batch
+        let columns = chunk
             .columns()
             .par_iter()
             .zip(parquet_schema.fields().to_vec())
@@ -140,7 +140,7 @@ fn main() -> Result<()> {
     let schema = Schema {
         fields: vec![
             Field::new("c1", DataType::Int32, true),
-            Field::new("c1", DataType::Utf8, true),
+            Field::new("c2", DataType::Utf8, true),
         ],
         metadata: Default::default(),
     };
diff --git a/src/array/fixed_size_list/mod.rs b/src/array/fixed_size_list/mod.rs
index d50518190f..d61a64c258 100644
--- a/src/array/fixed_size_list/mod.rs
+++ b/src/array/fixed_size_list/mod.rs
@@ -84,6 +84,11 @@ impl FixedSizeListArray {
         Self::try_new(data_type, values, validity).unwrap()
     }
 
+    /// Returns the size (number of elements per slot) of this [`FixedSizeListArray`].
+    pub const fn size(&self) -> usize {
+        self.size
+    }
+
     /// Alias for `new`
     pub fn from_data(
         data_type: DataType,
@@ -101,10 +106,9 @@ impl FixedSizeListArray {
 
     /// Returns a new null [`FixedSizeListArray`].
     pub fn new_null(data_type: DataType, length: usize) -> Self {
-        let values = new_null_array(
-            Self::get_child_and_size(&data_type).0.data_type().clone(),
-            length,
-        );
+        let (field, size) = Self::get_child_and_size(&data_type);
+
+        let values = new_null_array(field.data_type().clone(), length * size);
         Self::new(data_type, values, Some(Bitmap::new_zeroed(length)))
     }
 
diff --git a/src/array/utf8/mod.rs b/src/array/utf8/mod.rs
index b6b56e14cc..69c5af6e8f 100644
--- a/src/array/utf8/mod.rs
+++ b/src/array/utf8/mod.rs
@@ -486,8 +486,7 @@ impl<O: Offset> Utf8Array<O> {
         P: AsRef<str>,
         I: TrustedLen<Item = Option<P>>,
     {
-        // soundness: I is `TrustedLen`
-        unsafe { Self::from_trusted_len_iter_unchecked(iterator) }
+        MutableUtf8Array::<O>::from_trusted_len_iter(iterator).into()
     }
 
     /// Creates a [`Utf8Array`] from an falible iterator of trusted length.
@@ -512,8 +511,7 @@ impl<O: Offset> Utf8Array<O> {
         P: AsRef<str>,
         I: TrustedLen<Item = std::result::Result<Option<P>, E>>,
     {
-        // soundness: I: TrustedLen
-        unsafe { Self::try_from_trusted_len_iter_unchecked(iter) }
+        MutableUtf8Array::<O>::try_from_trusted_len_iter(iter).map(|x| x.into())
     }
 
     /// Alias for `new`
diff --git a/src/io/flight/mod.rs b/src/io/flight/mod.rs
index 8aa4a1de0e..9ed0835bb8 100644
--- a/src/io/flight/mod.rs
+++ b/src/io/flight/mod.rs
@@ -142,6 +142,7 @@ pub fn deserialize_batch(
             &mut reader,
             0,
             length as u64,
+            &mut Default::default(),
         ),
         _ => Err(Error::nyi(
             "flight currently only supports reading RecordBatch messages",
diff --git a/src/io/ipc/append/mod.rs b/src/io/ipc/append/mod.rs
index 4a225c4db6..9e084a4589 100644
--- a/src/io/ipc/append/mod.rs
+++ b/src/io/ipc/append/mod.rs
@@ -32,7 +32,8 @@ impl<R: Read + Seek + Write> FileWriter<R> {
             ));
         }
 
-        let dictionaries = read::read_file_dictionaries(&mut writer, &metadata)?;
+        let dictionaries =
+            read::read_file_dictionaries(&mut writer, &metadata, &mut Default::default())?;
 
         let last_block = metadata.blocks.last().ok_or_else(|| {
             Error::oos("An Arrow IPC file must have at least 1 message (the schema message)")
diff --git a/src/io/ipc/read/array/binary.rs b/src/io/ipc/read/array/binary.rs
index f68471f699..51d3d7d26b 100644
--- a/src/io/ipc/read/array/binary.rs
+++ b/src/io/ipc/read/array/binary.rs
@@ -7,8 +7,9 @@ use crate::datatypes::DataType;
 use crate::error::{Error, Result};
 
 use super::super::read_basic::*;
-use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind};
+use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind, ReadBuffer};
 
+#[allow(clippy::too_many_arguments)]
 pub fn read_binary<O: Offset, R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
     data_type: DataType,
@@ -17,6 +18,7 @@ pub fn read_binary<O: Offset, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    scratch: &mut ReadBuffer,
 ) -> Result<BinaryArray<O>> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
@@ -46,6 +48,7 @@ pub fn read_binary<O: Offset, R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        scratch,
     )
     // Older versions of the IPC format sometimes do not report an offset
     .or_else(|_| Result::Ok(Buffer::<O>::from(vec![O::default()])))?;
@@ -58,6 +61,7 @@ pub fn read_binary<O: Offset, R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        scratch,
     )?;
 
     BinaryArray::<O>::try_new(data_type, offsets, values, validity)
diff --git a/src/io/ipc/read/array/dictionary.rs b/src/io/ipc/read/array/dictionary.rs
index 6589790c94..0b212c4501 100644
--- a/src/io/ipc/read/array/dictionary.rs
+++ b/src/io/ipc/read/array/dictionary.rs
@@ -6,7 +6,7 @@ use crate::array::{DictionaryArray, DictionaryKey};
 use crate::error::{Error, Result};
 
 use super::super::Dictionaries;
-use super::super::{Compression, IpcBuffer, Node};
+use super::super::{Compression, IpcBuffer, Node, ReadBuffer};
 use super::{read_primitive, skip_primitive};
 
 #[allow(clippy::too_many_arguments)]
@@ -19,6 +19,7 @@ pub fn read_dictionary<T: DictionaryKey, R: Read + Seek>(
     block_offset: u64,
     compression: Option<Compression>,
     is_little_endian: bool,
+    scratch: &mut ReadBuffer,
 ) -> Result<DictionaryArray<T>>
 where
     Vec<u8>: TryInto<T::Bytes>,
@@ -47,6 +48,7 @@ where
         block_offset,
         is_little_endian,
         compression,
+        scratch,
     )?;
 
     Ok(DictionaryArray::<T>::from_data(keys, values))
diff --git a/src/io/ipc/read/array/fixed_size_binary.rs b/src/io/ipc/read/array/fixed_size_binary.rs
index 04029cd7c8..5dc1b2ab0b 100644
--- a/src/io/ipc/read/array/fixed_size_binary.rs
+++ b/src/io/ipc/read/array/fixed_size_binary.rs
@@ -6,8 +6,9 @@ use crate::datatypes::DataType;
 use crate::error::{Error, Result};
 
 use super::super::read_basic::*;
-use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind};
+use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind, ReadBuffer};
 
+#[allow(clippy::too_many_arguments)]
 pub fn read_fixed_size_binary<R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
     data_type: DataType,
@@ -16,6 +17,7 @@ pub fn read_fixed_size_binary<R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    scratch: &mut ReadBuffer,
 ) -> Result<FixedSizeBinaryArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
@@ -46,6 +48,7 @@ pub fn read_fixed_size_binary<R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        scratch,
     )?;
 
     FixedSizeBinaryArray::try_new(data_type, values, validity)
diff --git a/src/io/ipc/read/array/fixed_size_list.rs b/src/io/ipc/read/array/fixed_size_list.rs
index f8e314fc08..ac169ed3f6 100644
--- a/src/io/ipc/read/array/fixed_size_list.rs
+++ b/src/io/ipc/read/array/fixed_size_list.rs
@@ -8,8 +8,7 @@ use crate::error::{Error, Result};
 use super::super::super::IpcField;
 use super::super::deserialize::{read, skip};
 use super::super::read_basic::*;
-use super::super::Dictionaries;
-use super::super::{Compression, IpcBuffer, Node, Version};
+use super::super::{Compression, Dictionaries, IpcBuffer, Node, ReadBuffer, Version};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_fixed_size_list<R: Read + Seek>(
@@ -23,6 +22,7 @@ pub fn read_fixed_size_list<R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<Compression>,
     version: Version,
+    scratch: &mut ReadBuffer,
 ) -> Result<FixedSizeListArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
@@ -53,6 +53,7 @@ pub fn read_fixed_size_list<R: Read + Seek>(
         is_little_endian,
         compression,
         version,
+        scratch,
     )?;
     FixedSizeListArray::try_new(data_type, values, validity)
 }
diff --git a/src/io/ipc/read/array/list.rs b/src/io/ipc/read/array/list.rs
index f5d2eb5465..bb85f233cf 100644
--- a/src/io/ipc/read/array/list.rs
+++ b/src/io/ipc/read/array/list.rs
@@ -10,8 +10,9 @@ use crate::error::{Error, Result};
 use super::super::super::IpcField;
 use super::super::deserialize::{read, skip};
 use super::super::read_basic::*;
-use super::super::Dictionaries;
-use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind, Version};
+use super::super::{
+    Compression, Dictionaries, IpcBuffer, Node, OutOfSpecKind, ReadBuffer, Version,
+};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_list<O: Offset, R: Read + Seek>(
@@ -25,6 +26,7 @@ pub fn read_list<O: Offset, R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<Compression>,
     version: Version,
+    scratch: &mut ReadBuffer,
 ) -> Result<ListArray<O>>
 where
     Vec<u8>: TryInto<O::Bytes>,
@@ -57,6 +59,7 @@ where
         block_offset,
         is_little_endian,
         compression,
+        scratch,
     )
     // Older versions of the IPC format sometimes do not report an offset
     .or_else(|_| Result::Ok(Buffer::<O>::from(vec![O::default()])))?;
@@ -74,6 +77,7 @@ where
         is_little_endian,
         compression,
         version,
+        scratch,
     )?;
     ListArray::try_new(data_type, offsets, values, validity)
 }
diff --git a/src/io/ipc/read/array/map.rs b/src/io/ipc/read/array/map.rs
index 594f8a495b..40b4f5c038 100644
--- a/src/io/ipc/read/array/map.rs
+++ b/src/io/ipc/read/array/map.rs
@@ -9,8 +9,9 @@ use crate::error::{Error, Result};
 use super::super::super::IpcField;
 use super::super::deserialize::{read, skip};
 use super::super::read_basic::*;
-use super::super::Dictionaries;
-use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind, Version};
+use super::super::{
+    Compression, Dictionaries, IpcBuffer, Node, OutOfSpecKind, ReadBuffer, Version,
+};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_map<R: Read + Seek>(
@@ -24,6 +25,7 @@ pub fn read_map<R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<Compression>,
     version: Version,
+    scratch: &mut ReadBuffer,
 ) -> Result<MapArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
@@ -53,6 +55,7 @@ pub fn read_map<R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        scratch,
     )
     // Older versions of the IPC format sometimes do not report an offset
     .or_else(|_| Result::Ok(Buffer::<i32>::from(vec![0i32])))?;
@@ -70,6 +73,7 @@ pub fn read_map<R: Read + Seek>(
         is_little_endian,
         compression,
         version,
+        scratch,
     )?;
     MapArray::try_new(data_type, offsets, field, validity)
 }
diff --git a/src/io/ipc/read/array/primitive.rs b/src/io/ipc/read/array/primitive.rs
index c7c7b780d8..1e4000123b 100644
--- a/src/io/ipc/read/array/primitive.rs
+++ b/src/io/ipc/read/array/primitive.rs
@@ -6,8 +6,9 @@ use crate::error::{Error, Result};
 use crate::{array::PrimitiveArray, types::NativeType};
 
 use super::super::read_basic::*;
-use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind};
+use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind, ReadBuffer};
 
+#[allow(clippy::too_many_arguments)]
 pub fn read_primitive<T: NativeType, R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
     data_type: DataType,
@@ -16,6 +17,7 @@ pub fn read_primitive<T: NativeType, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    scratch: &mut ReadBuffer,
 ) -> Result<PrimitiveArray<T>>
 where
     Vec<u8>: TryInto<T::Bytes>,
@@ -48,6 +50,7 @@ where
         block_offset,
         is_little_endian,
         compression,
+        scratch,
     )?;
     PrimitiveArray::<T>::try_new(data_type, values, validity)
 }
diff --git a/src/io/ipc/read/array/struct_.rs b/src/io/ipc/read/array/struct_.rs
index c87440a978..ae23c0f882 100644
--- a/src/io/ipc/read/array/struct_.rs
+++ b/src/io/ipc/read/array/struct_.rs
@@ -8,8 +8,7 @@ use crate::error::{Error, Result};
 use super::super::super::IpcField;
 use super::super::deserialize::{read, skip};
 use super::super::read_basic::*;
-use super::super::Dictionaries;
-use super::super::{Compression, IpcBuffer, Node, Version};
+use super::super::{Compression, Dictionaries, IpcBuffer, Node, ReadBuffer, Version};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_struct<R: Read + Seek>(
@@ -23,6 +22,7 @@ pub fn read_struct<R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<Compression>,
     version: Version,
+    scratch: &mut ReadBuffer,
 ) -> Result<StructArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
@@ -57,6 +57,7 @@ pub fn read_struct<R: Read + Seek>(
                 is_little_endian,
                 compression,
                 version,
+                scratch,
             )
         })
         .collect::<Result<Vec<_>>>()?;
diff --git a/src/io/ipc/read/array/union.rs b/src/io/ipc/read/array/union.rs
index 7c86f5e30b..fc907a3239 100644
--- a/src/io/ipc/read/array/union.rs
+++ b/src/io/ipc/read/array/union.rs
@@ -9,8 +9,9 @@ use crate::error::{Error, Result};
 use super::super::super::IpcField;
 use super::super::deserialize::{read, skip};
 use super::super::read_basic::*;
-use super::super::Dictionaries;
-use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind, Version};
+use super::super::{
+    Compression, Dictionaries, IpcBuffer, Node, OutOfSpecKind, ReadBuffer, Version,
+};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_union<R: Read + Seek>(
@@ -24,6 +25,7 @@ pub fn read_union<R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<Compression>,
     version: Version,
+    scratch: &mut ReadBuffer,
 ) -> Result<UnionArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
@@ -50,6 +52,7 @@ pub fn read_union<R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        scratch,
     )?;
 
     let offsets = if let DataType::Union(_, _, mode) = data_type {
@@ -61,6 +64,7 @@ pub fn read_union<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                scratch,
             )?)
         } else {
             None
@@ -86,6 +90,7 @@ pub fn read_union<R: Read + Seek>(
                 is_little_endian,
                 compression,
                 version,
+                scratch,
             )
         })
         .collect::<Result<Vec<_>>>()?;
diff --git a/src/io/ipc/read/array/utf8.rs b/src/io/ipc/read/array/utf8.rs
index 8424fbb2e7..10f1260f5a 100644
--- a/src/io/ipc/read/array/utf8.rs
+++ b/src/io/ipc/read/array/utf8.rs
@@ -7,8 +7,9 @@ use crate::datatypes::DataType;
 use crate::error::{Error, Result};
 
 use super::super::read_basic::*;
-use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind};
+use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind, ReadBuffer};
 
+#[allow(clippy::too_many_arguments)]
 pub fn read_utf8<O: Offset, R: Read + Seek>(
     field_nodes: &mut VecDeque<Node>,
     data_type: DataType,
@@ -17,6 +18,7 @@ pub fn read_utf8<O: Offset, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    scratch: &mut ReadBuffer,
 ) -> Result<Utf8Array<O>> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
@@ -46,6 +48,7 @@ pub fn read_utf8<O: Offset, R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        scratch,
     )
     // Older versions of the IPC format sometimes do not report an offset
     .or_else(|_| Result::Ok(Buffer::<O>::from(vec![O::default()])))?;
@@ -58,6 +61,7 @@ pub fn read_utf8<O: Offset, R: Read + Seek>(
         block_offset,
         is_little_endian,
         compression,
+        scratch,
     )?;
 
     Utf8Array::<O>::try_new(data_type, offsets, values, validity)
diff --git a/src/io/ipc/read/common.rs b/src/io/ipc/read/common.rs
index 7e32cde5d0..18ddd8cbb8 100644
--- a/src/io/ipc/read/common.rs
+++ b/src/io/ipc/read/common.rs
@@ -9,6 +9,7 @@ use crate::datatypes::{DataType, Field};
 use crate::error::{Error, Result};
 use crate::io::ipc::read::OutOfSpecKind;
 use crate::io::ipc::{IpcField, IpcSchema};
+use crate::io::ReadBuffer;
 
 use super::deserialize::{read, skip};
 use super::Dictionaries;
@@ -85,6 +86,7 @@ pub fn read_record_batch<R: Read + Seek>(
     reader: &mut R,
     block_offset: u64,
     file_size: u64,
+    scratch: &mut ReadBuffer,
 ) -> Result<Chunk<Box<dyn Array>>> {
     assert_eq!(fields.len(), ipc_schema.fields.len());
     let buffers = batch
@@ -136,6 +138,7 @@ pub fn read_record_batch<R: Read + Seek>(
                         Error::from(OutOfSpecKind::InvalidFlatbufferCompression(err))
                     })?,
                     version,
+                    scratch,
                 )?)),
                 ProjectionResult::NotSelected((field, _)) => {
                     skip(&mut field_nodes, &field.data_type, &mut buffers)?;
@@ -162,6 +165,7 @@ pub fn read_record_batch<R: Read + Seek>(
                         Error::from(OutOfSpecKind::InvalidFlatbufferCompression(err))
                     })?,
                     version,
+                    scratch,
                 )
             })
             .collect::<Result<Vec<_>>>()?
@@ -221,6 +225,7 @@ fn first_dict_field<'a>(
 
 /// Read the dictionary from the buffer and provided metadata,
 /// updating the `dictionaries` with the resulting dictionary
+#[allow(clippy::too_many_arguments)]
 pub fn read_dictionary<R: Read + Seek>(
     batch: arrow_format::ipc::DictionaryBatchRef,
     fields: &[Field],
@@ -229,6 +234,7 @@ pub fn read_dictionary<R: Read + Seek>(
     reader: &mut R,
     block_offset: u64,
     file_size: u64,
+    scratch: &mut ReadBuffer,
 ) -> Result<()> {
     if batch
         .is_delta()
@@ -270,6 +276,7 @@ pub fn read_dictionary<R: Read + Seek>(
                 reader,
                 block_offset,
                 file_size,
+                scratch,
             )?;
             let mut arrays = columns.into_arrays();
             arrays.pop().unwrap()
diff --git a/src/io/ipc/read/deserialize.rs b/src/io/ipc/read/deserialize.rs
index f5c8b97c00..3378f99a89 100644
--- a/src/io/ipc/read/deserialize.rs
+++ b/src/io/ipc/read/deserialize.rs
@@ -10,7 +10,7 @@ use crate::error::Result;
 use crate::io::ipc::IpcField;
 
 use super::{array::*, Dictionaries};
-use super::{IpcBuffer, Node};
+use super::{IpcBuffer, Node, ReadBuffer};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read<R: Read + Seek>(
@@ -24,6 +24,7 @@ pub fn read<R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<BodyCompressionRef>,
     version: MetadataVersion,
+    scratch: &mut ReadBuffer,
 ) -> Result<Box<dyn Array>> {
     use PhysicalType::*;
     let data_type = field.data_type.clone();
@@ -49,6 +50,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                scratch
             )
             .map(|x| x.boxed())
         }),
@@ -61,6 +63,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                scratch,
             )?;
             Ok(Box::new(array))
         }
@@ -73,6 +76,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                scratch,
             )?;
             Ok(Box::new(array))
         }
@@ -85,6 +89,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                scratch,
             )?;
             Ok(Box::new(array))
         }
@@ -97,6 +102,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                scratch,
             )?;
             Ok(Box::new(array))
         }
@@ -109,6 +115,7 @@ pub fn read<R: Read + Seek>(
                 block_offset,
                 is_little_endian,
                 compression,
+                scratch,
             )?;
             Ok(Box::new(array))
         }
@@ -123,6 +130,7 @@ pub fn read<R: Read + Seek>(
             is_little_endian,
             compression,
             version,
+            scratch,
         )
         .map(|x| x.boxed()),
         LargeList => read_list::<i64, _>(
@@ -136,6 +144,7 @@ pub fn read<R: Read + Seek>(
             is_little_endian,
             compression,
             version,
+            scratch,
         )
         .map(|x| x.boxed()),
         FixedSizeList => read_fixed_size_list(
@@ -149,6 +158,7 @@ pub fn read<R: Read + Seek>(
             is_little_endian,
             compression,
             version,
+            scratch,
         )
         .map(|x| x.boxed()),
         Struct => read_struct(
@@ -162,6 +172,7 @@ pub fn read<R: Read + Seek>(
             is_little_endian,
             compression,
             version,
+            scratch,
         )
         .map(|x| x.boxed()),
         Dictionary(key_type) => {
@@ -175,6 +186,7 @@ pub fn read<R: Read + Seek>(
                     block_offset,
                     compression,
                     is_little_endian,
+                    scratch
                 )
                 .map(|x| x.boxed())
             })
@@ -190,6 +202,7 @@ pub fn read<R: Read + Seek>(
             is_little_endian,
             compression,
             version,
+            scratch,
         )
         .map(|x| x.boxed()),
         Map => read_map(
@@ -203,6 +216,7 @@ pub fn read<R: Read + Seek>(
             is_little_endian,
             compression,
             version,
+            scratch,
         )
         .map(|x| x.boxed()),
     }
diff --git a/src/io/ipc/read/file_async.rs b/src/io/ipc/read/file_async.rs
index 8397eb4a01..cabfdd43b7 100644
--- a/src/io/ipc/read/file_async.rs
+++ b/src/io/ipc/read/file_async.rs
@@ -18,6 +18,7 @@ use super::reader::{deserialize_footer, get_serialized_batch};
 use super::Dictionaries;
 use super::FileMetadata;
 use super::OutOfSpecKind;
+use super::ReadBuffer;
 
 /// Async reader for Arrow IPC files
 pub struct FileStream<'a> {
@@ -77,8 +78,9 @@ impl<'a> FileStream<'a> {
             // read dictionaries
             cached_read_dictionaries(&mut reader, &metadata, &mut dictionaries).await?;
 
-            let mut meta_buffer = vec![];
-            let mut block_buffer = vec![];
+            let mut meta_buffer = Default::default();
+            let mut block_buffer = Default::default();
+            let mut scratch = Default::default();
             for block in 0..metadata.blocks.len() {
                 let chunk = read_batch(
                     &mut reader,
@@ -88,6 +90,7 @@ impl<'a> FileStream<'a> {
                     block,
                     &mut meta_buffer,
                     &mut block_buffer,
+                    &mut scratch
                 ).await?;
 
                 let chunk = if let Some((_, map)) = &projection {
@@ -146,14 +149,16 @@ where
     deserialize_footer(&footer, u64::MAX)
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn read_batch<R>(
     mut reader: R,
     dictionaries: &mut Dictionaries,
     metadata: &FileMetadata,
     projection: Option<&[usize]>,
     block: usize,
-    meta_buffer: &mut Vec<u8>,
-    block_buffer: &mut Vec<u8>,
+    meta_buffer: &mut ReadBuffer,
+    block_buffer: &mut ReadBuffer,
+    scratch: &mut ReadBuffer,
 ) -> Result<Chunk<Box<dyn Array>>>
 where
     R: AsyncRead + AsyncSeek + Unpin,
@@ -176,11 +181,10 @@ where
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
 
-    meta_buffer.clear();
-    meta_buffer.resize(meta_len, 0);
-    reader.read_exact(meta_buffer).await?;
+    meta_buffer.set_len(meta_len);
+    reader.read_exact(meta_buffer.as_mut()).await?;
 
-    let message = arrow_format::ipc::MessageRef::read_as_root(meta_buffer)
+    let message = arrow_format::ipc::MessageRef::read_as_root(meta_buffer.as_ref())
         .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
 
     let batch = get_serialized_batch(&message)?;
@@ -191,10 +195,9 @@ where
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
 
-    block_buffer.clear();
-    block_buffer.resize(block_length, 0);
-    reader.read_exact(block_buffer).await?;
-    let mut cursor = std::io::Cursor::new(block_buffer);
+    block_buffer.set_len(block_length);
+    reader.read_exact(block_buffer.as_mut()).await?;
+    let mut cursor = std::io::Cursor::new(block_buffer.as_ref());
 
     read_record_batch(
         batch,
@@ -208,6 +211,7 @@ where
         &mut cursor,
         0,
         metadata.size,
+        scratch,
     )
 }
 
@@ -216,13 +220,14 @@ async fn read_dictionaries<R>(
     fields: &[Field],
     ipc_schema: &IpcSchema,
     blocks: &[Block],
+    scratch: &mut ReadBuffer,
 ) -> Result<Dictionaries>
 where
     R: AsyncRead + AsyncSeek + Unpin,
 {
     let mut dictionaries = Default::default();
-    let mut data = vec![];
-    let mut buffer = vec![];
+    let mut data: ReadBuffer = vec![].into();
+    let mut buffer: ReadBuffer = vec![].into();
 
     for block in blocks {
         let offset: u64 = block
@@ -237,7 +242,7 @@ where
 
         read_dictionary_message(&mut reader, offset, &mut data).await?;
 
-        let message = arrow_format::ipc::MessageRef::read_as_root(&data)
+        let message = arrow_format::ipc::MessageRef::read_as_root(data.as_ref())
             .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
 
         let header = message
@@ -245,12 +250,11 @@ where
             .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferHeader(err)))?
             .ok_or_else(|| Error::from(OutOfSpecKind::MissingMessageHeader))?;
 
+        buffer.set_len(length);
         match header {
             MessageHeaderRef::DictionaryBatch(batch) => {
-                buffer.clear();
-                buffer.resize(length, 0);
-                reader.read_exact(&mut buffer).await?;
-                let mut cursor = std::io::Cursor::new(&mut buffer);
+                reader.read_exact(buffer.as_mut()).await?;
+                let mut cursor = std::io::Cursor::new(buffer.as_ref());
                 read_dictionary(
                     batch,
                     fields,
@@ -259,6 +263,7 @@ where
                     &mut cursor,
                     0,
                     u64::MAX,
+                    scratch,
                 )?;
             }
             _ => return Err(Error::from(OutOfSpecKind::UnexpectedMessageType)),
@@ -267,7 +272,7 @@ where
     Ok(dictionaries)
 }
 
-async fn read_dictionary_message<R>(mut reader: R, offset: u64, data: &mut Vec<u8>) -> Result<()>
+async fn read_dictionary_message<R>(mut reader: R, offset: u64, data: &mut ReadBuffer) -> Result<()>
 where
     R: AsyncRead + AsyncSeek + Unpin,
 {
@@ -283,9 +288,8 @@ where
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
 
-    data.clear();
-    data.resize(footer_size, 0);
-    reader.read_exact(data).await?;
+    data.set_len(footer_size);
+    reader.read_exact(data.as_mut()).await?;
 
     Ok(())
 }
@@ -302,6 +306,7 @@ async fn cached_read_dictionaries<R: AsyncRead + AsyncSeek + Unpin>(
                 &metadata.schema.fields,
                 &metadata.ipc_schema,
                 blocks,
+                &mut Default::default(),
             )
             .await?;
             *dictionaries = Some(new_dictionaries);
diff --git a/src/io/ipc/read/mod.rs b/src/io/ipc/read/mod.rs
index 5ffe6426e2..f25da7a4a1 100644
--- a/src/io/ipc/read/mod.rs
+++ b/src/io/ipc/read/mod.rs
@@ -27,6 +27,8 @@ pub mod stream_async;
 #[cfg_attr(docsrs, doc(cfg(feature = "io_ipc_read_async")))]
 pub mod file_async;
 
+use super::super::ReadBuffer;
+
 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/read_basic.rs b/src/io/ipc/read/read_basic.rs
index 378b2976e4..cc04035eec 100644
--- a/src/io/ipc/read/read_basic.rs
+++ b/src/io/ipc/read/read_basic.rs
@@ -7,7 +7,7 @@ use crate::{bitmap::Bitmap, types::NativeType};
 
 use super::super::compression;
 use super::super::endianess::is_native_little_endian;
-use super::{Compression, IpcBuffer, Node, OutOfSpecKind};
+use super::{Compression, IpcBuffer, Node, OutOfSpecKind, ReadBuffer};
 
 fn read_swapped<T: NativeType, R: Read + Seek>(
     reader: &mut R,
@@ -93,6 +93,7 @@ fn read_compressed_buffer<T: NativeType, R: Read + Seek>(
     length: usize,
     is_little_endian: bool,
     compression: Compression,
+    scratch: &mut ReadBuffer,
 ) -> Result<Vec<T>> {
     if is_little_endian != is_native_little_endian() {
         return Err(Error::NotYetImplemented(
@@ -105,9 +106,8 @@ fn read_compressed_buffer<T: NativeType, R: Read + Seek>(
     let mut buffer = vec![T::default(); length];
 
     // decompress first
-    // todo: move this allocation to an external buffer for re-use
-    let mut slice = vec![0u8; buffer_length];
-    reader.read_exact(&mut slice)?;
+    scratch.set_len(buffer_length);
+    reader.read_exact(scratch.as_mut())?;
 
     let out_slice = bytemuck::cast_slice_mut(&mut buffer);
 
@@ -117,10 +117,10 @@ fn read_compressed_buffer<T: NativeType, R: Read + Seek>(
 
     match compression {
         arrow_format::ipc::CompressionType::Lz4Frame => {
-            compression::decompress_lz4(&slice[8..], out_slice)?;
+            compression::decompress_lz4(&scratch.as_ref()[8..], out_slice)?;
         }
         arrow_format::ipc::CompressionType::Zstd => {
-            compression::decompress_zstd(&slice[8..], out_slice)?;
+            compression::decompress_zstd(&scratch.as_ref()[8..], out_slice)?;
         }
     }
     Ok(buffer)
@@ -133,6 +133,7 @@ pub fn read_buffer<T: NativeType, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
+    scratch: &mut ReadBuffer,
 ) -> Result<Buffer<T>> {
     let buf = buf
         .pop_front()
@@ -151,10 +152,15 @@ pub fn read_buffer<T: NativeType, R: Read + Seek>(
     reader.seek(SeekFrom::Start(block_offset + offset))?;
 
     if let Some(compression) = compression {
-        Ok(
-            read_compressed_buffer(reader, buffer_length, length, is_little_endian, compression)?
-                .into(),
-        )
+        Ok(read_compressed_buffer(
+            reader,
+            buffer_length,
+            length,
+            is_little_endian,
+            compression,
+            scratch,
+        )?
+        .into())
     } else {
         Ok(read_uncompressed_buffer(reader, buffer_length, length, is_little_endian)?.into())
     }
diff --git a/src/io/ipc/read/reader.rs b/src/io/ipc/read/reader.rs
index 9807c19849..b01d85cf49 100644
--- a/src/io/ipc/read/reader.rs
+++ b/src/io/ipc/read/reader.rs
@@ -13,6 +13,7 @@ use super::common::*;
 use super::schema::fb_to_schema;
 use super::Dictionaries;
 use super::OutOfSpecKind;
+use super::ReadBuffer;
 use arrow_format::ipc::planus::ReadAsRoot;
 
 /// Metadata of an Arrow IPC file, written in the footer of the file.
@@ -44,13 +45,14 @@ pub struct FileReader<R: Read + Seek> {
     dictionaries: Option<Dictionaries>,
     current_block: usize,
     projection: Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
-    buffer: Vec<u8>,
+    data_scratch: ReadBuffer,
+    message_scratch: ReadBuffer,
 }
 
 fn read_dictionary_message<R: Read + Seek>(
     reader: &mut R,
     offset: u64,
-    data: &mut Vec<u8>,
+    data: &mut ReadBuffer,
 ) -> Result<()> {
     let mut message_size: [u8; 4] = [0; 4];
     reader.seek(SeekFrom::Start(offset))?;
@@ -64,11 +66,8 @@ fn read_dictionary_message<R: Read + Seek>(
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
 
-    // prepare `data` to read the message
-    data.clear();
-    data.resize(message_length, 0);
-
-    reader.read_exact(data)?;
+    data.set_len(message_length);
+    reader.read_exact(data.as_mut())?;
     Ok(())
 }
 
@@ -77,7 +76,8 @@ fn read_dictionary_block<R: Read + Seek>(
     metadata: &FileMetadata,
     block: &arrow_format::ipc::Block,
     dictionaries: &mut Dictionaries,
-    scratch: &mut Vec<u8>,
+    message_scratch: &mut ReadBuffer,
+    dictionary_scratch: &mut ReadBuffer,
 ) -> Result<()> {
     let offset: u64 = block
         .offset
@@ -87,9 +87,9 @@ fn read_dictionary_block<R: Read + Seek>(
         .meta_data_length
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
-    read_dictionary_message(reader, offset, scratch)?;
+    read_dictionary_message(reader, offset, message_scratch)?;
 
-    let message = arrow_format::ipc::MessageRef::read_as_root(scratch)
+    let message = arrow_format::ipc::MessageRef::read_as_root(message_scratch.as_ref())
         .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
 
     let header = message
@@ -108,6 +108,7 @@ fn read_dictionary_block<R: Read + Seek>(
                 reader,
                 block_offset,
                 metadata.size,
+                dictionary_scratch,
             )
         }
         _ => Err(Error::from(OutOfSpecKind::UnexpectedMessageType)),
@@ -119,18 +120,27 @@ fn read_dictionary_block<R: Read + Seek>(
 pub fn read_file_dictionaries<R: Read + Seek>(
     reader: &mut R,
     metadata: &FileMetadata,
+    scratch: &mut ReadBuffer,
 ) -> Result<Dictionaries> {
     let mut dictionaries = Default::default();
-    let mut data = vec![];
 
     let blocks = if let Some(blocks) = metadata.dictionaries.as_deref() {
         blocks
     } else {
         return Ok(HashMap::new());
     };
+    // use a temporary smaller scratch for the messages
+    let mut message_scratch = Default::default();
 
     for block in blocks {
-        read_dictionary_block(reader, metadata, block, &mut dictionaries, &mut data)?;
+        read_dictionary_block(
+            reader,
+            metadata,
+            block,
+            &mut dictionaries,
+            &mut message_scratch,
+            scratch,
+        )?;
     }
     Ok(dictionaries)
 }
@@ -249,7 +259,8 @@ pub fn read_batch<R: Read + Seek>(
     metadata: &FileMetadata,
     projection: Option<&[usize]>,
     index: usize,
-    stratch: &mut Vec<u8>,
+    message_scratch: &mut ReadBuffer,
+    data_scratch: &mut ReadBuffer,
 ) -> Result<Chunk<Box<dyn Array>>> {
     let block = metadata.blocks[index];
 
@@ -270,11 +281,10 @@ pub fn read_batch<R: Read + Seek>(
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
 
-    stratch.clear();
-    stratch.resize(meta_len, 0);
-    reader.read_exact(stratch)?;
+    message_scratch.set_len(meta_len);
+    reader.read_exact(message_scratch.as_mut())?;
 
-    let message = arrow_format::ipc::MessageRef::read_as_root(stratch)
+    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)?;
@@ -301,6 +311,7 @@ pub fn read_batch<R: Read + Seek>(
         reader,
         offset + length,
         metadata.size,
+        data_scratch,
     )
 }
 
@@ -323,7 +334,8 @@ impl<R: Read + Seek> FileReader<R> {
             dictionaries: Default::default(),
             projection,
             current_block: 0,
-            buffer: vec![],
+            data_scratch: Default::default(),
+            message_scratch: Default::default(),
         }
     }
 
@@ -347,7 +359,11 @@ impl<R: Read + Seek> FileReader<R> {
 
     fn read_dictionaries(&mut self) -> Result<()> {
         if self.dictionaries.is_none() {
-            self.dictionaries = Some(read_file_dictionaries(&mut self.reader, &self.metadata)?);
+            self.dictionaries = Some(read_file_dictionaries(
+                &mut self.reader,
+                &self.metadata,
+                &mut self.data_scratch,
+            )?);
         };
         Ok(())
     }
@@ -376,7 +392,8 @@ impl<R: Read + Seek> Iterator for FileReader<R> {
             &self.metadata,
             self.projection.as_ref().map(|x| x.0.as_ref()),
             block,
-            &mut self.buffer,
+            &mut self.message_scratch,
+            &mut self.data_scratch,
         );
 
         let chunk = if let Some((_, map, _)) = &self.projection {
diff --git a/src/io/ipc/read/stream.rs b/src/io/ipc/read/stream.rs
index 3149e5e5b1..5dd7b5cddd 100644
--- a/src/io/ipc/read/stream.rs
+++ b/src/io/ipc/read/stream.rs
@@ -15,6 +15,7 @@ use super::common::*;
 use super::schema::deserialize_stream_metadata;
 use super::Dictionaries;
 use super::OutOfSpecKind;
+use super::ReadBuffer;
 
 /// Metadata of an Arrow IPC stream, written at the start of the stream
 #[derive(Debug, Clone)]
@@ -90,9 +91,10 @@ fn read_next<R: Read>(
     reader: &mut R,
     metadata: &StreamMetadata,
     dictionaries: &mut Dictionaries,
-    message_buffer: &mut Vec<u8>,
-    data_buffer: &mut Vec<u8>,
+    message_buffer: &mut ReadBuffer,
+    data_buffer: &mut ReadBuffer,
     projection: &Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
+    scratch: &mut ReadBuffer,
 ) -> Result<Option<StreamState>> {
     // determine metadata length
     let mut meta_length: [u8; 4] = [0; 4];
@@ -129,11 +131,10 @@ fn read_next<R: Read>(
         return Ok(None);
     }
 
-    message_buffer.clear();
-    message_buffer.resize(meta_length, 0);
-    reader.read_exact(message_buffer)?;
+    message_buffer.set_len(meta_length);
+    reader.read_exact(message_buffer.as_mut())?;
 
-    let message = arrow_format::ipc::MessageRef::read_as_root(message_buffer)
+    let message = arrow_format::ipc::MessageRef::read_as_root(message_buffer.as_ref())
         .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
 
     let header = message
@@ -147,13 +148,12 @@ fn read_next<R: Read>(
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
 
+    data_buffer.set_len(block_length);
     match header {
         arrow_format::ipc::MessageHeaderRef::RecordBatch(batch) => {
-            data_buffer.clear();
-            data_buffer.resize(block_length, 0);
-            reader.read_exact(data_buffer)?;
+            reader.read_exact(data_buffer.as_mut())?;
 
-            let file_size = data_buffer.len() as u64;
+            let file_size = data_buffer.as_ref().len() as u64;
 
             let mut reader = std::io::Cursor::new(data_buffer);
 
@@ -167,6 +167,7 @@ fn read_next<R: Read>(
                 &mut reader,
                 0,
                 file_size,
+                scratch,
             );
 
             if let Some((_, map, _)) = projection {
@@ -192,6 +193,7 @@ fn read_next<R: Read>(
                 &mut dict_reader,
                 0,
                 buf.len() as u64,
+                scratch,
             )?;
 
             // read the next message until we encounter a RecordBatch message
@@ -202,6 +204,7 @@ fn read_next<R: Read>(
                 message_buffer,
                 data_buffer,
                 projection,
+                scratch,
             )
         }
         _ => Err(Error::from(OutOfSpecKind::UnexpectedMessageType)),
@@ -219,9 +222,10 @@ pub struct StreamReader<R: Read> {
     metadata: StreamMetadata,
     dictionaries: Dictionaries,
     finished: bool,
-    data_buffer: Vec<u8>,
-    message_buffer: Vec<u8>,
+    data_buffer: ReadBuffer,
+    message_buffer: ReadBuffer,
     projection: Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
+    scratch: ReadBuffer,
 }
 
 impl<R: Read> StreamReader<R> {
@@ -245,9 +249,10 @@ impl<R: Read> StreamReader<R> {
             metadata,
             dictionaries: Default::default(),
             finished: false,
-            data_buffer: vec![],
-            message_buffer: vec![],
+            data_buffer: Default::default(),
+            message_buffer: Default::default(),
             projection,
+            scratch: Default::default(),
         }
     }
 
@@ -280,6 +285,7 @@ impl<R: Read> StreamReader<R> {
             &mut self.message_buffer,
             &mut self.data_buffer,
             &self.projection,
+            &mut self.scratch,
         )?;
         if batch.is_none() {
             self.finished = true;
diff --git a/src/io/ipc/read/stream_async.rs b/src/io/ipc/read/stream_async.rs
index 1e130b4849..cfbfe5ad3e 100644
--- a/src/io/ipc/read/stream_async.rs
+++ b/src/io/ipc/read/stream_async.rs
@@ -16,6 +16,7 @@ use super::common::{read_dictionary, read_record_batch};
 use super::schema::deserialize_stream_metadata;
 use super::Dictionaries;
 use super::OutOfSpecKind;
+use super::ReadBuffer;
 use super::StreamMetadata;
 
 /// A (private) state of stream messages
@@ -24,9 +25,9 @@ struct ReadState<R> {
     pub metadata: StreamMetadata,
     pub dictionaries: Dictionaries,
     /// The internal buffer to read data inside the messages (records and dictionaries) to
-    pub data_buffer: Vec<u8>,
+    pub data_buffer: ReadBuffer,
     /// The internal buffer to read messages to
-    pub message_buffer: Vec<u8>,
+    pub message_buffer: ReadBuffer,
 }
 
 /// The state of an Arrow stream
@@ -68,6 +69,7 @@ pub async fn read_stream_metadata_async<R: AsyncRead + Unpin + Send>(
 async fn maybe_next<R: AsyncRead + Unpin + Send>(
     mut state: ReadState<R>,
 ) -> Result<Option<StreamState<R>>> {
+    let mut scratch = Default::default();
     // determine metadata length
     let mut meta_length: [u8; 4] = [0; 4];
 
@@ -103,11 +105,13 @@ async fn maybe_next<R: AsyncRead + Unpin + Send>(
         return Ok(None);
     }
 
-    state.message_buffer.clear();
-    state.message_buffer.resize(meta_length, 0);
-    state.reader.read_exact(&mut state.message_buffer).await?;
+    state.message_buffer.set_len(meta_length);
+    state
+        .reader
+        .read_exact(state.message_buffer.as_mut())
+        .await?;
 
-    let message = arrow_format::ipc::MessageRef::read_as_root(&state.message_buffer)
+    let message = arrow_format::ipc::MessageRef::read_as_root(state.message_buffer.as_ref())
         .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
 
     let header = message
@@ -121,11 +125,11 @@ async fn maybe_next<R: AsyncRead + Unpin + Send>(
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
 
+    state.data_buffer.set_len(block_length);
+
     match header {
         arrow_format::ipc::MessageHeaderRef::RecordBatch(batch) => {
-            state.data_buffer.clear();
-            state.data_buffer.resize(block_length, 0);
-            state.reader.read_exact(&mut state.data_buffer).await?;
+            state.reader.read_exact(state.data_buffer.as_mut()).await?;
 
             read_record_batch(
                 batch,
@@ -136,7 +140,8 @@ async fn maybe_next<R: AsyncRead + Unpin + Send>(
                 state.metadata.version,
                 &mut std::io::Cursor::new(&state.data_buffer),
                 0,
-                state.data_buffer.len() as u64,
+                state.data_buffer.as_ref().len() as u64,
+                &mut scratch,
             )
             .map(|chunk| Some(StreamState::Some((state, chunk))))
         }
@@ -156,6 +161,7 @@ async fn maybe_next<R: AsyncRead + Unpin + Send>(
                 &mut dict_reader,
                 0,
                 file_size,
+                &mut scratch,
             )?;
 
             // read the next message until we encounter a Chunk<Box<dyn Array>> message
diff --git a/src/io/mod.rs b/src/io/mod.rs
index 9343d4281c..95a17ee8bf 100644
--- a/src/io/mod.rs
+++ b/src/io/mod.rs
@@ -53,3 +53,9 @@ pub mod print;
 
 #[cfg(any(feature = "io_csv_write", feature = "io_avro", feature = "io_json"))]
 mod iterator;
+
+#[cfg(feature = "io_ipc")]
+mod readbuf;
+
+#[cfg(feature = "io_ipc")]
+pub use readbuf::ReadBuffer;
diff --git a/src/io/parquet/read/deserialize/binary/basic.rs b/src/io/parquet/read/deserialize/binary/basic.rs
index f678ba204a..63aa8ff642 100644
--- a/src/io/parquet/read/deserialize/binary/basic.rs
+++ b/src/io/parquet/read/deserialize/binary/basic.rs
@@ -4,7 +4,7 @@ use std::default::Default;
 use parquet2::{
     deserialize::SliceFilteredIter,
     encoding::{hybrid_rle, Encoding},
-    page::{BinaryPageDict, DataPage},
+    page::{split_buffer, BinaryPageDict, DataPage},
     schema::Repetition,
 };
 
@@ -67,11 +67,11 @@ pub(super) struct Required<'a> {
 }
 
 impl<'a> Required<'a> {
-    pub fn new(page: &'a DataPage) -> Self {
-        let (_, _, values) = utils::split_buffer(page);
+    pub fn try_new(page: &'a DataPage) -> Result<Self> {
+        let (_, _, values) = split_buffer(page)?;
         let values = SizedBinaryIter::new(values, page.num_values());
 
-        Self { values }
+        Ok(Self { values })
     }
 
     pub fn len(&self) -> usize {
@@ -106,10 +106,10 @@ pub(super) struct RequiredDictionary<'a> {
 }
 
 impl<'a> RequiredDictionary<'a> {
-    pub fn new(page: &'a DataPage, dict: &'a BinaryPageDict) -> Self {
-        let values = utils::dict_indices_decoder(page);
+    pub fn try_new(page: &'a DataPage, dict: &'a BinaryPageDict) -> Result<Self> {
+        let values = utils::dict_indices_decoder(page)?;
 
-        Self { dict, values }
+        Ok(Self { dict, values })
     }
 
     #[inline]
@@ -125,13 +125,13 @@ pub(super) struct FilteredRequiredDictionary<'a> {
 }
 
 impl<'a> FilteredRequiredDictionary<'a> {
-    pub fn new(page: &'a DataPage, dict: &'a BinaryPageDict) -> Self {
-        let values = utils::dict_indices_decoder(page);
+    pub fn try_new(page: &'a DataPage, dict: &'a BinaryPageDict) -> Result<Self> {
+        let values = utils::dict_indices_decoder(page)?;
 
         let rows = get_selected_rows(page);
         let values = SliceFilteredIter::new(values, rows);
 
-        Self { values, dict }
+        Ok(Self { values, dict })
     }
 
     #[inline]
@@ -147,10 +147,10 @@ pub(super) struct ValuesDictionary<'a> {
 }
 
 impl<'a> ValuesDictionary<'a> {
-    pub fn new(page: &'a DataPage, dict: &'a BinaryPageDict) -> Self {
-        let values = utils::dict_indices_decoder(page);
+    pub fn try_new(page: &'a DataPage, dict: &'a BinaryPageDict) -> Result<Self> {
+        let values = utils::dict_indices_decoder(page)?;
 
-        Self { dict, values }
+        Ok(Self { dict, values })
     }
 
     #[inline]
@@ -246,50 +246,52 @@ impl<'a, O: Offset> utils::Decoder<'a> for BinaryDecoder<O> {
             is_filtered,
         ) {
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), false, false) => {
-                Ok(State::RequiredDictionary(RequiredDictionary::new(
+                Ok(State::RequiredDictionary(RequiredDictionary::try_new(
                     page,
                     dict.as_any().downcast_ref().unwrap(),
-                )))
+                )?))
             }
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true, false) => {
                 let dict = dict.as_any().downcast_ref().unwrap();
 
                 Ok(State::OptionalDictionary(
-                    OptionalPageValidity::new(page),
-                    ValuesDictionary::new(page, dict),
+                    OptionalPageValidity::try_new(page)?,
+                    ValuesDictionary::try_new(page, dict)?,
                 ))
             }
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), false, true) => {
                 let dict = dict.as_any().downcast_ref().unwrap();
 
-                Ok(State::FilteredRequiredDictionary(
-                    FilteredRequiredDictionary::new(page, dict),
-                ))
+                FilteredRequiredDictionary::try_new(page, dict)
+                    .map(State::FilteredRequiredDictionary)
             }
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true, true) => {
                 let dict = dict.as_any().downcast_ref().unwrap();
 
                 Ok(State::FilteredOptionalDictionary(
-                    FilteredOptionalPageValidity::new(page),
-                    ValuesDictionary::new(page, dict),
+                    FilteredOptionalPageValidity::try_new(page)?,
+                    ValuesDictionary::try_new(page, dict)?,
                 ))
             }
             (Encoding::Plain, _, true, false) => {
-                let (_, _, values) = utils::split_buffer(page);
+                let (_, _, values) = split_buffer(page)?;
 
                 let values = BinaryIter::new(values);
 
-                Ok(State::Optional(OptionalPageValidity::new(page), values))
+                Ok(State::Optional(
+                    OptionalPageValidity::try_new(page)?,
+                    values,
+                ))
             }
-            (Encoding::Plain, _, false, false) => Ok(State::Required(Required::new(page))),
+            (Encoding::Plain, _, false, false) => Ok(State::Required(Required::try_new(page)?)),
             (Encoding::Plain, _, false, true) => {
                 Ok(State::FilteredRequired(FilteredRequired::new(page)))
             }
             (Encoding::Plain, _, true, true) => {
-                let (_, _, values) = utils::split_buffer(page);
+                let (_, _, values) = split_buffer(page)?;
 
                 Ok(State::FilteredOptional(
-                    FilteredOptionalPageValidity::new(page),
+                    FilteredOptionalPageValidity::try_new(page)?,
                     BinaryIter::new(values),
                 ))
             }
diff --git a/src/io/parquet/read/deserialize/binary/nested.rs b/src/io/parquet/read/deserialize/binary/nested.rs
index aeb87a352c..8dbf1ba22d 100644
--- a/src/io/parquet/read/deserialize/binary/nested.rs
+++ b/src/io/parquet/read/deserialize/binary/nested.rs
@@ -1,6 +1,10 @@
 use std::collections::VecDeque;
 
-use parquet2::{encoding::Encoding, page::DataPage, schema::Repetition};
+use parquet2::{
+    encoding::Encoding,
+    page::{split_buffer, DataPage},
+    schema::Repetition,
+};
 
 use crate::{
     array::Offset, bitmap::MutableBitmap, datatypes::DataType, error::Result,
@@ -58,23 +62,25 @@ impl<'a, O: Offset> utils::Decoder<'a> for BinaryDecoder<O> {
         ) {
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), false, false) => {
                 let dict = dict.as_any().downcast_ref().unwrap();
-                Ok(State::RequiredDictionary(ValuesDictionary::new(page, dict)))
+                Ok(State::RequiredDictionary(ValuesDictionary::try_new(
+                    page, dict,
+                )?))
             }
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true, false) => {
                 let dict = dict.as_any().downcast_ref().unwrap();
                 Ok(State::OptionalDictionary(
-                    Optional::new(page),
-                    ValuesDictionary::new(page, dict),
+                    Optional::try_new(page)?,
+                    ValuesDictionary::try_new(page, dict)?,
                 ))
             }
             (Encoding::Plain, _, true, false) => {
-                let (_, _, values) = utils::split_buffer(page);
+                let (_, _, values) = split_buffer(page)?;
 
                 let values = BinaryIter::new(values);
 
-                Ok(State::Optional(Optional::new(page), values))
+                Ok(State::Optional(Optional::try_new(page)?, values))
             }
-            (Encoding::Plain, _, false, false) => Ok(State::Required(Required::new(page))),
+            (Encoding::Plain, _, false, false) => Ok(State::Required(Required::try_new(page)?)),
             _ => Err(utils::not_implemented(page)),
         }
     }
diff --git a/src/io/parquet/read/deserialize/boolean/basic.rs b/src/io/parquet/read/deserialize/boolean/basic.rs
index cca21bae28..d3189b23f6 100644
--- a/src/io/parquet/read/deserialize/boolean/basic.rs
+++ b/src/io/parquet/read/deserialize/boolean/basic.rs
@@ -1,7 +1,10 @@
 use std::collections::VecDeque;
 
 use parquet2::{
-    deserialize::SliceFilteredIter, encoding::Encoding, page::DataPage, schema::Repetition,
+    deserialize::SliceFilteredIter,
+    encoding::Encoding,
+    page::{split_buffer, DataPage},
+    schema::Repetition,
 };
 
 use crate::{
@@ -13,7 +16,7 @@ use crate::{
 
 use super::super::utils;
 use super::super::utils::{
-    extend_from_decoder, get_selected_rows, next, split_buffer, DecodedState, Decoder,
+    extend_from_decoder, get_selected_rows, next, DecodedState, Decoder,
     FilteredOptionalPageValidity, MaybeNext, OptionalPageValidity,
 };
 use super::super::DataPages;
@@ -22,10 +25,10 @@ use super::super::DataPages;
 struct Values<'a>(BitmapIter<'a>);
 
 impl<'a> Values<'a> {
-    pub fn new(page: &'a DataPage) -> Self {
-        let (_, _, values) = split_buffer(page);
+    pub fn try_new(page: &'a DataPage) -> Result<Self> {
+        let (_, _, values) = split_buffer(page)?;
 
-        Self(BitmapIter::new(values, 0, values.len() * 8))
+        Ok(Self(BitmapIter::new(values, 0, values.len() * 8)))
     }
 }
 
@@ -54,15 +57,15 @@ struct FilteredRequired<'a> {
 }
 
 impl<'a> FilteredRequired<'a> {
-    pub fn new(page: &'a DataPage) -> Self {
-        let (_, _, values) = utils::split_buffer(page);
+    pub fn try_new(page: &'a DataPage) -> Result<Self> {
+        let (_, _, values) = split_buffer(page)?;
         // todo: replace this by an iterator over slices, for faster deserialization
         let values = BitmapIter::new(values, 0, page.num_values());
 
         let rows = get_selected_rows(page);
         let values = SliceFilteredIter::new(values, rows);
 
-        Self { values }
+        Ok(Self { values })
     }
 
     #[inline]
@@ -117,16 +120,16 @@ impl<'a> Decoder<'a> for BooleanDecoder {
 
         match (page.encoding(), is_optional, is_filtered) {
             (Encoding::Plain, true, false) => Ok(State::Optional(
-                OptionalPageValidity::new(page),
-                Values::new(page),
+                OptionalPageValidity::try_new(page)?,
+                Values::try_new(page)?,
             )),
             (Encoding::Plain, false, false) => Ok(State::Required(Required::new(page))),
             (Encoding::Plain, true, true) => Ok(State::FilteredOptional(
-                FilteredOptionalPageValidity::new(page),
-                Values::new(page),
+                FilteredOptionalPageValidity::try_new(page)?,
+                Values::try_new(page)?,
             )),
             (Encoding::Plain, false, true) => {
-                Ok(State::FilteredRequired(FilteredRequired::new(page)))
+                Ok(State::FilteredRequired(FilteredRequired::try_new(page)?))
             }
             _ => Err(utils::not_implemented(page)),
         }
diff --git a/src/io/parquet/read/deserialize/boolean/nested.rs b/src/io/parquet/read/deserialize/boolean/nested.rs
index 8b162908cb..163f23a9f3 100644
--- a/src/io/parquet/read/deserialize/boolean/nested.rs
+++ b/src/io/parquet/read/deserialize/boolean/nested.rs
@@ -1,6 +1,10 @@
 use std::collections::VecDeque;
 
-use parquet2::{encoding::Encoding, page::DataPage, schema::Repetition};
+use parquet2::{
+    encoding::Encoding,
+    page::{split_buffer, DataPage},
+    schema::Repetition,
+};
 
 use crate::{
     array::BooleanArray,
@@ -24,13 +28,13 @@ struct Required<'a> {
 }
 
 impl<'a> Required<'a> {
-    pub fn new(page: &'a DataPage) -> Self {
-        let (_, _, values) = utils::split_buffer(page);
-        Self {
+    pub fn try_new(page: &'a DataPage) -> Result<Self> {
+        let (_, _, values) = split_buffer(page)?;
+        Ok(Self {
             values,
             offset: 0,
             length: page.num_values(),
-        }
+        })
     }
 }
 
@@ -71,12 +75,12 @@ impl<'a> Decoder<'a> for BooleanDecoder {
 
         match (page.encoding(), is_optional, is_filtered) {
             (Encoding::Plain, true, false) => {
-                let (_, _, values) = utils::split_buffer(page);
+                let (_, _, values) = split_buffer(page)?;
                 let values = BitmapIter::new(values, 0, values.len() * 8);
 
-                Ok(State::Optional(Optional::new(page), values))
+                Ok(State::Optional(Optional::try_new(page)?, values))
             }
-            (Encoding::Plain, false, false) => Ok(State::Required(Required::new(page))),
+            (Encoding::Plain, false, false) => Ok(State::Required(Required::try_new(page)?)),
             _ => Err(utils::not_implemented(page)),
         }
     }
diff --git a/src/io/parquet/read/deserialize/dictionary.rs b/src/io/parquet/read/deserialize/dictionary.rs
index d62260f5b5..62ef51adc9 100644
--- a/src/io/parquet/read/deserialize/dictionary.rs
+++ b/src/io/parquet/read/deserialize/dictionary.rs
@@ -36,9 +36,9 @@ pub struct Required<'a> {
 }
 
 impl<'a> Required<'a> {
-    fn new(page: &'a DataPage) -> Self {
-        let values = dict_indices_decoder(page);
-        Self { values }
+    fn try_new(page: &'a DataPage) -> Result<Self> {
+        let values = dict_indices_decoder(page)?;
+        Ok(Self { values })
     }
 }
 
@@ -48,13 +48,13 @@ pub struct FilteredRequired<'a> {
 }
 
 impl<'a> FilteredRequired<'a> {
-    fn new(page: &'a DataPage) -> Self {
-        let values = dict_indices_decoder(page);
+    fn try_new(page: &'a DataPage) -> Result<Self> {
+        let values = dict_indices_decoder(page)?;
 
         let rows = get_selected_rows(page);
         let values = SliceFilteredIter::new(values, rows);
 
-        Self { values }
+        Ok(Self { values })
     }
 }
 
@@ -65,13 +65,13 @@ pub struct Optional<'a> {
 }
 
 impl<'a> Optional<'a> {
-    fn new(page: &'a DataPage) -> Self {
-        let values = dict_indices_decoder(page);
+    fn try_new(page: &'a DataPage) -> Result<Self> {
+        let values = dict_indices_decoder(page)?;
 
-        Self {
+        Ok(Self {
             values,
-            validity: OptionalPageValidity::new(page),
-        }
+            validity: OptionalPageValidity::try_new(page)?,
+        })
     }
 }
 
@@ -120,18 +120,18 @@ where
 
         match (page.encoding(), is_optional, is_filtered) {
             (Encoding::PlainDictionary | Encoding::RleDictionary, false, false) => {
-                Ok(State::Required(Required::new(page)))
+                Required::try_new(page).map(State::Required)
             }
             (Encoding::PlainDictionary | Encoding::RleDictionary, true, false) => {
-                Ok(State::Optional(Optional::new(page)))
+                Optional::try_new(page).map(State::Optional)
             }
             (Encoding::PlainDictionary | Encoding::RleDictionary, false, true) => {
-                Ok(State::FilteredRequired(FilteredRequired::new(page)))
+                FilteredRequired::try_new(page).map(State::FilteredRequired)
             }
             (Encoding::PlainDictionary | Encoding::RleDictionary, true, true) => {
                 Ok(State::FilteredOptional(
-                    FilteredOptionalPageValidity::new(page),
-                    dict_indices_decoder(page),
+                    FilteredOptionalPageValidity::try_new(page)?,
+                    dict_indices_decoder(page)?,
                 ))
             }
             _ => Err(utils::not_implemented(page)),
diff --git a/src/io/parquet/read/deserialize/fixed_size_binary/basic.rs b/src/io/parquet/read/deserialize/fixed_size_binary/basic.rs
index bbf8abb00b..5f260967e7 100644
--- a/src/io/parquet/read/deserialize/fixed_size_binary/basic.rs
+++ b/src/io/parquet/read/deserialize/fixed_size_binary/basic.rs
@@ -3,7 +3,7 @@ use std::collections::VecDeque;
 use parquet2::{
     deserialize::SliceFilteredIter,
     encoding::{hybrid_rle, Encoding},
-    page::{DataPage, FixedLenByteArrayPageDict},
+    page::{split_buffer, DataPage, FixedLenByteArrayPageDict},
     schema::Repetition,
 };
 
@@ -13,8 +13,8 @@ use crate::{
 
 use super::super::utils::{
     dict_indices_decoder, extend_from_decoder, get_selected_rows, next, not_implemented,
-    split_buffer, DecodedState, Decoder, FilteredOptionalPageValidity, MaybeNext,
-    OptionalPageValidity, PageState, Pushable,
+    DecodedState, Decoder, FilteredOptionalPageValidity, MaybeNext, OptionalPageValidity,
+    PageState, Pushable,
 };
 use super::super::DataPages;
 use super::utils::FixedSizeBinary;
@@ -26,15 +26,15 @@ struct Optional<'a> {
 }
 
 impl<'a> Optional<'a> {
-    fn new(page: &'a DataPage, size: usize) -> Self {
-        let (_, _, values_buffer) = split_buffer(page);
+    fn try_new(page: &'a DataPage, size: usize) -> Result<Self> {
+        let (_, _, values) = split_buffer(page)?;
 
-        let values = values_buffer.chunks_exact(size);
+        let values = values.chunks_exact(size);
 
-        Self {
+        Ok(Self {
             values,
-            validity: OptionalPageValidity::new(page),
-        }
+            validity: OptionalPageValidity::try_new(page)?,
+        })
     }
 }
 
@@ -87,10 +87,10 @@ struct RequiredDictionary<'a> {
 }
 
 impl<'a> RequiredDictionary<'a> {
-    fn new(page: &'a DataPage, dict: &'a FixedLenByteArrayPageDict) -> Self {
-        let values = dict_indices_decoder(page);
+    fn try_new(page: &'a DataPage, dict: &'a FixedLenByteArrayPageDict) -> Result<Self> {
+        let values = dict_indices_decoder(page)?;
 
-        Self { dict, values }
+        Ok(Self { dict, values })
     }
 
     #[inline]
@@ -107,14 +107,14 @@ struct OptionalDictionary<'a> {
 }
 
 impl<'a> OptionalDictionary<'a> {
-    fn new(page: &'a DataPage, dict: &'a FixedLenByteArrayPageDict) -> Self {
-        let values = dict_indices_decoder(page);
+    fn try_new(page: &'a DataPage, dict: &'a FixedLenByteArrayPageDict) -> Result<Self> {
+        let values = dict_indices_decoder(page)?;
 
-        Self {
+        Ok(Self {
             values,
-            validity: OptionalPageValidity::new(page),
+            validity: OptionalPageValidity::try_new(page)?,
             dict,
-        }
+        })
     }
 }
 
@@ -170,31 +170,27 @@ impl<'a> Decoder<'a> for BinaryDecoder {
             is_filtered,
         ) {
             (Encoding::Plain, None, true, false) => {
-                Ok(State::Optional(Optional::new(page, self.size)))
+                Ok(State::Optional(Optional::try_new(page, self.size)?))
             }
             (Encoding::Plain, None, false, false) => {
                 Ok(State::Required(Required::new(page, self.size)))
             }
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), false, false) => {
-                Ok(State::RequiredDictionary(RequiredDictionary::new(
-                    page,
-                    dict.as_any().downcast_ref().unwrap(),
-                )))
+                RequiredDictionary::try_new(page, dict.as_any().downcast_ref().unwrap())
+                    .map(State::RequiredDictionary)
             }
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true, false) => {
-                Ok(State::OptionalDictionary(OptionalDictionary::new(
-                    page,
-                    dict.as_any().downcast_ref().unwrap(),
-                )))
+                OptionalDictionary::try_new(page, dict.as_any().downcast_ref().unwrap())
+                    .map(State::OptionalDictionary)
             }
             (Encoding::Plain, None, false, true) => Ok(State::FilteredRequired(
                 FilteredRequired::new(page, self.size),
             )),
             (Encoding::Plain, _, true, true) => {
-                let (_, _, values) = split_buffer(page);
+                let (_, _, values) = split_buffer(page)?;
 
                 Ok(State::FilteredOptional(
-                    FilteredOptionalPageValidity::new(page),
+                    FilteredOptionalPageValidity::try_new(page)?,
                     values.chunks_exact(self.size),
                 ))
             }
diff --git a/src/io/parquet/read/deserialize/nested_utils.rs b/src/io/parquet/read/deserialize/nested_utils.rs
index c829ca43ea..5c7f062f96 100644
--- a/src/io/parquet/read/deserialize/nested_utils.rs
+++ b/src/io/parquet/read/deserialize/nested_utils.rs
@@ -1,14 +1,16 @@
 use std::collections::VecDeque;
 
 use parquet2::{
-    encoding::hybrid_rle::HybridRleDecoder, page::DataPage, read::levels::get_bit_width,
+    encoding::hybrid_rle::HybridRleDecoder,
+    page::{split_buffer, DataPage},
+    read::levels::get_bit_width,
 };
 
 use crate::{array::Array, bitmap::MutableBitmap, error::Result};
 
 use super::super::DataPages;
 pub use super::utils::Zip;
-use super::utils::{split_buffer, DecodedState, Decoder, MaybeNext, Pushable};
+use super::utils::{DecodedState, Decoder, MaybeNext, Pushable};
 
 /// trait describing deserialized repetition and definition levels
 pub trait Nested: std::fmt::Debug + Send + Sync {
@@ -271,8 +273,8 @@ pub struct NestedPage<'a> {
 }
 
 impl<'a> NestedPage<'a> {
-    pub fn new(page: &'a DataPage) -> Self {
-        let (rep_levels, def_levels, _) = split_buffer(page);
+    pub fn try_new(page: &'a DataPage) -> Result<Self> {
+        let (rep_levels, def_levels, _) = split_buffer(page)?;
 
         let max_rep_level = page.descriptor.max_rep_level;
         let max_def_level = page.descriptor.max_def_level;
@@ -284,7 +286,7 @@ impl<'a> NestedPage<'a> {
 
         let iter = reps.zip(defs).peekable();
 
-        Self { iter }
+        Ok(Self { iter })
     }
 
     // number of values (!= number of rows)
@@ -458,15 +460,15 @@ impl<'a> Iterator for Optional<'a> {
 }
 
 impl<'a> Optional<'a> {
-    pub fn new(page: &'a DataPage) -> Self {
-        let (_, def_levels, _) = split_buffer(page);
+    pub fn try_new(page: &'a DataPage) -> Result<Self> {
+        let (_, def_levels, _) = split_buffer(page)?;
 
         let max_def = page.descriptor.max_def_level;
 
-        Self {
+        Ok(Self {
             iter: HybridRleDecoder::new(def_levels, get_bit_width(max_def), page.num_values()),
             max_def: max_def as u32,
-        }
+        })
     }
 
     #[inline]
@@ -508,7 +510,11 @@ where
         }
         Ok(Some(page)) => {
             // there is a new page => consume the page from the start
-            let mut nested_page = NestedPage::new(page);
+            let nested_page = NestedPage::try_new(page);
+            let mut nested_page = match nested_page {
+                Ok(page) => page,
+                Err(e) => return MaybeNext::Some(Err(e)),
+            };
 
             extend_offsets1(&mut nested_page, init, nested_items, chunk_size);
 
diff --git a/src/io/parquet/read/deserialize/primitive/basic.rs b/src/io/parquet/read/deserialize/primitive/basic.rs
index 51aac9736b..6e947b5643 100644
--- a/src/io/parquet/read/deserialize/primitive/basic.rs
+++ b/src/io/parquet/read/deserialize/primitive/basic.rs
@@ -3,7 +3,7 @@ use std::collections::VecDeque;
 use parquet2::{
     deserialize::SliceFilteredIter,
     encoding::{hybrid_rle, Encoding},
-    page::{DataPage, PrimitivePageDict},
+    page::{split_buffer, DataPage, PrimitivePageDict},
     schema::Repetition,
     types::decode,
     types::NativeType as ParquetNativeType,
@@ -24,8 +24,8 @@ struct FilteredRequiredValues<'a> {
 }
 
 impl<'a> FilteredRequiredValues<'a> {
-    pub fn new<P: ParquetNativeType>(page: &'a DataPage) -> Self {
-        let (_, _, values) = utils::split_buffer(page);
+    pub fn try_new<P: ParquetNativeType>(page: &'a DataPage) -> Result<Self> {
+        let (_, _, values) = split_buffer(page)?;
         assert_eq!(values.len() % std::mem::size_of::<P>(), 0);
 
         let values = values.chunks_exact(std::mem::size_of::<P>());
@@ -33,7 +33,7 @@ impl<'a> FilteredRequiredValues<'a> {
         let rows = get_selected_rows(page);
         let values = SliceFilteredIter::new(values, rows);
 
-        Self { values }
+        Ok(Self { values })
     }
 
     #[inline]
@@ -48,12 +48,12 @@ pub(super) struct Values<'a> {
 }
 
 impl<'a> Values<'a> {
-    pub fn new<P: ParquetNativeType>(page: &'a DataPage) -> Self {
-        let (_, _, values) = utils::split_buffer(page);
+    pub fn try_new<P: ParquetNativeType>(page: &'a DataPage) -> Result<Self> {
+        let (_, _, values) = split_buffer(page)?;
         assert_eq!(values.len() % std::mem::size_of::<P>(), 0);
-        Self {
+        Ok(Self {
             values: values.chunks_exact(std::mem::size_of::<P>()),
-        }
+        })
     }
 
     #[inline]
@@ -75,13 +75,13 @@ impl<'a, P> ValuesDictionary<'a, P>
 where
     P: ParquetNativeType,
 {
-    pub fn new(page: &'a DataPage, dict: &'a PrimitivePageDict<P>) -> Self {
-        let values = utils::dict_indices_decoder(page);
+    pub fn try_new(page: &'a DataPage, dict: &'a PrimitivePageDict<P>) -> Result<Self> {
+        let values = utils::dict_indices_decoder(page)?;
 
-        Self {
+        Ok(Self {
             dict: dict.values(),
             values,
-        }
+        })
     }
 
     #[inline]
@@ -176,29 +176,29 @@ where
         ) {
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), false, false) => {
                 let dict = dict.as_any().downcast_ref().unwrap();
-                Ok(State::RequiredDictionary(ValuesDictionary::new(page, dict)))
+                ValuesDictionary::try_new(page, dict).map(State::RequiredDictionary)
             }
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true, false) => {
                 let dict = dict.as_any().downcast_ref().unwrap();
 
                 Ok(State::OptionalDictionary(
-                    OptionalPageValidity::new(page),
-                    ValuesDictionary::new(page, dict),
+                    OptionalPageValidity::try_new(page)?,
+                    ValuesDictionary::try_new(page, dict)?,
                 ))
             }
             (Encoding::Plain, _, true, false) => {
-                let validity = OptionalPageValidity::new(page);
-                let values = Values::new::<P>(page);
+                let validity = OptionalPageValidity::try_new(page)?;
+                let values = Values::try_new::<P>(page)?;
 
                 Ok(State::Optional(validity, values))
             }
-            (Encoding::Plain, _, false, false) => Ok(State::Required(Values::new::<P>(page))),
+            (Encoding::Plain, _, false, false) => Ok(State::Required(Values::try_new::<P>(page)?)),
             (Encoding::Plain, _, false, true) => Ok(State::FilteredRequired(
-                FilteredRequiredValues::new::<P>(page),
+                FilteredRequiredValues::try_new::<P>(page)?,
             )),
             (Encoding::Plain, _, true, true) => Ok(State::FilteredOptional(
-                FilteredOptionalPageValidity::new(page),
-                Values::new::<P>(page),
+                FilteredOptionalPageValidity::try_new(page)?,
+                Values::try_new::<P>(page)?,
             )),
             _ => Err(utils::not_implemented(page)),
         }
diff --git a/src/io/parquet/read/deserialize/primitive/nested.rs b/src/io/parquet/read/deserialize/primitive/nested.rs
index 841ef1719b..9263137298 100644
--- a/src/io/parquet/read/deserialize/primitive/nested.rs
+++ b/src/io/parquet/read/deserialize/primitive/nested.rs
@@ -92,19 +92,20 @@ where
         ) {
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), false, false) => {
                 let dict = dict.as_any().downcast_ref().unwrap();
-                Ok(State::RequiredDictionary(ValuesDictionary::new(page, dict)))
+                ValuesDictionary::try_new(page, dict).map(State::RequiredDictionary)
             }
             (Encoding::PlainDictionary | Encoding::RleDictionary, Some(dict), true, false) => {
                 let dict = dict.as_any().downcast_ref().unwrap();
                 Ok(State::OptionalDictionary(
-                    Optional::new(page),
-                    ValuesDictionary::new(page, dict),
+                    Optional::try_new(page)?,
+                    ValuesDictionary::try_new(page, dict)?,
                 ))
             }
-            (Encoding::Plain, _, true, false) => {
-                Ok(State::Optional(Optional::new(page), Values::new::<P>(page)))
-            }
-            (Encoding::Plain, _, false, false) => Ok(State::Required(Values::new::<P>(page))),
+            (Encoding::Plain, _, true, false) => Ok(State::Optional(
+                Optional::try_new(page)?,
+                Values::try_new::<P>(page)?,
+            )),
+            (Encoding::Plain, _, false, false) => Ok(State::Required(Values::try_new::<P>(page)?)),
             _ => Err(utils::not_implemented(page)),
         }
     }
diff --git a/src/io/parquet/read/deserialize/simple.rs b/src/io/parquet/read/deserialize/simple.rs
index aaab8aee9f..20dba38065 100644
--- a/src/io/parquet/read/deserialize/simple.rs
+++ b/src/io/parquet/read/deserialize/simple.rs
@@ -467,6 +467,12 @@ fn dict_read<'a, K: DictionaryKey, I: 'a + DataPages>(
             chunk_size,
             |x: i32| x as u32,
         )),
+        UInt64 => dyn_iter(primitive::DictIter::<K, _, _, _, _>::new(
+            iter,
+            data_type,
+            chunk_size,
+            |x: i64| x as u64,
+        )),
         Int8 => dyn_iter(primitive::DictIter::<K, _, _, _, _>::new(
             iter,
             data_type,
diff --git a/src/io/parquet/read/deserialize/utils.rs b/src/io/parquet/read/deserialize/utils.rs
index dd2174c418..b29bc48ba4 100644
--- a/src/io/parquet/read/deserialize/utils.rs
+++ b/src/io/parquet/read/deserialize/utils.rs
@@ -5,7 +5,7 @@ use parquet2::deserialize::{
 };
 use parquet2::encoding::hybrid_rle;
 use parquet2::indexes::Interval;
-use parquet2::page::{split_buffer as _split_buffer, DataPage};
+use parquet2::page::{split_buffer, DataPage};
 use parquet2::schema::Repetition;
 
 use crate::bitmap::utils::BitmapIter;
@@ -34,11 +34,6 @@ pub fn not_implemented(page: &DataPage) -> Error {
     ))
 }
 
-#[inline]
-pub fn split_buffer(page: &DataPage) -> (&[u8], &[u8], &[u8]) {
-    _split_buffer(page)
-}
-
 /// A private trait representing structs that can receive elements.
 pub(super) trait Pushable<T>: Sized {
     //fn reserve(&mut self, additional: usize);
@@ -104,18 +99,18 @@ pub struct FilteredOptionalPageValidity<'a> {
 }
 
 impl<'a> FilteredOptionalPageValidity<'a> {
-    pub fn new(page: &'a DataPage) -> Self {
-        let (_, validity, _) = split_buffer(page);
+    pub fn try_new(page: &'a DataPage) -> Result<Self, Error> {
+        let (_, validity, _) = split_buffer(page)?;
 
         let iter = hybrid_rle::Decoder::new(validity, 1);
         let iter = HybridDecoderBitmapIter::new(iter, page.num_values());
         let selected_rows = get_selected_rows(page);
         let iter = FilteredHybridRleDecoderIter::new(iter, selected_rows);
 
-        Self {
+        Ok(Self {
             iter,
             current: None,
-        }
+        })
     }
 
     pub fn len(&self) -> usize {
@@ -219,15 +214,15 @@ pub struct OptionalPageValidity<'a> {
 }
 
 impl<'a> OptionalPageValidity<'a> {
-    pub fn new(page: &'a DataPage) -> Self {
-        let (_, validity, _) = split_buffer(page);
+    pub fn try_new(page: &'a DataPage) -> Result<Self, Error> {
+        let (_, validity, _) = split_buffer(page)?;
 
         let iter = hybrid_rle::Decoder::new(validity, 1);
         let iter = HybridDecoderBitmapIter::new(iter, page.num_values());
-        Self {
+        Ok(Self {
             iter,
             current: None,
-        }
+        })
     }
 
     /// Number of items remaining
@@ -459,13 +454,17 @@ pub(super) fn next<'a, I: DataPages, D: Decoder<'a>>(
 }
 
 #[inline]
-pub(super) fn dict_indices_decoder(page: &DataPage) -> hybrid_rle::HybridRleDecoder {
-    let (_, _, indices_buffer) = split_buffer(page);
+pub(super) fn dict_indices_decoder(page: &DataPage) -> Result<hybrid_rle::HybridRleDecoder, Error> {
+    let (_, _, indices_buffer) = split_buffer(page)?;
 
     // SPEC: Data page format: the bit width used to encode the entry ids stored as 1 byte (max bit width = 32),
     // SPEC: followed by the values encoded using RLE/Bit packed described above (with the given bit width).
     let bit_width = indices_buffer[0];
     let indices_buffer = &indices_buffer[1..];
 
-    hybrid_rle::HybridRleDecoder::new(indices_buffer, bit_width as u32, page.num_values())
+    Ok(hybrid_rle::HybridRleDecoder::new(
+        indices_buffer,
+        bit_width as u32,
+        page.num_values(),
+    ))
 }
diff --git a/src/io/parquet/read/row_group.rs b/src/io/parquet/read/row_group.rs
index 0c7222343c..5d9198e6f4 100644
--- a/src/io/parquet/read/row_group.rs
+++ b/src/io/parquet/read/row_group.rs
@@ -88,7 +88,7 @@ impl Iterator for RowGroupDeserializer {
 
 /// Returns all [`ColumnChunkMetaData`] associated to `field_name`.
 /// For non-nested parquet types, this returns a single column
-pub(super) fn get_field_columns<'a>(
+pub fn get_field_columns<'a>(
     columns: &'a [ColumnChunkMetaData],
     field_name: &str,
 ) -> Vec<&'a ColumnChunkMetaData> {
diff --git a/src/io/parquet/write/dictionary.rs b/src/io/parquet/write/dictionary.rs
index 65f13a4904..05c56207bf 100644
--- a/src/io/parquet/write/dictionary.rs
+++ b/src/io/parquet/write/dictionary.rs
@@ -149,7 +149,7 @@ pub fn array_to_pages<K: DictionaryKey>(
                 DataType::UInt8 => dyn_prim!(u8, i32, array, options, type_),
                 DataType::UInt16 => dyn_prim!(u16, i32, array, options, type_),
                 DataType::UInt32 => dyn_prim!(u32, i32, array, options, type_),
-                DataType::UInt64 => dyn_prim!(i64, i64, array, options, type_),
+                DataType::UInt64 => dyn_prim!(u64, i64, array, options, type_),
                 DataType::Float32 => dyn_prim!(f32, f32, array, options, type_),
                 DataType::Float64 => dyn_prim!(f64, f64, array, options, type_),
                 DataType::Utf8 => {
diff --git a/src/io/readbuf.rs b/src/io/readbuf.rs
new file mode 100644
index 0000000000..d30396a828
--- /dev/null
+++ b/src/io/readbuf.rs
@@ -0,0 +1,50 @@
+/// A small wrapper around [`Vec<u8>`] that allows us to reuse memory once it is initialized.
+/// This may improve performance of the [`Read`] trait.
+#[derive(Clone, Default)]
+pub struct ReadBuffer {
+    data: Vec<u8>,
+    // length to be read or is read
+    length: usize,
+}
+
+impl ReadBuffer {
+    /// Set the minimal length of the [`ReadBuf`]. Contrary to the
+    /// method on `Vec` this is `safe` because this function guarantees that
+    /// the underlying data always is initialized.
+    pub fn set_len(&mut self, length: usize) {
+        if length > self.data.capacity() {
+            // exponential growing strategy
+            // benchmark showed it was ~5% faster
+            // in reading lz4 yellow-trip dataset
+            self.data = vec![0; length * 2];
+        } else if length > self.data.len() {
+            self.data.resize(length, 0);
+        }
+        self.length = length;
+    }
+}
+
+impl AsRef<[u8]> for ReadBuffer {
+    fn as_ref(&self) -> &[u8] {
+        &self.data[..self.length]
+    }
+}
+
+impl AsMut<[u8]> for ReadBuffer {
+    fn as_mut(&mut self) -> &mut [u8] {
+        &mut self.data[..self.length]
+    }
+}
+
+impl From<Vec<u8>> for ReadBuffer {
+    fn from(data: Vec<u8>) -> Self {
+        let length = data.len();
+        Self { data, length }
+    }
+}
+
+impl From<ReadBuffer> for Vec<u8> {
+    fn from(buf: ReadBuffer) -> Self {
+        buf.data
+    }
+}
diff --git a/src/types/native.rs b/src/types/native.rs
index fbc8ac408b..4d4f34eb60 100644
--- a/src/types/native.rs
+++ b/src/types/native.rs
@@ -33,9 +33,6 @@ pub trait NativeType:
     /// To bytes in little endian
     fn to_le_bytes(&self) -> Self::Bytes;
 
-    /// To bytes in native endian
-    fn to_ne_bytes(&self) -> Self::Bytes;
-
     /// To bytes in big endian
     fn to_be_bytes(&self) -> Self::Bytes;
 
@@ -59,11 +56,6 @@ macro_rules! native_type {
                 Self::to_be_bytes(*self)
             }
 
-            #[inline]
-            fn to_ne_bytes(&self) -> Self::Bytes {
-                Self::to_ne_bytes(*self)
-            }
-
             #[inline]
             fn from_be_bytes(bytes: Self::Bytes) -> Self {
                 Self::from_be_bytes(bytes)
@@ -88,25 +80,25 @@ native_type!(i128, PrimitiveType::Int128);
 #[derive(Debug, Copy, Clone, Default, PartialEq, Eq, Hash, Zeroable, Pod)]
 #[allow(non_camel_case_types)]
 #[repr(C)]
-pub struct days_ms([i32; 2]);
+pub struct days_ms(pub i32, pub i32);
 
 impl days_ms {
     /// A new [`days_ms`].
     #[inline]
     pub fn new(days: i32, milliseconds: i32) -> Self {
-        Self([days, milliseconds])
+        Self(days, milliseconds)
     }
 
     /// The number of days
     #[inline]
     pub fn days(&self) -> i32 {
-        self.0[0]
+        self.0
     }
 
     /// The number of milliseconds
     #[inline]
     pub fn milliseconds(&self) -> i32 {
-        self.0[1]
+        self.1
     }
 }
 
@@ -115,24 +107,8 @@ impl NativeType for days_ms {
     type Bytes = [u8; 8];
     #[inline]
     fn to_le_bytes(&self) -> Self::Bytes {
-        let days = self.0[0].to_le_bytes();
-        let ms = self.0[1].to_le_bytes();
-        let mut result = [0; 8];
-        result[0] = days[0];
-        result[1] = days[1];
-        result[2] = days[2];
-        result[3] = days[3];
-        result[4] = ms[0];
-        result[5] = ms[1];
-        result[6] = ms[2];
-        result[7] = ms[3];
-        result
-    }
-
-    #[inline]
-    fn to_ne_bytes(&self) -> Self::Bytes {
-        let days = self.0[0].to_ne_bytes();
-        let ms = self.0[1].to_ne_bytes();
+        let days = self.0.to_le_bytes();
+        let ms = self.1.to_le_bytes();
         let mut result = [0; 8];
         result[0] = days[0];
         result[1] = days[1];
@@ -147,8 +123,8 @@ impl NativeType for days_ms {
 
     #[inline]
     fn to_be_bytes(&self) -> Self::Bytes {
-        let days = self.0[0].to_be_bytes();
-        let ms = self.0[1].to_be_bytes();
+        let days = self.0.to_be_bytes();
+        let ms = self.1.to_be_bytes();
         let mut result = [0; 8];
         result[0] = days[0];
         result[1] = days[1];
@@ -173,7 +149,7 @@ impl NativeType for days_ms {
         ms[1] = bytes[5];
         ms[2] = bytes[6];
         ms[3] = bytes[7];
-        Self([i32::from_be_bytes(days), i32::from_be_bytes(ms)])
+        Self(i32::from_be_bytes(days), i32::from_be_bytes(ms))
     }
 }
 
@@ -181,7 +157,7 @@ impl NativeType for days_ms {
 #[derive(Debug, Copy, Clone, Default, PartialEq, Eq, Hash, Zeroable, Pod)]
 #[allow(non_camel_case_types)]
 #[repr(C)]
-pub struct months_days_ns(i32, i32, i64);
+pub struct months_days_ns(pub i32, pub i32, pub i64);
 
 impl months_days_ns {
     /// A new [`months_days_ns`].
@@ -232,26 +208,6 @@ impl NativeType for months_days_ns {
         result
     }
 
-    #[inline]
-    fn to_ne_bytes(&self) -> Self::Bytes {
-        let months = self.months().to_ne_bytes();
-        let days = self.days().to_ne_bytes();
-        let ns = self.ns().to_ne_bytes();
-        let mut result = [0; 16];
-        result[0] = months[0];
-        result[1] = months[1];
-        result[2] = months[2];
-        result[3] = months[3];
-        result[4] = days[0];
-        result[5] = days[1];
-        result[6] = days[2];
-        result[7] = days[3];
-        (0..8).for_each(|i| {
-            result[8 + i] = ns[i];
-        });
-        result
-    }
-
     #[inline]
     fn to_be_bytes(&self) -> Self::Bytes {
         let months = self.months().to_be_bytes();
@@ -420,11 +376,6 @@ impl NativeType for f16 {
         self.0.to_le_bytes()
     }
 
-    #[inline]
-    fn to_ne_bytes(&self) -> Self::Bytes {
-        self.0.to_ne_bytes()
-    }
-
     #[inline]
     fn to_be_bytes(&self) -> Self::Bytes {
         self.0.to_be_bytes()


Test patch:
diff --git a/integration-testing/src/flight_client_scenarios/integration_test.rs b/integration-testing/src/flight_client_scenarios/integration_test.rs
index 19a4b1a4ef..a731a06eab 100644
--- a/integration-testing/src/flight_client_scenarios/integration_test.rs
+++ b/integration-testing/src/flight_client_scenarios/integration_test.rs
@@ -276,7 +276,7 @@ async fn receive_batch_flight_data(
     {
         let length = data.data_body.len();
         let mut reader = std::io::Cursor::new(&data.data_body);
-        read::read_dictionary(batch, fields, ipc_schema, dictionaries, &mut reader, 0, length as u64)
+        read::read_dictionary(batch, fields, ipc_schema, dictionaries, &mut reader, 0, length as u64, &mut Default::default())
             .expect("Error reading dictionary");
 
         data = resp.next().await?.ok()?;
diff --git a/integration-testing/src/flight_server_scenarios/integration_test.rs b/integration-testing/src/flight_server_scenarios/integration_test.rs
index 07453f5c72..10424759e8 100644
--- a/integration-testing/src/flight_server_scenarios/integration_test.rs
+++ b/integration-testing/src/flight_server_scenarios/integration_test.rs
@@ -296,6 +296,7 @@ async fn record_batch_from_message(
         &mut reader,
         0,
         length as u64,
+        &mut Default::default()
     );
 
     arrow_batch_result.map_err(|e| Status::internal(format!("Could not convert to Chunk: {:?}", e)))
@@ -312,7 +313,7 @@ async fn dictionary_from_message(
     let mut reader = std::io::Cursor::new(data_body);
 
     let dictionary_batch_result =
-        ipc::read::read_dictionary(dict_batch, fields, ipc_schema, dictionaries, &mut reader, 0, length as u64);
+        ipc::read::read_dictionary(dict_batch, fields, ipc_schema, dictionaries, &mut reader, 0, length as u64, &mut Default::default());
     dictionary_batch_result
         .map_err(|e| Status::internal(format!("Could not convert to Dictionary: {:?}", e)))
 }
diff --git a/tests/it/array/fixed_size_list/mod.rs b/tests/it/array/fixed_size_list/mod.rs
index f7a9b61988..8281aee519 100644
--- a/tests/it/array/fixed_size_list/mod.rs
+++ b/tests/it/array/fixed_size_list/mod.rs
@@ -1,1 +1,104 @@
 mod mutable;
+
+use arrow2::{
+    array::*,
+    bitmap::Bitmap,
+    datatypes::{DataType, Field},
+};
+
+fn data() -> FixedSizeListArray {
+    let values = Int32Array::from_slice([10, 20, 0, 0]);
+
+    FixedSizeListArray::try_new(
+        DataType::FixedSizeList(
+            Box::new(Field::new("a", values.data_type().clone(), true)),
+            2,
+        ),
+        values.boxed(),
+        Some([true, false].into()),
+    )
+    .unwrap()
+}
+
+#[test]
+fn basics() {
+    let array = data();
+    assert_eq!(array.size(), 2);
+    assert_eq!(array.len(), 2);
+    assert_eq!(array.validity(), Some(&Bitmap::from([true, false])));
+
+    assert_eq!(array.value(0).as_ref(), Int32Array::from_slice([10, 20]));
+    assert_eq!(array.value(1).as_ref(), Int32Array::from_slice([0, 0]));
+
+    let array = array.slice(1, 1);
+
+    assert_eq!(array.value(0).as_ref(), Int32Array::from_slice([0, 0]));
+}
+
+#[test]
+fn with_validity() {
+    let array = data();
+
+    let a = array.with_validity(None);
+    assert!(a.validity().is_none());
+}
+
+#[test]
+fn debug() {
+    let array = data();
+
+    assert_eq!(format!("{:?}", array), "FixedSizeListArray[[10, 20], None]");
+}
+
+#[test]
+fn empty() {
+    let array = FixedSizeListArray::new_empty(DataType::FixedSizeList(
+        Box::new(Field::new("a", DataType::Int32, true)),
+        2,
+    ));
+    assert_eq!(array.values().len(), 0);
+    assert_eq!(array.validity(), None);
+}
+
+#[test]
+fn null() {
+    let array = FixedSizeListArray::new_null(
+        DataType::FixedSizeList(Box::new(Field::new("a", DataType::Int32, true)), 2),
+        2,
+    );
+    assert_eq!(array.values().len(), 4);
+    assert_eq!(array.validity().cloned(), Some([false, false].into()));
+}
+
+#[test]
+fn wrong_size() {
+    let values = Int32Array::from_slice([10, 20, 0]);
+    assert!(FixedSizeListArray::try_new(
+        DataType::FixedSizeList(Box::new(Field::new("a", DataType::Int32, true)), 2),
+        values.boxed(),
+        None
+    )
+    .is_err());
+}
+
+#[test]
+fn wrong_len() {
+    let values = Int32Array::from_slice([10, 20, 0]);
+    assert!(FixedSizeListArray::try_new(
+        DataType::FixedSizeList(Box::new(Field::new("a", DataType::Int32, true)), 2),
+        values.boxed(),
+        Some([true, false, false].into()), // it should be 2
+    )
+    .is_err());
+}
+
+#[test]
+fn wrong_data_type() {
+    let values = Int32Array::from_slice([10, 20, 0]);
+    assert!(FixedSizeListArray::try_new(
+        DataType::Binary,
+        values.boxed(),
+        Some([true, false, false].into()), // it should be 2
+    )
+    .is_err());
+}
diff --git a/tests/it/array/mod.rs b/tests/it/array/mod.rs
index ef14b993d2..8b400901d7 100644
--- a/tests/it/array/mod.rs
+++ b/tests/it/array/mod.rs
@@ -99,8 +99,8 @@ fn test_with_validity() {
     assert_eq!(arr_ref, &expected);
 }
 
-// check that `PartialEq` can be derived
-#[derive(PartialEq)]
+// check that we ca derive stuff
+#[derive(PartialEq, Clone, Debug)]
 struct A {
     array: Box<dyn Array>,
 }
diff --git a/tests/it/array/primitive/mod.rs b/tests/it/array/primitive/mod.rs
index 058d503f47..5a618cba25 100644
--- a/tests/it/array/primitive/mod.rs
+++ b/tests/it/array/primitive/mod.rs
@@ -72,7 +72,7 @@ fn from() {
 }
 
 #[test]
-fn months_days_ns() {
+fn months_days_ns_from_slice() {
     let data = &[
         months_days_ns::new(1, 1, 2),
         months_days_ns::new(1, 1, 3),
diff --git a/tests/it/array/utf8/mod.rs b/tests/it/array/utf8/mod.rs
index fc55ebccde..c6a0acec03 100644
--- a/tests/it/array/utf8/mod.rs
+++ b/tests/it/array/utf8/mod.rs
@@ -215,3 +215,21 @@ fn into_mut_4() {
     let array = Utf8Array::<i32>::new(DataType::Utf8, offsets, values, validity);
     assert!(array.into_mut().is_right());
 }
+
+#[test]
+fn rev_iter() {
+    let array = Utf8Array::<i32>::from(&[Some("hello"), Some(" "), None]);
+
+    assert_eq!(
+        array.into_iter().rev().collect::<Vec<_>>(),
+        vec![None, Some(" "), Some("hello")]
+    );
+}
+
+#[test]
+fn iter_nth() {
+    let array = Utf8Array::<i32>::from(&[Some("hello"), Some(" "), None]);
+
+    assert_eq!(array.iter().nth(1), Some(Some(" ")));
+    assert_eq!(array.iter().nth(10), None);
+}
diff --git a/tests/it/array/utf8/mutable.rs b/tests/it/array/utf8/mutable.rs
index 8ee76476cb..80cc24ca3c 100644
--- a/tests/it/array/utf8/mutable.rs
+++ b/tests/it/array/utf8/mutable.rs
@@ -141,3 +141,29 @@ fn test_extend_values() {
     assert_eq!(array.offsets().as_slice(), &[0, 2, 7, 12, 17]);
     assert_eq!(array.validity(), None,);
 }
+
+#[test]
+fn test_extend() {
+    let mut array = MutableUtf8Array::<i32>::new();
+
+    array.extend([Some("hi"), None, Some("there"), None].into_iter());
+
+    let array: Utf8Array<i32> = array.into();
+
+    assert_eq!(
+        array,
+        Utf8Array::<i32>::from([Some("hi"), None, Some("there"), None])
+    );
+}
+
+#[test]
+fn as_arc() {
+    let mut array = MutableUtf8Array::<i32>::new();
+
+    array.extend([Some("hi"), None, Some("there"), None].into_iter());
+
+    assert_eq!(
+        Utf8Array::<i32>::from([Some("hi"), None, Some("there"), None]),
+        array.as_arc().as_ref()
+    );
+}
diff --git a/tests/it/compute/arithmetics/time.rs b/tests/it/compute/arithmetics/time.rs
index cbe53381be..9f31dd60a9 100644
--- a/tests/it/compute/arithmetics/time.rs
+++ b/tests/it/compute/arithmetics/time.rs
@@ -2,6 +2,7 @@ use arrow2::array::*;
 use arrow2::compute::arithmetics::time::*;
 use arrow2::datatypes::{DataType, TimeUnit};
 use arrow2::scalar::*;
+use arrow2::types::months_days_ns;
 
 #[test]
 fn test_adding_timestamp() {
@@ -317,3 +318,68 @@ fn test_date64() {
 
     assert_eq!(result, expected);
 }
+
+#[test]
+fn test_add_interval() {
+    let timestamp =
+        PrimitiveArray::from_slice([1i64]).to(DataType::Timestamp(TimeUnit::Second, None));
+
+    let interval = months_days_ns::new(0, 1, 0);
+
+    let intervals = PrimitiveArray::from_slice([interval]);
+
+    let expected = PrimitiveArray::from_slice([1i64 + 24 * 60 * 60])
+        .to(DataType::Timestamp(TimeUnit::Second, None));
+
+    let result = add_interval(&timestamp, &intervals).unwrap();
+    assert_eq!(result, expected);
+
+    let result = add_interval_scalar(&timestamp, &Some(interval).into()).unwrap();
+    assert_eq!(result, expected);
+}
+
+#[test]
+fn test_add_interval_offset() {
+    let timestamp = PrimitiveArray::from_slice([1i64]).to(DataType::Timestamp(
+        TimeUnit::Second,
+        Some("+01:00".to_string()),
+    ));
+
+    let interval = months_days_ns::new(0, 1, 0);
+
+    let intervals = PrimitiveArray::from_slice([interval]);
+
+    let expected = PrimitiveArray::from_slice([1i64 + 24 * 60 * 60]).to(DataType::Timestamp(
+        TimeUnit::Second,
+        Some("+01:00".to_string()),
+    ));
+
+    let result = add_interval(&timestamp, &intervals).unwrap();
+    assert_eq!(result, expected);
+
+    let result = add_interval_scalar(&timestamp, &Some(interval).into()).unwrap();
+    assert_eq!(result, expected);
+}
+
+#[cfg(feature = "chrono-tz")]
+#[test]
+fn test_add_interval_tz() {
+    let timestamp = PrimitiveArray::from_slice([1i64]).to(DataType::Timestamp(
+        TimeUnit::Second,
+        Some("GMT".to_string()),
+    ));
+
+    let interval = months_days_ns::new(0, 1, 0);
+    let intervals = PrimitiveArray::from_slice([interval]);
+
+    let expected = PrimitiveArray::from_slice([1i64 + 24 * 60 * 60]).to(DataType::Timestamp(
+        TimeUnit::Second,
+        Some("GMT".to_string()),
+    ));
+
+    let result = add_interval(&timestamp, &intervals).unwrap();
+    assert_eq!(result, expected);
+
+    let result = add_interval_scalar(&timestamp, &Some(interval).into()).unwrap();
+    assert_eq!(result, expected);
+}
diff --git a/tests/it/io/parquet/mod.rs b/tests/it/io/parquet/mod.rs
index 955df23e66..3c8b7bafd6 100644
--- a/tests/it/io/parquet/mod.rs
+++ b/tests/it/io/parquet/mod.rs
@@ -962,8 +962,8 @@ fn integration_read(data: &[u8]) -> Result<IntegrationRead> {
 /// logical types.
 #[test]
 fn arrow_type() -> Result<()> {
-    let dt1 = DataType::Duration(TimeUnit::Second);
-    let array = PrimitiveArray::<i64>::from([Some(1), None, Some(2)]).to(dt1.clone());
+    let array1 = PrimitiveArray::<i64>::from([Some(1), None, Some(2)])
+        .to(DataType::Duration(TimeUnit::Second));
     let array2 = Utf8Array::<i64>::from([Some("a"), None, Some("bb")]);
 
     let indices = PrimitiveArray::from_values((0..3u64).map(|x| x % 2));
@@ -978,23 +978,61 @@ fn arrow_type() -> Result<()> {
         vec![b'a', b'b', b'a', b'c'].into(),
         None,
     );
-    let array5 = DictionaryArray::from_data(indices, Box::new(values));
+    let array5 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+
+    let values = PrimitiveArray::from_slice([1i16, 3]);
+    let array6 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+
+    let values = PrimitiveArray::from_slice([1i64, 3]).to(DataType::Timestamp(
+        TimeUnit::Millisecond,
+        Some("UTC".to_string()),
+    ));
+    let array7 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+
+    let values = PrimitiveArray::from_slice([1.0f64, 3.0]);
+    let array8 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+
+    let values = PrimitiveArray::from_slice([1u8, 3]);
+    let array9 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+
+    let values = PrimitiveArray::from_slice([1u16, 3]);
+    let array10 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+
+    let values = PrimitiveArray::from_slice([1u32, 3]);
+    let array11 = DictionaryArray::from_data(indices.clone(), Box::new(values));
+
+    let values = PrimitiveArray::from_slice([1u64, 3]);
+    let array12 = DictionaryArray::from_data(indices, Box::new(values));
 
     let schema = Schema::from(vec![
-        Field::new("a1", dt1, true),
+        Field::new("a1", array1.data_type().clone(), true),
         Field::new("a2", array2.data_type().clone(), true),
         Field::new("a3", array3.data_type().clone(), true),
         Field::new("a4", array4.data_type().clone(), true),
         Field::new("a5", array5.data_type().clone(), true),
-        Field::new("a6", array5.data_type().clone(), false),
+        Field::new("a5a", array5.data_type().clone(), false),
+        Field::new("a6", array6.data_type().clone(), true),
+        Field::new("a7", array7.data_type().clone(), true),
+        Field::new("a8", array8.data_type().clone(), true),
+        Field::new("a9", array9.data_type().clone(), true),
+        Field::new("a10", array10.data_type().clone(), true),
+        Field::new("a11", array11.data_type().clone(), true),
+        Field::new("a12", array12.data_type().clone(), true),
     ]);
     let batch = Chunk::try_new(vec![
-        array.boxed(),
+        array1.boxed(),
         array2.boxed(),
         array3.boxed(),
         array4.boxed(),
         array5.clone().boxed(),
         array5.boxed(),
+        array6.boxed(),
+        array7.boxed(),
+        array8.boxed(),
+        array9.boxed(),
+        array10.boxed(),
+        array11.boxed(),
+        array12.boxed(),
     ])?;
 
     let r = integration_write(&schema, &[batch.clone()])?;
diff --git a/tests/it/types.rs b/tests/it/types.rs
index 7dfa07cc16..ee337e901f 100644
--- a/tests/it/types.rs
+++ b/tests/it/types.rs
@@ -1,4 +1,4 @@
-use arrow2::types::{BitChunkIter, BitChunkOnes};
+use arrow2::types::{days_ms, months_days_ns, BitChunkIter, BitChunkOnes, NativeType};
 
 #[test]
 fn test_basic1() {
@@ -18,3 +18,23 @@ fn test_ones() {
     assert_eq!(iter.next(), Some(0));
     assert_eq!(iter.next(), Some(12));
 }
+
+#[test]
+fn months_days_ns_roundtrip() {
+    let a = months_days_ns(1, 2, 3);
+    let bytes = a.to_le_bytes();
+    assert_eq!(bytes, [1, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0]);
+
+    let a = months_days_ns(1, 1, 1);
+    assert_eq!(a, months_days_ns::from_be_bytes(a.to_be_bytes()));
+}
+
+#[test]
+fn days_ms_roundtrip() {
+    let a = days_ms(1, 2);
+    let bytes = a.to_le_bytes();
+    assert_eq!(bytes, [1, 0, 0, 0, 2, 0, 0, 0]);
+
+    let a = days_ms(1, 2);
+    assert_eq!(a, days_ms::from_be_bytes(a.to_be_bytes()));
+}