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/10/05 11:42:24 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-5804-a4e72fd9d4e3bb8f459fc7950a98536724daf859 created (now 4b679c68c5)
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-5804-a4e72fd9d4e3bb8f459fc7950a98536724daf859
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 4b679c68c5 fix: keep parquet view types out of the shuffle benchmark input (#5804)
This branch includes the following new commits:
new 4b679c68c5 fix: keep parquet view types out of the shuffle benchmark input (#5804)
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: fix: keep parquet view types out of the shuffle benchmark input (#5804)
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-5804-a4e72fd9d4e3bb8f459fc7950a98536724daf859
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 4b679c68c54ae1171fdb651adbd0b7c43932cde0
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Mon Oct 5 11:41:59 2026 +0000
fix: keep parquet view types out of the shuffle benchmark input (#5804)
* fix: keep parquet view types out of the shuffle benchmark input
shuffle_bench built its input with a default SessionConfig, and DataFusion
defaults schema_force_view_types to true, so string and binary columns reached
ShuffleWriterExec as Utf8View and BinaryView. Comet does not hand the shuffle
writer those types: the serde maps Spark String to Utf8, and the planner casts
UDF results back from the view variants. The benchmark was measuring the writer
on a data shape production does not produce.
View arrays are also the input this writer handles worst. BufBatchWriter
configures its BatchCoalescer with biggest_coalesce_batch_size of batch_size - 1,
so a batch at or above batch_size rows bypasses the coalescer and is serialized
verbatim, and an interleaved view array still references the backing data buffers
of every input batch it drew rows from. On a 5 column input with 2 string
columns, 4M rows and 200 partitions, the output file measured against the
writer's own data size metric was:
batch size 2048 33.17 GiB / 236.95 MiB (~143x)
batch size 4096 14.86 GiB / 242.07 MiB (~63x)
batch size 8192 7.62 GiB / 242.33 MiB (~32x)
batch size 16384 4.00 GiB / 243.96 MiB (~17x)
batch size 20480 294.92 MiB / 245.45 MiB (~1.2x)
The cliff at 20480 is where a partition's 20000 rows stop exceeding the
threshold and go through the coalescer, which compacts the views. With view
types disabled the output is flat at ~238 MiB across that whole sweep and encode
time at batch size 2048 drops from 7.580s to 0.019s.
The header schema is read separately through ParquetRecordBatchReaderBuilder, so
it reported the file's Utf8 rather than the Utf8View that actually executed and
the mismatch was invisible. Check the schema that reaches the writer instead, and
name the view types in describe_schema rather than folding them into "other".
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>
---
native/shuffle/src/bin/shuffle_bench.rs | 113 +++++++++++++++++++++++++++++++-
1 file changed, 111 insertions(+), 2 deletions(-)
diff --git a/native/shuffle/src/bin/shuffle_bench.rs b/native/shuffle/src/bin/shuffle_bench.rs
index c2c9db30f0..ec638fbf96 100644
--- a/native/shuffle/src/bin/shuffle_bench.rs
+++ b/native/shuffle/src/bin/shuffle_bench.rs
@@ -35,7 +35,7 @@
//! --partitions 200 --codec lz4
//! ```
-use arrow::datatypes::{DataType, SchemaRef};
+use arrow::datatypes::{DataType, Schema, SchemaRef};
use clap::Parser;
use datafusion::execution::config::SessionConfig;
use datafusion::execution::runtime_env::RuntimeEnvBuilder;
@@ -442,7 +442,19 @@ async fn execute_shuffle_write(
limit: usize,
data_file: String,
) -> datafusion::common::Result<(MetricsSet, MetricsSet)> {
- let config = SessionConfig::new().with_batch_size(batch_size);
+ let mut config = SessionConfig::new().with_batch_size(batch_size);
+ // Comet never hands the shuffle writer view types: the serde maps Spark `String` to `Utf8`,
+ // and the planner casts UDF results back from `Utf8View`/`BinaryView` to the non-view
+ // variants. DataFusion's parquet reader defaults `schema_force_view_types` to true, which
+ // would feed this benchmark a data shape production never produces -- and one the writer
+ // handles far worse, since a batch at or above `batch_size` rows bypasses the
+ // `BatchCoalescer` and an interleaved view array is serialized with the backing data
+ // buffers of every input batch it drew rows from.
+ config
+ .options_mut()
+ .execution
+ .parquet
+ .schema_force_view_types = false;
let mut runtime_builder = RuntimeEnvBuilder::new();
if let Some(mem_limit) = memory_limit {
runtime_builder = runtime_builder.with_memory_limit(mem_limit, 1.0);
@@ -463,6 +475,9 @@ async fn execute_shuffle_write(
.await
.expect("Failed to create physical plan");
+ // The header schema is read straight from the file, so check what reaches the writer.
+ reject_view_types(&parquet_plan.schema());
+
let input: Arc<dyn ExecutionPlan> = if parquet_plan
.properties()
.output_partitioning()
@@ -626,6 +641,38 @@ fn parse_hash_columns(s: &str) -> Vec<usize> {
.collect()
}
+/// Fails the run if the schema reaching the shuffle writer carries Arrow view types, which
+/// Comet does not produce and this writer serializes with every backing buffer attached.
+fn reject_view_types(schema: &Schema) {
+ fn find_view(data_type: &DataType) -> Option<&DataType> {
+ match data_type {
+ DataType::Utf8View | DataType::BinaryView => Some(data_type),
+ DataType::List(field)
+ | DataType::LargeList(field)
+ | DataType::FixedSizeList(field, _)
+ | DataType::Map(field, _) => find_view(field.data_type()),
+ DataType::Struct(fields) => fields.iter().find_map(|f| find_view(f.data_type())),
+ DataType::Dictionary(_, values) => find_view(values),
+ _ => None,
+ }
+ }
+
+ let offenders = schema
+ .fields()
+ .iter()
+ .filter_map(|field| {
+ find_view(field.data_type()).map(|found| format!("{} ({found})", field.name()))
+ })
+ .collect::<Vec<_>>();
+
+ assert!(
+ offenders.is_empty(),
+ "Execution schema carries Arrow view types that Comet does not produce: {}. \
+ Re-check `schema_force_view_types` before trusting any measurement from this run.",
+ offenders.join(", ")
+ );
+}
+
fn describe_schema(schema: &arrow::datatypes::Schema) -> String {
let mut counts = std::collections::HashMap::new();
for field in schema.fields() {
@@ -640,6 +687,9 @@ fn describe_schema(schema: &arrow::datatypes::Schema) -> String {
| DataType::UInt64 => "int",
DataType::Float16 | DataType::Float32 | DataType::Float64 => "float",
DataType::Utf8 | DataType::LargeUtf8 => "string",
+ // named, not folded into "string"/"binary", so a view schema is visible in the header
+ DataType::Utf8View => "stringview",
+ DataType::BinaryView => "binaryview",
DataType::Boolean => "bool",
DataType::Date32 | DataType::Date64 => "date",
DataType::Decimal128(_, _) | DataType::Decimal256(_, _) => "decimal",
@@ -680,3 +730,62 @@ fn format_bytes(bytes: usize) -> String {
format!("{bytes} B")
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::reject_view_types;
+ use arrow::datatypes::{DataType, Field, Fields, Schema};
+ use std::sync::Arc;
+
+ fn schema_of(data_type: DataType) -> Schema {
+ Schema::new(vec![
+ Field::new("plain", DataType::Int32, false),
+ Field::new("subject", data_type, true),
+ ])
+ }
+
+ #[test]
+ fn plain_schema_is_accepted() {
+ reject_view_types(&schema_of(DataType::Utf8));
+ }
+
+ #[test]
+ #[should_panic(expected = "subject (Utf8View)")]
+ fn top_level_string_view_is_rejected() {
+ reject_view_types(&schema_of(DataType::Utf8View));
+ }
+
+ #[test]
+ #[should_panic(expected = "subject (BinaryView)")]
+ fn top_level_binary_view_is_rejected() {
+ reject_view_types(&schema_of(DataType::BinaryView));
+ }
+
+ #[test]
+ #[should_panic(expected = "subject (Utf8View)")]
+ fn view_nested_in_a_list_is_rejected() {
+ reject_view_types(&schema_of(DataType::List(Arc::new(Field::new(
+ "item",
+ DataType::Utf8View,
+ true,
+ )))));
+ }
+
+ #[test]
+ #[should_panic(expected = "subject (Utf8View)")]
+ fn view_nested_in_a_struct_is_rejected() {
+ reject_view_types(&schema_of(DataType::Struct(Fields::from(vec![
+ Field::new("a", DataType::Int64, true),
+ Field::new("b", DataType::Utf8View, true),
+ ]))));
+ }
+
+ #[test]
+ #[should_panic(expected = "subject (Utf8View)")]
+ fn view_behind_a_dictionary_is_rejected() {
+ reject_view_types(&schema_of(DataType::Dictionary(
+ Box::new(DataType::Int32),
+ Box::new(DataType::Utf8View),
+ )));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org