[SPARK-58064][SQL] Add filter pushdown for BIN BY#57159
Conversation
cloud-fan
left a comment
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.
| child = child, | ||
| timeZoneId = Some("UTC")) | ||
|
|
||
| // (a) A predicate on a forwarded pass-through column (label) must push BELOW BinBy. |
There was a problem hiding this comment.
This assertion pushes a predicate on a pass-through column (label) below BinBy, but not on a range column (ts_start/ts_end). The range-column case is the subtler safety claim the PR description leans on: range columns are semantically consumed by the binning logic yet forwarded unchanged in output, which is exactly why a predicate on them is safe to push. Both take the same code path (both are child attributes), so this is coverage-completeness rather than a correctness gap — but adding a range-column predicate to this "pushes below" case would exercise that claim directly. Non-blocking.
What changes were proposed in this pull request?
This PR adds
BinBytoPushPredicateThroughNonJoin.canPushThrough, so the generic unary-node pushdown arm relocates deterministic predicates belowBIN BY.BIN BYreplicates 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.ExprIds not in the child output, sopushDownPredicatenever 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 BYis 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).Generateis in the same allowlist for the same reason.Does this PR introduce any user-facing change?
No.
BIN BYis 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 whetherBIN_BY_INVALID_RANGEis 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 belowBinBy; 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)