From 0faf9ca5c3a4c156d5acf9194d75386543ea8fe1 Mon Sep 17 00:00:00 2001 From: Reid Kaufmann Date: Wed, 12 Aug 2026 11:36:58 -0500 Subject: [PATCH 1/6] fix: fast paths for UnionExec schema recomputation UnionExec::try_new recomputes the union schema via union_schema() on every construction, including with_new_children rebuilds performed by optimizer passes. For wide unions whose children share one schema (generated UNION ALL, unions of per-partition scans) this makes physical planning O(n^2) in child count. Two fast paths: - union_schema(): if every input schema is pointer- or content-equal to the first, return the first schema (merging N identical schemas is the identity operation). - UnionExec::with_new_children(): when child count and per-position child schemas are unchanged, reuse the existing schema. Plan properties are always recomputed, since they can legitimately change when schemas do not. (DataFusion 53 has an analogous properties optimization upstream, apache/datafusion#19792; this variant targets the 51 line.) Adds a union_schema benchmark covering shared-Arc, content-equal, and adversarial (last child differs) shapes. All existing union unit tests pass. Co-Authored-By: Claude Fable 5 --- datafusion/physical-plan/Cargo.toml | 4 + .../physical-plan/benches/union_schema.rs | 131 ++++++++++++++++++ datafusion/physical-plan/src/union.rs | 42 ++++++ 3 files changed, 177 insertions(+) create mode 100644 datafusion/physical-plan/benches/union_schema.rs diff --git a/datafusion/physical-plan/Cargo.toml b/datafusion/physical-plan/Cargo.toml index 5858deb83c83c..956e7f8d4b530 100644 --- a/datafusion/physical-plan/Cargo.toml +++ b/datafusion/physical-plan/Cargo.toml @@ -88,6 +88,10 @@ tokio = { workspace = true, features = [ harness = false name = "partial_ordering" +[[bench]] +harness = false +name = "union_schema" + [[bench]] harness = false name = "spill_io" diff --git a/datafusion/physical-plan/benches/union_schema.rs b/datafusion/physical-plan/benches/union_schema.rs new file mode 100644 index 0000000000000..49a5410a9c97a --- /dev/null +++ b/datafusion/physical-plan/benches/union_schema.rs @@ -0,0 +1,131 @@ +// 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. + +//! Benchmark for `UnionExec` construction cost as a function of child count. +//! +//! `UnionExec::try_new` recomputes the union schema by merging every child's +//! schema (fields, nullability, and per-field metadata). Optimizer passes +//! rebuild unions via `with_new_children`, so for wide unions this +//! construction cost is paid many times during planning. The common wide +//! union — thousands of children that all share one table schema (generated +//! UNION ALL, unions of per-partition scans) — should be cheap to construct. +//! +//! Scenarios: +//! - `shared_arc`: every child returns the same `Arc` (pointer-equal) +//! - `content_equal`: every child holds its own `Arc` with identical +//! contents (pointer-distinct; the shape produced by per-partition scan +//! subtrees, since each subtree stores its own schema handle) +//! - `last_differs`: all children content-equal except the last, whose final +//! field carries different metadata (adversarial worst case: any fast-path +//! equality scan is wasted, then the full merge runs anyway) + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; +use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion}; +use datafusion_physical_plan::empty::EmptyExec; +use datafusion_physical_plan::union::UnionExec; +use datafusion_physical_plan::ExecutionPlan; + +const NUM_FIELDS: usize = 10; +const METADATA_PER_FIELD: usize = 2; + +fn base_schema() -> Schema { + let fields: Vec = (0..NUM_FIELDS) + .map(|i| { + let mut md = HashMap::new(); + for m in 0..METADATA_PER_FIELD { + md.insert(format!("key_{m}"), format!("value_{i}_{m}")); + } + Field::new(format!("col_{i}"), DataType::Int64, true).with_metadata(md) + }) + .collect(); + Schema::new(fields) +} + +fn children_shared_arc(n: usize) -> Vec> { + let schema: SchemaRef = Arc::new(base_schema()); + (0..n) + .map(|_| Arc::new(EmptyExec::new(Arc::clone(&schema))) as Arc) + .collect() +} + +fn children_content_equal(n: usize) -> Vec> { + let schema = base_schema(); + (0..n) + .map(|_| { + Arc::new(EmptyExec::new(Arc::new(schema.clone()))) as Arc + }) + .collect() +} + +fn children_last_differs(n: usize) -> Vec> { + let mut children = children_content_equal(n - 1); + let schema = base_schema(); + let mut fields: Vec = + schema.fields().iter().map(|f| f.as_ref().clone()).collect(); + let last = fields.pop().unwrap(); + let mut md = last.metadata().clone(); + md.insert("divergent".to_string(), "true".to_string()); + fields.push(last.with_metadata(md)); + children.push(Arc::new(EmptyExec::new(Arc::new(Schema::new(fields)))) as _); + children +} + +fn bench_union_construction(c: &mut Criterion) { + let mut group = c.benchmark_group("union_exec_try_new"); + for n in [100, 1000, 4000] { + let shared = children_shared_arc(n); + let content = children_content_equal(n); + let differs = children_last_differs(n); + + group.bench_with_input(BenchmarkId::new("shared_arc", n), &shared, |b, ch| { + b.iter(|| UnionExec::try_new(ch.clone()).unwrap()) + }); + group.bench_with_input( + BenchmarkId::new("content_equal", n), + &content, + |b, ch| b.iter(|| UnionExec::try_new(ch.clone()).unwrap()), + ); + group.bench_with_input(BenchmarkId::new("last_differs", n), &differs, |b, ch| { + b.iter(|| UnionExec::try_new(ch.clone()).unwrap()) + }); + } + group.finish(); + + // Rebuild storm: the pattern optimizer passes produce — reconstruct the + // union once per child edit via `with_new_children`. + let mut group = c.benchmark_group("union_exec_rebuild_per_child"); + for n in [100, 1000] { + let children = children_content_equal(n); + let union: Arc = UnionExec::try_new(children.clone()).unwrap(); + group.bench_with_input(BenchmarkId::new("content_equal", n), &n, |b, _| { + b.iter(|| { + let mut plan = Arc::clone(&union); + for _ in 0..n { + plan = plan.with_new_children(children.clone()).unwrap(); + } + plan + }) + }); + } + group.finish(); +} + +criterion_group!(benches, bench_union_construction); +criterion_main!(benches); diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs index c95678dac9cdd..ff73289b69479 100644 --- a/datafusion/physical-plan/src/union.rs +++ b/datafusion/physical-plan/src/union.rs @@ -257,6 +257,32 @@ impl ExecutionPlan for UnionExec { self: Arc, children: Vec>, ) -> Result> { + // Fast path: optimizer passes frequently rebuild a `UnionExec` with the + // same number of children whose schemas are unchanged. In that case the + // union schema cannot change either, so we can skip the expensive + // `union_schema` recomputation, which is O(children x fields x + // metadata clones) and makes repeated rebuilds of wide unions O(n^2) + // over the whole optimization run. + // + // Note that the plan properties (partitioning, orderings, equivalences, + // boundedness) CAN legitimately change even when the schemas do not, so + // they are always recomputed from the new children. + if children.len() == self.inputs.len() + && children.len() >= 2 + && children.iter().zip(self.inputs.iter()).all(|(new, old)| { + let new_schema = new.schema(); + let old_schema = old.schema(); + Arc::ptr_eq(&new_schema, &old_schema) || new_schema == old_schema + }) + { + let schema = self.schema(); + let cache = Self::compute_properties(&children, schema)?; + return Ok(Arc::new(UnionExec { + inputs: children, + metrics: ExecutionPlanMetricsSet::new(), + cache, + })); + } UnionExec::try_new(children) } @@ -584,6 +610,22 @@ fn union_schema(inputs: &[Arc]) -> Result { let first_schema = inputs[0].schema(); + // Fast path: if every input reports a schema identical to the first — + // either the same `Arc` or content-equal (`Schema` equality covers field + // names, types, nullability, field metadata, and schema metadata, i.e. + // every property the merge below reads) — then merging N identical + // schemas is the identity operation, so return the first schema directly. + // This skips the O(inputs^2 x fields) per-field metadata merge below, + // which otherwise dominates planning time for unions of thousands of + // structurally-identical children (e.g. generated UNION ALL, or unions + // of per-partition scans that all share one table schema). + if inputs.iter().all(|input| { + let schema = input.schema(); + Arc::ptr_eq(&schema, &first_schema) || schema == first_schema + }) { + return Ok(first_schema); + } + let fields = (0..first_schema.fields().len()) .map(|i| { // We take the name from the left side of the union to match how names are coerced during logical planning, From 0f6a1ab3cb15599f11daa3e6684ea93d0a507d3f Mon Sep 17 00:00:00 2001 From: Reid Kaufmann Date: Wed, 12 Aug 2026 12:33:47 -0500 Subject: [PATCH 2/6] chore: tighten fast-path comments Co-Authored-By: Claude Fable 5 --- .../physical-plan/benches/union_schema.rs | 21 ++++----------- datafusion/physical-plan/src/union.rs | 27 ++++++------------- 2 files changed, 13 insertions(+), 35 deletions(-) diff --git a/datafusion/physical-plan/benches/union_schema.rs b/datafusion/physical-plan/benches/union_schema.rs index 49a5410a9c97a..98ae5b886f0f0 100644 --- a/datafusion/physical-plan/benches/union_schema.rs +++ b/datafusion/physical-plan/benches/union_schema.rs @@ -17,21 +17,11 @@ //! Benchmark for `UnionExec` construction cost as a function of child count. //! -//! `UnionExec::try_new` recomputes the union schema by merging every child's -//! schema (fields, nullability, and per-field metadata). Optimizer passes -//! rebuild unions via `with_new_children`, so for wide unions this -//! construction cost is paid many times during planning. The common wide -//! union — thousands of children that all share one table schema (generated -//! UNION ALL, unions of per-partition scans) — should be cheap to construct. -//! //! Scenarios: -//! - `shared_arc`: every child returns the same `Arc` (pointer-equal) -//! - `content_equal`: every child holds its own `Arc` with identical -//! contents (pointer-distinct; the shape produced by per-partition scan -//! subtrees, since each subtree stores its own schema handle) -//! - `last_differs`: all children content-equal except the last, whose final -//! field carries different metadata (adversarial worst case: any fast-path -//! equality scan is wasted, then the full merge runs anyway) +//! - `shared_arc`: every child returns the same `Arc` +//! - `content_equal`: pointer-distinct but identical schemas per child +//! - `last_differs`: identical except the last child (worst case for any +//! equality fast path: the scan is wasted, then the full merge runs) use std::collections::HashMap; use std::sync::Arc; @@ -108,8 +98,7 @@ fn bench_union_construction(c: &mut Criterion) { } group.finish(); - // Rebuild storm: the pattern optimizer passes produce — reconstruct the - // union once per child edit via `with_new_children`. + // Reconstruct the union once per child, as optimizer rewrites do. let mut group = c.benchmark_group("union_exec_rebuild_per_child"); for n in [100, 1000] { let children = children_content_equal(n); diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs index ff73289b69479..a4eeb8b498560 100644 --- a/datafusion/physical-plan/src/union.rs +++ b/datafusion/physical-plan/src/union.rs @@ -257,16 +257,10 @@ impl ExecutionPlan for UnionExec { self: Arc, children: Vec>, ) -> Result> { - // Fast path: optimizer passes frequently rebuild a `UnionExec` with the - // same number of children whose schemas are unchanged. In that case the - // union schema cannot change either, so we can skip the expensive - // `union_schema` recomputation, which is O(children x fields x - // metadata clones) and makes repeated rebuilds of wide unions O(n^2) - // over the whole optimization run. - // - // Note that the plan properties (partitioning, orderings, equivalences, - // boundedness) CAN legitimately change even when the schemas do not, so - // they are always recomputed from the new children. + // Fast path: if the children's schemas are unchanged the union schema + // cannot change, so skip the O(children x fields) `union_schema` + // recomputation. Plan properties can change even when schemas do not, + // so they are always recomputed. if children.len() == self.inputs.len() && children.len() >= 2 && children.iter().zip(self.inputs.iter()).all(|(new, old)| { @@ -610,15 +604,10 @@ fn union_schema(inputs: &[Arc]) -> Result { let first_schema = inputs[0].schema(); - // Fast path: if every input reports a schema identical to the first — - // either the same `Arc` or content-equal (`Schema` equality covers field - // names, types, nullability, field metadata, and schema metadata, i.e. - // every property the merge below reads) — then merging N identical - // schemas is the identity operation, so return the first schema directly. - // This skips the O(inputs^2 x fields) per-field metadata merge below, - // which otherwise dominates planning time for unions of thousands of - // structurally-identical children (e.g. generated UNION ALL, or unions - // of per-partition scans that all share one table schema). + // Merging N identical schemas is the identity operation: if every input + // schema is pointer- or content-equal to the first (`Schema` equality + // covers every property the merge below reads), return it directly + // instead of paying the per-field metadata merge. if inputs.iter().all(|input| { let schema = input.schema(); Arc::ptr_eq(&schema, &first_schema) || schema == first_schema From ccb548edec70519b6dfa3182ff4c718f3ceae2c6 Mon Sep 17 00:00:00 2001 From: Reid Kaufmann Date: Wed, 12 Aug 2026 15:09:31 -0500 Subject: [PATCH 3/6] bench: add nested-schema scenarios to union_schema Same three scenarios against a struct schema (10 fields x 5 subfields, metadata at both levels), with the last_differs divergence buried in the deepest last field. Co-Authored-By: Claude Fable 5 --- .../physical-plan/benches/union_schema.rs | 148 ++++++++++++------ 1 file changed, 100 insertions(+), 48 deletions(-) diff --git a/datafusion/physical-plan/benches/union_schema.rs b/datafusion/physical-plan/benches/union_schema.rs index 98ae5b886f0f0..fc0801d6317c6 100644 --- a/datafusion/physical-plan/benches/union_schema.rs +++ b/datafusion/physical-plan/benches/union_schema.rs @@ -17,11 +17,12 @@ //! Benchmark for `UnionExec` construction cost as a function of child count. //! -//! Scenarios: +//! Scenarios (run against a flat and a nested/struct schema): //! - `shared_arc`: every child returns the same `Arc` //! - `content_equal`: pointer-distinct but identical schemas per child -//! - `last_differs`: identical except the last child (worst case for any -//! equality fast path: the scan is wasted, then the full merge runs) +//! - `last_differs`: identical except the last child's deepest field (worst +//! case for any equality fast path: the scan is wasted, then the full +//! merge runs) use std::collections::HashMap; use std::sync::Arc; @@ -33,75 +34,126 @@ use datafusion_physical_plan::union::UnionExec; use datafusion_physical_plan::ExecutionPlan; const NUM_FIELDS: usize = 10; +const NESTED_CHILDREN: usize = 5; const METADATA_PER_FIELD: usize = 2; -fn base_schema() -> Schema { +fn metadata(tag: &str, i: usize) -> HashMap { + (0..METADATA_PER_FIELD) + .map(|m| (format!("key_{m}"), format!("value_{tag}_{i}_{m}"))) + .collect() +} + +fn flat_schema() -> Schema { let fields: Vec = (0..NUM_FIELDS) .map(|i| { - let mut md = HashMap::new(); - for m in 0..METADATA_PER_FIELD { - md.insert(format!("key_{m}"), format!("value_{i}_{m}")); - } - Field::new(format!("col_{i}"), DataType::Int64, true).with_metadata(md) + Field::new(format!("col_{i}"), DataType::Int64, true) + .with_metadata(metadata("f", i)) }) .collect(); Schema::new(fields) } -fn children_shared_arc(n: usize) -> Vec> { - let schema: SchemaRef = Arc::new(base_schema()); - (0..n) - .map(|_| Arc::new(EmptyExec::new(Arc::clone(&schema))) as Arc) - .collect() -} - -fn children_content_equal(n: usize) -> Vec> { - let schema = base_schema(); - (0..n) - .map(|_| { - Arc::new(EmptyExec::new(Arc::new(schema.clone()))) as Arc +fn nested_schema() -> Schema { + let fields: Vec = (0..NUM_FIELDS) + .map(|i| { + let children: Vec = (0..NESTED_CHILDREN) + .map(|c| { + Field::new(format!("sub_{i}_{c}"), DataType::Int64, true) + .with_metadata(metadata("n", i * NESTED_CHILDREN + c)) + }) + .collect(); + Field::new(format!("col_{i}"), DataType::Struct(children.into()), true) + .with_metadata(metadata("s", i)) }) - .collect() + .collect(); + Schema::new(fields) } -fn children_last_differs(n: usize) -> Vec> { - let mut children = children_content_equal(n - 1); - let schema = base_schema(); +/// Clone `schema` with extra metadata on its deepest last field, so equality +/// checks succeed on everything before failing at the very end. +fn divergent(schema: &Schema) -> Schema { let mut fields: Vec = schema.fields().iter().map(|f| f.as_ref().clone()).collect(); let last = fields.pop().unwrap(); - let mut md = last.metadata().clone(); - md.insert("divergent".to_string(), "true".to_string()); - fields.push(last.with_metadata(md)); - children.push(Arc::new(EmptyExec::new(Arc::new(Schema::new(fields)))) as _); + let last = match last.data_type() { + DataType::Struct(children) => { + let mut children: Vec = + children.iter().map(|f| f.as_ref().clone()).collect(); + let sub = children.pop().unwrap(); + let mut md = sub.metadata().clone(); + md.insert("divergent".to_string(), "true".to_string()); + children.push(sub.with_metadata(md)); + Field::new( + last.name(), + DataType::Struct(children.into()), + last.is_nullable(), + ) + .with_metadata(last.metadata().clone()) + } + _ => { + let mut md = last.metadata().clone(); + md.insert("divergent".to_string(), "true".to_string()); + last.with_metadata(md) + } + }; + fields.push(last); + Schema::new(fields) +} + +fn child(schema: SchemaRef) -> Arc { + Arc::new(EmptyExec::new(schema)) +} + +fn children_shared_arc(schema: &Schema, n: usize) -> Vec> { + let schema: SchemaRef = Arc::new(schema.clone()); + (0..n).map(|_| child(Arc::clone(&schema))).collect() +} + +fn children_content_equal(schema: &Schema, n: usize) -> Vec> { + (0..n).map(|_| child(Arc::new(schema.clone()))).collect() +} + +fn children_last_differs(schema: &Schema, n: usize) -> Vec> { + let mut children = children_content_equal(schema, n - 1); + children.push(child(Arc::new(divergent(schema)))); children } fn bench_union_construction(c: &mut Criterion) { - let mut group = c.benchmark_group("union_exec_try_new"); - for n in [100, 1000, 4000] { - let shared = children_shared_arc(n); - let content = children_content_equal(n); - let differs = children_last_differs(n); - - group.bench_with_input(BenchmarkId::new("shared_arc", n), &shared, |b, ch| { - b.iter(|| UnionExec::try_new(ch.clone()).unwrap()) - }); - group.bench_with_input( - BenchmarkId::new("content_equal", n), - &content, - |b, ch| b.iter(|| UnionExec::try_new(ch.clone()).unwrap()), - ); - group.bench_with_input(BenchmarkId::new("last_differs", n), &differs, |b, ch| { - b.iter(|| UnionExec::try_new(ch.clone()).unwrap()) - }); + for (suffix, schema, sizes) in [ + ("", flat_schema(), &[100usize, 1000, 4000][..]), + ("_nested", nested_schema(), &[1000, 4000][..]), + ] { + let mut group = c.benchmark_group(format!("union_exec_try_new{suffix}")); + for &n in sizes { + let shared = children_shared_arc(&schema, n); + let content = children_content_equal(&schema, n); + let differs = children_last_differs(&schema, n); + + group.bench_with_input( + BenchmarkId::new("shared_arc", n), + &shared, + |b, ch| b.iter(|| UnionExec::try_new(ch.clone()).unwrap()), + ); + group.bench_with_input( + BenchmarkId::new("content_equal", n), + &content, + |b, ch| b.iter(|| UnionExec::try_new(ch.clone()).unwrap()), + ); + group.bench_with_input( + BenchmarkId::new("last_differs", n), + &differs, + |b, ch| b.iter(|| UnionExec::try_new(ch.clone()).unwrap()), + ); + } + group.finish(); } - group.finish(); // Reconstruct the union once per child, as optimizer rewrites do. let mut group = c.benchmark_group("union_exec_rebuild_per_child"); + let schema = flat_schema(); for n in [100, 1000] { - let children = children_content_equal(n); + let children = children_content_equal(&schema, n); let union: Arc = UnionExec::try_new(children.clone()).unwrap(); group.bench_with_input(BenchmarkId::new("content_equal", n), &n, |b, _| { b.iter(|| { From a722cedbe38fe18eb32fba27b268a7fed60b8a2a Mon Sep 17 00:00:00 2001 From: Reid Kaufmann Date: Wed, 12 Aug 2026 15:18:07 -0500 Subject: [PATCH 4/6] bench: keep nested last_differs divergence at the top level Divergence inside a nested type changes the field DataType, which UnionExec::try_new rejects ('Schemas have to be aligned to rewrite equivalences') -- the union is unconstructible, so only top-level metadata divergence is a reachable worst case. Co-Authored-By: Claude Fable 5 --- .../physical-plan/benches/union_schema.rs | 30 ++++--------------- 1 file changed, 6 insertions(+), 24 deletions(-) diff --git a/datafusion/physical-plan/benches/union_schema.rs b/datafusion/physical-plan/benches/union_schema.rs index fc0801d6317c6..8ab1350a226c9 100644 --- a/datafusion/physical-plan/benches/union_schema.rs +++ b/datafusion/physical-plan/benches/union_schema.rs @@ -69,34 +69,16 @@ fn nested_schema() -> Schema { Schema::new(fields) } -/// Clone `schema` with extra metadata on its deepest last field, so equality -/// checks succeed on everything before failing at the very end. +/// Clone `schema` with extra metadata on its last field. Divergence must stay +/// at the top level: differing nested fields change the field `DataType` +/// itself, which `UnionExec::try_new` rejects ("Schemas have to be aligned"). fn divergent(schema: &Schema) -> Schema { let mut fields: Vec = schema.fields().iter().map(|f| f.as_ref().clone()).collect(); let last = fields.pop().unwrap(); - let last = match last.data_type() { - DataType::Struct(children) => { - let mut children: Vec = - children.iter().map(|f| f.as_ref().clone()).collect(); - let sub = children.pop().unwrap(); - let mut md = sub.metadata().clone(); - md.insert("divergent".to_string(), "true".to_string()); - children.push(sub.with_metadata(md)); - Field::new( - last.name(), - DataType::Struct(children.into()), - last.is_nullable(), - ) - .with_metadata(last.metadata().clone()) - } - _ => { - let mut md = last.metadata().clone(); - md.insert("divergent".to_string(), "true".to_string()); - last.with_metadata(md) - } - }; - fields.push(last); + let mut md = last.metadata().clone(); + md.insert("divergent".to_string(), "true".to_string()); + fields.push(last.with_metadata(md)); Schema::new(fields) } From e4c24be146a8732b6ae93f31ed077950c5b6fa11 Mon Sep 17 00:00:00 2001 From: Reid Kaufmann Date: Wed, 12 Aug 2026 15:30:12 -0500 Subject: [PATCH 5/6] bench: add names_differ scenario (typical unequal union) Same types, different field names -- the common SELECT a UNION ALL SELECT b shape; equality guards fail on the second child's first field. Co-Authored-By: Claude Fable 5 --- .../physical-plan/benches/union_schema.rs | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/datafusion/physical-plan/benches/union_schema.rs b/datafusion/physical-plan/benches/union_schema.rs index 8ab1350a226c9..91525530c1cc5 100644 --- a/datafusion/physical-plan/benches/union_schema.rs +++ b/datafusion/physical-plan/benches/union_schema.rs @@ -95,6 +95,21 @@ fn children_content_equal(schema: &Schema, n: usize) -> Vec Vec> { + let mut fields: Vec = + schema.fields().iter().map(|f| f.as_ref().clone()).collect(); + let first = fields.remove(0); + let renamed = first.clone().with_name(format!("renamed_{}", first.name())); + fields.insert(0, renamed); + let alt = Schema::new(fields); + let mut children = vec![child(Arc::new(schema.clone()))]; + children.extend((1..n).map(|_| child(Arc::new(alt.clone())))); + children +} + fn children_last_differs(schema: &Schema, n: usize) -> Vec> { let mut children = children_content_equal(schema, n - 1); children.push(child(Arc::new(divergent(schema)))); @@ -127,6 +142,12 @@ fn bench_union_construction(c: &mut Criterion) { &differs, |b, ch| b.iter(|| UnionExec::try_new(ch.clone()).unwrap()), ); + let names = children_names_differ(&schema, n); + group.bench_with_input( + BenchmarkId::new("names_differ", n), + &names, + |b, ch| b.iter(|| UnionExec::try_new(ch.clone()).unwrap()), + ); } group.finish(); } From 6754e2939b945499283fba7283a66d249f68c6ba Mon Sep 17 00:00:00 2001 From: Reid Kaufmann Date: Wed, 12 Aug 2026 15:57:03 -0500 Subject: [PATCH 6/6] bench: add rebuild_per_child last_differs scenario Measures with_new_children rebuild cost for heterogeneous unions: the positional guard hits ptr-eq on unchanged child Arcs regardless of cross-child schema differences. Co-Authored-By: Claude Fable 5 --- .../physical-plan/benches/union_schema.rs | 27 +++++++++++-------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/datafusion/physical-plan/benches/union_schema.rs b/datafusion/physical-plan/benches/union_schema.rs index 91525530c1cc5..d25fe2c0bdee8 100644 --- a/datafusion/physical-plan/benches/union_schema.rs +++ b/datafusion/physical-plan/benches/union_schema.rs @@ -156,17 +156,22 @@ fn bench_union_construction(c: &mut Criterion) { let mut group = c.benchmark_group("union_exec_rebuild_per_child"); let schema = flat_schema(); for n in [100, 1000] { - let children = children_content_equal(&schema, n); - let union: Arc = UnionExec::try_new(children.clone()).unwrap(); - group.bench_with_input(BenchmarkId::new("content_equal", n), &n, |b, _| { - b.iter(|| { - let mut plan = Arc::clone(&union); - for _ in 0..n { - plan = plan.with_new_children(children.clone()).unwrap(); - } - plan - }) - }); + for (label, children) in [ + ("content_equal", children_content_equal(&schema, n)), + ("last_differs", children_last_differs(&schema, n)), + ] { + let union: Arc = + UnionExec::try_new(children.clone()).unwrap(); + group.bench_with_input(BenchmarkId::new(label, n), &n, |b, _| { + b.iter(|| { + let mut plan = Arc::clone(&union); + for _ in 0..n { + plan = plan.with_new_children(children.clone()).unwrap(); + } + plan + }) + }); + } } group.finish(); }