[SPARK-58003][SQL] Add physical execution for BIN BY

[vranes]

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 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 rescaled by the bin's overlap fraction. Row assembly follows GenerateExec: a JoinedRow(childRow, appendedRow) run through a per-partition UnsafeProjection, with each DISTRIBUTE column's expression Cast(Multiply(Cast(col, DoubleType), ratio), col.dataType). The inner cast keeps FLOAT columns from forming a mismatched Multiply(FloatType, DoubleType), since these expressions bind at execution time and skip 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.

producedAttributes declares the scaled and appended attributes, so missingInput is empty and EXPLAIN does not flag the plan invalid.

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 BIN_BY_INVALID_RANGE; a NULL in either range column emits one row with the scaled DISTRIBUTE values and all three appended columns NULL (no valid bin to scale into). Passthrough columns are unaffected in every case. To cover the NULL case, BinByResolution.scaledDistributeAttributes marks the scaled produced attributes nullable = true.

The two BIN BY optimizer follow-ons are not in this PR: column pruning (SPARK-58063, #57157) and filter pushdown (SPARK-58064, #57159) shipped separately, both already merged. This PR is the operator + planning only and does not touch Optimizer.scala or the optimizer test suites.

Why are the changes needed?

BIN BY parsed, resolved, and type-checked but had no physical execution: planning threw UNSUPPORTED_FEATURE.BIN_BY. This 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 stubbed before this PR. Execution is now behind that gate, which stays off by default, so released and default behavior are unchanged.

How was this patch tested?

• BinBySuite (sql/core): end-to-end execution: multi-bin proportional split, single-bin passthrough, mixed FLOAT + DOUBLE DISTRIBUTE columns, downstream GROUP BY, a zero-length range (ratio 1.0), civil-time boundaries across a DST spring-forward (the 23-hour day gets a smaller ratio than the 24-hour day), a NULL in either range column, inverted-range error (checkError), renamed output columns, and the gated-off rejection.
• bin-by.sql golden file: executing scenarios plus representative analysis errors.
• 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)

[cloud-fan]

0 blocking, 2 non-blocking, 0 nits.
A clean, well-tested operator faithfully following the GenerateExec pattern; the correctness core checks out. Two non-blocking suggestions.

Design / architecture (1)

• BinByExec.scala:78: outputPartitioning not overridden (falls back to UnknownPartitioning); unlike GenerateExec, can force a needless shuffle downstream — see inline

Correctness (1)

• BinByExec.scala:87: NTZ execution path (timeZoneId=None -> UTC) has no end-to-end test coverage — see inline

Verification

Traced the bin-tiling in doExecute: curStart starts at timeBucketDTInterval(rs) <= rs and advances by timestampAddDayTime(width) (strictly increasing since width > 0), hasNext = curStart < re, so bins tile [rs, re) and overlaps sum to total = re - rs, hence ratios sum to 1.0. The NULL-range, zero-length (rs == re, ratio 1.0), and inverted (rs > re, error) branches each check out. Scaled-attribute nullable = true is required and correct: the null-range path emits Literal null for a possibly-non-null input column. FLOAT narrowing via the inner Cast(col, Double) avoids a bind-time Multiply(FloatType, DoubleType) mismatch, since these expressions bind post-analysis.

[cloud-fan]

BinByExec doesn't override outputPartitioning, so it falls back to the SparkPlan default UnknownPartitioning(0). The GenerateExec analogue this operator otherwise mirrors sets outputPartitioning = child.outputPartitioning (GenerateExec.scala:74). As a result, a downstream operator clustered on a passthrough column the child already partitions on would get an unnecessary shuffle here.

Non-blocking. Note the naive fix (copy child.outputPartitioning verbatim) would be unsafe: DISTRIBUTE columns are rescaled with fresh exprIds, so a child HashPartitioning on a DISTRIBUTE column would reference an attribute that no longer exists in output. The safe enhancement preserves the child's partitioning restricted to passthrough columns (those whose exprId survives in output).

[cloud-fan]

The timeZoneId = None -> ZoneOffset.UTC branch is the NTZ execution path, but BinBySuite's NTZ coverage only calls .assertAnalyzed() — NTZ inputs are never actually executed via checkAnswer/collect. So the NTZ branch of the boundary arithmetic and error formatting has no runtime coverage, even though the LTZ paths are covered thoroughly.

Non-blocking: consider adding one end-to-end NTZ checkAnswer (NTZ differs from LTZ in the origin default and the absence of DST, both worth one executed assertion).