Repository: jorgecarleitao/arrow2
Pull request #1125 (2022-06-29 14:21:01)

Improved performance of reading from IPC
Reads are no longer zeroed.

Based on the excellent tip by [Kornel](https://users.rust-lang.org/u/kornel/summary) at https://users.rust-lang.org/t/how-to-efficiently-re-use-a-vec-u8-for-multiple-read-read/77628/6 


Patch:
diff --git a/examples/cow.rs b/examples/cow.rs
index 5548f453a1..65e3920727 100644
--- a/examples/cow.rs
+++ b/examples/cow.rs
@@ -19,4 +19,20 @@ fn main() {
 
     // confirm that it gives the right result :)
     assert_eq!(array, &PrimitiveArray::from_vec(vec![10i32, 20]));
+
+    // alternatively, you can use `get_mut_values`. Unwrap works because it is single owned
+    let values = array.get_mut_values().unwrap();
+    values[0] = 0;
+
+    assert_eq!(array, &PrimitiveArray::from_vec(vec![0, 20]));
+
+    // you can also modify the validity:
+    array.set_validity(Some([true, false].into()));
+    array.apply_validity(|bitmap| {
+        let mut mut_bitmap = bitmap.into_mut().right().unwrap();
+        mut_bitmap.set(1, true);
+        mut_bitmap.into()
+    });
+
+    assert_eq!(array, &PrimitiveArray::from_vec(vec![0, 20]));
 }
diff --git a/examples/ipc_file_read.rs b/examples/ipc_file_read.rs
index 6478b25d48..18ceed0ba8 100644
--- a/examples/ipc_file_read.rs
+++ b/examples/ipc_file_read.rs
@@ -8,6 +8,7 @@ use arrow2::io::ipc::read;
 use arrow2::io::print;
 
 /// Simplest way: read all record batches from the file. This can be used e.g. for random access.
+#[allow(clippy::type_complexity)]
 fn read_batches(path: &str) -> Result<(Schema, Vec<Chunk<Box<dyn Array>>>)> {
     let mut file = File::open(path)?;
 
diff --git a/examples/parquet_read_parallel/Cargo.toml b/examples/parquet_read_parallel/Cargo.toml
index 46f10ff204..f2af17c352 100644
--- a/examples/parquet_read_parallel/Cargo.toml
+++ b/examples/parquet_read_parallel/Cargo.toml
@@ -1,5 +1,5 @@
 [package]
-name = "parquet_write_parallel"
+name = "parquet_read_parallel"
 version = "0.1.0"
 edition = "2021"
 
diff --git a/examples/parquet_write_parallel/src/main.rs b/examples/parquet_write_parallel/src/main.rs
index 4973d0e7a7..3d6a0fe0d4 100644
--- a/examples/parquet_write_parallel/src/main.rs
+++ b/examples/parquet_write_parallel/src/main.rs
@@ -71,7 +71,7 @@ fn parallel_write(path: &str, schema: Schema, chunks: &[Chunk]) -> Result<()> {
             .columns()
             .par_iter()
             .zip(parquet_schema.fields().to_vec())
-            .zip(encodings.clone())
+            .zip(encodings.par_iter())
             .flat_map(move |((array, type_), encoding)| {
                 let encoded_columns = array_to_columns(array, type_, options, encoding).unwrap();
                 encoded_columns
@@ -114,7 +114,7 @@ fn parallel_write(path: &str, schema: Schema, chunks: &[Chunk]) -> Result<()> {
     Ok(())
 }
 
-fn create_batch(size: usize) -> Result<Chunk> {
+fn create_chunk(size: usize) -> Result<Chunk> {
     let c1: Int32Array = (0..size)
         .map(|x| if x % 9 == 0 { None } else { Some(x as i32) })
         .collect();
@@ -143,10 +143,10 @@ fn main() -> Result<()> {
         Field::new("c3", DataType::Int32, true),
         Field::new("c4", DataType::LargeUtf8, true),
     ];
-    let batch = create_batch(100_000_000)?;
+    let chunk = create_chunk(100_000_000)?;
 
     let start = std::time::SystemTime::now();
-    parallel_write("example.parquet", fields.into(), &[batch])?;
+    parallel_write("example.parquet", fields.into(), &[chunk])?;
     println!("took: {} ms", start.elapsed().unwrap().as_millis());
     Ok(())
 }
diff --git a/src/bitmap/assign_ops.rs b/src/bitmap/assign_ops.rs
index 81ded04578..fe8f280598 100644
--- a/src/bitmap/assign_ops.rs
+++ b/src/bitmap/assign_ops.rs
@@ -118,14 +118,14 @@ fn or_assign<T: BitChunk>(lhs: &mut MutableBitmap, rhs: &Bitmap) {
     }
 }
 
-impl<'a, 'b> std::ops::BitOrAssign<&'a Bitmap> for &'b mut MutableBitmap {
+impl<'a> std::ops::BitOrAssign<&'a Bitmap> for &mut MutableBitmap {
     #[inline]
     fn bitor_assign(&mut self, rhs: &'a Bitmap) {
         or_assign::<u64>(self, rhs)
     }
 }
 
-impl<'a, 'b> std::ops::BitOr<&'a Bitmap> for MutableBitmap {
+impl<'a> std::ops::BitOr<&'a Bitmap> for MutableBitmap {
     type Output = Self;
 
     #[inline]
@@ -150,14 +150,14 @@ fn and_assign<T: BitChunk>(lhs: &mut MutableBitmap, rhs: &Bitmap) {
     }
 }
 
-impl<'a, 'b> std::ops::BitAndAssign<&'a Bitmap> for &'b mut MutableBitmap {
+impl<'a> std::ops::BitAndAssign<&'a Bitmap> for &mut MutableBitmap {
     #[inline]
     fn bitand_assign(&mut self, rhs: &'a Bitmap) {
         and_assign::<u64>(self, rhs)
     }
 }
 
-impl<'a, 'b> std::ops::BitAnd<&'a Bitmap> for MutableBitmap {
+impl<'a> std::ops::BitAnd<&'a Bitmap> for MutableBitmap {
     type Output = Self;
 
     #[inline]
@@ -173,14 +173,14 @@ fn xor_assign<T: BitChunk>(lhs: &mut MutableBitmap, rhs: &Bitmap) {
     binary_assign(lhs, rhs, |x: T, y| x ^ y)
 }
 
-impl<'a, 'b> std::ops::BitXorAssign<&'a Bitmap> for &'b mut MutableBitmap {
+impl<'a> std::ops::BitXorAssign<&'a Bitmap> for &mut MutableBitmap {
     #[inline]
     fn bitxor_assign(&mut self, rhs: &'a Bitmap) {
         xor_assign::<u64>(self, rhs)
     }
 }
 
-impl<'a, 'b> std::ops::BitXor<&'a Bitmap> for MutableBitmap {
+impl<'a> std::ops::BitXor<&'a Bitmap> for MutableBitmap {
     type Output = Self;
 
     #[inline]
diff --git a/src/chunk.rs b/src/chunk.rs
index 499f0f27fa..ffc857bcc1 100644
--- a/src/chunk.rs
+++ b/src/chunk.rs
@@ -6,7 +6,7 @@ use crate::error::{Error, Result};
 
 /// A vector of trait objects of [`Array`] where every item has
 /// the same length, [`Chunk::len`].
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Chunk<A: AsRef<dyn Array>> {
     arrays: Vec<A>,
 }
diff --git a/src/error.rs b/src/error.rs
index 3bf4d8b650..52a8cd9062 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -62,6 +62,12 @@ impl From<simdutf8::basic::Utf8Error> for Error {
     }
 }
 
+impl From<std::collections::TryReserveError> for Error {
+    fn from(_: std::collections::TryReserveError) -> Error {
+        Error::Overflow
+    }
+}
+
 impl Display for Error {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match self {
diff --git a/src/io/avro/mod.rs b/src/io/avro/mod.rs
index d502e3aeb8..852915d900 100644
--- a/src/io/avro/mod.rs
+++ b/src/io/avro/mod.rs
@@ -93,7 +93,7 @@ macro_rules! read_metadata {
 pub(crate) use {avro_decode, read_header, read_metadata};
 
 /// A compressed Avro block.
-#[derive(Debug, Clone, Default, PartialEq)]
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct CompressedBlock {
     /// The number of rows
     pub number_of_rows: usize,
@@ -112,7 +112,7 @@ impl CompressedBlock {
 }
 
 /// An uncompressed Avro block.
-#[derive(Debug, Clone, Default, PartialEq)]
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct Block {
     /// The number of rows
     pub number_of_rows: usize,
diff --git a/src/io/avro/read/block.rs b/src/io/avro/read/block.rs
index c2e74ec883..62cc5a88ef 100644
--- a/src/io/avro/read/block.rs
+++ b/src/io/avro/read/block.rs
@@ -40,8 +40,8 @@ fn read_block<R: Read>(
     };
 
     block.data.clear();
-    block.data.resize(bytes, 0);
-    reader.read_exact(&mut block.data)?;
+    block.data.try_reserve(bytes)?;
+    reader.take(bytes as u64).read_to_end(&mut block.data)?;
 
     let mut marker = [0u8; 16];
     reader.read_exact(&mut marker)?;
diff --git a/src/io/avro/read/decompress.rs b/src/io/avro/read/decompress.rs
index 05f5fd43d9..bc5fcbd582 100644
--- a/src/io/avro/read/decompress.rs
+++ b/src/io/avro/read/decompress.rs
@@ -96,7 +96,7 @@ impl<R: Read> Decompressor<R> {
     }
 }
 
-impl<'a, R: Read> FallibleStreamingIterator for Decompressor<R> {
+impl<R: Read> FallibleStreamingIterator for Decompressor<R> {
     type Error = Error;
     type Item = Block;
 
diff --git a/src/io/avro/read/util.rs b/src/io/avro/read/util.rs
index 7bda9d84fa..92cbe3db2b 100644
--- a/src/io/avro/read/util.rs
+++ b/src/io/avro/read/util.rs
@@ -23,8 +23,9 @@ fn decode_variable<R: Read>(reader: &mut R) -> Result<u64> {
 
 fn _read_binary<R: Read>(reader: &mut R) -> Result<Vec<u8>> {
     let len: usize = zigzag_i64(reader)? as usize;
-    let mut buf = vec![0u8; len];
-    reader.read_exact(&mut buf)?;
+    let mut buf = vec![];
+    buf.try_reserve(len)?;
+    reader.take(len as u64).read_to_end(&mut buf)?;
     Ok(buf)
 }
 
diff --git a/src/io/avro/read_async/block.rs b/src/io/avro/read_async/block.rs
index 89069cad83..ff92d6a97e 100644
--- a/src/io/avro/read_async/block.rs
+++ b/src/io/avro/read_async/block.rs
@@ -42,8 +42,11 @@ async fn read_block<R: AsyncRead + Unpin + Send>(
     };
 
     block.data.clear();
-    block.data.resize(bytes, 0);
-    reader.read_exact(&mut block.data).await?;
+    block.data.try_reserve(bytes)?;
+    reader
+        .take(bytes as u64)
+        .read_to_end(&mut block.data)
+        .await?;
 
     let mut marker = [0u8; 16];
     reader.read_exact(&mut marker).await?;
diff --git a/src/io/avro/read_async/metadata.rs b/src/io/avro/read_async/metadata.rs
index 7d8df6888f..4609509376 100644
--- a/src/io/avro/read_async/metadata.rs
+++ b/src/io/avro/read_async/metadata.rs
@@ -48,8 +48,9 @@ async fn read_file_marker<R: AsyncRead + Unpin + Send>(reader: &mut R) -> Result
 
 async fn _read_binary<R: AsyncRead + Unpin + Send>(reader: &mut R) -> Result<Vec<u8>> {
     let len: usize = zigzag_i64(reader).await? as usize;
-    let mut buf = vec![0u8; len];
-    reader.read_exact(&mut buf).await?;
+    let mut buf = vec![];
+    buf.try_reserve(len)?;
+    reader.take(len as u64).read_to_end(&mut buf).await?;
     Ok(buf)
 }
 
diff --git a/src/io/csv/write/mod.rs b/src/io/csv/write/mod.rs
index 1f124ef676..62f228ef14 100644
--- a/src/io/csv/write/mod.rs
+++ b/src/io/csv/write/mod.rs
@@ -27,16 +27,16 @@ fn new_serializers<'a, A: AsRef<dyn Array>>(
 /// The vector is guaranteed to have `columns.len()` entries.
 /// Each `row` is guaranteed to have `columns.array().len()` fields.
 pub fn serialize<A: AsRef<dyn Array>>(
-    columns: &Chunk<A>,
+    chunk: &Chunk<A>,
     options: &SerializeOptions,
 ) -> Result<Vec<Vec<u8>>> {
-    let mut serializers = new_serializers(columns, options)?;
+    let mut serializers = new_serializers(chunk, options)?;
 
-    let mut rows = Vec::with_capacity(columns.len());
+    let mut rows = Vec::with_capacity(chunk.len());
     let mut row = vec![];
 
     // this is where the (expensive) transposition happens: the outer loop is on rows, the inner on columns
-    (0..columns.len()).try_for_each(|_| {
+    (0..chunk.len()).try_for_each(|_| {
         serializers
             .iter_mut()
             // `unwrap` is infalible because `array.len()` equals `Chunk::len`
@@ -49,8 +49,7 @@ pub fn serialize<A: AsRef<dyn Array>>(
             // replace last delimiter with new line
             let last_byte = row.len() - 1;
             row[last_byte] = b'\n';
-            rows.push(row.clone());
-            row.clear();
+            rows.push(std::mem::take(&mut row));
         }
         Result::Ok(())
     })?;
diff --git a/src/io/ipc/read/array/binary.rs b/src/io/ipc/read/array/binary.rs
index 51d3d7d26b..917f46b401 100644
--- a/src/io/ipc/read/array/binary.rs
+++ b/src/io/ipc/read/array/binary.rs
@@ -7,7 +7,7 @@ use crate::datatypes::DataType;
 use crate::error::{Error, Result};
 
 use super::super::read_basic::*;
-use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind, ReadBuffer};
+use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_binary<O: Offset, R: Read + Seek>(
@@ -18,7 +18,7 @@ pub fn read_binary<O: Offset, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<BinaryArray<O>> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
diff --git a/src/io/ipc/read/array/dictionary.rs b/src/io/ipc/read/array/dictionary.rs
index 0b212c4501..63a0cb0b80 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, ReadBuffer};
+use super::super::{Compression, IpcBuffer, Node};
 use super::{read_primitive, skip_primitive};
 
 #[allow(clippy::too_many_arguments)]
@@ -19,7 +19,7 @@ pub fn read_dictionary<T: DictionaryKey, R: Read + Seek>(
     block_offset: u64,
     compression: Option<Compression>,
     is_little_endian: bool,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<DictionaryArray<T>>
 where
     Vec<u8>: TryInto<T::Bytes>,
diff --git a/src/io/ipc/read/array/fixed_size_binary.rs b/src/io/ipc/read/array/fixed_size_binary.rs
index 5dc1b2ab0b..e4cb0480de 100644
--- a/src/io/ipc/read/array/fixed_size_binary.rs
+++ b/src/io/ipc/read/array/fixed_size_binary.rs
@@ -6,7 +6,7 @@ use crate::datatypes::DataType;
 use crate::error::{Error, Result};
 
 use super::super::read_basic::*;
-use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind, ReadBuffer};
+use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_fixed_size_binary<R: Read + Seek>(
@@ -17,7 +17,7 @@ pub fn read_fixed_size_binary<R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<FixedSizeBinaryArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
diff --git a/src/io/ipc/read/array/fixed_size_list.rs b/src/io/ipc/read/array/fixed_size_list.rs
index ac169ed3f6..d8eb9725ed 100644
--- a/src/io/ipc/read/array/fixed_size_list.rs
+++ b/src/io/ipc/read/array/fixed_size_list.rs
@@ -8,7 +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::{Compression, Dictionaries, IpcBuffer, Node, ReadBuffer, Version};
+use super::super::{Compression, Dictionaries, IpcBuffer, Node, Version};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_fixed_size_list<R: Read + Seek>(
@@ -22,7 +22,7 @@ pub fn read_fixed_size_list<R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<Compression>,
     version: Version,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<FixedSizeListArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
diff --git a/src/io/ipc/read/array/list.rs b/src/io/ipc/read/array/list.rs
index bb85f233cf..44fb17bd2a 100644
--- a/src/io/ipc/read/array/list.rs
+++ b/src/io/ipc/read/array/list.rs
@@ -10,9 +10,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::{
-    Compression, Dictionaries, IpcBuffer, Node, OutOfSpecKind, ReadBuffer, Version,
-};
+use super::super::{Compression, Dictionaries, IpcBuffer, Node, OutOfSpecKind, Version};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_list<O: Offset, R: Read + Seek>(
@@ -26,7 +24,7 @@ pub fn read_list<O: Offset, R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<Compression>,
     version: Version,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<ListArray<O>>
 where
     Vec<u8>: TryInto<O::Bytes>,
diff --git a/src/io/ipc/read/array/map.rs b/src/io/ipc/read/array/map.rs
index 40b4f5c038..5020a157a0 100644
--- a/src/io/ipc/read/array/map.rs
+++ b/src/io/ipc/read/array/map.rs
@@ -9,9 +9,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::{
-    Compression, Dictionaries, IpcBuffer, Node, OutOfSpecKind, ReadBuffer, Version,
-};
+use super::super::{Compression, Dictionaries, IpcBuffer, Node, OutOfSpecKind, Version};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_map<R: Read + Seek>(
@@ -25,7 +23,7 @@ pub fn read_map<R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<Compression>,
     version: Version,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<MapArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
diff --git a/src/io/ipc/read/array/primitive.rs b/src/io/ipc/read/array/primitive.rs
index 1e4000123b..75a62e5a83 100644
--- a/src/io/ipc/read/array/primitive.rs
+++ b/src/io/ipc/read/array/primitive.rs
@@ -6,7 +6,7 @@ use crate::error::{Error, Result};
 use crate::{array::PrimitiveArray, types::NativeType};
 
 use super::super::read_basic::*;
-use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind, ReadBuffer};
+use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_primitive<T: NativeType, R: Read + Seek>(
@@ -17,7 +17,7 @@ pub fn read_primitive<T: NativeType, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<PrimitiveArray<T>>
 where
     Vec<u8>: TryInto<T::Bytes>,
diff --git a/src/io/ipc/read/array/struct_.rs b/src/io/ipc/read/array/struct_.rs
index ae23c0f882..469689d247 100644
--- a/src/io/ipc/read/array/struct_.rs
+++ b/src/io/ipc/read/array/struct_.rs
@@ -8,7 +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::{Compression, Dictionaries, IpcBuffer, Node, ReadBuffer, Version};
+use super::super::{Compression, Dictionaries, IpcBuffer, Node, Version};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_struct<R: Read + Seek>(
@@ -22,7 +22,7 @@ pub fn read_struct<R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<Compression>,
     version: Version,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<StructArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
diff --git a/src/io/ipc/read/array/union.rs b/src/io/ipc/read/array/union.rs
index fc907a3239..678d65ded9 100644
--- a/src/io/ipc/read/array/union.rs
+++ b/src/io/ipc/read/array/union.rs
@@ -9,9 +9,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::{
-    Compression, Dictionaries, IpcBuffer, Node, OutOfSpecKind, ReadBuffer, Version,
-};
+use super::super::{Compression, Dictionaries, IpcBuffer, Node, OutOfSpecKind, Version};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_union<R: Read + Seek>(
@@ -25,7 +23,7 @@ pub fn read_union<R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<Compression>,
     version: Version,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<UnionArray> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
diff --git a/src/io/ipc/read/array/utf8.rs b/src/io/ipc/read/array/utf8.rs
index 10f1260f5a..af25eaed89 100644
--- a/src/io/ipc/read/array/utf8.rs
+++ b/src/io/ipc/read/array/utf8.rs
@@ -7,7 +7,7 @@ use crate::datatypes::DataType;
 use crate::error::{Error, Result};
 
 use super::super::read_basic::*;
-use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind, ReadBuffer};
+use super::super::{Compression, IpcBuffer, Node, OutOfSpecKind};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read_utf8<O: Offset, R: Read + Seek>(
@@ -18,7 +18,7 @@ pub fn read_utf8<O: Offset, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<Utf8Array<O>> {
     let field_node = field_nodes.pop_front().ok_or_else(|| {
         Error::oos(format!(
diff --git a/src/io/ipc/read/common.rs b/src/io/ipc/read/common.rs
index 18ddd8cbb8..26252dd183 100644
--- a/src/io/ipc/read/common.rs
+++ b/src/io/ipc/read/common.rs
@@ -9,7 +9,6 @@ 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;
@@ -86,7 +85,7 @@ pub fn read_record_batch<R: Read + Seek>(
     reader: &mut R,
     block_offset: u64,
     file_size: u64,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<Chunk<Box<dyn Array>>> {
     assert_eq!(fields.len(), ipc_schema.fields.len());
     let buffers = batch
@@ -234,7 +233,7 @@ pub fn read_dictionary<R: Read + Seek>(
     reader: &mut R,
     block_offset: u64,
     file_size: u64,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<()> {
     if batch
         .is_delta()
diff --git a/src/io/ipc/read/deserialize.rs b/src/io/ipc/read/deserialize.rs
index 3378f99a89..3ed5c9349c 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, ReadBuffer};
+use super::{IpcBuffer, Node};
 
 #[allow(clippy::too_many_arguments)]
 pub fn read<R: Read + Seek>(
@@ -24,7 +24,7 @@ pub fn read<R: Read + Seek>(
     is_little_endian: bool,
     compression: Option<BodyCompressionRef>,
     version: MetadataVersion,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<Box<dyn Array>> {
     use PhysicalType::*;
     let data_type = field.data_type.clone();
diff --git a/src/io/ipc/read/file_async.rs b/src/io/ipc/read/file_async.rs
index cabfdd43b7..100990a650 100644
--- a/src/io/ipc/read/file_async.rs
+++ b/src/io/ipc/read/file_async.rs
@@ -18,7 +18,6 @@ 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> {
@@ -142,9 +141,14 @@ where
 {
     let footer_size = read_footer_len(reader).await?;
     // Read footer
-    let mut footer = vec![0; footer_size];
     reader.seek(SeekFrom::End(-10 - footer_size as i64)).await?;
-    reader.read_exact(&mut footer).await?;
+
+    let mut footer = vec![];
+    footer.try_reserve(footer_size)?;
+    reader
+        .take(footer_size as u64)
+        .read_to_end(&mut footer)
+        .await?;
 
     deserialize_footer(&footer, u64::MAX)
 }
@@ -156,9 +160,9 @@ async fn read_batch<R>(
     metadata: &FileMetadata,
     projection: Option<&[usize]>,
     block: usize,
-    meta_buffer: &mut ReadBuffer,
-    block_buffer: &mut ReadBuffer,
-    scratch: &mut ReadBuffer,
+    meta_buffer: &mut Vec<u8>,
+    block_buffer: &mut Vec<u8>,
+    scratch: &mut Vec<u8>,
 ) -> Result<Chunk<Box<dyn Array>>>
 where
     R: AsyncRead + AsyncSeek + Unpin,
@@ -181,10 +185,14 @@ where
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
 
-    meta_buffer.set_len(meta_len);
-    reader.read_exact(meta_buffer.as_mut()).await?;
+    meta_buffer.clear();
+    meta_buffer.try_reserve(meta_len)?;
+    (&mut reader)
+        .take(meta_len as u64)
+        .read_to_end(meta_buffer)
+        .await?;
 
-    let message = arrow_format::ipc::MessageRef::read_as_root(meta_buffer.as_ref())
+    let message = arrow_format::ipc::MessageRef::read_as_root(meta_buffer)
         .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
 
     let batch = get_serialized_batch(&message)?;
@@ -195,9 +203,14 @@ where
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
 
-    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());
+    block_buffer.clear();
+    block_buffer.try_reserve(block_length)?;
+    reader
+        .take(block_length as u64)
+        .read_to_end(block_buffer)
+        .await?;
+
+    let mut cursor = std::io::Cursor::new(&block_buffer);
 
     read_record_batch(
         batch,
@@ -220,14 +233,14 @@ async fn read_dictionaries<R>(
     fields: &[Field],
     ipc_schema: &IpcSchema,
     blocks: &[Block],
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<Dictionaries>
 where
     R: AsyncRead + AsyncSeek + Unpin,
 {
     let mut dictionaries = Default::default();
-    let mut data: ReadBuffer = vec![].into();
-    let mut buffer: ReadBuffer = vec![].into();
+    let mut data: Vec<u8> = vec![];
+    let mut buffer: Vec<u8> = vec![];
 
     for block in blocks {
         let offset: u64 = block
@@ -250,11 +263,15 @@ 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) => {
-                reader.read_exact(buffer.as_mut()).await?;
-                let mut cursor = std::io::Cursor::new(buffer.as_ref());
+                buffer.clear();
+                buffer.try_reserve(length)?;
+                (&mut reader)
+                    .take(length as u64)
+                    .read_to_end(&mut buffer)
+                    .await?;
+                let mut cursor = std::io::Cursor::new(&buffer);
                 read_dictionary(
                     batch,
                     fields,
@@ -272,7 +289,7 @@ where
     Ok(dictionaries)
 }
 
-async fn read_dictionary_message<R>(mut reader: R, offset: u64, data: &mut ReadBuffer) -> Result<()>
+async fn read_dictionary_message<R>(mut reader: R, offset: u64, data: &mut Vec<u8>) -> Result<()>
 where
     R: AsyncRead + AsyncSeek + Unpin,
 {
@@ -288,8 +305,12 @@ where
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
 
-    data.set_len(footer_size);
-    reader.read_exact(data.as_mut()).await?;
+    data.clear();
+    data.try_reserve(footer_size)?;
+    (&mut reader)
+        .take(footer_size as u64)
+        .read_to_end(data)
+        .await?;
 
     Ok(())
 }
diff --git a/src/io/ipc/read/mod.rs b/src/io/ipc/read/mod.rs
index f25da7a4a1..5ffe6426e2 100644
--- a/src/io/ipc/read/mod.rs
+++ b/src/io/ipc/read/mod.rs
@@ -27,8 +27,6 @@ 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 cc04035eec..7a5ae859a6 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, ReadBuffer};
+use super::{Compression, IpcBuffer, Node, OutOfSpecKind};
 
 fn read_swapped<T: NativeType, R: Read + Seek>(
     reader: &mut R,
@@ -93,7 +93,7 @@ fn read_compressed_buffer<T: NativeType, R: Read + Seek>(
     length: usize,
     is_little_endian: bool,
     compression: Compression,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<Vec<T>> {
     if is_little_endian != is_native_little_endian() {
         return Err(Error::NotYetImplemented(
@@ -106,8 +106,12 @@ fn read_compressed_buffer<T: NativeType, R: Read + Seek>(
     let mut buffer = vec![T::default(); length];
 
     // decompress first
-    scratch.set_len(buffer_length);
-    reader.read_exact(scratch.as_mut())?;
+    scratch.clear();
+    scratch.try_reserve(buffer_length)?;
+    reader
+        .by_ref()
+        .take(buffer_length as u64)
+        .read_to_end(scratch)?;
 
     let out_slice = bytemuck::cast_slice_mut(&mut buffer);
 
@@ -117,10 +121,10 @@ fn read_compressed_buffer<T: NativeType, R: Read + Seek>(
 
     match compression {
         arrow_format::ipc::CompressionType::Lz4Frame => {
-            compression::decompress_lz4(&scratch.as_ref()[8..], out_slice)?;
+            compression::decompress_lz4(&scratch[8..], out_slice)?;
         }
         arrow_format::ipc::CompressionType::Zstd => {
-            compression::decompress_zstd(&scratch.as_ref()[8..], out_slice)?;
+            compression::decompress_zstd(&scratch[8..], out_slice)?;
         }
     }
     Ok(buffer)
@@ -133,7 +137,7 @@ pub fn read_buffer<T: NativeType, R: Read + Seek>(
     block_offset: u64,
     is_little_endian: bool,
     compression: Option<Compression>,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<Buffer<T>> {
     let buf = buf
         .pop_front()
diff --git a/src/io/ipc/read/reader.rs b/src/io/ipc/read/reader.rs
index b01d85cf49..776ee684e8 100644
--- a/src/io/ipc/read/reader.rs
+++ b/src/io/ipc/read/reader.rs
@@ -13,7 +13,6 @@ 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.
@@ -45,14 +44,14 @@ pub struct FileReader<R: Read + Seek> {
     dictionaries: Option<Dictionaries>,
     current_block: usize,
     projection: Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
-    data_scratch: ReadBuffer,
-    message_scratch: ReadBuffer,
+    data_scratch: Vec<u8>,
+    message_scratch: Vec<u8>,
 }
 
 fn read_dictionary_message<R: Read + Seek>(
     reader: &mut R,
     offset: u64,
-    data: &mut ReadBuffer,
+    data: &mut Vec<u8>,
 ) -> Result<()> {
     let mut message_size: [u8; 4] = [0; 4];
     reader.seek(SeekFrom::Start(offset))?;
@@ -66,8 +65,13 @@ fn read_dictionary_message<R: Read + Seek>(
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
 
-    data.set_len(message_length);
-    reader.read_exact(data.as_mut())?;
+    data.clear();
+    data.try_reserve(message_length)?;
+    reader
+        .by_ref()
+        .take(message_length as u64)
+        .read_to_end(data)?;
+
     Ok(())
 }
 
@@ -76,8 +80,8 @@ fn read_dictionary_block<R: Read + Seek>(
     metadata: &FileMetadata,
     block: &arrow_format::ipc::Block,
     dictionaries: &mut Dictionaries,
-    message_scratch: &mut ReadBuffer,
-    dictionary_scratch: &mut ReadBuffer,
+    message_scratch: &mut Vec<u8>,
+    dictionary_scratch: &mut Vec<u8>,
 ) -> Result<()> {
     let offset: u64 = block
         .offset
@@ -120,7 +124,7 @@ fn read_dictionary_block<R: Read + Seek>(
 pub fn read_file_dictionaries<R: Read + Seek>(
     reader: &mut R,
     metadata: &FileMetadata,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<Dictionaries> {
     let mut dictionaries = Default::default();
 
@@ -259,8 +263,8 @@ pub fn read_batch<R: Read + Seek>(
     metadata: &FileMetadata,
     projection: Option<&[usize]>,
     index: usize,
-    message_scratch: &mut ReadBuffer,
-    data_scratch: &mut ReadBuffer,
+    message_scratch: &mut Vec<u8>,
+    data_scratch: &mut Vec<u8>,
 ) -> Result<Chunk<Box<dyn Array>>> {
     let block = metadata.blocks[index];
 
@@ -281,8 +285,12 @@ pub fn read_batch<R: Read + Seek>(
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::UnexpectedNegativeInteger))?;
 
-    message_scratch.set_len(meta_len);
-    reader.read_exact(message_scratch.as_mut())?;
+    message_scratch.clear();
+    message_scratch.try_reserve(meta_len)?;
+    reader
+        .by_ref()
+        .take(meta_len as u64)
+        .read_to_end(message_scratch)?;
 
     let message = arrow_format::ipc::MessageRef::read_as_root(message_scratch.as_ref())
         .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
diff --git a/src/io/ipc/read/stream.rs b/src/io/ipc/read/stream.rs
index 5dd7b5cddd..e7d3d590aa 100644
--- a/src/io/ipc/read/stream.rs
+++ b/src/io/ipc/read/stream.rs
@@ -15,7 +15,6 @@ 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)]
@@ -44,14 +43,18 @@ pub fn read_stream_metadata<R: Read>(reader: &mut R) -> Result<StreamMetadata> {
         i32::from_le_bytes(meta_size)
     };
 
-    let meta_length: usize = meta_length
+    let length: usize = meta_length
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
 
-    let mut meta_buffer = vec![0; meta_length];
-    reader.read_exact(&mut meta_buffer)?;
+    let mut buffer = vec![];
+    buffer.try_reserve(length as usize)?;
+    reader
+        .by_ref()
+        .take(length as u64)
+        .read_to_end(&mut buffer)?;
 
-    deserialize_stream_metadata(&meta_buffer)
+    deserialize_stream_metadata(&buffer)
 }
 
 /// Encodes the stream's status after each read.
@@ -91,10 +94,10 @@ fn read_next<R: Read>(
     reader: &mut R,
     metadata: &StreamMetadata,
     dictionaries: &mut Dictionaries,
-    message_buffer: &mut ReadBuffer,
-    data_buffer: &mut ReadBuffer,
+    message_buffer: &mut Vec<u8>,
+    data_buffer: &mut Vec<u8>,
     projection: &Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
-    scratch: &mut ReadBuffer,
+    scratch: &mut Vec<u8>,
 ) -> Result<Option<StreamState>> {
     // determine metadata length
     let mut meta_length: [u8; 4] = [0; 4];
@@ -131,8 +134,12 @@ fn read_next<R: Read>(
         return Ok(None);
     }
 
-    message_buffer.set_len(meta_length);
-    reader.read_exact(message_buffer.as_mut())?;
+    message_buffer.clear();
+    message_buffer.try_reserve(meta_length as usize)?;
+    reader
+        .by_ref()
+        .take(meta_length as u64)
+        .read_to_end(message_buffer)?;
 
     let message = arrow_format::ipc::MessageRef::read_as_root(message_buffer.as_ref())
         .map_err(|err| Error::from(OutOfSpecKind::InvalidFlatbufferMessage(err)))?;
@@ -148,12 +155,16 @@ 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) => {
-            reader.read_exact(data_buffer.as_mut())?;
+            data_buffer.clear();
+            data_buffer.try_reserve(block_length as usize)?;
+            reader
+                .by_ref()
+                .take(block_length as u64)
+                .read_to_end(data_buffer)?;
 
-            let file_size = data_buffer.as_ref().len() as u64;
+            let file_size = data_buffer.len() as u64;
 
             let mut reader = std::io::Cursor::new(data_buffer);
 
@@ -180,10 +191,15 @@ fn read_next<R: Read>(
             }
         }
         arrow_format::ipc::MessageHeaderRef::DictionaryBatch(batch) => {
-            let mut buf = vec![0; block_length];
-            reader.read_exact(&mut buf)?;
+            data_buffer.clear();
+            data_buffer.try_reserve(block_length as usize)?;
+            reader
+                .by_ref()
+                .take(block_length as u64)
+                .read_to_end(data_buffer)?;
 
-            let mut dict_reader = std::io::Cursor::new(&buf);
+            let file_size = data_buffer.len() as u64;
+            let mut dict_reader = std::io::Cursor::new(&data_buffer);
 
             read_dictionary(
                 batch,
@@ -192,7 +208,7 @@ fn read_next<R: Read>(
                 dictionaries,
                 &mut dict_reader,
                 0,
-                buf.len() as u64,
+                file_size,
                 scratch,
             )?;
 
@@ -222,10 +238,10 @@ pub struct StreamReader<R: Read> {
     metadata: StreamMetadata,
     dictionaries: Dictionaries,
     finished: bool,
-    data_buffer: ReadBuffer,
-    message_buffer: ReadBuffer,
+    data_buffer: Vec<u8>,
+    message_buffer: Vec<u8>,
     projection: Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
-    scratch: ReadBuffer,
+    scratch: Vec<u8>,
 }
 
 impl<R: Read> StreamReader<R> {
diff --git a/src/io/ipc/read/stream_async.rs b/src/io/ipc/read/stream_async.rs
index cfbfe5ad3e..b637dc0a2d 100644
--- a/src/io/ipc/read/stream_async.rs
+++ b/src/io/ipc/read/stream_async.rs
@@ -16,7 +16,6 @@ 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
@@ -25,9 +24,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: ReadBuffer,
+    pub data_buffer: Vec<u8>,
     /// The internal buffer to read messages to
-    pub message_buffer: ReadBuffer,
+    pub message_buffer: Vec<u8>,
 }
 
 /// The state of an Arrow stream
@@ -58,8 +57,12 @@ pub async fn read_stream_metadata_async<R: AsyncRead + Unpin + Send>(
         .try_into()
         .map_err(|_| Error::from(OutOfSpecKind::NegativeFooterLength))?;
 
-    let mut meta_buffer = vec![0; meta_len];
-    reader.read_exact(&mut meta_buffer).await?;
+    let mut meta_buffer = vec![];
+    meta_buffer.try_reserve(meta_len as usize)?;
+    reader
+        .take(meta_len as u64)
+        .read_to_end(&mut meta_buffer)
+        .await?;
 
     deserialize_stream_metadata(&meta_buffer)
 }
@@ -105,10 +108,11 @@ async fn maybe_next<R: AsyncRead + Unpin + Send>(
         return Ok(None);
     }
 
-    state.message_buffer.set_len(meta_length);
-    state
-        .reader
-        .read_exact(state.message_buffer.as_mut())
+    state.message_buffer.clear();
+    state.message_buffer.try_reserve(meta_length as usize)?;
+    (&mut state.reader)
+        .take(meta_length as u64)
+        .read_to_end(&mut state.message_buffer)
         .await?;
 
     let message = arrow_format::ipc::MessageRef::read_as_root(state.message_buffer.as_ref())
@@ -125,11 +129,14 @@ 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.reader.read_exact(state.data_buffer.as_mut()).await?;
+            state.data_buffer.clear();
+            state.data_buffer.try_reserve(block_length as usize)?;
+            (&mut state.reader)
+                .take(block_length as u64)
+                .read_to_end(&mut state.data_buffer)
+                .await?;
 
             read_record_batch(
                 batch,
@@ -140,18 +147,22 @@ async fn maybe_next<R: AsyncRead + Unpin + Send>(
                 state.metadata.version,
                 &mut std::io::Cursor::new(&state.data_buffer),
                 0,
-                state.data_buffer.as_ref().len() as u64,
+                state.data_buffer.len() as u64,
                 &mut scratch,
             )
             .map(|chunk| Some(StreamState::Some((state, chunk))))
         }
         arrow_format::ipc::MessageHeaderRef::DictionaryBatch(batch) => {
-            let mut body = vec![0; block_length];
-            state.reader.read_exact(&mut body).await?;
+            state.data_buffer.clear();
+            state.data_buffer.try_reserve(block_length as usize)?;
+            (&mut state.reader)
+                .take(block_length as u64)
+                .read_to_end(&mut state.data_buffer)
+                .await?;
 
-            let file_size = body.len() as u64;
+            let file_size = state.data_buffer.len() as u64;
 
-            let mut dict_reader = std::io::Cursor::new(body);
+            let mut dict_reader = std::io::Cursor::new(&state.data_buffer);
 
             read_dictionary(
                 batch,
diff --git a/src/io/mod.rs b/src/io/mod.rs
index 95a17ee8bf..9343d4281c 100644
--- a/src/io/mod.rs
+++ b/src/io/mod.rs
@@ -53,9 +53,3 @@ 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/nested_utils.rs b/src/io/parquet/read/deserialize/nested_utils.rs
index 5c7f062f96..637cdf8919 100644
--- a/src/io/parquet/read/deserialize/nested_utils.rs
+++ b/src/io/parquet/read/deserialize/nested_utils.rs
@@ -235,7 +235,7 @@ where
     }
 }
 
-#[derive(Debug, Clone, Copy, PartialEq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum InitNested {
     Primitive(bool),
     List(bool),
diff --git a/src/io/parquet/read/deserialize/primitive/basic.rs b/src/io/parquet/read/deserialize/primitive/basic.rs
index 6e947b5643..eddf95872d 100644
--- a/src/io/parquet/read/deserialize/primitive/basic.rs
+++ b/src/io/parquet/read/deserialize/primitive/basic.rs
@@ -132,7 +132,7 @@ where
     op: F,
 }
 
-impl<'a, T, P, F> PrimitiveDecoder<T, P, F>
+impl<T, P, F> PrimitiveDecoder<T, P, F>
 where
     T: NativeType,
     P: ParquetNativeType,
diff --git a/src/io/parquet/read/deserialize/primitive/nested.rs b/src/io/parquet/read/deserialize/primitive/nested.rs
index 9263137298..4a2192e623 100644
--- a/src/io/parquet/read/deserialize/primitive/nested.rs
+++ b/src/io/parquet/read/deserialize/primitive/nested.rs
@@ -54,7 +54,7 @@ where
     op: F,
 }
 
-impl<'a, T, P, F> PrimitiveDecoder<T, P, F>
+impl<T, P, F> PrimitiveDecoder<T, P, F>
 where
     T: NativeType,
     P: ParquetNativeType,
diff --git a/src/io/parquet/read/row_group.rs b/src/io/parquet/read/row_group.rs
index 5d9198e6f4..deefaea364 100644
--- a/src/io/parquet/read/row_group.rs
+++ b/src/io/parquet/read/row_group.rs
@@ -120,10 +120,15 @@ fn _read_single_column<'a, R>(
 where
     R: Read + Seek,
 {
-    let (start, len) = meta.byte_range();
+    let (start, length) = meta.byte_range();
     reader.seek(std::io::SeekFrom::Start(start))?;
-    let mut chunk = vec![0; len as usize];
-    reader.read_exact(&mut chunk)?;
+
+    let mut chunk = vec![];
+    chunk.try_reserve(length as usize)?;
+    reader
+        .by_ref()
+        .take(length as u64)
+        .read_to_end(&mut chunk)?;
     Ok((meta, chunk))
 }
 
@@ -136,10 +141,12 @@ where
     F: Fn() -> BoxFuture<'b, std::io::Result<R>>,
 {
     let mut reader = factory().await?;
-    let (start, len) = meta.byte_range();
+    let (start, length) = meta.byte_range();
     reader.seek(std::io::SeekFrom::Start(start)).await?;
-    let mut chunk = vec![0; len as usize];
-    reader.read_exact(&mut chunk).await?;
+
+    let mut chunk = vec![];
+    chunk.try_reserve(length as usize)?;
+    reader.take(length as u64).read_to_end(&mut chunk).await?;
     Result::Ok((meta, chunk))
 }
 
diff --git a/src/io/parquet/read/statistics/mod.rs b/src/io/parquet/read/statistics/mod.rs
index a5916a9879..ff9449f0be 100644
--- a/src/io/parquet/read/statistics/mod.rs
+++ b/src/io/parquet/read/statistics/mod.rs
@@ -414,12 +414,10 @@ fn push(
             // some implementations of parquet write arrow's u32 into i64.
             ParquetPhysicalType::Int64 => primitive::push(from, min, max, |x: i64| Ok(x as u32)),
             ParquetPhysicalType::Int32 => primitive::push(from, min, max, |x: i32| Ok(x as u32)),
-            other => {
-                return Err(Error::NotYetImplemented(format!(
-                    "Can't decode UInt32 type from parquet type {:?}",
-                    other
-                )))
-            }
+            other => Err(Error::NotYetImplemented(format!(
+                "Can't decode UInt32 type from parquet type {:?}",
+                other
+            ))),
         },
         Int32 => primitive::push(from, min, max, |x: i32| Ok(x as i32)),
         Int64 | Date64 | Time64(_) | Duration(_) => {
@@ -442,7 +440,7 @@ fn push(
             ParquetPhysicalType::Int32 => primitive::push(from, min, max, |x: i32| Ok(x as i128)),
             ParquetPhysicalType::Int64 => primitive::push(from, min, max, |x: i64| Ok(x as i128)),
             ParquetPhysicalType::FixedLenByteArray(n) if *n > 16 => {
-                return Err(Error::NotYetImplemented(format!(
+                Err(Error::NotYetImplemented(format!(
                     "Can't decode Decimal128 type from Fixed Size Byte Array of len {:?}",
                     n
                 )))
diff --git a/src/io/parquet/write/pages.rs b/src/io/parquet/write/pages.rs
index 28f45a06fd..64f6da9a0f 100644
--- a/src/io/parquet/write/pages.rs
+++ b/src/io/parquet/write/pages.rs
@@ -198,7 +198,7 @@ pub fn array_to_columns<A: AsRef<dyn Array> + Send + Sync>(
     array: A,
     type_: ParquetType,
     options: WriteOptions,
-    encoding: Vec<Encoding>,
+    encoding: &[Encoding],
 ) -> Result<Vec<DynIter<'static, Result<EncodedPage>>>> {
     let array = array.as_ref();
     let nested = to_nested(array, &type_)?;
@@ -213,9 +213,9 @@ pub fn array_to_columns<A: AsRef<dyn Array> + Send + Sync>(
         .iter()
         .zip(nested.into_iter())
         .zip(types.into_iter())
-        .zip(encoding.into_iter())
+        .zip(encoding.iter())
         .map(|(((values, nested), type_), encoding)| {
-            array_to_pages(*values, type_, nested, options, encoding)
+            array_to_pages(*values, type_, nested, options, *encoding)
         })
         .collect()
 }
diff --git a/src/io/parquet/write/row_group.rs b/src/io/parquet/write/row_group.rs
index b9960b5600..5558bab49e 100644
--- a/src/io/parquet/write/row_group.rs
+++ b/src/io/parquet/write/row_group.rs
@@ -36,7 +36,7 @@ pub fn row_group_iter<A: AsRef<dyn Array> + 'static + Send + Sync>(
             .zip(fields.into_iter())
             .zip(encodings.into_iter())
             .flat_map(move |((array, type_), encoding)| {
-                let encoded_columns = array_to_columns(array, type_, options, encoding).unwrap();
+                let encoded_columns = array_to_columns(array, type_, options, &encoding).unwrap();
                 encoded_columns
                     .into_iter()
                     .map(|encoded_pages| {
diff --git a/src/io/readbuf.rs b/src/io/readbuf.rs
deleted file mode 100644
index d30396a828..0000000000
--- a/src/io/readbuf.rs
+++ /dev/null
@@ -1,50 +0,0 @@
-/// 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/lib.rs b/src/lib.rs
index ec01cec875..03963841b5 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -6,6 +6,8 @@
 #![allow(clippy::len_without_is_empty)]
 // Trait objects must be returned as a &Box<dyn Array> so that they can be cloned
 #![allow(clippy::borrowed_box)]
+// Allow type complexity warning to avoid API break.
+#![allow(clippy::type_complexity)]
 #![cfg_attr(docsrs, feature(doc_cfg))]
 #![cfg_attr(feature = "simd", feature(portable_simd))]
 
diff --git a/src/scalar/binary.rs b/src/scalar/binary.rs
index 7b6c0b9824..70a9be4037 100644
--- a/src/scalar/binary.rs
+++ b/src/scalar/binary.rs
@@ -3,7 +3,7 @@ use crate::{array::*, datatypes::DataType};
 use super::Scalar;
 
 /// The [`Scalar`] implementation of binary ([`Option<Vec<u8>>`]).
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct BinaryScalar<O: Offset> {
     value: Option<Vec<u8>>,
     phantom: std::marker::PhantomData<O>,
diff --git a/src/scalar/boolean.rs b/src/scalar/boolean.rs
index 47c3f9a9bc..a31cf1947f 100644
--- a/src/scalar/boolean.rs
+++ b/src/scalar/boolean.rs
@@ -3,7 +3,7 @@ use crate::datatypes::DataType;
 use super::Scalar;
 
 /// The [`Scalar`] implementation of a boolean.
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct BooleanScalar {
     value: Option<bool>,
 }
diff --git a/src/scalar/fixed_size_binary.rs b/src/scalar/fixed_size_binary.rs
index c09466f9eb..dfa4ace341 100644
--- a/src/scalar/fixed_size_binary.rs
+++ b/src/scalar/fixed_size_binary.rs
@@ -2,7 +2,7 @@ use crate::datatypes::DataType;
 
 use super::Scalar;
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 /// The [`Scalar`] implementation of fixed size binary ([`Option<Box<[u8]>>`]).
 pub struct FixedSizeBinaryScalar {
     value: Option<Box<[u8]>>,
diff --git a/src/scalar/primitive.rs b/src/scalar/primitive.rs
index 431906f759..158701c6bd 100644
--- a/src/scalar/primitive.rs
+++ b/src/scalar/primitive.rs
@@ -4,7 +4,7 @@ use super::Scalar;
 
 /// The implementation of [`Scalar`] for primitive, semantically equivalent to [`Option<T>`]
 /// with [`DataType`].
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct PrimitiveScalar<T: NativeType> {
     value: Option<T>,
     data_type: DataType,
diff --git a/src/scalar/utf8.rs b/src/scalar/utf8.rs
index 3fad8a4a48..f62581eccb 100644
--- a/src/scalar/utf8.rs
+++ b/src/scalar/utf8.rs
@@ -3,7 +3,7 @@ use crate::{array::*, datatypes::DataType};
 use super::Scalar;
 
 /// The implementation of [`Scalar`] for utf8, semantically equivalent to [`Option<String>`].
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Utf8Scalar<O: Offset> {
     value: Option<String>,
     phantom: std::marker::PhantomData<O>,


Test patch:
diff --git a/tests/it/io/csv/write.rs b/tests/it/io/csv/write.rs
index 88b502adfb..44bec50497 100644
--- a/tests/it/io/csv/write.rs
+++ b/tests/it/io/csv/write.rs
@@ -423,3 +423,23 @@ fn write_escaping_resize_local_buf() {
         test_generic(chunk,  &format!("\"{}\"\n", payload));
     }
 }
+
+#[test]
+fn serialize_vec() -> Result<()> {
+    let columns = data();
+
+    let options = SerializeOptions::default();
+
+    let data = serialize(&columns, &options)?;
+
+    // check
+    assert_eq!(
+        vec![
+            b"a b,123.564532,3,true,,00:20:34,d\n".to_vec(),
+            b"c,,2,false,2019-04-18 10:54:47.378,06:51:20,a b\n".to_vec(),
+            b"d,-556132.25,1,,2019-04-18 02:45:55.555,23:46:03,c\n".to_vec(),
+        ],
+        data
+    );
+    Ok(())
+}