xavlee opened a new pull request, #24497:
URL: https://github.com/apache/datafusion/pull/24497
## Which issue does this PR close?
- Closes #24438.
## Rationale for this change
Data can be range-partitioned into logical runs that are each locally
grouped by `(key, time)` and non-overlapping on `(key, date_bin(time))`.
Concatenating several runs into one DataFusion execution partition resets the
global sort order, so `AggregateExec` currently falls back to blocking hash
aggregation even though every group remains one contiguous range.
This change lets a data source communicate that stronger contiguity
guarantee independently of physical sort order, allowing the first aggregate
after an optional projection to emit completed groups incrementally.
## What changes are included in this PR?
- Add `ExecutionPlan::partition_disjoint_exprs` and the corresponding
`DataSource` hook. The returned expressions form one composite tuple whose
values must occur in a single contiguous range within each output partition.
- Propagate the complete tuple through `ProjectionExec`; if any component
cannot be projected, clear the property.
- Let `AggregateExec` derive a private group-completion mode from the
property while retaining `InputOrderMode::Linear`, so execution can reuse the
existing ordered aggregate paths without advertising a false output ordering.
- Support exact group-key matches and strict subsets:
- exact match uses full group completion;
- a disjoint-key subset uses partial group completion for additional
grouping keys.
- Consume the property at the first aggregate and keep grouping sets on
their conservative non-streaming path.
- Prevent round-robin repartitioning from destroying the source guarantee
when it is enabling streaming.
- Propagate the property across FFI execution plans and add source/test
helpers plus optimizer and cross-library coverage.
- Add a Criterion benchmark for concatenated, descending logical runs whose
global key order resets.
A source may declare a derived expression such as `(key, date_bin(time))`
after verifying that logical-run boundaries do not split a bin. The projection
path maps that expression but deliberately does not infer it from `(key, time)`.
## Are these changes tested?
Yes. Coverage includes:
- early output before end-of-input;
- order resets between logical runs;
- exact and partial composite-key implication;
- `date_bin` projection;
- all-or-nothing projection propagation;
- aggregate combination and distribution optimizer behavior;
- spill replay and both aggregate implementations;
- local and cross-library FFI round trips.
Validation run locally:
- `cargo test -p datafusion-physical-plan --lib` (1,765 tests)
- `cargo test -p datafusion-datasource --lib` (178 tests)
- `cargo test -p datafusion --test core_integration physical_optimizer` (553
tests)
- FFI unit/integration tests, including the cross-library execution-plan test
- proto-feature focused tests
- `./dev/rust_lint.sh`
- `cargo clippy --all-targets --all-features -- -D warnings`
The added benchmark measured approximately 885 us for unordered hash
aggregation versus 663 us for partition-disjoint streaming on the development
machine (about 25% faster for this workload).
## Are there any user-facing changes?
Yes. Data source and execution-plan implementors can opt into the new
property. It is a correctness contract: an invalid declaration can cause an
aggregate group to be emitted before all of its rows have been observed.
The new FFI callback adds a field to the versioned `FFI_ExecutionPlan`
struct. `cargo-semver-checks` therefore reports the expected major-version FFI
layout change; `datafusion-physical-plan` and `datafusion-datasource` otherwise
pass all semver checks.
--
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]
