[SPARK-58063][SQL] Test that BIN BY is column-pruned by the generic ColumnPruning fallback

[vranes]

What changes were proposed in this pull request?

This PR adds test coverage asserting that the logical BinBy operator is column-pruned, and a binBy test DSL helper to construct the node.

• ColumnPruningSuite: a Project over BinBy prunes the unused pass-through column while keeping the range and DISTRIBUTE inputs in the child.
• dsl/package.scala: a binBy test DSL helper (alongside generate) used by the test.

Why are the changes needed?

BIN BY is row-multiplying, so forwarding unused pass-through columns through the operator wastes work proportional to the fan-out. The generic ColumnPruning fallback already prunes them; this PR adds coverage so that behavior is not silently lost.

Does this PR introduce any user-facing change?

No. BIN BY is gated off by default (spark.sql.binByRelationOperator.enabled, SPARK-57440). This is test-only, plus a test DSL helper.

How was this patch tested?

• ColumnPruningSuite: the new Column pruning for BinBy test.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Anthropic)

[cloud-fan]

0 blocking, 1 non-blocking, 0 nits. Correct and result-safe; one non-blocking design suggestion on the pruning mechanism.

Design / architecture (1)

• Optimizer.scala:1139: unrequiredChildIndex machinery buys only <=16 bytes/row and the Generate analogy doesn't transfer to BinBy — see inline

Verification

Traced the Project(_, BinBy) rewrite: the pruning Project (prunedChild) keeps every kernel-read input (requiredAttrs ⊇ b.references), and the unrequired set reduces to {rangeStart, rangeEnd} -- p.references — DISTRIBUTE columns are gated out, so no kernel input is ever pruned. Result-equivalent for empty / single / many-row input, and the synthesized Project forwards child attributes with unchanged nullability/dataType/exprId (no metadata mismatch). The rewrite is correct; the finding below is a simplification, not a correctness concern.

[cloud-fan]

Re-review: 1 addressed, 0 remaining, 1 new (late catch). The prior suggestion (drop the unrequiredChildIndex field / output override) is addressed — switched to the plain prunedChild form. 0 blocking, 1 non-blocking, 0 nits. Correct and result-safe; one non-blocking simplification question on the surviving pruning arm.

Design / architecture (1)

• Optimizer.scala:1140: the dedicated Project(_, b: BinBy) arm looks redundant with the generic Project(_, child) fallback at :1199 — BinBy isn't intercepted in between, and both reduce to the same prunedChild(b.child, ...). See inline.

Verification

Traced Project(_, BinBy) routing: no arm between :1140 and :1199 matches it (GeneratorNestedColumnAliasing handles only Generate; NestedColumnAliasing's canProjectPushThrough/canPruneOn both exclude BinBy), so absent the new arm the pattern falls through to the generic fallback. The arm's requiredAttrs = (p.references -- b.producedAttributes) ++ b.references and the fallback's child.references ++ p.references filter b.child.output to the same set — b.producedAttributes carry fresh ExprIds and are never in b.child.output, so subtracting them changes nothing — and the two guards fire on the same plans. The arm is therefore result-equivalent to the fallback (hence a redundancy finding, not a correctness one), and result-safe across empty / single / many-row input with no output-attribute metadata change.

[cloud-fan]

This arm is correct, but it looks functionally redundant with the generic Project(_, child) fallback at :1199.

Project(_, BinBy) isn't intercepted by any arm between here and :1199: GeneratorNestedColumnAliasing (:1146) matches only Project(_, g: Generate), and NestedColumnAliasing (:1195) matches Project(projectList, child) only when canProjectPushThrough(child) or canPruneOn(child) — and BinBy is in neither list (NestedColumnAliasing.scala:199-219). So without this arm, Project(_, BinBy) falls through to the generic fallback:

case p @ Project(_, child) if !child.isInstanceOf[Project] =>
  val required = child.references ++ p.references
  ...
  val newChildren = child.children.map(c => prunedChild(c, required))

For child = b: BinBy that computes required = b.references ++ p.references and calls the same prunedChild(b.child, required). Compared to this arm's requiredAttrs = (p.references -- b.producedAttributes) ++ b.references: b.producedAttributes (scaled DISTRIBUTE + appended attrs) all carry fresh ExprIds, so none are in b.child.output; subtracting them from p.references can't change which child attributes survive prunedChild's filter. The two reference sets prune the child identically, and the two guards fire on the same plans.

Unlike the Generate arm (which must set unrequiredChildIndex, a field the generic fallback can't populate) or the Expand arm (which rewrites projections), this arm does no irreducible work beyond prunedChild — so it appears eliminable. The new ColumnPruningSuite test would also still pass with the arm deleted, so it doesn't actually distinguish the arm from the fallback.

Is there a reason to keep the explicit arm — a planned future field on BinBy, or a deliberate choice to make the pruning discoverable at this call site? If not, consider dropping it and relying on the generic fallback. Non-blocking either way.

[vranes]

Good catch, you're right. Removed the arm. I kept the test (it still passes) and retitled/reframed the PR + JIRA since it's now test-only.

[cloud-fan]

Merge Summary:

• merged into master da4c3c2
• merged into branch-4.x 39f7837

Posted by merge_spark_pr.py