This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 1e6aef1c07 bench(parquet): broaden Arrow writer benchmark coverage
(#11267)
1e6aef1c07 is described below
commit 1e6aef1c07dc1e36758374ab28bdafb4fd93bead
Author: Hippolyte Barraud <[email protected]>
AuthorDate: Fri Oct 9 19:42:35 2026 +0000
bench(parquet): broaden Arrow writer benchmark coverage (#11267)
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax.
-->
- Contributes to #9731.
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
Writer changes need a representative baseline across Arrow's physical
representations, null distributions, and nested layouts. Dense scalar
inputs alone do not capture the costs of dictionaries, run-end encoding,
sparse values, or fragmented nested traversal.
Expand the timing suite so future changes can be compared across these
workloads on a consistent basis.
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
Expand the Arrow writer timing suite to establish a representative
baseline ahead of subsequent writer changes. The matrix now covers
dictionary, run-end-encoded, nested, sparse, fixed-size binary, and
decimal inputs, including combinations such as REE over dictionaries,
nested REE values, and fragmented list reachability patterns.
Report throughput in logical rows so dense and compact Arrow
representations can be compared on the same basis, and make
low-cardinality fixtures reproducible across runs.
**On Linux, use a constrained jemalloc configuration to reduce
allocator-driven noise** (which would create measurable drift depending
on benchmark execution order).
# Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
If this PR claims a performance improvement, please include evidence
such as benchmark results.
-->
NA
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
If there are any breaking changes to public APIs, please call them out.
-->
NA
Signed-off-by: Hippolyte Barraud <[email protected]>
---
parquet/benches/arrow_writer.rs | 1002 +++++++++++++++++++++++++++++++++++----
1 file changed, 898 insertions(+), 104 deletions(-)
diff --git a/parquet/benches/arrow_writer.rs b/parquet/benches/arrow_writer.rs
index 54fdd27397..d399c0a593 100644
--- a/parquet/benches/arrow_writer.rs
+++ b/parquet/benches/arrow_writer.rs
@@ -24,15 +24,22 @@ use parquet::arrow::ArrowWriter;
use parquet::basic::{Compression, Encoding, ZstdLevel};
use rand::{RngExt, distr::Alphanumeric};
+use std::cell::OnceCell;
use std::hint::black_box;
use std::io::Empty;
use std::sync::Arc;
use arrow::datatypes::*;
use arrow::util::bench_util::{create_f16_array, create_f32_array,
create_f64_array};
+use arrow::util::test_util::seedable_rng;
use arrow::{record_batch::RecordBatch, util::data_gen::*};
use arrow_array::builder::{FixedSizeBinaryBuilder, ListBuilder, StringBuilder};
-use arrow_array::{Int32Array, LargeBinaryArray, RecordBatchOptions,
StringArray, StringViewArray};
+use arrow_array::{
+ Array, ArrayRef, BooleanArray, Decimal128Array, DictionaryArray,
FixedSizeBinaryArray,
+ Float64Array, Int8Array, Int32Array, Int64Array, LargeBinaryArray,
ListArray,
+ RecordBatchOptions, RunArray, StringArray, StringViewArray, StructArray,
+};
+use arrow_buffer::{NullBuffer, OffsetBuffer};
use parquet::errors::Result;
use parquet::file::properties::{CdcOptions, WriterProperties, WriterVersion};
@@ -82,6 +89,99 @@ fn create_primitive_bench_batch_non_null(
)?)
}
+fn create_primitive_dictionary_bench_batch(
+ size: usize,
+ null_density: f32,
+ true_density: f32,
+) -> Result<RecordBatch> {
+ let fields = vec![Field::new(
+ "_1",
+ DataType::Dictionary(Box::new(DataType::Int32),
Box::new(DataType::Int32)),
+ true,
+ )];
+ let schema = Schema::new(fields);
+ Ok(create_random_batch(
+ Arc::new(schema),
+ size,
+ null_density,
+ true_density,
+ )?)
+}
+
+fn create_primitive_dictionary_bench_batch_1pct_cardinality(
+ size: usize,
+ null_density: f32,
+) -> Result<RecordBatch> {
+ let cardinality = dictionary_cardinality_1pct(size);
+ let schema = dictionary_schema(DataType::Int32);
+ let keys = dictionary_keys(size, cardinality, null_density);
+ let values = Int32Array::from_iter_values((0..cardinality).map(|i| i as
i32));
+ let array = DictionaryArray::<Int32Type>::new(keys, Arc::new(values));
+
+ Ok(RecordBatch::try_new(
+ schema,
+ vec![Arc::new(array) as ArrayRef],
+ )?)
+}
+
+fn create_int64_dictionary_bench_batch(
+ size: usize,
+ null_density: f32,
+ true_density: f32,
+) -> Result<RecordBatch> {
+ Ok(create_random_batch(
+ dictionary_schema(DataType::Int64),
+ size,
+ null_density,
+ true_density,
+ )?)
+}
+
+fn create_int64_dictionary_bench_batch_1pct_cardinality(
+ size: usize,
+ null_density: f32,
+) -> Result<RecordBatch> {
+ let cardinality = dictionary_cardinality_1pct(size);
+ let schema = dictionary_schema(DataType::Int64);
+ let keys = dictionary_keys(size, cardinality, null_density);
+ let values = Int64Array::from_iter_values((0..cardinality).map(|i| i as
i64));
+ let array = DictionaryArray::<Int32Type>::new(keys, Arc::new(values));
+
+ Ok(RecordBatch::try_new(
+ schema,
+ vec![Arc::new(array) as ArrayRef],
+ )?)
+}
+
+fn create_float64_dictionary_bench_batch(
+ size: usize,
+ null_density: f32,
+ true_density: f32,
+) -> Result<RecordBatch> {
+ Ok(create_random_batch(
+ dictionary_schema(DataType::Float64),
+ size,
+ null_density,
+ true_density,
+ )?)
+}
+
+fn create_float64_dictionary_bench_batch_1pct_cardinality(
+ size: usize,
+ null_density: f32,
+) -> Result<RecordBatch> {
+ let cardinality = dictionary_cardinality_1pct(size);
+ let schema = dictionary_schema(DataType::Float64);
+ let keys = dictionary_keys(size, cardinality, null_density);
+ let values = Float64Array::from_iter_values((0..cardinality).map(|i| i as
f64));
+ let array = DictionaryArray::<Int32Type>::new(keys, Arc::new(values));
+
+ Ok(RecordBatch::try_new(
+ schema,
+ vec![Arc::new(array) as ArrayRef],
+ )?)
+}
+
fn create_string_bench_batch(
size: usize,
null_density: f32,
@@ -101,7 +201,7 @@ fn create_string_bench_batch(
}
// Creates a DictionaryArray with target cardinality
fn create_low_card_dictionary_bench_batch(size: usize, cardinality: usize) ->
Result<RecordBatch> {
- let mut rng = rand::rng();
+ let mut rng = seedable_rng();
// Generate `cardinality` unique random strings.
let categories: Vec<String> = (0..cardinality)
@@ -310,6 +410,68 @@ fn create_string_dictionary_bench_batch(
true_density,
)?)
}
+
+/// Run-end-encoded column whose run values are a low-cardinality
+/// `Dictionary<Int32, Utf8>` (run length 8, 16 distinct dictionary entries).
+fn create_ree_of_dict_bench_batch(size: usize) -> RecordBatch {
+ let run_len = 8usize;
+ let num_runs = size / run_len;
+ let run_ends = Int32Array::from(
+ (1..=num_runs)
+ .map(|i| (i * run_len) as i32)
+ .collect::<Vec<_>>(),
+ );
+ let dict_values = StringArray::from((0..16).map(|i|
format!("val_{i:04}")).collect::<Vec<_>>());
+ let keys = Int32Array::from((0..num_runs).map(|i| (i % 16) as
i32).collect::<Vec<_>>());
+ let dict = DictionaryArray::<Int32Type>::new(keys, Arc::new(dict_values));
+ let ree = RunArray::<Int32Type>::try_new(&run_ends, &dict).unwrap();
+ let field = Field::new("_1", ree.data_type().clone(), true);
+ let schema = Arc::new(Schema::new(vec![field]));
+ RecordBatch::try_new(schema, vec![Arc::new(ree)]).unwrap()
+}
+
+/// A run-end column whose values are a sized `Dictionary<Int32, Int64>`.
+fn create_ree_of_numeric_dict_bench_batch(size: usize) -> RecordBatch {
+ let run_len = 8usize;
+ let num_runs = size / run_len;
+ let run_ends = Int32Array::from(
+ (1..=num_runs)
+ .map(|i| (i * run_len) as i32)
+ .collect::<Vec<_>>(),
+ );
+ let dict_values = Int64Array::from((0..16).map(|i| i as i64 *
1_000_000).collect::<Vec<_>>());
+ let keys = Int32Array::from((0..num_runs).map(|i| (i % 16) as
i32).collect::<Vec<_>>());
+ let dict = DictionaryArray::<Int32Type>::new(keys, Arc::new(dict_values));
+ let ree = RunArray::<Int32Type>::try_new(&run_ends, &dict).unwrap();
+ let field = Field::new("_1", ree.data_type().clone(), true);
+ let schema = Arc::new(Schema::new(vec![field]));
+ RecordBatch::try_new(schema, vec![Arc::new(ree)]).unwrap()
+}
+
+/// A run-end-encoded struct with a dictionary string field and an Int64 field.
+fn create_ree_struct_of_dict_bench_batch(size: usize) -> RecordBatch {
+ let run_len = 8usize;
+ let num_runs = size / run_len;
+ let run_ends = Int32Array::from(
+ (1..=num_runs)
+ .map(|i| (i * run_len) as i32)
+ .collect::<Vec<_>>(),
+ );
+ let dict_values = StringArray::from((0..16).map(|i|
format!("tag_{i:04}")).collect::<Vec<_>>());
+ let keys = Int32Array::from((0..num_runs).map(|i| (i % 16) as
i32).collect::<Vec<_>>());
+ let tag = DictionaryArray::<Int32Type>::new(keys, Arc::new(dict_values));
+ let val = Int64Array::from((0..num_runs).map(|i| i as
i64).collect::<Vec<_>>());
+ let fields = Fields::from(vec![
+ Field::new("tag", tag.data_type().clone(), true),
+ Field::new("val", DataType::Int64, true),
+ ]);
+ let values = StructArray::new(fields, vec![Arc::new(tag), Arc::new(val)],
None);
+ let ree = RunArray::<Int32Type>::try_new(&run_ends, &values).unwrap();
+ let field = Field::new("_1", ree.data_type().clone(), true);
+ let schema = Arc::new(Schema::new(vec![field]));
+ RecordBatch::try_new(schema, vec![Arc::new(ree)]).unwrap()
+}
+
fn create_ree_bench_batch(
value_dt: DataType,
size: usize,
@@ -343,6 +505,256 @@ fn create_ree_bench_batch(
)?)
}
+/// A single run-end-encoded column with configurable run length and run-end
+/// index width. `make_values` produces one value per run, allowing benchmarks
+/// to vary value type and distinct-value cardinality.
+fn create_run_end_encoded_bench_batch(
+ run_ends_type: DataType,
+ size: usize,
+ run_len: usize,
+ make_values: impl Fn(usize) -> ArrayRef,
+) -> RecordBatch {
+ let num_runs = size.div_ceil(run_len);
+ let mut acc = 0usize;
+ let run_ends: Vec<i64> = (0..num_runs)
+ .map(|_| {
+ acc = (acc + run_len).min(size);
+ acc as i64
+ })
+ .collect();
+ let values = make_values(num_runs);
+ let run_array: ArrayRef = match run_ends_type {
+ DataType::Int32 => {
+ let ends = Int32Array::from_iter_values(run_ends.iter().map(|&v| v
as i32));
+ Arc::new(RunArray::<Int32Type>::try_new(&ends, &values).unwrap())
+ }
+ DataType::Int64 => {
+ let ends = Int64Array::from_iter_values(run_ends.iter().copied());
+ Arc::new(RunArray::<Int64Type>::try_new(&ends, &values).unwrap())
+ }
+ other => panic!("unsupported REE run-ends type for bench: {other:?}"),
+ };
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "_1",
+ run_array.data_type().clone(),
+ true,
+ )]));
+ RecordBatch::try_new(schema, vec![run_array]).unwrap()
+}
+
+/// Struct-of-leaves run values for the non-leaf REE benches: 16 distinct
+/// (int64, utf8) pairs cycled over `n` runs, one in eight struct-null.
+fn make_ree_struct_values(n: usize) -> ArrayRef {
+ let a = Int64Array::from_iter_values((0..n).map(|i| (i % 16) as i64));
+ let b = StringArray::from_iter_values((0..n).map(|i|
format!("category_{:02}", i % 16)));
+ let fields = Fields::from(vec![
+ Field::new("a", DataType::Int64, true),
+ Field::new("b", DataType::Utf8, true),
+ ]);
+ let nulls = NullBuffer::from_iter((0..n).map(|i| i % 8 != 7));
+ Arc::new(StructArray::new(
+ fields,
+ vec![Arc::new(a), Arc::new(b)],
+ Some(nulls),
+ ))
+}
+
+/// String-list run values: each run value is a list of 4 strings drawn from
+/// 16 distinct lists.
+fn make_ree_string_list_values(n: usize) -> ArrayRef {
+ let values = StringArray::from_iter_values(
+ (0..n * 4).map(|i| format!("category_{:02}_{}", (i / 4) % 16, i % 4)),
+ );
+ let offsets = OffsetBuffer::from_lengths(std::iter::repeat_n(4usize, n));
+ let field = Arc::new(Field::new_list_field(DataType::Utf8, true));
+ Arc::new(ListArray::new(field, offsets, Arc::new(values), None))
+}
+
+/// A batch of `size` lists of 4 elements over a run-end-encoded child with
+/// 256-row runs (16 distinct values, one run in eight null). `null_every`
+/// controls the frequency of null list rows.
+fn create_list_of_int32_ree_batch(size: usize, null_every: Option<usize>) ->
RecordBatch {
+ let child_len = size * 4;
+ let num_runs = child_len.div_ceil(256);
+ let run_ends = Int32Array::from_iter_values(
+ (0..num_runs).map(|i| (((i + 1) * 256).min(child_len)) as i32),
+ );
+ let values =
+ Int32Array::from_iter((0..num_runs).map(|i| (i % 8 != 7).then_some((i
% 16) as i32)));
+ let ree: ArrayRef = Arc::new(RunArray::<Int32Type>::try_new(&run_ends,
&values).unwrap());
+ let offsets = OffsetBuffer::from_lengths(std::iter::repeat_n(4usize,
size));
+ let field = Arc::new(Field::new_list_field(ree.data_type().clone(), true));
+ let nulls = null_every.map(|n| NullBuffer::from_iter((0..size).map(|i| i %
n != n - 1)));
+ let list: ArrayRef = Arc::new(ListArray::new(field, offsets, ree, nulls));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "_1",
+ list.data_type().clone(),
+ true,
+ )]));
+ RecordBatch::try_new(schema, vec![list]).unwrap()
+}
+
+/// A list-of-runs batch whose rows alternate between empty and four values,
+/// without null list rows.
+fn create_list_of_int32_ree_alternating_empty_batch(size: usize) ->
RecordBatch {
+ let child_len = size / 2 * 4;
+ let num_runs = child_len.div_ceil(256);
+ let run_ends = Int32Array::from_iter_values(
+ (0..num_runs).map(|i| (((i + 1) * 256).min(child_len)) as i32),
+ );
+ let values =
+ Int32Array::from_iter((0..num_runs).map(|i| (i % 8 != 7).then_some((i
% 16) as i32)));
+ let ree: ArrayRef = Arc::new(RunArray::<Int32Type>::try_new(&run_ends,
&values).unwrap());
+ let offsets = OffsetBuffer::from_lengths((0..size).map(|i| if i % 2 == 0 {
0 } else { 4 }));
+ let field = Arc::new(Field::new_list_field(ree.data_type().clone(), true));
+ let list: ArrayRef = Arc::new(ListArray::new(field, offsets, ree, None));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "_1",
+ list.data_type().clone(),
+ true,
+ )]));
+ RecordBatch::try_new(schema, vec![list]).unwrap()
+}
+
+/// A uniform list-of-runs column under a nullable struct. Parent validity can
+/// divide the child into multiple write ranges even though the list itself has
+/// no null or empty rows.
+fn create_struct_of_list_of_int32_ree_batch(size: usize, nulls: NullBuffer) ->
RecordBatch {
+ let list_batch = create_list_of_int32_ree_batch(size, None);
+ let list = list_batch.column(0).clone();
+ let fields = Fields::from(vec![Field::new("items",
list.data_type().clone(), true)]);
+ let structs: ArrayRef = Arc::new(StructArray::new(fields, vec![list],
Some(nulls)));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "_1",
+ structs.data_type().clone(),
+ true,
+ )]));
+ RecordBatch::try_new(schema, vec![structs]).unwrap()
+}
+
+/// A batch of `size` outer lists of 2 elements over a run-end-encoded child
+/// (128-element runs) whose run values are inner lists of 4 ints.
+fn create_list_of_list_int32_ree_batch(size: usize) -> RecordBatch {
+ let child_len = size * 2;
+ let num_runs = child_len.div_ceil(128);
+ let run_ends = Int32Array::from_iter_values(
+ (0..num_runs).map(|i| (((i + 1) * 128).min(child_len)) as i32),
+ );
+ let inner_values =
+ Int32Array::from_iter_values((0..num_runs * 4).map(|i| ((i / 4) % 16 *
4 + i % 4) as i32));
+ let inner_offsets = OffsetBuffer::from_lengths(std::iter::repeat_n(4usize,
num_runs));
+ let inner_field = Arc::new(Field::new_list_field(DataType::Int32, true));
+ let inner = ListArray::new(inner_field, inner_offsets,
Arc::new(inner_values), None);
+ let ree: ArrayRef = Arc::new(RunArray::<Int32Type>::try_new(&run_ends,
&inner).unwrap());
+ let offsets = OffsetBuffer::from_lengths(std::iter::repeat_n(2usize,
size));
+ let field = Arc::new(Field::new_list_field(ree.data_type().clone(), true));
+ let list: ArrayRef = Arc::new(ListArray::new(field, offsets, ree, None));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "_1",
+ list.data_type().clone(),
+ true,
+ )]));
+ RecordBatch::try_new(schema, vec![list]).unwrap()
+}
+
+/// `RunEndEncoded<List<RunEndEncoded<List<Int32>>>>` with 256-row outer runs,
+/// two inner entries per physical outer value, and 128-element inner runs.
+fn create_ree_list_ree_list_batch(size: usize) -> RecordBatch {
+ let outer_runs = size.div_ceil(256);
+ let inner_len = outer_runs * 2;
+ let inner_runs = inner_len.div_ceil(128);
+
+ let values = Int32Array::from_iter_values(
+ (0..inner_runs * 4).map(|i| ((i / 4) % 16 * 4 + i % 4) as i32),
+ );
+ let value_offsets = OffsetBuffer::from_lengths(std::iter::repeat_n(4usize,
inner_runs));
+ let value_field = Arc::new(Field::new_list_field(DataType::Int32, true));
+ let value_lists = ListArray::new(value_field, value_offsets,
Arc::new(values), None);
+
+ let inner_ends = Int32Array::from_iter_values(
+ (0..inner_runs).map(|i| (((i + 1) * 128).min(inner_len)) as i32),
+ );
+ let inner: ArrayRef =
+ Arc::new(RunArray::<Int32Type>::try_new(&inner_ends,
&value_lists).unwrap());
+ let outer_value_offsets =
OffsetBuffer::from_lengths(std::iter::repeat_n(2usize, outer_runs));
+ let outer_value_field =
Arc::new(Field::new_list_field(inner.data_type().clone(), true));
+ let outer_values: ArrayRef = Arc::new(ListArray::new(
+ outer_value_field,
+ outer_value_offsets,
+ inner,
+ None,
+ ));
+
+ let outer_ends =
+ Int32Array::from_iter_values((0..outer_runs).map(|i| (((i + 1) *
256).min(size)) as i32));
+ let array: ArrayRef =
+ Arc::new(RunArray::<Int32Type>::try_new(&outer_ends,
&outer_values).unwrap());
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "_1",
+ array.data_type().clone(),
+ true,
+ )]));
+ RecordBatch::try_new(schema, vec![array]).unwrap()
+}
+
+/// List run values for the non-leaf REE benches: each run value is a list of
+/// 8 ints drawn from 16 distinct lists.
+fn make_ree_list_values(n: usize) -> ArrayRef {
+ let values =
+ Int32Array::from_iter_values((0..n * 8).map(|i| ((i / 8) % 16 * 8 + i
% 8) as i32));
+ let offsets = OffsetBuffer::from_lengths(std::iter::repeat_n(8usize, n));
+ let field = Arc::new(Field::new_list_field(DataType::Int32, true));
+ Arc::new(ListArray::new(field, offsets, Arc::new(values), None))
+}
+
+fn create_string_dictionary_bench_batch_1pct_cardinality(
+ size: usize,
+ null_density: f32,
+) -> Result<RecordBatch> {
+ let cardinality = dictionary_cardinality_1pct(size);
+ let schema = dictionary_schema(DataType::Utf8);
+ let keys = dictionary_keys(size, cardinality, null_density);
+ let values = StringArray::from_iter_values((0..cardinality).map(|i|
format!("value_{i:08}")));
+ let array = DictionaryArray::<Int32Type>::new(keys, Arc::new(values));
+
+ Ok(RecordBatch::try_new(
+ schema,
+ vec![Arc::new(array) as ArrayRef],
+ )?)
+}
+
+fn dictionary_schema(value_type: DataType) -> Arc<Schema> {
+ Arc::new(Schema::new(vec![Field::new(
+ "_1",
+ DataType::Dictionary(Box::new(DataType::Int32), Box::new(value_type)),
+ true,
+ )]))
+}
+
+fn dictionary_cardinality_1pct(size: usize) -> usize {
+ (size / 100).max(1)
+}
+
+fn dictionary_keys(size: usize, cardinality: usize, null_density: f32) ->
Int32Array {
+ let keys = (0..size)
+ .map(|i| (i % cardinality) as i32)
+ .collect::<Vec<_>>();
+ Int32Array::new(keys.into(), nulls_for_density(size, null_density))
+}
+
+fn nulls_for_density(size: usize, null_density: f32) -> Option<NullBuffer> {
+ if null_density == 0. {
+ return None;
+ }
+
+ let null_threshold = (null_density.clamp(0., 1.) * 10_000.) as usize;
+ Some(NullBuffer::from(
+ (0..size)
+ .map(|i| (i * 9973) % 10_000 >= null_threshold)
+ .collect::<Vec<_>>(),
+ ))
+}
+
fn create_string_bench_batch_non_null(
size: usize,
null_density: f32,
@@ -539,8 +951,6 @@ fn create_nested_list_bench_batch(size: usize,
null_density: f32) -> Result<Reco
fn create_list_struct_with_list_batch(size: usize, null_density: f32) ->
Result<RecordBatch> {
// List<Struct<a:Int32, b:Float32, c:List<Int32>>>
- // The struct child contains a nested list, so child_has_no_nested_rep() =
false.
- // This exercises the per-slot (non-batched) write path in level
computation.
let fields = vec![Field::new(
"_1",
DataType::List(Arc::new(Field::new_list_field(
@@ -653,111 +1063,492 @@ fn write_batch_with_option(
Ok(())
}
-fn create_batches() -> Vec<(&'static str, RecordBatch)> {
- const BATCH_SIZE: usize = 1024 * 1024;
-
- let mut batches = vec![];
-
- let batch = create_primitive_bench_batch(BATCH_SIZE, 0.25, 0.75).unwrap();
- batches.push(("primitive", batch));
-
- let batch = create_primitive_bench_batch_non_null(BATCH_SIZE, 0.25,
0.75).unwrap();
- batches.push(("primitive_non_null", batch));
-
- let batch = create_bool_bench_batch(BATCH_SIZE, 0.25, 0.75).unwrap();
- batches.push(("bool", batch));
-
- let batch = create_bool_bench_batch_non_null(BATCH_SIZE, 0.25,
0.75).unwrap();
- batches.push(("bool_non_null", batch));
-
- let batch = create_string_bench_batch(BATCH_SIZE, 0.25, 0.75).unwrap();
- batches.push(("string", batch));
-
- let batch = create_short_string_bench_batch(BATCH_SIZE).unwrap();
- batches.push(("short_string_non_null", batch));
-
- // 1024 rows × 256 KiB = 256 MiB total. With the default 1 MiB page byte
- // limit, this is the case where the page-size fix kicks in: each value
- // needs its own page, and `write_batch_size = 1024` would otherwise
- // buffer all 256 MiB before the post-write check runs.
- let batch = create_large_string_bench_batch(1024, 256 * 1024).unwrap();
- batches.push(("large_string_non_null", batch));
-
- let batch = create_string_and_binary_view_bench_batch(BATCH_SIZE, 0.25,
0.75).unwrap();
- batches.push(("string_and_binary_view", batch));
-
- let batch = create_string_dictionary_bench_batch(BATCH_SIZE, 0.25,
0.75).unwrap();
- batches.push(("string_dictionary", batch));
-
- let batch = create_low_card_dictionary_bench_batch(BATCH_SIZE,
20).unwrap();
- batches.push(("string_dictionary_low_cardinality_20", batch));
-
- let batch = create_low_card_dictionary_bench_batch(BATCH_SIZE,
100).unwrap();
- batches.push(("string_dictionary_low_cardinality_100", batch));
-
- let batch = create_low_card_dictionary_bench_batch(BATCH_SIZE,
400).unwrap();
- batches.push(("string_dictionary_low_cardinality_400", batch));
-
- let batch = create_string_bench_batch_non_null(BATCH_SIZE, 0.25,
0.75).unwrap();
- batches.push(("string_non_null", batch));
-
- let batch = create_ree_bench_batch(DataType::Utf8, BATCH_SIZE, None,
0.75).unwrap();
- batches.push(("string_ree", batch));
-
- let batch = create_ree_bench_batch(DataType::Int32, BATCH_SIZE, None,
0.75).unwrap();
- batches.push(("int32_ree", batch));
-
- let batch = create_ree_bench_batch(DataType::Boolean, BATCH_SIZE, None,
0.75).unwrap();
- batches.push(("bool_ree", batch));
-
- let batch =
- create_ree_bench_batch(DataType::FixedSizeBinary(16), BATCH_SIZE,
None, 0.75).unwrap();
- batches.push(("fixed_size_binary_ree", batch));
-
- let batch = create_ree_bench_batch(DataType::Utf8, BATCH_SIZE, Some(95),
0.75).unwrap();
- batches.push(("string_ree_95pct_null", batch));
+/// High-null byte-array columns (Utf8 / LargeUtf8 / Utf8View / BinaryView).
+/// Varies sparsity across the supported byte-array representations.
+fn create_byte_array_sparse_bench_batch(size: usize, null_density: f32) ->
Result<RecordBatch> {
+ let fields = vec![
+ Field::new("_1", DataType::Utf8, true),
+ Field::new("_2", DataType::LargeUtf8, true),
+ Field::new("_3", DataType::Utf8View, true),
+ Field::new("_4", DataType::BinaryView, true),
+ ];
+ let schema = Schema::new(fields);
+ Ok(create_random_batch(
+ Arc::new(schema),
+ size,
+ null_density,
+ 0.5,
+ )?)
+}
- let batch = create_ree_bench_batch(DataType::Int32, BATCH_SIZE, Some(95),
0.75).unwrap();
- batches.push(("int32_ree_95pct_null", batch));
+/// FixedSizeBinary column mapped to FIXED_LEN_BYTE_ARRAY, with configurable
+/// null density.
+fn create_fixed_size_binary_bench_batch(
+ size: usize,
+ byte_width: i32,
+ null_density: f32,
+) -> Result<RecordBatch> {
+ let width = byte_width as usize;
+ let nulls = nulls_for_density(size, null_density);
+ let array = FixedSizeBinaryArray::try_from_sparse_iter_with_size(
+ (0..size).map(|i| {
+ nulls.as_ref().is_none_or(|n| n.is_valid(i)).then(|| {
+ let mut v = vec![0u8; width];
+ let b = (i as u64).to_le_bytes();
+ let n = width.min(8);
+ v[..n].copy_from_slice(&b[..n]);
+ v
+ })
+ }),
+ byte_width,
+ )?;
+ Ok(RecordBatch::try_from_iter([(
+ "col",
+ Arc::new(array) as ArrayRef,
+ )])?)
+}
- let batch = create_float_bench_batch_with_nans(BATCH_SIZE, 0.5).unwrap();
- batches.push(("float_with_nans", batch));
+/// Decimal128(38, 10) column mapped to FIXED_LEN_BYTE_ARRAY(16), with
+/// configurable null density. Precision 38 requires fixed-width storage.
+fn create_decimal128_bench_batch(size: usize, null_density: f32) ->
Result<RecordBatch> {
+ let values: Vec<i128> = (0..size).map(|i| i as i128).collect();
+ let nulls = nulls_for_density(size, null_density);
+ let array = Decimal128Array::new(values.into(),
nulls).with_precision_and_scale(38, 10)?;
+ Ok(RecordBatch::try_from_iter([(
+ "col",
+ Arc::new(array) as ArrayRef,
+ )])?)
+}
- let batch = create_decimal_bench_batch(BATCH_SIZE, 0.75).unwrap();
- batches.push(("decimal", batch));
+fn create_decimal128_dictionary_bench_batch(
+ size: usize,
+ null_density: f32,
+ true_density: f32,
+) -> Result<RecordBatch> {
+ Ok(create_random_batch(
+ dictionary_schema(DataType::Decimal128(38, 10)),
+ size,
+ null_density,
+ true_density,
+ )?)
+}
- let batch = create_list_primitive_bench_batch(BATCH_SIZE, 0.25,
0.75).unwrap();
- batches.push(("list_primitive", batch));
+fn create_decimal128_dictionary_bench_batch_1pct_cardinality(
+ size: usize,
+ null_density: f32,
+) -> Result<RecordBatch> {
+ let cardinality = dictionary_cardinality_1pct(size);
+ let keys = dictionary_keys(size, cardinality, null_density);
+ let values = Decimal128Array::from_iter_values((0..cardinality).map(|i| i
as i128))
+ .with_precision_and_scale(38, 10)?;
+ let array: ArrayRef = Arc::new(DictionaryArray::<Int32Type>::new(keys,
Arc::new(values)));
+ Ok(RecordBatch::try_from_iter([("col", array)])?)
+}
- let batch = create_list_primitive_bench_batch_non_null(BATCH_SIZE, 0.25,
0.75).unwrap();
- batches.push(("list_primitive_non_null", batch));
+struct BatchBenchmark {
+ name: &'static str,
+ logical_rows: u64,
+ make_batch: Box<dyn Fn() -> RecordBatch>,
+}
- let batch = create_primitive_bench_batch(BATCH_SIZE, 0.99, 0.75).unwrap();
- batches.push(("primitive_sparse_99pct_null", batch));
+impl BatchBenchmark {
+ fn new(
+ name: &'static str,
+ logical_rows: usize,
+ make_batch: impl Fn() -> RecordBatch + 'static,
+ ) -> Self {
+ Self {
+ name,
+ logical_rows: logical_rows as u64,
+ make_batch: Box::new(make_batch),
+ }
+ }
+}
- let batch = create_list_primitive_bench_batch(BATCH_SIZE, 0.99,
0.75).unwrap();
- batches.push(("list_primitive_sparse_99pct_null", batch));
+fn create_batches() -> Vec<BatchBenchmark> {
+ const BATCH_SIZE: usize = 1024 * 1024;
- let batch = create_primitive_bench_batch(BATCH_SIZE, 1.0, 0.75).unwrap();
- batches.push(("primitive_all_null", batch));
+ let mut batches = vec![];
- let batch = create_struct_bench_batch(BATCH_SIZE, 0.0).unwrap();
- batches.push(("struct_non_null", batch));
+ macro_rules! push_batch {
+ ($name:expr, $batch:expr) => {
+ batches.push(BatchBenchmark::new($name, BATCH_SIZE, || $batch));
+ };
+ ($name:expr, $logical_rows:expr, $batch:expr) => {
+ batches.push(BatchBenchmark::new($name, $logical_rows, || $batch));
+ };
+ }
- let batch = create_struct_bench_batch(BATCH_SIZE, 0.99).unwrap();
- batches.push(("struct_sparse_99pct_null", batch));
+ push_batch!(
+ "primitive",
+ create_primitive_bench_batch(BATCH_SIZE, 0.25, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "primitive_non_null",
+ create_primitive_bench_batch_non_null(BATCH_SIZE, 0.25, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "primitive_dictionary",
+ create_primitive_dictionary_bench_batch(BATCH_SIZE, 0.25,
0.75).unwrap()
+ );
+
+ push_batch!(
+ "primitive_dictionary_1pct_cardinality",
+ create_primitive_dictionary_bench_batch_1pct_cardinality(BATCH_SIZE,
0.25).unwrap()
+ );
+
+ push_batch!(
+ "int64_dictionary",
+ create_int64_dictionary_bench_batch(BATCH_SIZE, 0.25, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "int64_dictionary_1pct_cardinality",
+ create_int64_dictionary_bench_batch_1pct_cardinality(BATCH_SIZE,
0.25).unwrap()
+ );
+
+ push_batch!(
+ "float64_dictionary",
+ create_float64_dictionary_bench_batch(BATCH_SIZE, 0.25, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "float64_dictionary_1pct_cardinality",
+ create_float64_dictionary_bench_batch_1pct_cardinality(BATCH_SIZE,
0.25).unwrap()
+ );
+
+ push_batch!(
+ "bool",
+ create_bool_bench_batch(BATCH_SIZE, 0.25, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "bool_non_null",
+ create_bool_bench_batch_non_null(BATCH_SIZE, 0.25, 0.75).unwrap()
+ );
+
+ // Dictionary-encoded booleans exercise the native bool-dictionary write
path.
+ push_batch!("bool_dictionary", {
+ let keys = Int8Array::from_iter((0..BATCH_SIZE).map(|i| {
+ (i % 10 != 9).then_some((i % 2) as i8) // 10% null keys
+ }));
+ let values = BooleanArray::from(vec![false, true]);
+ let dict: ArrayRef = Arc::new(DictionaryArray::new(keys,
Arc::new(values) as ArrayRef));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "_1",
+ dict.data_type().clone(),
+ true,
+ )]));
+ RecordBatch::try_new(schema, vec![dict]).unwrap()
+ });
- let batch = create_struct_bench_batch(BATCH_SIZE, 1.0).unwrap();
- batches.push(("struct_all_null", batch));
+ push_batch!(
+ "string",
+ create_string_bench_batch(BATCH_SIZE, 0.25, 0.75).unwrap()
+ );
- let batch = create_fsb_bench_batch(BATCH_SIZE, 0.9, 16).unwrap();
- batches.push(("fsb", batch));
+ push_batch!(
+ "short_string_non_null",
+ create_short_string_bench_batch(BATCH_SIZE).unwrap()
+ );
- let batch = create_nested_list_bench_batch(BATCH_SIZE, 0.25).unwrap();
- batches.push(("list_nested", batch));
+ // 1024 rows × 256 KiB = 256 MiB total. With the default 1 MiB page byte
+ // limit, byte-budget sub-batching avoids admitting all 256 MiB before
+ // the post-write check when `write_batch_size = 1024`.
+ push_batch!(
+ "large_string_non_null",
+ 1024,
+ create_large_string_bench_batch(1024, 256 * 1024).unwrap()
+ );
+
+ push_batch!(
+ "string_and_binary_view",
+ create_string_and_binary_view_bench_batch(BATCH_SIZE, 0.25,
0.75).unwrap()
+ );
+
+ push_batch!(
+ "string_dictionary",
+ create_string_dictionary_bench_batch(BATCH_SIZE, 0.25, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "string_dictionary_low_cardinality_20",
+ create_low_card_dictionary_bench_batch(BATCH_SIZE, 20).unwrap()
+ );
+ push_batch!(
+ "string_dictionary_low_cardinality_100",
+ create_low_card_dictionary_bench_batch(BATCH_SIZE, 100).unwrap()
+ );
+ push_batch!(
+ "string_dictionary_low_cardinality_400",
+ create_low_card_dictionary_bench_batch(BATCH_SIZE, 400).unwrap()
+ );
+ push_batch!(
+ "string_dictionary_1pct_cardinality",
+ create_string_dictionary_bench_batch_1pct_cardinality(BATCH_SIZE,
0.25).unwrap()
+ );
+
+ push_batch!(
+ "string_non_null",
+ create_string_bench_batch_non_null(BATCH_SIZE, 0.25, 0.75).unwrap()
+ );
+
+ // Run-end-encoded — high-cardinality / short-run regime (random runs).
+ push_batch!(
+ "string_ree",
+ create_ree_bench_batch(DataType::Utf8, BATCH_SIZE, None, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "int32_ree",
+ create_ree_bench_batch(DataType::Int32, BATCH_SIZE, None,
0.75).unwrap()
+ );
+
+ push_batch!(
+ "string_ree_of_dict",
+ create_ree_of_dict_bench_batch(BATCH_SIZE)
+ );
+
+ push_batch!(
+ "numeric_ree_of_dict",
+ create_ree_of_numeric_dict_bench_batch(BATCH_SIZE)
+ );
+
+ push_batch!(
+ "ree_struct_of_dict",
+ create_ree_struct_of_dict_bench_batch(BATCH_SIZE)
+ );
+
+ push_batch!(
+ "bool_ree",
+ create_ree_bench_batch(DataType::Boolean, BATCH_SIZE, None,
0.75).unwrap()
+ );
+
+ push_batch!(
+ "decimal",
+ create_decimal_bench_batch(BATCH_SIZE, 0.75).unwrap()
+ );
+ push_batch!(
+ "fixed_size_binary_ree",
+ create_ree_bench_batch(DataType::FixedSizeBinary(16), BATCH_SIZE,
None, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "string_ree_95pct_null",
+ create_ree_bench_batch(DataType::Utf8, BATCH_SIZE, Some(95),
0.75).unwrap()
+ );
+
+ push_batch!(
+ "int32_ree_95pct_null",
+ create_ree_bench_batch(DataType::Int32, BATCH_SIZE, Some(95),
0.75).unwrap()
+ );
+
+ // Run-end-encoded batches with 256-row runs and 16 distinct values, across
+ // value families and run-end widths.
+ push_batch!(
+ "string_ree_low_cardinality",
+ create_run_end_encoded_bench_batch(DataType::Int32, BATCH_SIZE, 256,
|n| {
+ Arc::new(StringArray::from_iter_values(
+ (0..n).map(|i| format!("category_{:02}", i % 16)),
+ ))
+ })
+ );
+ // String run values with Int64 run ends.
+ push_batch!(
+ "string_ree_int64_run_ends",
+ create_run_end_encoded_bench_batch(DataType::Int64, BATCH_SIZE, 256,
|n| {
+ Arc::new(StringArray::from_iter_values(
+ (0..n).map(|i| format!("category_{:02}", i % 16)),
+ ))
+ })
+ );
+ // Numeric (Int64) run values.
+ push_batch!(
+ "int64_ree_low_cardinality",
+ create_run_end_encoded_bench_batch(DataType::Int32, BATCH_SIZE, 256,
|n| {
+ Arc::new(Int64Array::from_iter_values(
+ (0..n).map(|i| (i % 16) as i64),
+ ))
+ })
+ );
+ // Fixed-length byte-array run values (FLBA path).
+ push_batch!(
+ "fixed_size_binary_ree_low_cardinality",
+ create_run_end_encoded_bench_batch(DataType::Int32, BATCH_SIZE, 256,
|n| {
+ Arc::new(
+ FixedSizeBinaryArray::try_from_iter((0..n).map(|i| {
+ let mut b = [0u8; 16];
+ b[0] = (i % 16) as u8;
+ b
+ }))
+ .unwrap(),
+ )
+ })
+ );
- let batch = create_list_struct_with_list_batch(BATCH_SIZE, 0.25).unwrap();
- batches.push(("list_struct_with_list", batch));
+ // Non-leaf run values: struct-of-leaves with long runs.
+ push_batch!(
+ "struct_ree",
+ create_run_end_encoded_bench_batch(DataType::Int32, BATCH_SIZE, 256,
|n| {
+ make_ree_struct_values(n)
+ })
+ );
+ // Non-leaf struct run values with short runs (per-run overhead regime).
+ push_batch!(
+ "struct_ree_short_runs",
+ create_run_end_encoded_bench_batch(DataType::Int32, BATCH_SIZE, 8, |n|
{
+ make_ree_struct_values(n)
+ })
+ );
+ // Non-leaf run values: each run value is a list of 8 ints (block regime).
+ push_batch!(
+ "list_int32_ree",
+ create_run_end_encoded_bench_batch(DataType::Int32, BATCH_SIZE, 256,
|n| {
+ make_ree_list_values(n)
+ })
+ );
+ push_batch!(
+ "list_int32_ree_short_runs",
+ create_run_end_encoded_bench_batch(DataType::Int32, BATCH_SIZE, 8, |n|
{
+ make_ree_list_values(n)
+ })
+ );
+ // Byte-array leaves under run-encoded list values.
+ push_batch!(
+ "list_string_ree",
+ create_run_end_encoded_bench_batch(DataType::Int32, BATCH_SIZE, 256,
|n| {
+ make_ree_string_list_values(n)
+ })
+ );
+ // Lists whose run-end-encoded child contains lists.
+ push_batch!(
+ "list_of_list_int32_ree",
+ create_list_of_list_int32_ree_batch(BATCH_SIZE)
+ );
+ push_batch!(
+ "ree_list_ree_list",
+ create_ree_list_ree_list_batch(BATCH_SIZE)
+ );
+ // A list whose values child is run-end encoded.
+ push_batch!(
+ "list_of_int32_ree",
+ create_list_of_int32_ree_batch(BATCH_SIZE, None)
+ );
+ // Alternating null and valid list rows.
+ push_batch!(
+ "list_of_int32_ree_alt_null",
+ create_list_of_int32_ree_batch(BATCH_SIZE, Some(2))
+ );
+ push_batch!(
+ "list_of_int32_ree_alt_empty",
+ create_list_of_int32_ree_alternating_empty_batch(BATCH_SIZE)
+ );
+ push_batch!(
+ "struct_of_list_of_int32_ree_alt_null",
+ create_struct_of_list_of_int32_ree_batch(
+ BATCH_SIZE,
+ NullBuffer::from_iter((0..BATCH_SIZE).map(|i| i % 2 == 0)),
+ )
+ );
+ push_batch!(
+ "struct_of_list_of_int32_ree_single_null",
+ create_struct_of_list_of_int32_ree_batch(
+ BATCH_SIZE,
+ NullBuffer::from_iter((0..BATCH_SIZE).map(|i| i != BATCH_SIZE /
2)),
+ )
+ );
+ // Alternate ordinary and run-end-encoded input batches.
+ push_batch!(
+ "struct_of_list_of_int32_ree_fragmented",
+ create_struct_of_list_of_int32_ree_batch(
+ BATCH_SIZE,
+ NullBuffer::from_iter(
+ (0..BATCH_SIZE).map(|i| i != 0 && i != BATCH_SIZE / 2 && i + 1
!= BATCH_SIZE),
+ ),
+ )
+ );
+
+ push_batch!(
+ "float_with_nans",
+ create_float_bench_batch_with_nans(BATCH_SIZE, 0.5).unwrap()
+ );
+
+ push_batch!(
+ "list_primitive",
+ create_list_primitive_bench_batch(BATCH_SIZE, 0.25, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "list_primitive_non_null",
+ create_list_primitive_bench_batch_non_null(BATCH_SIZE, 0.25,
0.75).unwrap()
+ );
+
+ push_batch!("fsb", create_fsb_bench_batch(BATCH_SIZE, 0.9, 16).unwrap());
+ push_batch!(
+ "primitive_sparse_99pct_null",
+ create_primitive_bench_batch(BATCH_SIZE, 0.99, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "list_primitive_sparse_99pct_null",
+ create_list_primitive_bench_batch(BATCH_SIZE, 0.99, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "primitive_all_null",
+ create_primitive_bench_batch(BATCH_SIZE, 1.0, 0.75).unwrap()
+ );
+
+ push_batch!(
+ "struct_non_null",
+ create_struct_bench_batch(BATCH_SIZE, 0.0).unwrap()
+ );
+
+ push_batch!(
+ "struct_sparse_99pct_null",
+ create_struct_bench_batch(BATCH_SIZE, 0.99).unwrap()
+ );
+
+ push_batch!(
+ "struct_all_null",
+ create_struct_bench_batch(BATCH_SIZE, 1.0).unwrap()
+ );
+
+ push_batch!(
+ "list_nested",
+ create_nested_list_bench_batch(BATCH_SIZE, 0.25).unwrap()
+ );
+
+ push_batch!(
+ "list_struct_with_list",
+ create_list_struct_with_list_batch(BATCH_SIZE, 0.25).unwrap()
+ );
+
+ push_batch!(
+ "byte_array_sparse_99pct_null",
+ create_byte_array_sparse_bench_batch(BATCH_SIZE, 0.99).unwrap()
+ );
+
+ push_batch!(
+ "fixed_size_binary_non_null",
+ create_fixed_size_binary_bench_batch(BATCH_SIZE, 16, 0.0).unwrap()
+ );
+
+ push_batch!(
+ "decimal128_sparse",
+ create_decimal128_bench_batch(BATCH_SIZE, 0.25).unwrap()
+ );
+
+ push_batch!(
+ "decimal128_dictionary",
+ create_decimal128_dictionary_bench_batch(BATCH_SIZE, 0.25,
0.75).unwrap()
+ );
+ push_batch!(
+ "decimal128_dictionary_1pct_cardinality",
+ create_decimal128_dictionary_bench_batch_1pct_cardinality(BATCH_SIZE,
0.25).unwrap()
+ );
batches
}
@@ -805,18 +1596,21 @@ fn bench_all_writers(c: &mut Criterion) {
let batches = create_batches();
let props = create_writer_props();
- for (batch_name, batch) in &batches {
- let mut group = c.benchmark_group(*batch_name);
- group.throughput(Throughput::Bytes(
- batch
- .columns()
- .iter()
- .map(|f| f.get_array_memory_size() as u64)
- .sum(),
- ));
+ for benchmark in &batches {
+ let mut group = c.benchmark_group(benchmark.name);
+
+ // Arrow's physical buffer size is not comparable between a dense
+ // array and its run-end encoded equivalent. Logical rows are stable
+ // across representations and don't require eagerly building input.
+ group.throughput(Throughput::Elements(benchmark.logical_rows));
+ // Criterion doesn't invoke a benchmark closure for a filtered-out
+ // ID. Once any property is selected, retain one input for the whole
+ // workload group without including its construction in `b.iter`.
+ let batch = OnceCell::new();
for (prop_name, prop) in &props {
group.bench_function(*prop_name, |b| {
+ let batch = batch.get_or_init(|| (benchmark.make_batch)());
write_batch_with_option(b, batch, Some(prop.clone())).unwrap()
});
}
