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

common/utils/src/main/resources/error/error-conditions.json

@@ -156,6 +156,30 @@
     ],
     "sqlState" : "42713"
   },
+  "ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION" : {
+    "message" : [
+      "The MATCH_CONDITION operand contains an invalid expression: <expr>. Rewrite the query to avoid subqueries, aggregate functions, window functions, generator functions, or non-deterministic functions in MATCH_CONDITION operands."
+    ],
+    "sqlState" : "42903"
+  },
+  "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR" : {
+    "message" : [
+      "The MATCH_CONDITION operator must be one of >=, >, <=, <. Got: <operator>."
+    ],
+    "sqlState" : "42K0E"
+  },
+  "ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE" : {
+    "message" : [
+      "The MATCH_CONDITION operands must be of an orderable, mutually comparable type. Got: <type1> and <type2>."
+    ],
+    "sqlState" : "42K09"
+  },
+  "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE" : {
+    "message" : [
+      "Each operand of the MATCH_CONDITION comparison must reference only columns of one join input. Got: <refs1> and <refs2>."
+    ],
+    "sqlState" : "42K0E"
+  },
   "ASSIGNMENT_ARITY_MISMATCH" : {
     "message" : [
       "The number of columns or variables assigned or aliased: <numTarget> does not match the number of source expressions: <numExpr>."
@@ -167,6 +191,11 @@
       "Invalid as-of join."
     ],
     "subClass" : {
+      "SORT_MERGE_REQUIRED" : {
+        "message" : [
+          "SQL ASOF JOIN requires the sort-merge physical operator. Set <config> to true."
+        ]
+      },
       "TOLERANCE_IS_NON_NEGATIVE" : {
         "message" : [
           "The input argument `tolerance` must be non-negative."
@@ -8231,6 +8260,11 @@
           "The ANALYZE TABLE command does not support views."
         ]
       },
+      "ASOF_JOIN" : {
+        "message" : [
+          "SQL ASOF JOIN syntax. Set <config> to true to enable it."
+        ]
+      },
       "BIN_BY" : {
         "message" : [
           "The BIN BY relation operator is not yet supported."

docs/sql-ref-ansi-compliance.md

@@ -430,6 +430,7 @@ Below is a list of all the keywords in Spark SQL.
 |AS|reserved|non-reserved|reserved|
 |ASC|non-reserved|non-reserved|non-reserved|
 |ASENSITIVE|non-reserved|non-reserved|non-reserved|
+|ASOF|non-reserved|non-reserved|non-reserved|
 |AT|non-reserved|non-reserved|reserved|
 |ATOMIC|non-reserved|non-reserved|non-reserved|
 |AUTHORIZATION|reserved|non-reserved|reserved|
@@ -641,6 +642,7 @@ Below is a list of all the keywords in Spark SQL.
 |MACRO|non-reserved|non-reserved|non-reserved|
 |MAP|non-reserved|non-reserved|non-reserved|
 |MATCHED|non-reserved|non-reserved|non-reserved|
+|MATCH_CONDITION|non-reserved|non-reserved|non-reserved|
 |MATERIALIZED|non-reserved|non-reserved|non-reserved|
 |MAX|non-reserved|non-reserved|non-reserved|
 |MEASURE|non-reserved|non-reserved|non-reserved|

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4

@@ -145,6 +145,7 @@ APPROX: 'APPROX';
 ARCHIVE: 'ARCHIVE';
 ARRAY: 'ARRAY' {incComplexTypeLevelCounter();};
 AS: 'AS';
+ASOF: 'ASOF';
 ASC: 'ASC';
 ASENSITIVE: 'ASENSITIVE';
 AT: 'AT';
@@ -358,6 +359,7 @@ LOOP: 'LOOP';
 MACRO: 'MACRO';
 MAP: 'MAP' {incComplexTypeLevelCounter();};
 MATCHED: 'MATCHED';
+MATCH_CONDITION: 'MATCH_CONDITION';
 MATERIALIZED: 'MATERIALIZED';
 MAX: 'MAX';
 MEASURE: 'MEASURE';

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

@@ -75,7 +75,7 @@ options { tokenVocab = SqlBaseLexer; }
            la == AS || la == WHERE || la == PIVOT || la == UNPIVOT ||
            la == TABLESAMPLE || la == INNER || la == CROSS || la == LEFT ||
            la == RIGHT || la == FULL || la == NATURAL || la == SEMI ||
-           la == ANTI || la == JOIN || la == UNION || la == EXCEPT ||
+           la == ANTI || la == ASOF || la == JOIN || la == UNION || la == EXCEPT ||
            la == SETMINUS || la == INTERSECT || la == ORDER || la == CLUSTER ||
            la == DISTRIBUTE || la == SORT || la == LIMIT || la == OFFSET ||
            la == AGGREGATE || la == WINDOW || la == LATERAL || la == BIN;
@@ -1104,8 +1104,24 @@ relationExtension
     ;
 
 joinRelation
-    : (joinType) JOIN LATERAL? right=relationPrimary (joinCriteria | nearestByClause)?
+    : (joinType) JOIN LATERAL? right=relationPrimary asofJoinPostfix?
     | NATURAL joinType JOIN LATERAL? right=relationPrimary
+    | asofJoinType ASOF JOIN right=relationPrimary asofJoinCriteria
+    ;
+
+asofJoinType
+    : INNER?
+    | LEFT OUTER?
+    ;
+
+asofJoinPostfix
+    : joinCriteria
+    | nearestByClause
+    ;
+
+asofJoinCriteria
+    : MATCH_CONDITION LEFT_PAREN matchExpr=booleanExpression RIGHT_PAREN
+      ( ON onExpr=booleanExpression | USING identifierList )?
     ;
 
 joinType
@@ -2013,6 +2029,7 @@ ansiNonReserved
     | ARRAY
     | ASC
     | ASENSITIVE
+    | ASOF
     | AT
     | ATOMIC
     | AUTO
@@ -2180,6 +2197,7 @@ ansiNonReserved
     | MACRO
     | MAP
     | MATCHED
+    | MATCH_CONDITION
     | MATERIALIZED
     | MAX
     | MEASURE
@@ -2404,6 +2422,7 @@ nonReserved
     | ARRAY
     | AS
     | ASC
+    | ASOF
     | ASENSITIVE
     | AT
     | ATOMIC
@@ -2610,6 +2629,7 @@ nonReserved
     | MACRO
     | MAP
     | MATCHED
+    | MATCH_CONDITION
     | MATERIALIZED
     | MAX
     | MEASURE

sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala

@@ -902,6 +902,22 @@ private[sql] object QueryParsingErrors extends DataTypeErrorsBase {
       ctx)
   }
 
+  def sqlAsofJoinDisabled(configKey: String, ctx: ParserRuleContext): Throwable = {
+    new ParseException(
+      errorClass = "UNSUPPORTED_FEATURE.ASOF_JOIN",
+      messageParameters = Map("config" -> toSQLConf(configKey)),
+      ctx)
+  }
+
+  def sqlAsOfJoinMatchConditionInvalidOperator(
+      operator: String,
+      ctx: ParserRuleContext): Throwable = {
+    new ParseException(
+      errorClass = "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR",
+      messageParameters = Map("operator" -> operator),
+      ctx)
+  }
+
   def invalidNameForDropTempFunc(name: Seq[String], ctx: ParserRuleContext): Throwable = {
     new ParseException(
       errorClass = "INVALID_SQL_SYNTAX.MULTI_PART_NAME",

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -636,7 +636,8 @@ class Analyzer(
       typeCoercionRules() ++
       Seq(
         ResolveWithCTE,
-        ExtractDistributedSequenceID) ++
+        ExtractDistributedSequenceID,
+        ResolveAsOfJoin) ++
       Seq(ResolveUpdateEventTimeWatermarkColumn) ++
       extendedResolutionRules ++
       Seq(NameStreamingSources) : _*),
@@ -1931,6 +1932,29 @@ class Analyzer(
       case d: DataFrameDropColumns if !d.resolved =>
         resolveDataFrameDropColumns(d)
 
+      case j @ AsOfJoin(left, right, _, condition, _, _, _, _, matchCmp, _, _, _)
+          if left.resolved && right.resolved &&
+            (matchCmp.exists(mc => !mc.left.resolved || !mc.right.resolved) ||
+              condition.exists(!_.resolved)) =>
+        var updated = j
+        matchCmp.foreach { mc =>
+          val resolvedLeft = resolveExpressionByPlanChildren(mc.left, j, includeLastResort = true)
+          val resolvedRight = resolveExpressionByPlanChildren(mc.right, j, includeLastResort = true)
+          if (!resolvedLeft.fastEquals(mc.left) || !resolvedRight.fastEquals(mc.right)) {
+            updated = updated.copy(
+              matchComparison = Some(AsOfMatchCondition(resolvedLeft, mc.operator, resolvedRight)))
+          }
+        }
+        condition.foreach { cond =>
+          if (!cond.resolved) {
+            val resolvedCond = resolveExpressionByPlanChildren(cond, j, includeLastResort = true)
+            if (!resolvedCond.fastEquals(cond)) {
+              updated = updated.copy(condition = Some(resolvedCond))
+            }
+          }
+        }
+        if (updated.fastEquals(j)) j else updated
+
       case q: LogicalPlan =>
         logTrace(s"Attempting to resolve ${q.simpleString(conf.maxToStringFields)}")
         q.mapExpressions(resolveExpressionByPlanChildren(_, q, includeLastResort = true))

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala

@@ -676,15 +676,22 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
                 "joinCondition" -> toSQLExpr(condition),
                 "conditionType" -> toSQLType(condition.dataType)))
 
-          case j @ AsOfJoin(_, _, _, Some(condition), _, _, _)
+          case j @ AsOfJoin(_, _, _, _, _, _, _, _, _, _, _, true)
+              if !SQLConf.get.sortMergeAsOfJoinEnabled =>
+            j.failAnalysis(
+              errorClass = "AS_OF_JOIN.SORT_MERGE_REQUIRED",
+              messageParameters = Map(
+                "config" -> SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key))
+
+          case j @ AsOfJoin(_, _, _, Some(condition), _, _, _, _, _, _, _, _)
               if condition.dataType != BooleanType =>
             throw SparkException.internalError(
               msg = s"join condition '${toSQLExpr(condition)}' " +
                 s"of type ${toSQLType(condition.dataType)} is not a boolean.",
               context = j.origin.getQueryContext,
               summary = j.origin.context.summary)
 
-          case j @ AsOfJoin(_, _, _, _, _, _, Some(toleranceAssertion)) =>
+          case j @ AsOfJoin(_, _, _, _, _, _, Some(toleranceAssertion), _, _, _, _, _) =>
             if (!toleranceAssertion.foldable) {
               j.failAnalysis(
                 errorClass = "AS_OF_JOIN.TOLERANCE_IS_UNFOLDABLE",

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala

@@ -48,7 +48,7 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
           if right.resolved && !j.duplicateResolved && noMissingInput(right.plan) =>
         j.copy(right = right.withNewPlan(dedupRight(left, right.plan)))
       // Resolve duplicate output for AsOfJoin.
-      case j @ AsOfJoin(left, right, _, _, _, _, _)
+      case j @ AsOfJoin(left, right, _, _, _, _, _, _, _, _, _, _)
           if !j.duplicateResolved && noMissingInput(right) =>
         j.copy(right = dedupRight(left, right))
       // Resolve duplicate output for NearestByJoin.