[SPARK-59698][SQL][SS] Reject ASOF joins with streaming right side - #58950
HeartSaVioR wants to merge 6 commits into
Conversation
|
cc. @cloud-fan This is a blocker for Apache Spark 4.3.0. |
|
cc. @viirya as well for more visibility |
viirya
left a comment
viirya
left a comment
There was a problem hiding this comment.
The restriction makes sense to me. The current stateless ASOF operator cannot preserve closest-match semantics across micro-batches when the right side is streaming. Keeping stream-static joins supported is also reasonable, since each left row can independently match against the static right side.
Checking this in UnsupportedOperationChecker is consistent with the existing streaming join restrictions. I did not find a correctness issue in the change.
One documentation suggestion: could we add a short note to sql-ref-syntax-qry-select-asof-join.md explaining that micro-batch streaming supports stream-static ASOF joins, while a streaming right side is unsupported? This would help distinguish ASOF support from ordinary streaming inner joins.
I also left two non-blocking test suggestions below.
| asOfJoin(streamRelation, batchRelation), | ||
| outputMode = Append) | ||
|
|
||
| assertNotSupportedInStreamingPlan( |
There was a problem hiding this comment.
Could we also add a SQL-level rejection test in StreamingAsOfJoinSuite for static-stream and stream-stream ASOF joins, asserting that writeStream.start() fails with the expected message?
These tests cover the checker directly, while the execution tests cover only the supported stream-static path. A SQL-level negative test would also protect the integration between SQL analysis and the streaming unsupported-operation check. This could be parameterized over INNER and LEFT ASOF.
There was a problem hiding this comment.
Will do. Though maybe I'd take a shortcut on using temp view rather than SDP to create a streaming table.
There was a problem hiding this comment.
Addressed in d9a200e. Added SQL-level writeStream.start() rejection tests for INNER/LEFT across static-stream and stream-stream inputs, asserting the expected message.
DISCLAIMER: This reply was posted by an LLM (OpenAI Codex).
There was a problem hiding this comment.
Thanks, this covers the SQL-to-streaming-check integration I had in mind. Using temporary views over the streaming inputs is sufficient; no need to introduce SDP for this test.
| testStream(joined)( | ||
| 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))) |
There was a problem hiding this comment.
Could we include one matching row in this LEFT ASOF test as well, for example an AAPL trade at 10:00:11 expecting 18015?
Both input rows currently exercise unmatched cases. Mixing matched and unmatched rows would verify that LEFT ASOF both returns the selected quote and preserves unmatched rows; otherwise, a regression that null-pads every row in the LEFT path could still pass this test.
There was a problem hiding this comment.
I feel like this is slightly redundant, but not a big deal to add it. Will do.
There was a problem hiding this comment.
Addressed in d9a200e. The LEFT ASOF test now includes a matched AAPL row alongside both unmatched cases.
DISCLAIMER: This reply was posted by an LLM (OpenAI Codex).
There was a problem hiding this comment.
Thanks, the mixed matched/unmatched input addresses this suggestion.
|
Thanks for the review, @viirya !
Good suggestion and I missed it. Will do. |
|
Addressed the documentation suggestion in d9a200e by documenting the micro-batch stream-static allowance and streaming-right restriction. DISCLAIMER: This reply was posted by an LLM (OpenAI Codex). |
viirya
left a comment
viirya
left a comment
There was a problem hiding this comment.
Thanks for the updates. The SQL-level rejection tests now cover INNER and LEFT ASOF joins with both static-stream and stream-stream inputs, and the LEFT ASOF test covers matched and unmatched rows together.
The documentation clearly distinguishes micro-batch support from the real-time restriction, which is also covered by the new allowlist test. Using temporary views is sufficient here, since the tests still exercise SQL analysis and the streaming check through writeStream.start().
All my previous suggestions have been addressed. LGTM.
This branch has not been deployed
2 participants
What changes were proposed in this pull request?
Reject ASOF joins when the right side is streaming. Stream-static ASOF joins remain allowed because each streaming row can match against the available snapshot of the static right side.
Add UnsupportedOperationChecker coverage for stream-static, static-stream, and stream-stream ASOF joins. Add streaming execution coverage for inner and left stream-static ASOF joins.
Why are the changes needed?
The stateless ASOF join operator cannot provide correct semantics when the right side is streaming. Static-stream and stream-stream joins would need state to account for future rows that could become better matches.
The internal AsOfJoin logical node was introduced in Spark 3.3 for pandas-on-Spark merge_asof, but was not exposed as a public streaming operation. SQL ASOF JOIN and the stateless sort-merge operator are introduced in Spark 4.3, which is where this becomes user-reachable.
Does this PR introduce any user-facing change?
Yes. Static-stream and stream-stream ASOF joins now fail analysis with a clear unsupported-operation error. Stream-static ASOF joins remain supported.
This affects the unreleased Spark 4.3 line and master.
How was this patch tested?
Added focused UnsupportedOperationsSuite coverage for all three streaming/static combinations. Added StreamingAsOfJoinSuite coverage for matching across multiple micro-batches and preserving unmatched rows in a left ASOF join against a static snapshot.
Static whitespace, ASCII, and line-length checks pass. The focused SBT suites could not reach compilation locally because dependency resolution failed: the configured Maven mirror does not contain netty-tcnative-boringssl-static 2.0.84.Final for osx-x86_64, while public Maven access was unavailable.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: GPT 5.6 SOL