reidkaufmann opened a new pull request, #24389:
URL: https://github.com/apache/datafusion/pull/24389
## Which issue does this close?
Complements #19792. Part of the wide-`UnionExec` planning-cost work.
## 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 identically-shaped 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
incurs the planning latency penalty from `O(n^2 * f)` complexity.
### Relationship to #19792
`UnionExec` construction has two quadratic halves:
- **`with_new_children` / `PlanProperties`** -- addressed by #19792
(`with_new_children_and_same_properties`, `Arc<PlanProperties>`, the
properties fast path). Already on `main`.
- **`union_schema`** -- *not* covered by #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
The natural objection: doesn't the deep `==` make the *unequal* case slower?
Here's why it doesn't matter.
- **The equal case never runs 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 | patched | change |
|---|---|---|---|---|
| union_exec_try_new/shared_arc | 100 | 263 µs | 42 µs | 6.2× |
| union_exec_try_new/shared_arc | 1000 | 2.60 ms | 429 µs | 6.1× |
| union_exec_try_new/shared_arc | 4000 | 10.5 ms | 1.74 ms | 6.0× |
| union_exec_try_new/content_equal | 100 | 262 µs | 42 µs | 6.2× |
| union_exec_try_new/content_equal | 1000 | 2.61 ms | 433 µs | 6.0× |
| union_exec_try_new/content_equal | 4000 | 10.5 ms | 1.74 ms | 6.0× |
| union_exec_try_new/last_differs | 4000 | ~98 ms | ~97 ms | flat (tight
interleave: patched ≈ baseline) |
| union_exec_try_new/names_differ | 4000 | ~87 ms | ~87 ms | flat (±1%) |
| union_exec_try_new_nested/content_equal | 1000 | 2.68 ms | 431 µs | 6.2× |
| union_exec_try_new_nested/content_equal | 4000 | 10.8 ms | 1.73 ms | 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 | patched |
|---|---|---|
| physical_plan_tpcds_all | 995.9 ± 1.4 ms | 991.9 ± 1.9 ms |
| physical_plan_tpch_all | 60.4 ± 0.2 ms | 60.3 ± 0.1 ms |
| physical_sorted_union_order_by_50 | 349.9 ± 2.4 ms | 346.3 ± 2.7 ms |
| physical_sorted_union_order_by_10 | 12.3 ± 0.03 ms | 12.2 ± 0.07 ms |
| physical_select_all_from_1000 | 30.8 ± 0.25 ms | 30.7 ± 0.08 ms |
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
