vranes opened a new pull request, #57086: URL: https://github.com/apache/spark/pull/57086
### What changes were proposed in this pull request? This PR makes the `BIN BY` relation operator (SPARK-57133) execute end-to-end. It adds the `BinByExec` physical operator and wires it into planning, replacing the stub strategy that raised `UNSUPPORTED_FEATURE.BIN_BY`. `BinByExec` emits one output row per bin overlapping each input row's `[range_start, range_end)`: - The `DISTRIBUTE UNIFORM` columns are proportionally rescaled by the bin's overlap fraction. Row assembly follows the `GenerateExec` pattern: a `JoinedRow(childRow, appendedRow)` fed through a per-partition `UnsafeProjection`, where each DISTRIBUTE column's output expression is `Cast(Multiply(Cast(col, DoubleType), ratio), col.dataType)` (the inner cast keeps FLOAT columns from forming a mismatched `Multiply(FloatType, DoubleType)`, since these expressions are bound at execution time and never pass through the analyzer's operand coercion). - The other forwarded columns, including the range columns, are replicated unchanged. - `bin_start`, `bin_end`, and `bin_distribute_ratio` are appended. Bin boundaries reuse `DateTimeUtils.timeBucketDTInterval` / `timestampAddDayTime`, matching `time_bucket`: sub-day widths use UTC microsecond arithmetic, multi-day widths use civil-time arithmetic in the session zone (UTC for `TIMESTAMP_NTZ`). Per-row edge cases: a zero-length range emits one row with ratio `1.0`; an inverted range (`range_start > range_end`) raises the runtime error `BIN_BY_INVALID_RANGE`; a NULL in either range column emits one row with all computed columns NULL (the scaled DISTRIBUTE values and all three appended columns), since no valid bin exists to scale into. Non-DISTRIBUTE passthrough columns are unaffected in every case. `BinByResolution.scaledDistributeAttributes` marks the scaled produced attributes `nullable = true` to reflect the NULL-range case. Two optimizer changes ride along, both enabled by the produced-attributes shape from SPARK-57858: - **Column pruning**: `BinBy` gains `unrequiredChildIndex` / `requiredChildOutput` (mirroring `Generate`) and a `ColumnPruning` arm that trims forwarded pass-through columns while keeping the range and DISTRIBUTE inputs the kernel reads. - **Filter pushdown**: `BinBy` is added to `PushPredicateThroughNonJoin.canPushThrough`, so deterministic predicates on forwarded pass-through / range columns push below the operator. Predicates on the scaled DISTRIBUTE or appended columns cannot push (fresh `ExprId`s not in the child output), which is what makes this safe. ### Why are the changes needed? `BIN BY` parsed, resolved, and type-checked but had no physical execution: planning threw `UNSUPPORTED_FEATURE.BIN_BY`. This PR provides the JVM execution path so the operator runs end-to-end. It is the physical-execution step of the `BIN BY` feature (SPARK-57133), following the parser/analyzer (SPARK-57133), the config gate (SPARK-57440), and the produced-attributes plan shape (SPARK-57858). ### Does this PR introduce _any_ user-facing change? No. `BIN BY` is gated off by default (`spark.sql.binByRelationOperator.enabled`, SPARK-57440) and was not usable end-to-end before this PR (execution was stubbed). This enables execution behind that gate; the gate remains off by default, so there is no change to released or default behavior. ### How was this patch tested? - `BinBySuite` (`sql/core`): end-to-end execution over the wired operator: multi-bin proportional split, single-bin passthrough, mixed FLOAT + DOUBLE DISTRIBUTE columns, downstream `GROUP BY` composability, NULL-range rows, inverted-range error (`checkError`), renamed output columns, and the gated-off rejection. - `bin-by.sql` golden file: executing scenarios plus representative analysis errors. - `ColumnPruningSuite`: a `Project` over `BinBy` prunes unused pass-through columns while keeping range and DISTRIBUTE inputs required. - `FilterPushdownSuite`: a predicate on a pass-through column pushes below `BinBy`; a predicate on a produced (fresh-`ExprId`) appended column stays above. - `ResolveBinBySuite`: existing analyzer coverage, green after the output-shape and nullability changes. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Anthropic) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
