[SPARK-59698][SQL][SS] Reject ASOF joins with streaming right side

docs/sql-ref-syntax-qry-select-asof-join.md

@@ -102,6 +102,14 @@ comparison_operator
 
 ### Notes
 
+* **Structured Streaming - micro-batch mode.** Micro-batch queries support stream-static
+  `ASOF JOIN`,
+  with the streaming relation on the left and the static relation on the right. Static-stream
+  and stream-stream `ASOF JOIN` are not supported because a streaming right side requires state
+  to account for future, closer matches.
+
+* **Structured Streaming - real-time mode.** `ASOF JOIN` is not supported in real-time mode.
+
 * **Direction of match.** Let *L* be the operand of `MATCH_CONDITION` that references
   the left table and *R* the operand that references the right table. The operator
   determines which row on the right is closest:

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

@@ -486,6 +486,12 @@ object UnsupportedOperationChecker extends Logging {
             }
           }
 
+        // Allow stream-static ASOF joins: the streaming left side matches against the current
+        // snapshot of the static right side. Reject static-stream and stream-stream ASOF joins:
+        // a streaming right side would require state to account for future, better matches.
+        case j: AsOfJoin if j.right.isStreaming =>
+          throwError("ASOF join with a streaming DataFrame/Dataset on the right is not supported")
+
         case j @ Join(left, right, joinType, condition, _) =>
           if (left.isStreaming && right.isStreaming) {
             joinType match {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala

@@ -396,6 +396,34 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper {
     outputMode = Append
   )
 
+  def asOfJoin(left: LogicalPlan, right: LogicalPlan): AsOfJoin = {
+    AsOfJoin(
+      left,
+      right,
+      left.output.head >= right.output.head,
+      condition = None,
+      joinType = Inner,
+      orderExpression = left.output.head - right.output.head,
+      toleranceAssertion = None)
+  }
+
+  assertSupportedInStreamingPlan(
+    "ASOF join with stream-static relations",
+    asOfJoin(streamRelation, batchRelation),
+    outputMode = Append)
+
+  assertNotSupportedInStreamingPlan(
+    "ASOF join with static-stream relations",
+    asOfJoin(batchRelation, streamRelation),
+    outputMode = Append,
+    expectedMsgs = Seq("ASOF join", "streaming DataFrame/Dataset on the right"))
+
+  assertNotSupportedInStreamingPlan(
+    "ASOF join with stream-stream relations",
+    asOfJoin(streamRelation, streamRelation),
+    outputMode = Append,
+    expectedMsgs = Seq("ASOF join", "streaming DataFrame/Dataset on the right"))
+
   // Inner joins: Multiple stream-stream joins supported only in append mode
   testBinaryOperationInStreamingPlan(
     "single inner join in append mode",

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeAllowlistSuite.scala

@@ -78,6 +78,41 @@ class StreamRealTimeModeAllowlistSuite extends StreamRealTimeModeE2ESuiteBase {
     }
   }
 
+  test("rtm does not support stream-static ASOF join") {
+    withSQLConf(SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "true") {
+      withTempView("trades", "quotes") {
+        val inputData = LowLatencyMemoryStream[(Int, String)](2)
+        inputData.toDF().toDF("trade_time", "symbol").createOrReplaceTempView("trades")
+        Seq((1, "AAPL", 18010), (3, "AAPL", 18015))
+          .toDF("quote_time", "symbol", "bid_price")
+          .createOrReplaceTempView("quotes")
+
+        val df = sql(
+          """
+            |SELECT concat(t.trade_time, '-', t.symbol, '-', q.bid_price) AS output
+            |FROM trades t ASOF JOIN quotes q
+            |  MATCH_CONDITION (t.trade_time >= q.quote_time)
+            |  ON t.symbol = q.symbol
+            |""".stripMargin)
+        val query = runStreamingQuery("asof_join_allowlist", df)
+
+        eventually(timeout(60.seconds)) {
+          checkError(
+            exception = query.exception.get.getCause.asInstanceOf[SparkIllegalArgumentException],
+            condition = "STREAMING_REAL_TIME_MODE.OPERATOR_OR_SINK_NOT_IN_ALLOWLIST",
+            parameters = Map(
+              "errorType" -> "operator",
+              "message" -> (
+                "org.apache.spark.sql.execution.SortExec, " +
+                  "org.apache.spark.sql.execution.joins.SortMergeAsOfJoinExec are"
+                )
+            )
+          )
+        }
+      }
+    }
+  }
+
   test("rtm sink allowlist") {
     val read = LowLatencyMemoryStream[Int](2)
 

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala

@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.streaming
+
+import java.sql.Timestamp
+
+import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.internal.SQLConf
+
+class StreamingAsOfJoinSuite extends StreamTest {
+
+  import testImplicits._
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    spark.conf.set(SQLConf.SQL_ASOF_JOIN_ENABLED.key, "true")
+  }
+
+  override def afterAll(): Unit = {
+    spark.conf.unset(SQLConf.SQL_ASOF_JOIN_ENABLED.key)
+    super.afterAll()
+  }
+
+  private def timestamp(value: String): Timestamp = Timestamp.valueOf(value)
+
+  private def withTradeQuoteViews(
+      testCode: MemoryStream[(Timestamp, String, Int)] => Unit): Unit = {
+    withTempView("streaming_trades", "static_quotes") {
+      val input = MemoryStream[(Timestamp, String, Int)]
+      input.toDF().toDF("trade_time", "symbol", "quantity")
+        .createOrReplaceTempView("streaming_trades")
+      sql(
+        """
+          |CREATE TEMP VIEW static_quotes(quote_time, symbol, bid_price) AS
+          |VALUES (TIMESTAMP '2026-06-29 10:00:00', 'AAPL', 18010),
+          |       (TIMESTAMP '2026-06-29 10:00:07', 'AAPL', 18015),
+          |       (TIMESTAMP '2026-06-29 10:00:08', 'MSFT', 42050)
+          |""".stripMargin)
+      testCode(input)
+    }
+  }
+
+  test("stream-static ASOF join matches against the static snapshot") {
+    withTradeQuoteViews { input =>
+      val joined = sql(
+        """
+          |SELECT t.trade_time, t.symbol, t.quantity, q.bid_price
+          |FROM streaming_trades t ASOF JOIN static_quotes q
+          |  MATCH_CONDITION (t.trade_time >= q.quote_time)
+          |  ON t.symbol = q.symbol
+          |""".stripMargin)
+
+      testStream(joined)(
+        AddData(input,
+          (timestamp("2026-06-29 10:00:05"), "AAPL", 100),
+          (timestamp("2026-06-29 10:00:09"), "MSFT", 50)),
+        CheckNewAnswer(
+          (timestamp("2026-06-29 10:00:05"), "AAPL", 100, 18010),
+          (timestamp("2026-06-29 10:00:09"), "MSFT", 50, 42050)),
+        AddData(input, (timestamp("2026-06-29 10:00:11"), "AAPL", 200)),
+        CheckNewAnswer((timestamp("2026-06-29 10:00:11"), "AAPL", 200, 18015)))
+    }
+  }
+
+  test("stream-static LEFT ASOF join preserves unmatched streaming rows") {
+    withTradeQuoteViews { input =>
+      val joined = sql(
+        """
+          |SELECT t.trade_time, t.symbol, t.quantity, q.bid_price
+          |FROM streaming_trades t LEFT ASOF JOIN static_quotes q
+          |  MATCH_CONDITION (t.trade_time >= q.quote_time)
+          |  ON t.symbol = q.symbol
+          |""".stripMargin)
+
+      testStream(joined)(
+        AddData(input,
+          (timestamp("2026-06-29 09:59:59"), "AAPL", 30),
+          (timestamp("2026-06-29 10:00:09"), "GOOG", 40),
+          (timestamp("2026-06-29 10:00:11"), "AAPL", 200)),
+        CheckNewAnswer(
+          Row(timestamp("2026-06-29 09:59:59"), "AAPL", 30, null),
+          Row(timestamp("2026-06-29 10:00:09"), "GOOG", 40, null),
+          Row(timestamp("2026-06-29 10:00:11"), "AAPL", 200, 18015)))
+    }
+  }
+
+  Seq("INNER", "LEFT").foreach { joinType =>
+    Seq(false, true).foreach { isLeftStreaming =>
+      val inputType = if (isLeftStreaming) "stream-stream" else "static-stream"
+
+      test(s"$inputType $joinType ASOF join is not supported") {
+        withTempView("trades", "quotes") {
+          val quotesInput = MemoryStream[(Timestamp, String, Int)]
+          quotesInput.toDF().toDF("quote_time", "symbol", "bid_price")
+            .createOrReplaceTempView("quotes")
+
+          if (isLeftStreaming) {
+            val tradesInput = MemoryStream[(Timestamp, String, Int)]
+            tradesInput.toDF().toDF("trade_time", "symbol", "quantity")
+              .createOrReplaceTempView("trades")
+          } else {
+            Seq((timestamp("2026-06-29 10:00:05"), "AAPL", 100))
+              .toDF("trade_time", "symbol", "quantity")
+              .createOrReplaceTempView("trades")
+          }
+
+          val joined = sql(
+            s"""
+               |SELECT t.trade_time, t.symbol, t.quantity, q.bid_price
+               |FROM trades t $joinType ASOF JOIN quotes q
+               |  MATCH_CONDITION (t.trade_time >= q.quote_time)
+               |  ON t.symbol = q.symbol
+               |""".stripMargin)
+          val error = intercept[AnalysisException] {
+            joined.writeStream.format("noop").start()
+          }
+          assert(error.getMessage.contains(
+            "ASOF join with a streaming DataFrame/Dataset on the right is not supported"))
+        }
+      }
+    }
+  }
+}