You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/12 17:10:39 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-5805-de1eb4f86c12af0895784c93e1f152c705f6ef0e created (now da266854dc)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-5805-de1eb4f86c12af0895784c93e1f152c705f6ef0e
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at da266854dc bench: add a shuffle read benchmark covering the per-block schema parse (#5805)
This branch includes the following new commits:
new da266854dc bench: add a shuffle read benchmark covering the per-block schema parse (#5805)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: bench: add a shuffle read benchmark covering the per-block schema parse (#5805)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5805-de1eb4f86c12af0895784c93e1f152c705f6ef0e
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit da266854dcfb395e4ff4da882c800c4352762fed
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Sat Sep 12 16:55:46 2026 +0000
bench: add a shuffle read benchmark covering the per-block schema parse (#5805)
* bench: add a shuffle read benchmark covering the per-block schema parse
Every shuffle block is a self-contained Arrow IPC stream, so read_single_batch
builds a fresh StreamReader per block and parses the schema flatbuffer once per
block, even though every block in a shuffle carries the same schema. The write
side already avoids the mirror image of this, encoding the schema once in
ShuffleBlockWriter::try_new and writing the pre-encoded bytes verbatim, but there
was no read-side benchmark to say whether the reader's half is worth removing.
This adds one, parameterized by column count and rows per block, measuring the
schema parse separately from the full block decode. On an M-series laptop:
shape decode schema parse share
5 col x 64 row 1.93 us 1.14 us 59%
5 col x 512 row 2.38 us 0.91 us 38%
5 col x 8192 row 10.99 us 0.86 us 8%
50 col x 64 row 12.77 us 6.03 us 47%
50 col x 512 row 17.89 us 6.05 us 34%
50 col x 8192 row 218 us 6.05 us 3%
The parse cost is constant per block and independent of row count, so its share
is set by how many rows land in a block. That is largest exactly where the issue
predicted: wide shuffles, where rows per partition are few, and repeated
spilling, where each spill round emits its own block per partition.
Co-Authored-By: Claude Opus 5 <no...@anthropic.com>
* review: trim comments to what they need to say
Co-Authored-By: Claude Opus 5 <no...@anthropic.com>
---------
Co-authored-by: Claude Opus 5 <no...@anthropic.com>
Co-authored-by: Andy Grove <ag...@apache.org>
---
native/shuffle/Cargo.toml | 4 +
native/shuffle/benches/shuffle_reader.rs | 121 +++++++++++++++++++++++++++++++
2 files changed, 125 insertions(+)
diff --git a/native/shuffle/Cargo.toml b/native/shuffle/Cargo.toml
index 71be932422..9504834ef4 100644
--- a/native/shuffle/Cargo.toml
+++ b/native/shuffle/Cargo.toml
@@ -78,3 +78,7 @@ harness = false
[[bench]]
name = "row_columnar"
harness = false
+
+[[bench]]
+name = "shuffle_reader"
+harness = false
diff --git a/native/shuffle/benches/shuffle_reader.rs b/native/shuffle/benches/shuffle_reader.rs
new file mode 100644
index 0000000000..81efcda7aa
--- /dev/null
+++ b/native/shuffle/benches/shuffle_reader.rs
@@ -0,0 +1,121 @@
+// 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.
+
+//! Shuffle read benchmarks: the per-block schema parse measured against a full block decode,
+//! across column counts and rows per block.
+
+use arrow::array::{Int64Array, RecordBatch, StringArray};
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use arrow::ipc::reader::StreamReader;
+use arrow::ipc::writer::IpcWriteContext;
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
+use datafusion::physical_plan::metrics::Time;
+use datafusion_comet_shuffle::{read_ipc_compressed, CompressionCodec, ShuffleBlockWriter};
+use std::hint::black_box;
+use std::io::Cursor;
+use std::sync::Arc;
+
+/// 8-byte compressed length plus 8-byte field count; `read_ipc_compressed` expects what follows.
+const BLOCK_HEADER_LEN: usize = 16;
+
+/// Alternating `Int64` and `Utf8`.
+fn schema_of(num_columns: usize) -> SchemaRef {
+ Arc::new(Schema::new(
+ (0..num_columns)
+ .map(|i| {
+ let data_type = if i % 2 == 0 {
+ DataType::Int64
+ } else {
+ DataType::Utf8
+ };
+ Field::new(format!("column_{i}"), data_type, false)
+ })
+ .collect::<Vec<_>>(),
+ ))
+}
+
+fn batch_of(num_columns: usize, num_rows: usize) -> RecordBatch {
+ let schema = schema_of(num_columns);
+ let columns = (0..num_columns)
+ .map(|i| {
+ if i % 2 == 0 {
+ Arc::new(
+ (0..num_rows)
+ .map(|r| Some(r as i64))
+ .collect::<Int64Array>(),
+ ) as arrow::array::ArrayRef
+ } else {
+ Arc::new(
+ (0..num_rows)
+ .map(|r| Some(format!("value_{r}")))
+ .collect::<StringArray>(),
+ ) as arrow::array::ArrayRef
+ }
+ })
+ .collect::<Vec<_>>();
+ RecordBatch::try_new(schema, columns).unwrap()
+}
+
+/// One encoded block, with the 16-byte Comet header stripped.
+fn encode_block(batch: &RecordBatch, codec: CompressionCodec) -> Vec<u8> {
+ let writer = ShuffleBlockWriter::try_new(batch.schema().as_ref(), codec).unwrap();
+ let mut context = IpcWriteContext::default();
+ let mut buffer = Vec::new();
+ let mut cursor = Cursor::new(&mut buffer);
+ writer
+ .write_batch(batch, &mut cursor, &mut context, &Time::default())
+ .unwrap();
+ buffer[BLOCK_HEADER_LEN..].to_vec()
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+ let mut group = c.benchmark_group("shuffle_reader");
+
+ // rows per block shrink as partition count rises, so the small cases stand in for wide shuffles
+ for num_columns in [5usize, 50] {
+ for num_rows in [64usize, 512, 8192] {
+ let batch = batch_of(num_columns, num_rows);
+ let uncompressed = encode_block(&batch, CompressionCodec::None);
+
+ let id = format!("{num_columns}col_{num_rows}row");
+
+ // full decode: schema parse plus record batch
+ group.bench_with_input(
+ BenchmarkId::new("decode_block", &id),
+ &uncompressed,
+ |b, block| b.iter(|| black_box(read_ipc_compressed(black_box(block)).unwrap())),
+ );
+
+ // schema parse alone: `try_new` stops before the record batch. Skips the codec tag.
+ group.bench_with_input(
+ BenchmarkId::new("parse_schema_only", &id),
+ &uncompressed,
+ |b, block| {
+ b.iter(|| {
+ let mut ipc = &black_box(block)[4..];
+ black_box(StreamReader::try_new(&mut ipc, None).unwrap().schema())
+ })
+ },
+ );
+ }
+ }
+
+ group.finish();
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org