diff --git a/docs/sql-ref-syntax-qry-select-asof-join.md b/docs/sql-ref-syntax-qry-select-asof-join.md index 98cba2ad9e72c..ebbffd1a90054 100644 --- a/docs/sql-ref-syntax-qry-select-asof-join.md +++ b/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: 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", 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) 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..c074adc66bd89 --- /dev/null +++ b/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")) + } + } + } + } +}