From 7025df206f6941e08c2965ae17a8c08265766c64 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 13 Jan 2025 08:24:31 -0700 Subject: [PATCH 01/10] Add benchmarks for Arrow IPC writer --- arrow-ipc/Cargo.toml | 6 +++ arrow-ipc/benches/ipc.rs | 94 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 100 insertions(+) create mode 100644 arrow-ipc/benches/ipc.rs diff --git a/arrow-ipc/Cargo.toml b/arrow-ipc/Cargo.toml index 4988eed4a5e..5ab154f1670 100644 --- a/arrow-ipc/Cargo.toml +++ b/arrow-ipc/Cargo.toml @@ -47,4 +47,10 @@ default = [] lz4 = ["lz4_flex"] [dev-dependencies] +criterion = "0.5.1" tempfile = "3.3" +tokio = "1.43.0" + +[[bench]] +name = "ipc" +harness = false diff --git a/arrow-ipc/benches/ipc.rs b/arrow-ipc/benches/ipc.rs new file mode 100644 index 00000000000..00d6ac5cec6 --- /dev/null +++ b/arrow-ipc/benches/ipc.rs @@ -0,0 +1,94 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use arrow_array::builder::{Date32Builder, Decimal128Builder, Int32Builder}; +use arrow_array::{builder::StringBuilder, RecordBatch}; +use arrow_ipc::writer::StreamWriter; +use arrow_schema::{DataType, Field, Schema}; +use criterion::{criterion_group, criterion_main, Criterion}; +use std::sync::Arc; + +fn criterion_benchmark(c: &mut Criterion) { + let mut group = c.benchmark_group("arrow_ipc"); + + group.bench_function("write_single_batch", |b| { + let batch = create_batch(8192, true); + b.iter(|| { + let buffer = vec![]; + let mut writer = StreamWriter::try_new(buffer, batch.schema().as_ref()).unwrap(); + writer.write(&batch).unwrap(); + writer.finish().unwrap(); + }) + }); + + group.bench_function("write_multiple_batches", |b| { + let batch = create_batch(8192, true); + b.iter(|| { + let buffer = vec![]; + let mut writer = StreamWriter::try_new(buffer, batch.schema().as_ref()).unwrap(); + for _ in 0..10 { + writer.write(&batch).unwrap(); + } + writer.finish().unwrap(); + }) + }); +} + +fn create_batch(num_rows: usize, allow_nulls: bool) -> RecordBatch { + let schema = Arc::new(Schema::new(vec![ + Field::new("c0", DataType::Int32, true), + Field::new("c1", DataType::Utf8, true), + Field::new("c2", DataType::Date32, true), + Field::new("c3", DataType::Decimal128(11, 2), true), + ])); + let mut a = Int32Builder::new(); + let mut b = StringBuilder::new(); + let mut c = Date32Builder::new(); + let mut d = Decimal128Builder::new() + .with_precision_and_scale(11, 2) + .unwrap(); + for i in 0..num_rows { + a.append_value(i as i32); + c.append_value(i as i32); + d.append_value((i * 1000000) as i128); + if allow_nulls && i % 10 == 0 { + b.append_null(); + } else { + b.append_value(format!("this is string number {i}")); + } + } + let a = a.finish(); + let b = b.finish(); + let c = c.finish(); + let d = d.finish(); + RecordBatch::try_new( + schema.clone(), + vec![Arc::new(a), Arc::new(b), Arc::new(c), Arc::new(d)], + ) + .unwrap() +} + +fn config() -> Criterion { + Criterion::default() +} + +criterion_group! { + name = benches; + config = config(); + targets = criterion_benchmark +} +criterion_main!(benches); From e377abe073b18aed483af37a588db0a080b6bc40 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 13 Jan 2025 08:27:47 -0700 Subject: [PATCH 02/10] Add benchmarks for Arrow IPC writer --- arrow-ipc/Cargo.toml | 2 +- arrow-ipc/benches/{ipc.rs => stream_writer.rs} | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) rename arrow-ipc/benches/{ipc.rs => stream_writer.rs} (97%) diff --git a/arrow-ipc/Cargo.toml b/arrow-ipc/Cargo.toml index 5ab154f1670..701f4316111 100644 --- a/arrow-ipc/Cargo.toml +++ b/arrow-ipc/Cargo.toml @@ -52,5 +52,5 @@ tempfile = "3.3" tokio = "1.43.0" [[bench]] -name = "ipc" +name = "stream_writer" harness = false diff --git a/arrow-ipc/benches/ipc.rs b/arrow-ipc/benches/stream_writer.rs similarity index 97% rename from arrow-ipc/benches/ipc.rs rename to arrow-ipc/benches/stream_writer.rs index 00d6ac5cec6..95e1aec5f8d 100644 --- a/arrow-ipc/benches/ipc.rs +++ b/arrow-ipc/benches/stream_writer.rs @@ -23,7 +23,7 @@ use criterion::{criterion_group, criterion_main, Criterion}; use std::sync::Arc; fn criterion_benchmark(c: &mut Criterion) { - let mut group = c.benchmark_group("arrow_ipc"); + let mut group = c.benchmark_group("arrow_ipc_stream_writer"); group.bench_function("write_single_batch", |b| { let batch = create_batch(8192, true); From 1c999a6844817e2a7b25b44ad61d1a32f97d3fdd Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Tue, 14 Jan 2025 10:58:53 -0500 Subject: [PATCH 03/10] reuse target buffer --- arrow-ipc/benches/stream_writer.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/arrow-ipc/benches/stream_writer.rs b/arrow-ipc/benches/stream_writer.rs index 95e1aec5f8d..c794a91d6b0 100644 --- a/arrow-ipc/benches/stream_writer.rs +++ b/arrow-ipc/benches/stream_writer.rs @@ -27,9 +27,10 @@ fn criterion_benchmark(c: &mut Criterion) { group.bench_function("write_single_batch", |b| { let batch = create_batch(8192, true); - b.iter(|| { - let buffer = vec![]; - let mut writer = StreamWriter::try_new(buffer, batch.schema().as_ref()).unwrap(); + let mut buffer = Vec::with_capacity(2 * 1024 * 1024); + b.iter(move || { + buffer.clear(); + let mut writer = StreamWriter::try_new(&mut buffer, batch.schema().as_ref()).unwrap(); writer.write(&batch).unwrap(); writer.finish().unwrap(); }) @@ -37,9 +38,10 @@ fn criterion_benchmark(c: &mut Criterion) { group.bench_function("write_multiple_batches", |b| { let batch = create_batch(8192, true); - b.iter(|| { - let buffer = vec![]; - let mut writer = StreamWriter::try_new(buffer, batch.schema().as_ref()).unwrap(); + let mut buffer = Vec::with_capacity(2 * 1024 * 1024); + b.iter(move || { + buffer.clear(); + let mut writer = StreamWriter::try_new(&mut buffer, batch.schema().as_ref()).unwrap(); for _ in 0..10 { writer.write(&batch).unwrap(); } From 57e483807c35c06d031a4cc9c218cd842850df3f Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 6 Feb 2025 14:54:43 -0500 Subject: [PATCH 04/10] rename, etc --- arrow-ipc/Cargo.toml | 2 +- arrow-ipc/benches/{stream_writer.rs => ipc_writer.rs} | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) rename arrow-ipc/benches/{stream_writer.rs => ipc_writer.rs} (96%) diff --git a/arrow-ipc/Cargo.toml b/arrow-ipc/Cargo.toml index 701f4316111..9844ae9c657 100644 --- a/arrow-ipc/Cargo.toml +++ b/arrow-ipc/Cargo.toml @@ -52,5 +52,5 @@ tempfile = "3.3" tokio = "1.43.0" [[bench]] -name = "stream_writer" +name = "ipc_writer" harness = false diff --git a/arrow-ipc/benches/stream_writer.rs b/arrow-ipc/benches/ipc_writer.rs similarity index 96% rename from arrow-ipc/benches/stream_writer.rs rename to arrow-ipc/benches/ipc_writer.rs index c794a91d6b0..1207e98e928 100644 --- a/arrow-ipc/benches/stream_writer.rs +++ b/arrow-ipc/benches/ipc_writer.rs @@ -25,7 +25,7 @@ use std::sync::Arc; fn criterion_benchmark(c: &mut Criterion) { let mut group = c.benchmark_group("arrow_ipc_stream_writer"); - group.bench_function("write_single_batch", |b| { + group.bench_function("StreamWriter/write_single_batch", |b| { let batch = create_batch(8192, true); let mut buffer = Vec::with_capacity(2 * 1024 * 1024); b.iter(move || { @@ -36,7 +36,7 @@ fn criterion_benchmark(c: &mut Criterion) { }) }); - group.bench_function("write_multiple_batches", |b| { + group.bench_function("StreamWriter/write_10_batches", |b| { let batch = create_batch(8192, true); let mut buffer = Vec::with_capacity(2 * 1024 * 1024); b.iter(move || { From 92bdea3a49ec621071172e724d8efbba52457ee8 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 6 Feb 2025 15:10:53 -0500 Subject: [PATCH 05/10] Add compression type --- arrow-ipc/benches/ipc_writer.rs | 32 +++++++++++++++++++++++++++----- 1 file changed, 27 insertions(+), 5 deletions(-) diff --git a/arrow-ipc/benches/ipc_writer.rs b/arrow-ipc/benches/ipc_writer.rs index 1207e98e928..a0bfa41bfd6 100644 --- a/arrow-ipc/benches/ipc_writer.rs +++ b/arrow-ipc/benches/ipc_writer.rs @@ -17,7 +17,8 @@ use arrow_array::builder::{Date32Builder, Decimal128Builder, Int32Builder}; use arrow_array::{builder::StringBuilder, RecordBatch}; -use arrow_ipc::writer::StreamWriter; +use arrow_ipc::writer::{FileWriter, IpcWriteOptions, StreamWriter}; +use arrow_ipc::CompressionType; use arrow_schema::{DataType, Field, Schema}; use criterion::{criterion_group, criterion_main, Criterion}; use std::sync::Arc; @@ -25,23 +26,44 @@ use std::sync::Arc; fn criterion_benchmark(c: &mut Criterion) { let mut group = c.benchmark_group("arrow_ipc_stream_writer"); - group.bench_function("StreamWriter/write_single_batch", |b| { + + group.bench_function("StreamWriter/write_10", |b| { let batch = create_batch(8192, true); let mut buffer = Vec::with_capacity(2 * 1024 * 1024); b.iter(move || { buffer.clear(); let mut writer = StreamWriter::try_new(&mut buffer, batch.schema().as_ref()).unwrap(); - writer.write(&batch).unwrap(); + for _ in 0..10 { + writer.write(&batch).unwrap(); + } writer.finish().unwrap(); }) }); - group.bench_function("StreamWriter/write_10_batches", |b| { + group.bench_function("StreamWriter/write_10/zstd", |b| { let batch = create_batch(8192, true); let mut buffer = Vec::with_capacity(2 * 1024 * 1024); b.iter(move || { buffer.clear(); - let mut writer = StreamWriter::try_new(&mut buffer, batch.schema().as_ref()).unwrap(); + let options = IpcWriteOptions::default() + .try_with_compression(Some(CompressionType::ZSTD)) + .unwrap(); + let mut writer = + StreamWriter::try_new_with_options(&mut buffer, batch.schema().as_ref(), options) + .unwrap(); + for _ in 0..10 { + writer.write(&batch).unwrap(); + } + writer.finish().unwrap(); + }) + }); + + group.bench_function("FileWriter/write_10", |b| { + let batch = create_batch(8192, true); + let mut buffer = Vec::with_capacity(2 * 1024 * 1024); + b.iter(move || { + buffer.clear(); + let mut writer = FileWriter::try_new(&mut buffer, batch.schema().as_ref()).unwrap(); for _ in 0..10 { writer.write(&batch).unwrap(); } From 3d7d396ef1b72d41ed4c876a9515765e54de91b8 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 6 Feb 2025 15:11:10 -0500 Subject: [PATCH 06/10] update --- arrow-ipc/benches/ipc_writer.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/arrow-ipc/benches/ipc_writer.rs b/arrow-ipc/benches/ipc_writer.rs index a0bfa41bfd6..6b4d184b455 100644 --- a/arrow-ipc/benches/ipc_writer.rs +++ b/arrow-ipc/benches/ipc_writer.rs @@ -26,7 +26,6 @@ use std::sync::Arc; fn criterion_benchmark(c: &mut Criterion) { let mut group = c.benchmark_group("arrow_ipc_stream_writer"); - group.bench_function("StreamWriter/write_10", |b| { let batch = create_batch(8192, true); let mut buffer = Vec::with_capacity(2 * 1024 * 1024); From 96b454fec0f863610ee826b6458586f433c4a85c Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 6 Feb 2025 15:30:12 -0500 Subject: [PATCH 07/10] Add ipc reader benchmark --- arrow-ipc/Cargo.toml | 7 ++ arrow-ipc/benches/ipc_reader.rs | 140 ++++++++++++++++++++++++++++++++ 2 files changed, 147 insertions(+) create mode 100644 arrow-ipc/benches/ipc_reader.rs diff --git a/arrow-ipc/Cargo.toml b/arrow-ipc/Cargo.toml index 9844ae9c657..735f7a14a2f 100644 --- a/arrow-ipc/Cargo.toml +++ b/arrow-ipc/Cargo.toml @@ -50,7 +50,14 @@ lz4 = ["lz4_flex"] criterion = "0.5.1" tempfile = "3.3" tokio = "1.43.0" +# used in benches +memmap2 = "0.9.3" +bytes = "1.9" [[bench]] name = "ipc_writer" harness = false + +[[bench]] +name = "ipc_reader" +harness = false diff --git a/arrow-ipc/benches/ipc_reader.rs b/arrow-ipc/benches/ipc_reader.rs new file mode 100644 index 00000000000..c002dc6536d --- /dev/null +++ b/arrow-ipc/benches/ipc_reader.rs @@ -0,0 +1,140 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::io::Cursor; +use arrow_array::builder::{Date32Builder, Decimal128Builder, Int32Builder}; +use arrow_array::{builder::StringBuilder, RecordBatch}; +use arrow_ipc::reader::{FileReader, StreamReader}; +use arrow_ipc::writer::{FileWriter, IpcWriteOptions, StreamWriter}; +use arrow_ipc::CompressionType; +use arrow_schema::{DataType, Field, Schema}; +use criterion::{criterion_group, criterion_main, Criterion}; +use std::sync::Arc; + +fn criterion_benchmark(c: &mut Criterion) { + let mut group = c.benchmark_group("arrow_ipc_stream_writer"); + + group.bench_function("StreamReader/read_10", |b| { + let batch = create_batch(8192, true); + let mut buffer = Vec::with_capacity(2 * 1024 * 1024); + let mut writer = StreamWriter::try_new(&mut buffer, batch.schema().as_ref()).unwrap(); + for _ in 0..10 { + writer.write(&batch).unwrap(); + } + writer.finish().unwrap(); + + b.iter(move || { + let projection = None; + let mut reader = StreamReader::try_new(buffer.as_slice(), projection).unwrap(); + for _ in 0..10 { + reader.next().unwrap().unwrap(); + } + assert!(reader.next().is_none()); + }) + }); + + group.bench_function("StreamReader/read_10/zstd", |b| { + let batch = create_batch(8192, true); + let mut buffer = Vec::with_capacity(2 * 1024 * 1024); + let options = IpcWriteOptions::default() + .try_with_compression(Some(CompressionType::ZSTD)) + .unwrap(); + let mut writer = + StreamWriter::try_new_with_options(&mut buffer, batch.schema().as_ref(), options) + .unwrap(); + for _ in 0..10 { + writer.write(&batch).unwrap(); + } + writer.finish().unwrap(); + + b.iter(move || { + let projection = None; + let mut reader = StreamReader::try_new(buffer.as_slice(), projection).unwrap(); + for _ in 0..10 { + reader.next().unwrap().unwrap(); + } + assert!(reader.next().is_none()); + }) + }); + + group.bench_function("FileReader/read_10", |b| { + let batch = create_batch(8192, true); + let mut buffer = Vec::with_capacity(2 * 1024 * 1024); + let mut writer = FileWriter::try_new(&mut buffer, batch.schema().as_ref()).unwrap(); + for _ in 0..10 { + writer.write(&batch).unwrap(); + } + writer.finish().unwrap(); + + b.iter(move || { + let projection = None; + let cursor = Cursor::new(buffer.as_slice()); + let mut reader = FileReader::try_new(cursor, projection).unwrap(); + for _ in 0..10 { + reader.next().unwrap().unwrap(); + } + assert!(reader.next().is_none()); + }) + }); + + // mmap file read +} + +fn create_batch(num_rows: usize, allow_nulls: bool) -> RecordBatch { + let schema = Arc::new(Schema::new(vec![ + Field::new("c0", DataType::Int32, true), + Field::new("c1", DataType::Utf8, true), + Field::new("c2", DataType::Date32, true), + Field::new("c3", DataType::Decimal128(11, 2), true), + ])); + let mut a = Int32Builder::new(); + let mut b = StringBuilder::new(); + let mut c = Date32Builder::new(); + let mut d = Decimal128Builder::new() + .with_precision_and_scale(11, 2) + .unwrap(); + for i in 0..num_rows { + a.append_value(i as i32); + c.append_value(i as i32); + d.append_value((i * 1000000) as i128); + if allow_nulls && i % 10 == 0 { + b.append_null(); + } else { + b.append_value(format!("this is string number {i}")); + } + } + let a = a.finish(); + let b = b.finish(); + let c = c.finish(); + let d = d.finish(); + RecordBatch::try_new( + schema.clone(), + vec![Arc::new(a), Arc::new(b), Arc::new(c), Arc::new(d)], + ) + .unwrap() +} + +fn config() -> Criterion { + Criterion::default() +} + +criterion_group! { + name = benches; + config = config(); + targets = criterion_benchmark +} +criterion_main!(benches); From 7527f8e65fdb5501708d80f5a0e983971ffdda10 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 6 Feb 2025 15:50:48 -0500 Subject: [PATCH 08/10] Add mmap example --- arrow-ipc/benches/ipc_reader.rs | 93 +++++++++++++++++++++++++++++++-- 1 file changed, 90 insertions(+), 3 deletions(-) diff --git a/arrow-ipc/benches/ipc_reader.rs b/arrow-ipc/benches/ipc_reader.rs index c002dc6536d..f054b9354ab 100644 --- a/arrow-ipc/benches/ipc_reader.rs +++ b/arrow-ipc/benches/ipc_reader.rs @@ -18,12 +18,15 @@ use std::io::Cursor; use arrow_array::builder::{Date32Builder, Decimal128Builder, Int32Builder}; use arrow_array::{builder::StringBuilder, RecordBatch}; -use arrow_ipc::reader::{FileReader, StreamReader}; +use arrow_ipc::reader::{read_footer_length, FileDecoder, FileReader, StreamReader}; use arrow_ipc::writer::{FileWriter, IpcWriteOptions, StreamWriter}; -use arrow_ipc::CompressionType; +use arrow_ipc::{root_as_footer, Block, CompressionType}; use arrow_schema::{DataType, Field, Schema}; use criterion::{criterion_group, criterion_main, Criterion}; use std::sync::Arc; +use tempfile::{tempdir}; +use arrow_buffer::Buffer; +use arrow_ipc::convert::fb_to_schema; fn criterion_benchmark(c: &mut Criterion) { let mut group = c.benchmark_group("arrow_ipc_stream_writer"); @@ -91,9 +94,93 @@ fn criterion_benchmark(c: &mut Criterion) { }) }); - // mmap file read + group.bench_function("FileReader/read_10/mmap", |b| { + let batch = create_batch(8192, true); + // write to an actual file + let dir = tempdir().unwrap(); + let path = dir.path().join("test.arrow"); + let file = std::fs::File::create(&path).unwrap(); + let mut writer = FileWriter::try_new(file, batch.schema().as_ref()).unwrap(); + for _ in 0..10 { + writer.write(&batch).unwrap(); + } + writer.finish().unwrap(); + + b.iter(move || { + let ipc_file = std::fs::File::open(&path).expect("failed to open file"); + let mmap = unsafe { memmap2::Mmap::map(&ipc_file).expect("failed to mmap file") }; + + // Convert the mmap region to an Arrow `Buffer` to back the arrow arrays. + let bytes = bytes::Bytes::from_owner(mmap); + let buffer = Buffer::from(bytes); + let decoder = IPCBufferDecoder::new(buffer); + assert_eq!(decoder.num_batches(), 10); + + for i in 0..decoder.num_batches() { + decoder.get_batch(i); + } + }) + }); } +// copied from the zero_copy_ipc example. +// should we move this to an actual API? +/// Wrapper around the example in the `FileDecoder` which handles the +/// low level interaction with the Arrow IPC format. +struct IPCBufferDecoder { + /// Memory (or memory mapped) Buffer with the data + buffer: Buffer, + /// Decoder that reads Arrays that refers to the underlying buffers + decoder: FileDecoder, + /// Location of the batches within the buffer + batches: Vec, +} + +impl IPCBufferDecoder { + fn new(buffer: Buffer) -> Self { + let trailer_start = buffer.len() - 10; + let footer_len = read_footer_length(buffer[trailer_start..].try_into().unwrap()).unwrap(); + let footer = root_as_footer(&buffer[trailer_start - footer_len..trailer_start]).unwrap(); + + let schema = fb_to_schema(footer.schema().unwrap()); + + let mut decoder = FileDecoder::new(Arc::new(schema), footer.version()); + + // Read dictionaries + for block in footer.dictionaries().iter().flatten() { + let block_len = block.bodyLength() as usize + block.metaDataLength() as usize; + let data = buffer.slice_with_length(block.offset() as _, block_len); + decoder.read_dictionary(block, &data).unwrap(); + } + + // convert to Vec from the flatbuffers Vector to avoid having a direct dependency on flatbuffers + let batches = footer + .recordBatches() + .map(|b| b.iter().copied().collect()) + .unwrap_or_default(); + + Self { + buffer, + decoder, + batches, + } + } + + fn num_batches(&self) -> usize { + self.batches.len() + } + + fn get_batch(&self, i: usize) -> RecordBatch { + let block = &self.batches[i]; + let block_len = block.bodyLength() as usize + block.metaDataLength() as usize; + let data = self + .buffer + .slice_with_length(block.offset() as _, block_len); + self.decoder.read_record_batch(block, &data).unwrap().unwrap() + } +} + + fn create_batch(num_rows: usize, allow_nulls: bool) -> RecordBatch { let schema = Arc::new(Schema::new(vec![ Field::new("c0", DataType::Int32, true), From 8bc23a15d91297f0764724e399d16a3817cb2f51 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 6 Feb 2025 15:56:34 -0500 Subject: [PATCH 09/10] fmt --- arrow-ipc/benches/ipc_reader.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/arrow-ipc/benches/ipc_reader.rs b/arrow-ipc/benches/ipc_reader.rs index f054b9354ab..71482c4140e 100644 --- a/arrow-ipc/benches/ipc_reader.rs +++ b/arrow-ipc/benches/ipc_reader.rs @@ -15,18 +15,18 @@ // specific language governing permissions and limitations // under the License. -use std::io::Cursor; use arrow_array::builder::{Date32Builder, Decimal128Builder, Int32Builder}; use arrow_array::{builder::StringBuilder, RecordBatch}; +use arrow_buffer::Buffer; +use arrow_ipc::convert::fb_to_schema; use arrow_ipc::reader::{read_footer_length, FileDecoder, FileReader, StreamReader}; use arrow_ipc::writer::{FileWriter, IpcWriteOptions, StreamWriter}; use arrow_ipc::{root_as_footer, Block, CompressionType}; use arrow_schema::{DataType, Field, Schema}; use criterion::{criterion_group, criterion_main, Criterion}; +use std::io::Cursor; use std::sync::Arc; -use tempfile::{tempdir}; -use arrow_buffer::Buffer; -use arrow_ipc::convert::fb_to_schema; +use tempfile::tempdir; fn criterion_benchmark(c: &mut Criterion) { let mut group = c.benchmark_group("arrow_ipc_stream_writer"); @@ -176,11 +176,13 @@ impl IPCBufferDecoder { let data = self .buffer .slice_with_length(block.offset() as _, block_len); - self.decoder.read_record_batch(block, &data).unwrap().unwrap() + self.decoder + .read_record_batch(block, &data) + .unwrap() + .unwrap() } } - fn create_batch(num_rows: usize, allow_nulls: bool) -> RecordBatch { let schema = Arc::new(Schema::new(vec![ Field::new("c0", DataType::Int32, true), From 6cd7e5d48c1f20e8e6364b7cd5ce6a9695359527 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Fri, 7 Feb 2025 10:30:59 -0500 Subject: [PATCH 10/10] Update arrow-ipc/benches/ipc_reader.rs --- arrow-ipc/benches/ipc_reader.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/arrow-ipc/benches/ipc_reader.rs b/arrow-ipc/benches/ipc_reader.rs index 71482c4140e..7fc14664b4a 100644 --- a/arrow-ipc/benches/ipc_reader.rs +++ b/arrow-ipc/benches/ipc_reader.rs @@ -29,7 +29,7 @@ use std::sync::Arc; use tempfile::tempdir; fn criterion_benchmark(c: &mut Criterion) { - let mut group = c.benchmark_group("arrow_ipc_stream_writer"); + let mut group = c.benchmark_group("arrow_ipc_reader"); group.bench_function("StreamReader/read_10", |b| { let batch = create_batch(8192, true);