Skip to content

[SPARK-58064][SQL] Add filter pushdown for BIN BY#57159

Open
vranes wants to merge 1 commit into
apache:masterfrom
vranes:bin-by-filter-pushdown
Open

[SPARK-58064][SQL] Add filter pushdown for BIN BY#57159
vranes wants to merge 1 commit into
apache:masterfrom
vranes:bin-by-filter-pushdown

Conversation

@vranes

@vranes vranes commented Jul 9, 2026

Copy link
Copy Markdown
Contributor

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 ExprIds 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)

@vranes vranes changed the title [SPARK-58064][SQL] Push deterministic filters through BIN BY [SPARK-58064][SQL] Add filter pushdown for BIN BY Jul 9, 2026

@cloud-fan cloud-fan left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants