Improve planning speed: Fast path for union_schema when all children share a schema - #24389
Merged
Merged
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR improves planning-time performance for wide UnionExec / InterleaveExec plans by adding an early-return fast path in union_schema when all children already share an identical schema (by Arc::ptr_eq or structural ==), avoiding the existing quadratic metadata/nullability merge.
Changes:
- Add a
union_schemafast path that returns the first child’s schema when all children’s schemas are pointer-equal or structurally equal. - Add a regression test targeting the pointer-distinct-but-equal (
==) branch. - Add a new Criterion benchmark (
union_schema) and register it inCargo.toml.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| datafusion/physical-plan/src/union.rs | Adds the union_schema early-return fast path and a test for the content-equality branch. |
| datafusion/physical-plan/Cargo.toml | Registers the new union_schema benchmark target. |
| datafusion/physical-plan/benches/union_schema.rs | Adds a benchmark measuring UnionExec::try_new construction cost across schema-shape scenarios. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
`union_schema` (shared by `UnionExec` and `InterleaveExec`) coerces field metadata and nullability across every child. That merge is O(n^2) in the number of children -- for each field it scans all other children -- and dominates physical planning for wide unions whose children all carry the same schema. This shape is common when a union is built from repartitioned copies of a single plan (observed in InfluxDB). When every child already shares the first child's schema the merge is a no-op. Add a fast path that returns the first schema when all remaining children are either the same allocation (`Arc::ptr_eq`) or structurally equal (`==`), falling through to the full merge otherwise. Signed-off-by: Reid Kaufmann <reid.kaufmann@gmail.com>
Benchmark `UnionExec::try_new` construction cost as a function of child count, over both a flat and a nested/struct schema. Covers the shared-Arc and content-equal fast-path cases, the adversarial last-differs case (scan wasted, then full merge), and the names-differ case where equality fails immediately. Signed-off-by: Reid Kaufmann <reid.kaufmann@gmail.com>
reidkaufmann
force-pushed
the
union-schema-fast-path
branch
from
August 14, 2026 22:45
972e412 to
eb2cdf3
Compare
alamb
changed the title
union_schema when all children share a schemaunion_schema when all children share a schema
Contributor
|
run benchmark sql_planner |
Contributor
|
Thanks @reidkaufmann For anyone following along, this is porting a patch upstream we made in the influxdata fork for a performance issue we saw for some customer |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing union-schema-fast-path (eb2cdf3) to ec110ce (merge-base) diff Run configurationrun benchmark sql_plannerResults will be posted here when complete File an issue against this benchmark runner |
alamb
approved these changes
Aug 15, 2026
alamb
left a comment
alamb
left a comment
Contributor
There was a problem hiding this comment.
Thank you @reidkaufmann -- assuming the benchmark results look good I think this is a good addition
I left some comments to reduce the size of the diff / unecessary comments
| let first_schema = inputs[0].schema(); | ||
|
|
||
| // Fast path: when every input already shares the first input's schema, the | ||
| // field-by-field metadata/nullability merge below is redundant work that |
Contributor
There was a problem hiding this comment.
I think we should slim this comment down - the first sentence is probably enough. The last sentence is unecessary as it is restating in english what the code right below it clearly does so it is redundant
Contributor
|
run benchmark sql_planner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing union-schema-fast-path (eb2cdf3) to ec110ce (merge-base) diff Run configurationrun benchmark sql_plannerResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing union-schema-fast-path (eb2cdf3) to ec110ce (merge-base) diff Run configurationrun benchmark sql_plannerCPU Details (lscpu)Details
Resource Usagesql_planner — base (merge-base)
sql_planner — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing union-schema-fast-path (eb2cdf3) to ec110ce (merge-base) diff Run configurationrun benchmark sql_plannerCPU Details (lscpu)Details
Resource Usagesql_planner — base (merge-base)
sql_planner — branch
File an issue against this benchmark runner |
Dandandan
approved these changes
Aug 15, 2026
Dandandan
enabled auto-merge
August 15, 2026 12:49
Contributor
|
Amazing |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #24389 +/- ##
==========================================
- Coverage 81.19% 81.19% -0.01%
==========================================
Files 1110 1110
Lines 388618 388772 +154
Branches 388618 388772 +154
==========================================
+ Hits 315531 315648 +117
- Misses 54507 54529 +22
- Partials 18580 18595 +15 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
imtherealnaska
pushed a commit
to imtherealnaska/datafusion
that referenced
this pull request
Aug 16, 2026
…n share a schema (apache#24389) ## Which issue does this close? Complements apache#19792. Fits with the wide-`UnionExec` planning-cost work, but originates from an InfluxDB issue. ## Rationale for this change `union_schema` builds the output schema for `UnionExec` and `InterleaveExec` by coercing field metadata and nullability across **every** child. That merge is quadratic in the number of children: for each output field it walks all inputs, and for each input it walks *every other* input to union field-level metadata. For a union of `n` children with `f` fields the construction cost is `O(n^2 * f)` (worse when fields carry metadata). For narrow unions this is insignificant. It matters when a plan fans a single source out into many identical-schema children and unions them back together -- e.g. a union assembled from repartitioned copies of the same input. An instance like this occurred with InfluxDB: every child schema was the same, so the merge, guaranteed to reproduce the first child's schema, unnecessarily incurred the planning latency penalty from `O(n^2 * f)` complexity. ### Relationship to apache#19792 `UnionExec` construction has two quadratic halves: - **`with_new_children` / `PlanProperties`** -- addressed by apache#19792 (`with_new_children_and_same_properties`, `Arc<PlanProperties>`, the properties fast path). Already on `main`. - **`union_schema`** -- *not* covered by apache#19792 and still quadratic on `main`. This PR complements it by making `union_schema` skip the merge when it can't change the result. It deliberately doesn't touch `with_new_children`; that path is already handled. ## What changes are included? A fast path at the top of `union_schema`: after taking `inputs[0].schema()`, if every remaining child's schema is either the **same allocation** (`Arc::ptr_eq`) or **structurally equal** (`==`) to the first, return the first schema immediately. Otherwise we fall through to the existing full merge, so behavior for genuinely heterogeneous unions is byte-for-byte unchanged. ```rust let first_schema = inputs[0].schema(); if inputs[1..].iter().all(|input| { let schema = input.schema(); Arc::ptr_eq(&schema, &first_schema) || schema == first_schema }) { return Ok(first_schema); } ``` `InterleaveExec` shares `union_schema`, so it gets the same speedup for free. ## On the cost of the deep compare... The natural objection (which came up before this PR): **doesn't the deep `==` make the *unequal* case slower?** I'll paraphrase the prior conclusions, risking verbosity to avoid rehashing the discussion. Spoiler: it's not an issue. - The equal case avoids the merge, and its check is cheap. The shared-`Arc` case is settled by pointer comparison. The distinct-but-equal case runs `Schema::eq`, which is allocation-free and short-circuits on the first difference. Benchmarks show a small loss versus a pointer-equality-only control (the theoretical floor) but it still beats the full merge by a wide margin, and that advantage grows with schema complexity. - The adversarial worst case is bounded. The one shape where the scan is pure overhead is `last_differs`: children `0..n-1` are equal and the last diverges, so we scan `n` schemas, fail on the last, then merge anyway. That's a single linear `==` pass bounded by the merge that follows -- a constant fraction, not another factor of `n` -- and it takes *thousands* of near-identical children differing only in the last to hit. - Ordinary unequal unions fail fast. `SELECT a ... UNION ALL SELECT b ...` differs at field 0, so `==` rejects on the first field (see `names_differ`). And `UnionExec::try_new` already rejects misaligned children, so the only divergence `union_schema` ever sees is top-level (caught in the first pass). ## Benchmark results New bench `datafusion/physical-plan/benches/union_schema.rs` measures `UnionExec::try_new` construction over a flat schema and a nested/struct schema, for the four child shapes above. Run interleaved (baseline / patched alternated per cell) on a fixed-clock T2D VM. ### `union_schema` construction (lower is better) | scenario | n | baseline (ms) | patched (ms) | change | |---|---|---|---|---| | union_exec_try_new/shared_arc | 100 | 0.263 | 0.042 | 6.2× | | union_exec_try_new/shared_arc | 1000 | 2.60 | 0.429 | 6.1× | | union_exec_try_new/shared_arc | 4000 | 10.5 | 1.74 | 6.0× | | union_exec_try_new/content_equal | 100 | 0.262 | 0.042 | 6.2× | | union_exec_try_new/content_equal | 1000 | 2.61 | 0.433 | 6.0× | | union_exec_try_new/content_equal | 4000 | 10.5 | 1.74 | 6.0× | | union_exec_try_new/last_differs | 4000 | ~98 | ~97 | flat (±1.5%) | | union_exec_try_new/names_differ | 4000 | ~87 | ~87 | flat (±1%) | | union_exec_try_new_nested/content_equal | 1000 | 2.68 | 0.431 | 6.2× | | union_exec_try_new_nested/content_equal | 4000 | 10.8 | 1.73 | 6.2× | The `last_differs` (adversarial: N-1 children equal, deep compare then full merge) and `names_differ` (typical unequal: fails on the first field) cells were re-measured with tight per-cell interleaving (baseline/patched adjacent, 4 rounds) to control for variance: both are within ±1.5%, straddling zero. Interpretation: the **deep compare cost isn't observable end to end**. ### End-to-end planning: no regression (`sql_planner`) `cargo bench --bench sql_planner` (TPC-H + ClickBench) run baseline vs patched on the fixed-clock T2D VM. Every case lands within ±1% -- run-to-run noise -- with no case regressing beyond that noise. Notable rows, including the union-heavy `sorted_union` cases the fast path is meant to help: | case | baseline (ms) | patched (ms) | |---|---|---| | physical_plan_tpcds_all | 995.9 ± 1.4 | 991.9 ± 1.9 | | physical_plan_tpch_all | 60.4 ± 0.2 | 60.3 ± 0.1 | | physical_sorted_union_order_by_50 | 349.9 ± 2.4 | 346.3 ± 2.7 | | physical_sorted_union_order_by_10 | 12.3 ± 0.03 | 12.2 ± 0.07 | | physical_select_all_from_1000 | 30.8 ± 0.25 | 30.7 ± 0.08 | The full TPC-H q1-q22 and ClickBench sets are all flat (ratio is 1.00-1.01 in both directions). Separately, interleaving tests per-cell (baseline and patched back-to-back, so run-to-run variance -- e.g. thermal -- cancels rather than favoring one) for `physical_join_distinct` + eight ClickBench queries (4 rounds) confirmed the same thing: patched and baseline straddle zero; no systematic regression from the deep compare. ## Testing - `cargo test -p datafusion-physical-plan --lib union` -- all pass, including a new `test_union_schema_fast_path_content_equal` that exercises the `==` branch with pointer-distinct-but-equal schemas and asserts the result matches the shared schema (i.e. identical to the slow-path merge). - `cargo clippy -p datafusion-physical-plan --lib -- -D warnings` -- clean. - `cargo bench --bench union_schema` -- compiles and runs. ## Are there any user-facing changes? No: planning-time performance change only, results and schema are identical. --------- Signed-off-by: Reid Kaufmann <reid.kaufmann@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this close?
Complements #19792. Fits with the wide-
UnionExecplanning-cost work, but originates from an InfluxDB issue.Rationale for this change
union_schemabuilds the output schema forUnionExecandInterleaveExecby coercing field metadata and nullability across every child. That merge is quadratic in the number of children: for each output field it walks all inputs, and for each input it walks every other input to union field-level metadata. For a union ofnchildren withffields the construction cost isO(n^2 * f)(worse when fields carry metadata).For narrow unions this is insignificant. It matters when a plan fans a single source out into many identical-schema children and unions them back together -- e.g. a union assembled from repartitioned copies of the same input. An instance like this occurred with InfluxDB: every child schema was the same, so the merge, guaranteed to reproduce the first child's schema, unnecessarily incurred the planning latency penalty from
O(n^2 * f)complexity.Relationship to #19792
UnionExecconstruction has two quadratic halves:with_new_children/PlanProperties-- addressed by CachePlanProperties, add fast-path forwith_new_children#19792 (with_new_children_and_same_properties,Arc<PlanProperties>, the properties fast path). Already onmain.union_schema-- not covered by CachePlanProperties, add fast-path forwith_new_children#19792 and still quadratic onmain.This PR complements it by making
union_schemaskip the merge when it can't change the result. It deliberately doesn't touchwith_new_children; that path is already handled.What changes are included?
A fast path at the top of
union_schema: after takinginputs[0].schema(), if every remaining child's schema is either the same allocation (Arc::ptr_eq) or structurally equal (==) to the first, return the first schema immediately. Otherwise we fall through to the existing full merge, so behavior for genuinely heterogeneous unions is byte-for-byte unchanged.InterleaveExecsharesunion_schema, so it gets the same speedup for free.On the cost of the deep compare...
The natural objection (which came up before this PR): doesn't the deep
==make the unequal case slower? I'll paraphrase the prior conclusions, risking verbosity to avoid rehashing the discussion. Spoiler: it's not an issue.The equal case avoids the merge, and its check is cheap. The shared-
Arccase is settled by pointer comparison. The distinct-but-equal case runsSchema::eq, which is allocation-free and short-circuits on the first difference. Benchmarks show a small loss versus a pointer-equality-only control (the theoretical floor) but it still beats the full merge by a wide margin, and that advantage grows with schema complexity.The adversarial worst case is bounded. The one shape where the scan is pure overhead is
last_differs: children0..n-1are equal and the last diverges, so we scannschemas, fail on the last, then merge anyway. That's a single linear==pass bounded by the merge that follows -- a constant fraction, not another factor ofn-- and it takes thousands of near-identical children differing only in the last to hit.Ordinary unequal unions fail fast.
SELECT a ... UNION ALL SELECT b ...differs at field 0, so==rejects on the first field (seenames_differ). AndUnionExec::try_newalready rejects misaligned children, so the only divergenceunion_schemaever sees is top-level (caught in the first pass).Benchmark results
New bench
datafusion/physical-plan/benches/union_schema.rsmeasuresUnionExec::try_newconstruction over a flat schema and a nested/struct schema, for the four child shapes above. Run interleaved (baseline / patched alternated per cell) on a fixed-clock T2D VM.union_schemaconstruction (lower is better)The
last_differs(adversarial: N-1 children equal, deep compare then full merge) andnames_differ(typical unequal: fails on the first field) cells were re-measured with tight per-cell interleaving (baseline/patched adjacent, 4 rounds) to control for variance: both are within ±1.5%, straddling zero. Interpretation: the deep compare cost isn't observable end to end.End-to-end planning: no regression (
sql_planner)cargo bench --bench sql_planner(TPC-H + ClickBench) run baseline vs patched on the fixed-clock T2D VM. Every case lands within ±1% -- run-to-run noise -- with no case regressing beyond that noise. Notable rows, including the union-heavysorted_unioncases the fast path is meant to help:The full TPC-H q1-q22 and ClickBench sets are all flat (ratio is 1.00-1.01 in both directions). Separately, interleaving tests per-cell (baseline and patched back-to-back, so run-to-run variance -- e.g. thermal -- cancels rather than favoring one) for
physical_join_distinct+ eight ClickBench queries (4 rounds) confirmed the same thing: patched and baseline straddle zero; no systematic regression from the deep compare.Testing
cargo test -p datafusion-physical-plan --lib union-- all pass, including a newtest_union_schema_fast_path_content_equalthat exercises the==branch with pointer-distinct-but-equal schemas and asserts the result matches the shared schema (i.e. identical to the slow-path merge).cargo clippy -p datafusion-physical-plan --lib -- -D warnings-- clean.cargo bench --bench union_schema-- compiles and runs.Are there any user-facing changes?
No: planning-time performance change only, results and schema are identical.