This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 41b27393691a [SPARK-58064][SQL] Add filter pushdown for BIN BY
41b27393691a is described below
commit 41b27393691a0a9712960ab2784efbcd07eec720
Author: Nikolina Vraneš <[email protected]>
AuthorDate: Mon Jul 13 00:23:31 2026 +0800
[SPARK-58064][SQL] Add filter pushdown for BIN BY
### 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š <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../spark/sql/catalyst/optimizer/Optimizer.scala | 1 +
.../catalyst/optimizer/FilterPushdownSuite.scala | 46 ++++++++++++++++++++++
2 files changed, 47 insertions(+)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index 597a70e01a6e..53ddc53091fb 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -2311,6 +2311,7 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
case _: BatchEvalPython => true
case _: ArrowEvalPython => true
case _: Expand => true
+ case _: BinBy => true
case _ => false
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
index f457eced4d98..a43be9a1c0a6 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
@@ -1644,4 +1644,50 @@ class FilterPushdownSuite extends PlanTest {
.analyze
comparePlans(optimizedQueryWithoutStep, correctAnswer)
}
+
+ test("push down deterministic predicate through BinBy") {
+ // Relation: ts_start, ts_end, value (DISTRIBUTE), label (pass-through).
+ val tsStart = AttributeReference("ts_start", TimestampType, nullable =
false)()
+ val tsEnd = AttributeReference("ts_end", TimestampType, nullable = false)()
+ val value = AttributeReference("value", DoubleType, nullable = false)()
+ val label = AttributeReference("label", StringType, nullable = true)()
+ val relation = LocalRelation(tsStart, tsEnd, value, label)
+
+ // Produced attributes: scaled DISTRIBUTE output and appended BIN BY
columns.
+ val scaledValue = AttributeReference("value", DoubleType, nullable =
true)()
+ val binStart = AttributeReference("bin_start", TimestampType, nullable =
true)()
+ val binEnd = AttributeReference("bin_end", TimestampType, nullable =
true)()
+ val binRatio = AttributeReference("bin_distribute_ratio", DoubleType,
nullable = true)()
+
+ def binByOver(child: LogicalPlan): BinBy = BinBy(
+ binWidthMicros = 300000000L,
+ rangeStart = tsStart,
+ rangeEnd = tsEnd,
+ originMicros = 0L,
+ distributeColumns = Seq(value),
+ scaledDistributeColumns = Seq(scaledValue),
+ appendedAttributes = Seq(binStart, binEnd, binRatio),
+ child = child,
+ timeZoneId = Some("UTC"))
+
+ // (a) A predicate on a forwarded pass-through column (label) must push
BELOW BinBy.
+ val queryA = Filter(EqualTo(label, Literal("x")), binByOver(relation))
+ val optimizedA = Optimize.execute(queryA)
+ val expectedA = binByOver(Filter(EqualTo(label, Literal("x")), relation))
+ comparePlans(optimizedA, expectedA, checkAnalysis = false)
+
+ // (b) A predicate on a range column (ts_start) must push BELOW BinBy. The
range columns are
+ // consumed by the binning logic yet forwarded unchanged in output, so
their value is identical
+ // on every sub-row and filtering before vs. after binning is equivalent.
+ val tsLiteral = Literal(0L, TimestampType)
+ val queryB = Filter(GreaterThan(tsStart, tsLiteral), binByOver(relation))
+ val optimizedB = Optimize.execute(queryB)
+ val expectedB = binByOver(Filter(GreaterThan(tsStart, tsLiteral),
relation))
+ comparePlans(optimizedB, expectedB, checkAnalysis = false)
+
+ // (c) A predicate on a produced (fresh-ExprId) appended column must stay
ABOVE BinBy.
+ val queryC = Filter(GreaterThan(binRatio, Literal(0.5)),
binByOver(relation))
+ val optimizedC = Optimize.execute(queryC)
+ comparePlans(optimizedC, queryC, checkAnalysis = false)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
