[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š <nikolina.vranes@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: vranes

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
   }
 

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)
+  }
 }