[SPARK-56977][SQL] RewriteNearestByJoin should respect joinType in the synthetic join

[zhidongqu-db]

What changes were proposed in this pull request?

This PR changes RewriteNearestByJoin to construct its synthetic cross-join with the user's joinType (Inner or LeftOuter) instead of always using LeftOuter. The Generate operator's outer flag continues to be derived from joinType == LeftOuter, so the externally observable semantics are unchanged.

Why are the changes needed?

The original implementation hardcoded the synthetic join to LeftOuter and justified it on the grounds that LEFT OUTER and INNER are equivalent for an unconditioned join when the right side is non-empty, and Generate(outer = false) would drop unwanted rows for INNER when right is empty.

That reasoning holds for correctness but has a major performance cost:

• INNER NEAREST BY cannot be planned as a Cartesian product. Spark's strategy picks CartesianProductExec only for Inner joins with no condition; an unconditioned LeftOuter join falls back to BroadcastNestedLoopJoin, which tries to broadcast the right side. When the right relation is large, the broadcast either OOMs or exceeds spark.sql.autoBroadcastJoinThreshold and the planner is left with no good option. CartesianProductExec partitions both sides and streams pairs, so it scales naturally with right-side size. Respecting the user's INNER join type re-enables this strategy for the common INNER NEAREST BY case.
• It also makes the EXPLAIN output misleading (shows LeftOuter even though the user wrote INNER).
• For INNER with an empty right side, the old plan generates one row per left input and then filters them away via Generate(outer = false) and the size(matches) > 0 filter -- extra work that respecting joinType avoids at the source.

Does this PR introduce any user-facing change?

No change in query results. EXPLAIN output for INNER NEAREST BY queries now shows Inner rather than LeftOuter for the synthetic join node, and the physical plan for such queries can now use CartesianProductExec instead of BroadcastNestedLoopJoin when the right relation is too large to broadcast.

How was this patch tested?

• RewriteNearestByJoinSuite: expectedRewrite now takes a joinType: JoinType and the existing tests (similarity/distance x inner/leftouter, EXACT, boundary k, self-join, nondeterministic ranking) assert the synthetic join matches the user's join type.
• Golden file sql-tests/results/join-nearest-by.sql.out

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

Coauthored-by: Claude Code (Opus 4.7), human-reviewed and tested

[cloud-fan]

Summary

Prior state and problem. RewriteNearestByJoin hardcoded the synthetic cross-join to LeftOuter regardless of the user's joinType, reasoning that LeftOuter ≡ Inner for unconditioned joins with non-empty right, and that Generate(outer=false) + the inferred size(matches) > 0 filter would mop up the empty-right INNER case. That preserved semantics but hid a planner cliff: SparkStrategies only picks CartesianProductExec when joinType.isInstanceOf[InnerLike] (SparkStrategies.scala:394), so an unconditioned LeftOuter always falls through to BroadcastNestedLoopJoinExec. For large right sides, that means broadcast OOMs or the planner is wedged.

Design approach. Pass the user's joinType (already constrained to Inner or LeftOuter at NearestByJoin.scala:66) into the synthetic Join, and keep deriving Generate.outer as joinType == LeftOuter. Externally observable semantics are unchanged; EXPLAIN no longer misrepresents the user's INNER as LeftOuter, and INNER queries become eligible for CartesianProductExec. CheckCartesianProducts (Optimizer.scala:2548) already matches both Inner | LeftOuter, so the spark.sql.crossJoin.enabled = false gate behaves identically.

Key design decision. Routing the user's joinType directly into the synthetic Join (rather than e.g. introducing an Inner-only fast path) keeps the rewrite a single shape and lets the existing optimizer machinery do its job: EliminateOuterJoin can opportunistically narrow LeftOuter to Inner when downstream operators reject NULL right rows; the cross-join check fires uniformly; the SparkStrategies build-side logic picks the right physical operator per join type (the golden file's BuildLeft, Inner vs. previous BuildRight, LeftOuter reflects that).

Findings below are minor -- non-blocking.

General finding -- SQL coverage for INNER NEAREST BY against an empty right side

inputs/join-nearest-by.sql:35-37 covers LEFT OUTER NEAREST BY against an empty right side, but there's no analogous INNER NEAREST BY case. Before this PR, INNER against empty right ran through a join that materialized a row per left + NULL right and was dropped downstream via the Generate-side filter. After this PR, INNER empties at the join (0 rows -> 0 groups -> 0 output), which is a structurally different path. A trivial users INNER JOIN (SELECT * FROM products WHERE false) ... APPROX NEAREST 1 BY ... asserting an empty result would lock that in end-to-end.

[zhidongqu-db]

@cloud-fan Added test coverage for both comments

[dtenedor]

LGTM, thanks for adding the extra test coverage requested.

[dtenedor]

LGTM, merging to master and 4.2