From 9708a92f2a617005eebf13829a7c32e6cfa0cbb4 Mon Sep 17 00:00:00 2001 From: Jungtaek Lim Date: Mon, 21 Sep 2026 17:58:47 +0900 Subject: [PATCH 1/6] [MINOR][SQL] Reject ASOF joins with streaming right side --- .../UnsupportedOperationChecker.scala | 6 ++++ .../analysis/UnsupportedOperationsSuite.scala | 28 +++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala index 6db8234471ef5..9f4a58486e220 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala +++ b/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 { diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala index 720024357d9ef..83c7b58907da4 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala +++ b/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", From 0e1fe75c39400ad5de32e1f306cd3c015a0c215f Mon Sep 17 00:00:00 2001 From: Jungtaek Lim Date: Mon, 21 Sep 2026 21:32:15 +0900 Subject: [PATCH 2/6] [MINOR][SQL] Test stream-static ASOF joins --- .../streaming/StreamingAsOfJoinSuite.scala | 96 +++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala new file mode 100644 index 0000000000000..1e19a84dfa709 --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala @@ -0,0 +1,96 @@ +/* + * 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.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("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"), "GOOG", 30)), + CheckNewAnswer(Row(timestamp("2026-06-29 09:59:59"), "GOOG", 30, null))) + } + } +} From 4e5adba334ac2f86123b8b8e08365304c0ded6a0 Mon Sep 17 00:00:00 2001 From: Jungtaek Lim Date: Mon, 21 Sep 2026 23:26:49 +0900 Subject: [PATCH 3/6] [MINOR][SQL] Fix streaming ASOF test DataFrame conversion --- .../org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala index 1e19a84dfa709..dfbecc512b934 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala @@ -43,7 +43,7 @@ class StreamingAsOfJoinSuite extends StreamTest { testCode: MemoryStream[(Timestamp, String, Int)] => Unit): Unit = { withTempView("streaming_trades", "static_quotes") { val input = MemoryStream[(Timestamp, String, Int)] - input.toDF("trade_time", "symbol", "quantity") + input.toDF().toDF("trade_time", "symbol", "quantity") .createOrReplaceTempView("streaming_trades") sql( """ From 19ab66053b5aa51ee2bb08437bfbc5e9732fbf73 Mon Sep 17 00:00:00 2001 From: Jungtaek Lim Date: Tue, 22 Sep 2026 05:14:12 +0900 Subject: [PATCH 4/6] [MINOR][SQL] Test unmatched stream-static ASOF rows --- .../spark/sql/streaming/StreamingAsOfJoinSuite.scala | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala index dfbecc512b934..ebfd6c8ff2888 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala @@ -89,8 +89,12 @@ class StreamingAsOfJoinSuite extends StreamTest { |""".stripMargin) testStream(joined)( - AddData(input, (timestamp("2026-06-29 09:59:59"), "GOOG", 30)), - CheckNewAnswer(Row(timestamp("2026-06-29 09:59:59"), "GOOG", 30, null))) + AddData(input, + (timestamp("2026-06-29 09:59:59"), "AAPL", 30), + (timestamp("2026-06-29 10:00:09"), "GOOG", 40)), + CheckNewAnswer( + Row(timestamp("2026-06-29 09:59:59"), "AAPL", 30, null), + Row(timestamp("2026-06-29 10:00:09"), "GOOG", 40, null))) } } } From d9a200e08815fb206a906dfd10703bf98f5eb7ce Mon Sep 17 00:00:00 2001 From: Jungtaek Lim Date: Tue, 22 Sep 2026 16:38:52 +0900 Subject: [PATCH 5/6] [SPARK-59698][SQL][SS] Address ASOF join review comments --- docs/sql-ref-syntax-qry-select-asof-join.md | 5 +++ .../streaming/StreamingAsOfJoinSuite.scala | 45 +++++++++++++++++-- 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/docs/sql-ref-syntax-qry-select-asof-join.md b/docs/sql-ref-syntax-qry-select-asof-join.md index 98cba2ad9e72c..8a972fd11f9d8 100644 --- a/docs/sql-ref-syntax-qry-select-asof-join.md +++ b/docs/sql-ref-syntax-qry-select-asof-join.md @@ -102,6 +102,11 @@ comparison_operator ### Notes +* **Structured Streaming.** 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. + * **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: diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala index ebfd6c8ff2888..c074adc66bd89 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAsOfJoinSuite.scala @@ -19,7 +19,7 @@ package org.apache.spark.sql.streaming import java.sql.Timestamp -import org.apache.spark.sql.Row +import org.apache.spark.sql.{AnalysisException, Row} import org.apache.spark.sql.execution.streaming.runtime.MemoryStream import org.apache.spark.sql.internal.SQLConf @@ -91,10 +91,49 @@ class StreamingAsOfJoinSuite extends StreamTest { 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: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: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")) + } + } } } } From 97554675b312413f9b0f26eabb37b22c866b6a07 Mon Sep 17 00:00:00 2001 From: Jungtaek Lim Date: Tue, 22 Sep 2026 16:42:15 +0900 Subject: [PATCH 6/6] [SPARK-59698][SQL][SS] Document ASOF join streaming modes --- docs/sql-ref-syntax-qry-select-asof-join.md | 5 ++- .../StreamRealTimeModeAllowlistSuite.scala | 35 +++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/docs/sql-ref-syntax-qry-select-asof-join.md b/docs/sql-ref-syntax-qry-select-asof-join.md index 8a972fd11f9d8..ebbffd1a90054 100644 --- a/docs/sql-ref-syntax-qry-select-asof-join.md +++ b/docs/sql-ref-syntax-qry-select-asof-join.md @@ -102,11 +102,14 @@ comparison_operator ### Notes -* **Structured Streaming.** Micro-batch queries support stream-static `ASOF JOIN`, +* **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: diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeAllowlistSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeAllowlistSuite.scala index ee101cbede9d1..d2e1e54dcfec9 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeAllowlistSuite.scala +++ b/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)