[srielau]

What changes were proposed in this pull request?

Second stacked PR for SPARK-58092 / SPARK-58123.

Follows #57251 (parser, merged). Rebased onto current master; this PR is a 2-commit analysis-only delta.

Analysis rule

• Add ResolveAsOfJoin to materialize MATCH_CONDITION into asOfCondition, orderExpression, and per-side sort expressions
• Expand USING (...) into equi-join predicates via NaturalAndUsingJoinResolution
• Validate MATCH_CONDITION operands: cross-side references, determinism, and type compatibility (including struct/array)

Logical plan

• Extend AsOfJoin with leftSortExprs, rightSortExprs, and requiresSortMergeAsOfJoin
• Add resolveMatchComparison / matchSortExpressions helpers shared with the DataFrame API path
• Resolve matchLeftOperand / matchRightOperand via the standard analyzer expression walk before materialization

Gating

• CheckAnalysis rejects SQL ASOF plans when spark.sql.join.sortMergeAsOfJoin.enabled=false (AS_OF_JOIN.SORT_MERGE_REQUIRED)

Out of scope (follow-up PR)

• Sort-merge physical operator wiring for multi-column sort keys and SQL execution tests — [SPARK-58124][SQL] Wire sort-merge ASOF JOIN physical execution #57277

Why are the changes needed?

Parser (#57251) produces unresolved AsOfJoin nodes with parsed MATCH_CONDITION operands. Analysis must resolve operands, validate semantics, and produce the same logical shape the existing DataFrame joinAsOf path uses before physical planning can run.

Does this PR introduce any user-facing change?

Yes (behind spark.sql.join.asofJoin.enabled, default off). With the parser conf enabled, SQL ASOF JOIN now progresses through analysis: valid queries become analyzable logical plans; invalid MATCH_CONDITION expressions fail with structured analysis errors instead of reaching execution unresolved.

How was this patch tested?

Added analysis negative tests to AsOfJoinSQLSuite:

• AS_OF_JOIN.SORT_MERGE_REQUIRED
• ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE (cross-side and invalid refs)
• ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION
• ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE

Locally:

build/sbt "sql/testOnly org.apache.spark.sql.AsOfJoinSQLSuite" \
          "catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.RewriteAsOfJoinSuite"

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

Co-authored with Cursor.

[cloud-fan]

0 blocking, 2 non-blocking, 1 nit.
Clean analysis layer; the main issue is a validation/planner-capability gap surfaced as a question, behind off-by-default confs.

Correctness (1)

• ResolveAsOfJoin.scala:149: analysis accepts operand types whose orderExpression the planner's findFromOrder can't parse (STRING/struct/array) → cryptic IllegalStateException at planning — see inline

Suggestions (1)

• AsOfJoinSQLSuite.scala:41: no positive analyze-only test for a valid MATCH_CONDITION — see inline

Nits: 1 minor item (see inline comments).

Verification

Traced the SQL MATCH_CONDITION path end-to-end: validateMatchConditionOperands admits any orderable, type-compatible operands, but AsOfJoinSelection.findFromOrder (SparkStrategies.scala:514-528) recognizes only Subtract/If(_,Subtract,_). STRING operands yield If(EqualTo,0,...) and struct/array operands yield CreateStruct/ZipWith, none of which findFromOrder matches — so with both confs on these queries pass analysis, skip the rewrite, and throw IllegalStateException at planning. Subtractable-scalar operands (the DataFrame path's only shape) are unaffected.

[cloud-fan]

5 addressed, 0 remaining, 2 new. (2 = 2 newly introduced, 0 late catches.)

0 blocking, 2 non-blocking, 0 nits.
The planner-capability gap and both follow-ups are correctly resolved, and the two CI-gate breakages I was about to flag (the mis-ordered error-condition key and the over-length line) were fixed by 7d9bc48, which you pushed while I was reviewing. The fix is sound. What remains are two non-blocking cleanups left by the refactor.

Design / architecture (2)

• basicLogicalOperators.scala:2631: resolveMatchComparison is now dead code after the split — see inline
• ResolveAsOfJoin.scala:200: supportsSubtractForPlanner duplicates AsOfJoin.supportsSubtract — see inline

Verification

Traced the fix against the accepted (scalar) path: it inserts a type gate and splits resolveMatchComparison into normalizeMatchOperands + materializeMatchComparison, whose composition is byte-identical to the old body — no expression construction changed, so rows are unaffected. The gated operand domain (supportsSubtractForPlanner: numeric/datetime/interval) is exactly what the planner's findFromOrder consumes (Subtract / If(_, Subtract, _), SparkStrategies.scala:517-528) and what the DataFrame joinAsOf analogue produces, so the previously planner-unsupported STRING/struct/array operands now fail at analysis with a clear error instead of IllegalStateException at planning.

[cloud-fan]

2 addressed, 0 remaining, 0 new. (0 = 0 newly introduced, 0 late catches.)

0 blocking, 0 non-blocking, 0 nits.
Both prior non-blocking findings — the dead resolveMatchComparison and the duplicated supportsSubtractForPlanner — are correctly and completely resolved in e44becab19e. The removals leave no dangling callers or orphaned imports, and reusing AsOfJoin.supportsSubtract is byte-identical to the old predicate, so operand-gating behavior is unchanged. LGTM.

Verification

Traced the delta against the accepted path: it touches no expression construction — materializeMatchComparison / normalizeMatchOperands bodies are unchanged — and the reused AsOfJoin.supportsSubtract has the same 7-type arm (numeric / datetime / interval) as the removed local copy, so the operand-support gate's boolean output is identical for every dataType. The reuse keeps the analysis gate coupled to exactly what buildLeafOrderExpression routes down the Subtract path (basicLogicalOperators.scala:2762) and what the planner's findFromOrder can consume (Subtract / If(_, Subtract, _), SparkStrategies.scala:517-524), removing the silent-drift hazard the prior review flagged.

[cloud-fan]

Merge Summary:

• merged into master ea9d6a7
• merged into branch-4.x 081b4e0

Posted by merge_spark_pr.py