[SPARK-59628][SQL] Reject ASOF JOIN without MATCH_CONDITION instead of a silent plain join

docs/sql-ref-ansi-compliance.md

@@ -431,7 +431,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|
+|ASOF|reserved|strict-non-reserved|non-reserved|
 |AT|non-reserved|non-reserved|reserved|
 |ATOMIC|non-reserved|non-reserved|non-reserved|
 |AUTHORIZATION|reserved|non-reserved|reserved|

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

@@ -2156,7 +2156,6 @@ ansiNonReserved
     | ARRAY
     | ASC
     | ASENSITIVE
-    | ASOF
     | AT
     | ATOMIC
     | AUTO
@@ -2533,6 +2532,7 @@ ansiNonReserved
 // These 2 together contain all the keywords.
 strictNonReserved
     : ANTI
+    | ASOF
     | CROSS
     | EXCEPT
     | FULL
@@ -2571,7 +2571,6 @@ nonReserved
     | AS
     | ASC
     | ASENSITIVE
-    | ASOF
     | AT
     | ATOMIC
     | AUTHORIZATION

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala

@@ -1376,16 +1376,53 @@ class PlanParserSuite extends AnalysisTest {
     }
   }
 
+  test("asof join - missing MATCH_CONDITION is rejected, not parsed as a plain join") {
+    // A missing MATCH_CONDITION must fail to parse. If `asof` were read as t's alias, the first
+    // query would parse as a plain INNER join instead. Check both keyword modes.
+    Seq("false", "true").foreach { enforceReservedKeywords =>
+      withSQLConf(
+        SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "true",
+        SQLConf.ANSI_ENABLED.key -> "true",
+        SQLConf.ENFORCE_RESERVED_KEYWORDS.key -> enforceReservedKeywords) {
+        Seq(
+          "select * from t asof join u on t.a = u.a",
+          "select * from t x asof join u on t.a = u.a").foreach { query =>
+          checkError(
+            exception = parseException(query),
+            condition = "PARSE_SYNTAX_ERROR",
+            parameters = Map("error" -> "'on'", "hint" -> ""))
+        }
+        // A back-quoted `asof` is still a valid alias.
+        assertEqual(
+          "select * from t `asof` join u on t.a = u.a",
+          table("t").as("asof").join(table("u"), Inner, Option($"t.a" === $"u.a"))
+            .select(star()))
+      }
+    }
+  }
+
+  test("asof is strict-non-reserved in the default keyword mode") {
+    // Usable as a column and a table name, but not as an unquoted table alias.
+    assertEqual("select asof from t", table("t").select($"asof"))
+    assertEqual("select * from asof", table("asof").select(star()))
+    Seq("select * from t asof", "select * from t as asof").foreach { query =>
+      checkError(
+        exception = parseException(query),
+        condition = "PARSE_SYNTAX_ERROR",
+        parameters = Map("error" -> "end of input", "hint" -> ""))
+    }
+  }
+
   test("nearest-by keywords are non-reserved (usable as identifiers)") {
     // Spark-specific join keywords must remain non-reserved so they can be used as identifiers.
-    Seq("approx", "asof", "distance", "exact", "nearest", "similarity").foreach { kw =>
+    Seq("approx", "distance", "exact", "nearest", "similarity").foreach { kw =>
       // As a column identifier in the SELECT list.
       parsePlan(s"select $kw from t")
       // As a table identifier in the FROM clause.
       parsePlan(s"select * from $kw")
     }
-    // All six together in a single SELECT list.
-    parsePlan("select approx, asof, distance, exact, nearest, similarity from t")
+    // All five together in a single SELECT list.
+    parsePlan("select approx, distance, exact, nearest, similarity from t")
   }
 
   test("sampled relations") {

sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-grammar.sql.out

@@ -377,6 +377,20 @@ org.apache.spark.sql.catalyst.parser.ParseException
 }
 
 
+-- !query
+SELECT * FROM trades ASOF JOIN quotes USING (symbol)
+-- !query analysis
+org.apache.spark.sql.catalyst.parser.ParseException
+{
+  "errorClass" : "PARSE_SYNTAX_ERROR",
+  "sqlState" : "42601",
+  "messageParameters" : {
+    "error" : "'USING'",
+    "hint" : ""
+  }
+}
+
+
 -- !query
 SELECT * FROM trades t ASOF JOIN quotes q
   MATCH_CONDITION (t.trade_time >= q.quote_time AND t.symbol = q.symbol)

sql/core/src/test/resources/sql-tests/inputs/join-asof-grammar.sql

@@ -128,6 +128,9 @@ SELECT * FROM trades t LEFT SEMI ASOF JOIN quotes q
 -- FVT-ASOF-1-017: missing MATCH_CONDITION
 SELECT * FROM trades t ASOF JOIN quotes q ON t.symbol = q.symbol;
 
+-- FVT-ASOF-1-020: missing MATCH_CONDITION with no left alias is rejected, not a plain join
+SELECT * FROM trades ASOF JOIN quotes USING (symbol);
+
 -- FVT-ASOF-1-018: multiple comparisons in MATCH_CONDITION
 SELECT * FROM trades t ASOF JOIN quotes q
   MATCH_CONDITION (t.trade_time >= q.quote_time AND t.symbol = q.symbol)

sql/core/src/test/resources/sql-tests/results/join-asof-grammar.sql.out

@@ -304,6 +304,22 @@ org.apache.spark.sql.catalyst.parser.ParseException
 }
 
 
+-- !query
+SELECT * FROM trades ASOF JOIN quotes USING (symbol)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.sql.catalyst.parser.ParseException
+{
+  "errorClass" : "PARSE_SYNTAX_ERROR",
+  "sqlState" : "42601",
+  "messageParameters" : {
+    "error" : "'USING'",
+    "hint" : ""
+  }
+}
+
+
 -- !query
 SELECT * FROM trades t ASOF JOIN quotes q
   MATCH_CONDITION (t.trade_time >= q.quote_time AND t.symbol = q.symbol)

sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out

@@ -24,7 +24,7 @@ ARRAY	false
 AS	true
 ASC	false
 ASENSITIVE	false
-ASOF	false
+ASOF	true
 AT	false
 ATOMIC	false
 AUTHORIZATION	true
@@ -473,6 +473,7 @@ ALL
 AND
 ANY
 AS
+ASOF
 AUTHORIZATION
 BOTH
 CALL