[SPARK-58064][SQL] Add filter pushdown for BIN BY - #57159
Closed
vranes wants to merge 2 commits into
Closed
Conversation
vranes
changed the title
cloud-fan
approved these changes
Jul 9, 2026
cloud-fan
left a comment
cloud-fan
left a comment
Contributor
There was a problem hiding this comment.
0 blocking, 1 non-blocking, 0 nits.
Clean, minimal, correctly-reasoned optimizer change that follows the established row-multiplying-operator pushdown pattern (Generate/Expand).
Suggestions (1)
- FilterPushdownSuite.scala:1673: "pushes below" assertion covers a pass-through column but not a range column, the subtler safety claim — see inline (non-blocking)
Verification
Traced the Filter(p, BinBy(child)) -> BinBy(Filter(p, child)) rewrite: it is row-count-equivalent because filtering commutes with BIN BY's per-input-row expansion, for empty / one / many-row input. A predicate on a scaled DISTRIBUTE or appended column is gated out by the cond.references.subsetOf(grandchild.outputSet) check (Optimizer.scala:2329) because those are producedAttributes with fresh ExprIds (basicLogicalOperators.scala:1840), so it stays above; only child-forwarded pass-through / range columns push below. Nondeterministic predicates and operators are gated at 2326 and 2293. No input produces a different result.
cloud-fan
pushed a commit
that referenced
this pull request
Jul 12, 2026
### What changes were proposed in this pull request? This PR adds `BinBy` to `PushPredicateThroughNonJoin.canPushThrough`, so the generic unary-node pushdown arm relocates deterministic predicates below `BIN BY`. - A predicate on a forwarded pass-through or range column pushes below the operator: `BIN BY` replicates those columns unchanged across every sub-row of an input row, so filtering before binning is equivalent to filtering after, and pushing avoids expanding rows that would be discarded. - A predicate on a scaled DISTRIBUTE or appended column cannot push, and stays above: those are produced attributes with fresh `ExprId`s not in the child output, so `pushDownPredicate` never treats them as a subset of the child. This is safe because of the produced-attributes shape from SPARK-57858. Before that change the output DISTRIBUTE column carried the child's `ExprId`, so a predicate on it would have pushed below and filtered the unscaled value. ### Why are the changes needed? `BIN BY` is row-multiplying, so pushing a predicate on a pass-through / range column below the operator filters input rows before binning and avoids wasted expansion (e.g. `WHERE host = '...'` or a time-range predicate). `Generate` is in the same allowlist for the same reason. ### Does this PR introduce _any_ user-facing change? No. `BIN BY` is gated off by default (`spark.sql.binByRelationOperator.enabled`, SPARK-57440). This is an optimizer-only change and does not alter query results. A predicate pushed below the operator can filter out an inverted-range row before it is processed, so whether `BIN_BY_INVALID_RANGE` is raised may depend on the plan, as with any data-dependent error on a discarded row. ### How was this patch tested? - `FilterPushdownSuite`: a predicate on a pass-through column pushes below `BinBy`; a predicate on a produced (fresh-`ExprId`) appended column stays above. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Anthropic) Closes #57159 from vranes/bin-by-filter-pushdown. Authored-by: Nikolina Vraneš <nikolina.vranes@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> (cherry picked from commit 41b2739) Signed-off-by: Wenchen Fan <wenchen@databricks.com>
Contributor
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.
What changes were proposed in this pull request?
Adds
BinBytoPushPredicateThroughNonJoin.canPushThrough, letting predicates to be pushed belowBIN BY.By design predicates on pass-through columns are pushed down, predicates on the appended columns stay above.
Why are the changes needed?
BIN BYmultiplies rows, so filtering before binning avoids expanding rows that get discarded anyway (e.g.WHERE host = '...'). It's safe because pass-through columns are unchanged by binning.Does this PR introduce any user-facing change?
No.
BIN BYis off by default (spark.sql.binByRelationOperator.enabled).How was this patch tested?
FilterPushdownSuite: a filter on a pass-through column pushes belowBinBy; a filter on an appended column stays above.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic)