[SPARK-57039][SQL] Fold InnerJoin with single-row LocalRelation/OneRowRelation into Project

[yaooqinn]

What changes were proposed in this pull request?

Fold Inner Join where one side is a single-row LocalRelation or a OneRowRelation into a Project (or just the other side, when the single-row side has no columns).

Two rules, because the two relation types do not share a tree pattern:

• ConvertToLocalRelation gains two new cases for Join(LocalRelation(out, Seq(row), false, _), other, Inner, cond, JoinHint.NONE) and the symmetric right arm. The single row materializes as Alias(Literal.create(row.get(i, attr.dataType), attr.dataType), attr.name)(attr.exprId), preserving ExprId so downstream references survive. Result is Project(literals ++ other.output, other), wrapped in Filter(cond, _) if a condition exists.
• A new FoldInnerJoinWithOneRowRelation rule (tree pattern INNER_LIKE_JOIN) handles Join(OneRowRelation(), other, Inner, cond, JoinHint.NONE). OneRowRelation has zero columns, so no Project is needed — the result is just other (or Filter(cond, other)). Registered in RuleIdCollection and added to both LocalRelation early and LocalRelation batches.

A single combined rule using transformWithPruning(_.containsPattern(LOCAL_RELATION)) would miss the OneRowRelation subtrees, because OneRowRelation does not publish LOCAL_RELATION. Hence two rules.

Four guards keep both rules conservative:

• JoinType == Inner
• JoinHint == JoinHint.NONE
• LocalRelation.data.length == 1 (0 rows are handled by PropagateEmptyRelation; >1 row would Cartesian-explode)
• !cond.exists(hasUnevaluableExpr) && !other.isStreaming

The streaming guard is required because state-store ordering forbids rewriting a Join into a Project at plan time.

This fulfills the TODO in DecorrelateInnerQuery.scala:435:

// A special optimization for OneRowRelation.
// TODO: add a more general rule to optimize join with OneRowRelation.
case _: OneRowRelation => domain

This rule converges with OptimizeOneRowPlan: the new LocalRelation produced by Limit 1 over the OneRowRelation side is folded by EliminateOuterJoin/ConvertToLocalRelation in the next iteration.

Why are the changes needed?

The optimizer currently leaves a Join node in plans like the ones below, which blocks downstream column pruning and constant folding on the single-row side:

SELECT * FROM t CROSS JOIN (SELECT 1 AS c) s
SELECT * FROM t JOIN (VALUES (1, 'x')) AS s(a, b) ON t.k = s.a

Before:

Join Inner, (t.k = s.a)
:- Relation t
+- LocalRelation [a, b], [[1, x]]

After:

Project [t.*, 1 AS a, x AS b]
+- Filter (t.k = 1)
   +- Relation t

This pattern shows up in BI-tool-generated SQL and in views whose body collapses to a single row after other rules fold it.

Does this PR introduce any user-facing change?

No. Query results are unchanged; only the optimized logical plan shape changes.

How was this patch tested?

New unit tests:

• FoldInnerJoinWithOneRowRelationSuite — 9 cases: OneRow × table (left/right/no condition), LeftOuter not folded, Array/Map/Struct/nested-struct columns preserved on the other side, condition with Rand() (Unevaluable) not folded.
• ConvertToLocalRelationSuite — 4 new cases (T7–T10): LocalRelation × non-LocalRelation symmetric, condition referencing both sides folded into Filter, and ExprId preservation with collectFirst { case _: Join }.isEmpty && output.length == 4.

The existing PlanStability* suites pass with no plan diff for any TPC-DS / TPC-H query — none of them currently contain the targeted pattern.

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

Generated-by: Claude Opus 4.7

[dongjoon-hyun]

This looks like out-of-order. I guess this file is in the alphabetical order in general. May I ask why do you insert this at this line?

[dongjoon-hyun]

nit. Redundant s".

[dongjoon-hyun]

ditto.

[dongjoon-hyun]

In the PR description, there is !other.isStreaming, but the PR code seems to have none. Could you double-check?

Four guards keep both rules conservative:
...

• !cond.exists(hasUnevaluableExpr) && !other.isStreaming

In addition, FoldInnerJoinWithOneRowRelationSuite seems to have no Rand() test. Could you add a test coverage?

New unit tests:

• FoldInnerJoinWithOneRowRelationSuite — 9 cases: OneRow × table (left/right/no condition), LeftOuter not folded, Array/Map/Struct/nested-struct columns preserved on the other side, condition with Rand() (Unevaluable) not folded.

[shrirangmhalgi]

Minor question: the PR description mentions "condition with Rand() (Unevaluable) not folded" - but Rand is extending Nondeterministic, not Unevaluable, so hasUnevaluableExpr would return false and the join would still be folded. Is this intentional?

If so, might be worth updating the description to avoid confusion.