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

Improved examples

codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1109?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#1109](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1109?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (e171c2c) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/88f05bb1ea37c05688c19d062e10dd35e00c0987?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (88f05bb) will **increase** coverage by `0.00%`.
> The diff coverage is `n/a`.

```diff
@@           Coverage Diff           @@
##             main    #1109   +/-   ##
=======================================
  Coverage   81.24%   81.24%           
=======================================
  Files         366      366           
  Lines       35336    35336           
=======================================
+ Hits        28707    28708    +1     
+ Misses       6629     6628    -1     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1109?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/io/ipc/read/reader.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1109/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL3JlYWRlci5ycw==) | `95.41% <0.00%> (-0.77%)` | :arrow_down: |
| [src/chunk.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1109/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2NodW5rLnJz) | `90.47% <0.00%> (+7.14%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1109?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1109?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Last update [88f05bb...e171c2c](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1109?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).


Patch:
diff --git a/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/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/utils.rs b/src/io/parquet/read/deserialize/utils.rs
index 9e560e4301..dd2174c418 100644
--- a/src/io/parquet/read/deserialize/utils.rs
+++ b/src/io/parquet/read/deserialize/utils.rs
@@ -230,6 +230,7 @@ impl<'a> OptionalPageValidity<'a> {
         }
     }
 
+    /// Number of items remaining
     pub fn len(&self) -> usize {
         self.iter.len()
             + self
@@ -300,10 +301,9 @@ pub(super) fn extend_from_decoder<'a, T: Default, P: Pushable<T>, I: Iterator<It
 ) {
     let limit = limit.unwrap_or(usize::MAX);
 
-    // todo: remove `consumed_here` and compute next limit from `consumed`
-    let mut consumed_here = 0;
-    while consumed_here < limit {
-        let run = page_validity.next_limited(limit);
+    let mut remaining = limit;
+    while remaining > 0 {
+        let run = page_validity.next_limited(remaining);
         let run = if let Some(run) = run { run } else { break };
 
         match run {
@@ -325,7 +325,7 @@ pub(super) fn extend_from_decoder<'a, T: Default, P: Pushable<T>, I: Iterator<It
                 }
                 validity.extend_from_slice(values, offset, length);
 
-                consumed_here += length;
+                remaining -= length;
             }
             FilteredHybridEncoded::Repeated { is_set, length } => {
                 validity.extend_constant(length, is_set);
@@ -335,7 +335,7 @@ pub(super) fn extend_from_decoder<'a, T: Default, P: Pushable<T>, I: Iterator<It
                     pushable.extend_constant(length, T::default());
                 }
 
-                consumed_here += length;
+                remaining -= length;
             }
             FilteredHybridEncoded::Skipped(valids) => for _ in values_iter.by_ref().take(valids) {},
         };
diff --git a/src/io/parquet/read/row_group.rs b/src/io/parquet/read/row_group.rs
index 25c2165819..0c7222343c 100644
--- a/src/io/parquet/read/row_group.rs
+++ b/src/io/parquet/read/row_group.rs
@@ -74,7 +74,7 @@ impl Iterator for RowGroupDeserializer {
                 })
             })
             .collect::<Result<Vec<_>>>()
-            .map(Chunk::new);
+            .and_then(Chunk::try_new);
         self.remaining_rows = self.remaining_rows.saturating_sub(
             chunk
                 .as_ref()
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
+    }
+}


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)))
 }