[SPARK-58093][SQL] Add ASOF JOIN SQL syntax with MATCH_CONDITION

[srielau]

Wire the SQL Reference Spec ASOF JOIN surface into the existing AsOfJoin logical operator, reusing the DataFrame execution path (RewriteAsOfJoin / SortMergeAsOfJoinExec). Adds parser grammar, analysis resolution for USING and MATCH_CONDITION operands, spec error conditions, and parser/SQL integration tests.

What changes were proposed in this pull request?

This PR adds SQL syntax for ASOF JOIN with a required MATCH_CONDITION clause and optional ON / USING, per the SQL Language Reference spec (Snowflake / Calcite MATCH_CONDITION family).

Syntax supported (v1):

[ INNER | LEFT [ OUTER ] ] ASOF JOIN right_table
  MATCH_CONDITION ( left_expr comparison_operator right_expr )
  [ ON boolean_expression | USING ( column_name [, ...] ) ]

where comparison_operator is one of >=, >, <=, <.

Example:

SELECT t.trade_time, t.symbol, q.bid_price
FROM trades t
ASOF JOIN quotes q
  MATCH_CONDITION (t.trade_time >= q.quote_time)
  ON t.symbol = q.symbol;

Implementation layers (new vs reused):

Layer	Change
Parser	ASOF, MATCH_CONDITION keywords; asofJoinType, asofJoinCriteria, matchComparison grammar rules
AstBuilder	withAsOfJoin() builds AsOfJoin.fromMatchCondition(...)
Analysis	New ResolveAsOfJoin rule: expands USING via NaturalAndUsingJoinResolution, normalizes MATCH_CONDITION operands into asOfCondition + orderExpression, validates operand expressions/types
Logical plan	Extends AsOfJoin with deferred matchComparison / usingColumns fields; adds MatchComparisonOperator and AsOfJoin.fromMatchCondition
Execution	No changes — reuses existing RewriteAsOfJoin optimizer rule and SortMergeAsOfJoinExec physical operator already used by DataFrame.joinAsOf
Errors	Four new error classes: ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION, ..._INVALID_OPERATOR, ..._INVALID_TYPE, ..._TABLE_REFERENCE

Explicitly out of scope (deferred per spec): TOLERANCE, NEAREST direction, RIGHT / FULL / CROSS / SEMI / ANTI ASOF forms.

Related prior art in Spark: DataFrame.joinAsOf (PySpark/Scala), AsOfJoin logical node, SortMergeAsOfJoinSuite, DataFrameAsOfJoinSuite.

Why are the changes needed?

ASOF (as-of / closest-match) join is a standard temporal/analytics pattern — attaching the nearest preceding quote to each trade, finding the next maintenance window for an alert, etc. Spark already implements the semantics via the DataFrame API, but SQL users must hand-write LEFT JOIN + ROW_NUMBER + QUALIFY rewrites. Native SQL syntax lowers the barrier and aligns with industry dialects (Snowflake, Calcite/Pinot/Doris MATCH_CONDITION family).

Does this PR introduce any user-facing change?

Yes.

Before: ASOF JOIN was not valid SQL; closest-match joins were only available through DataFrame.joinAsOf.

After: SQL queries may use ASOF JOIN / LEFT ASOF JOIN with MATCH_CONDITION. Behavior matches the spec: asymmetric left-driven lookup, at most one right row per left row, INNER drops unmatched left rows, LEFT preserves them with NULL right-side columns.

Operand order inside MATCH_CONDITION is immaterial (t.ts >= q.ts ≡ q.ts <= t.ts); the analyzer normalizes to a canonical left-table-on-left form before planning.

How was this patch tested?

Added unit and integration tests; full compile/test suite to run on CI (local build blocked by missing netty 4.2.16 artifacts on the dev Maven proxy after rebasing onto latest master).

Parser tests (PlanParserSuite):

• INNER / LEFT ASOF JOIN with MATCH_CONDITION and ON
• USING clause
• Swapped operand order (u.a <= t.a)
• Rejection of = in MATCH_CONDITION (parse error)

SQL integration tests (AsOfJoinSQLSuite):

• Spec examples: trades/quotes with ON, LEFT ASOF preserving unmatched rows, USING equivalence, first-following match with <=

Existing coverage reused (unchanged): DataFrameAsOfJoinSuite, SortMergeAsOfJoinSuite, RewriteAsOfJoinSuite exercise the shared execution path.

Suggested CI commands for reviewers:

build/sbt "sql/catalyst/testOnly org.apache.spark.sql.catalyst.parser.PlanParserSuite -- -z \"asof join\""
build/sbt "sql/testOnly org.apache.spark.sql.AsOfJoinSQLSuite"
build/sbt "sql/testOnly org.apache.spark.sql.DataFrameAsOfJoinSuite"

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

Yes. Generated-by: Cursor (Claude agent), with human review and editing by the PR author.