Repository navigation
Conversation
`AggregateExec::try_new` takes six positional arguments, and the fields the physical optimizer changes afterwards were set through methods that each copied the remaining fields by hand: - `with_new_limit_options(&self, ..)` clones the node and resets metrics - `with_limit_options(self, ..)` consumes it and keeps them - `with_new_aggr_exprs(&self, ..)` clones it for a different field None of them validate the result, so it is easy to build an aggregate that only fails once it is executed. Pushing a limit into an aggregate that cannot execute one is the clearest case: with no MIN/MAX aggregate and no ordering direction `GroupedTopKAggregateStream` returns an internal error, with more than one group by expression it indexes out of bounds, and with a `FILTER` it silently drops the filter and returns wrong results. Replace them with `AggregateExecBuilder`, reachable as `AggregateExec::builder(mode, input)` for a new node and `AggregateExec::to_builder()` for a rewrite: - every argument is named, so `input`/`input_schema` and `aggr_expr`/`filter_expr` cannot be transposed - `filter_expr` defaults to "no filter per aggregate" - `build()` validates the node once: limits are checked against the shape of the aggregate, and a rewrite of the aggregate expressions is checked against the output schema it inherits - a rewrite carries over the derived state (output schema, plan properties, ordering requirements, dynamic filter) of the node it came from instead of asking each caller to copy the fields, so it cannot rename output fields The three optimizer rules that push a limit, `OptimizeAggregateOrder` and the protobuf decoder now go through the builder; the old methods are deprecated in favour of it. `AggregateExec::limit_options()` is unchanged: reading the limit was never the problem. Invalid limits are now an error at plan time rather than at execution time. Optimizer rules treat that as "skip this optimization", so plans that used to be built and then fail are simply not rewritten. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PwTc51ca2XHDCyVB7MbJoz
Building and rewriting an `AggregateExec` is how DataFusion's own physical optimizer rules work, not something external consumers should reach for. Hide that whole surface from the rendered docs, as apache#25257 does for the methods it deprecates: - the deprecated `with_limit_options`, `with_new_limit_options` and `with_new_aggr_exprs` - their replacements, `AggregateExec::builder`, `AggregateExec::to_builder` and `AggregateExecBuilder` itself - the `limit_options` getter `limit_options` is hidden but deliberately not deprecated: it has no replacement, reading the limit of an aggregate is safe, and deprecating it would only force `#[expect(deprecated)]` back into the optimizer rule that copies a limit between aggregates. Doc links from the still-public `AggregateExec::try_new` and `LimitOptions` into the hidden items are now plain code spans, since a link to a hidden item renders as a dead anchor. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PwTc51ca2XHDCyVB7MbJoz
github-actions
Bot
added
documentation
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
…write
Two defects found by review on the builder, both of which it is the
builder's job to catch:
A limit of zero on an aggregate that executes as a top-k builds fine and
then panics. `TopKHeap::new(0, ..)` has capacity 0 and no root, so
`is_full()` is immediately true and the first row read reaches
`worst_val().expect("Missing root")` in `PrimitiveHeap::is_worse`.
`build()` now rejects it. Reproduced before fixing: a `Single` aggregate
over `MIN(a)` with `LimitOptions::new(0)` panics at heap.rs:133.
A dynamic filter records which aggregate expressions are `MIN`/`MAX`, at
which index, and over which column: a `MIN` pushes `col < bound`, a `MAX`
pushes `col > bound`. `to_builder().with_aggr_exprs(..)` kept that state
verbatim, so rewriting `MIN(a)` into `MAX(a)` left the filter tagged
`Min` and pushed down a predicate that prunes the rows the aggregate
needs. The schema check does not catch it — both produce the same output
field. The state is now re-derived from the new expressions and kept only
while it still describes them, so a rewrite that merely reorders or
reverses them keeps the original filter, and with it the link to whichever
child accepted it during pushdown.
`init_dynamic_filter`'s body moves into `AggregateExec::derive_dynamic_filter`
so the builder can re-derive it; the logic is unchanged.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PwTc51ca2XHDCyVB7MbJoz
Member
Author
|
| PR | run | result |
|---|---|---|
| #76 (unrelated, not mine) | 19:03 | same job red, every other check green |
| #76 | 21:35 | same job red, every other check green |
| this PR | 04:50 | same job red, same four tests, same pull denial |
No fix exists to port: the image pull needs Docker Hub credentials (or a mirror) on the runner, which is CI configuration and outside what this PR should change. The tests are not flaky and a re-run will not help while the registry keeps denying the pull, so I have not spent one on it. A fresh run of that job is in flight anyway on the new head (674921f); if it comes back green there, this note is moot.
Everything else on this PR is mine to keep green, and I am still watching it.
Generated by Claude Code
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
2 participants
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 PR close?
AggregateExeclimit setters#[doc(hidden)] #[deprecated]and adds#[expect(deprecated)]at the call sites.Rationale for this change
The methods apache#25257 hides are hidden because they are dangerous. They are just as dangerous for the code inside DataFusion that keeps using them, and hiding them does nothing about that.
Every one of these plans can be built today, and each one only misbehaves once it is executed:
internal_err!("Ordering direction required for DISTINCT with limit")aggr.group_expr().expr()[0]panics0on a top-k aggregateworst_val().expect("Missing root")panics: a queue of capacity 0 reports itself full with an empty rootFILTERGroupedTopKAggregateStreamnever readsfilter_expr, so the filter is silently dropped and the query returns wrong resultsCOUNT/AVG/… aggregatedebug_assert!in debug builds onlyOn top of that,
AggregateExechad two nearly identical clone-with-one-change methods with different semantics (with_new_limit_optionsresets the metrics,with_limit_optionskeeps them), each hand-copying twelve fields, and atry_newwith six positional arguments, two of which are schemas that are easy to transpose (inputvsinput_schema).Validating the node when it is built makes these unrepresentable in a built plan, which means the API no longer needs to be dangerous — only internal.
What changes are included in this PR?
AggregateExecBuilder(datafusion/physical-plan/src/aggregates/builder.rs), reached throughAggregateExec::builder(mode, input)for a new node andAggregateExec::to_builder()for a rewrite:filter_exprdefaults to "no filter per aggregate"build()validates the node once: a limit is checked against the shape of the aggregate (the rows in the table above), and a replacement of the aggregate expressions is checked against both the output schema it inherits and the dynamic filter it inheritswith_mode,with_group_by,with_input,with_filter_exprs) drop that state and recompute; the others keep it, so rewrites behave exactly as they did before.A dynamic filter records which aggregate expressions are
MIN/MAX, at which index, and over which column — aMINpushes downcol < bound, aMAXpushescol > bound. Carrying it verbatim across a rewrite that turns aMINinto aMAXwould push down the wrong predicate and prune rows the aggregate needs, and the schema check cannot see it because both produce the same output field.with_aggr_exprstherefore re-derives the state and keeps the inherited filter only while it still describes the new expressions, so a rewrite that merely reorders or reverses them keeps the original filter (and with it the link to whichever child accepted it during pushdown).Migrated to the builder:
TopKAggregation,LimitedDistinctAggregation,CombinePartialFinalAggregate,OptimizeAggregateOrder, the protobuf decoder, and every test. The optimizer rules use.build().ok()?, so an invalid limit means "skip this optimization" instead of a plan that fails at runtime. No#[expect(deprecated)]anywhere.Deprecated and
#[doc(hidden)]:with_limit_options,with_new_limit_options,with_new_aggr_exprs.#[doc(hidden)]but not deprecated:AggregateExec::builder,AggregateExec::to_builder,AggregateExecBuilder, and thelimit_options()getter. Building and rewriting an aggregate is how DataFusion's own optimizer rules work, not a public API, so the whole surface is hidden — as apache#25257 does. The getter is not deprecated because reading a limit is safe and has no replacement; deprecating it would only push#[expect(deprecated)]back intoCombinePartialFinalAggregate, which lives in another crate and cannot reach the field directly.Two existing tests were building plans that cannot execute and had to be adjusted: a statistics test that put a limit on a
COUNT(a)aggregate, and the protobuf roundtrip test that put an ordered limit onAVG(b). Both now use shapes the optimizer actually produces.What is the testing strategy for this PR?
Eight new unit tests in
aggregates/builder.rscover the builder itself: defaulted filter expressions, mismatched filter arity, derived state being reused on a rewrite (asserted withArc::ptr_eqon the plan properties) and metrics being reset, the schema recompute when the mode changes, rejection of incompatible aggregate expressions, each limit validation rule (including the zero limit and the ordered-input case that used to produce an internal error at execution time), and all three dynamic-filter outcomes when the aggregate expressions are replaced — kept, rebuilt, dropped. The drop case usesSUM, whose output field matchesMIN's, so it gets past the schema check and actually reaches the dynamic filter. Two doctests on the builder and one onto_buildercover the documented usage.Both panics in the table above were reproduced before being fixed, not reasoned about: the zero limit panics at
heap.rs:133.Everything else is covered by the existing suites, which all pass:
datafusion-physical-planlib (1874) and doctestsdatafusion-physical-optimizer(33)datafusioncore_integration physical_optimizer(560)datafusion-protoaggregate roundtripssqllogictestsuitecargo clippy --all-targets --all-features --workspace -- -D warnings,cargo fmt --all, andcargo doc -p datafusion-physical-planunder-D warningsNot run: the
sql_plannerplanning benchmarks — the environment ran out of disk during the release build. The change is plan-time only (execution paths are untouched) and adds onecreate_schemacall per aggregate rewritten byOptimizeAggregateOrder, so it should be in the noise, but it has not been measured.Are there any user-facing changes?
Yes, and
docs/source/library-user-guide/upgrading/56.0.0.mdhas a section for them.#[doc(hidden)].FILTERexpressions at execution time. No plan DataFusion's own optimizer produces is affected — the rules already check these conditions before pushing a limit down.cargo-semver-checksclassifies this as requiring a major version:#[doc(hidden)]on the pre-existingAggregateExec::limit_optionsremoves it from the public API (major), and the three deprecations are a minor change. That is the intended consequence of hiding an internal API and is expected for the 56.0.0 release; theCheck semverjob reports it as a note and passes. Nothing else in the four checked crates moved.Follow-up notes
Validation catches the bad limit shapes, but
LimitOptionsstill lets you construct them. Splitting it intoSoftLimit { limit }/TopK { limit, descending }would make two of the failure modes unrepresentable rather than merely rejected. That's a wider rename, so it was left out of this PR — happy to do it here if a reviewer wants to go further.The deprecated
with_new_aggr_exprskeeps the dynamic-filter hazard described above: it copies the state verbatim, and this PR does not change its behaviour. Only the builder path re-derives it.🤖 Generated with Claude Code
https://claude.ai/code/session_01PwTc51ca2XHDCyVB7MbJoz