You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "ganeshashree (via GitHub)" <gi...@apache.org> on 2026/08/24 05:42:16 UTC
[PR] [SPARK-58904][SS] Enforce correct watermark placement for stream-stream left semi and left outer joins [spark]
ganeshashree opened a new pull request, #58238:
URL: https://github.com/apache/spark/pull/58238
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'common/utils/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Stream-stream left semi and left outer joins rely on the left-side state being evicted by the watermark, but the analyzer did not require the watermark to be placed where that eviction can happen. This adds a shared analyzer check, `checkStreamStreamJoinWatermarkPlacement`, with two requirements:
(1) left-state eviction for both types (the equi eviction key must be watermarked, or the range bound must sit between watermarked attributes on both sides);
(2) for left outer only, both eviction-ordinal join keys must be watermarked so a late row cannot invalidate an already-emitted unmatched row. The new helpers mirror the runtime's own attribute selection so the analyzer rejects exactly the placements the runtime cannot honor.
It is gated by a kill switch, `spark.sql.streaming.join.stricterWatermarkRequirements.enabled` (default true), to restore the previous behavior. This is the shared-code fix split out of #57813 (SPARK-58611, left anti).
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
With only the right side watermarked, or the watermark on the wrong join-key ordinal or an unrelated attribute, the runtime builds no left-eviction predicate. Left outer then silently drops its unmatched output, and left semi state grows without bound. These placements should be rejected at analysis time rather than run incorrectly.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
Yes. Some stream-stream left semi and left outer queries that previously ran now fail with an `AnalysisException` explaining the required watermark placement. The behavior can be reverted with `spark.sql.streaming.join.stricterWatermarkRequirements.enabled=false`.
Documented in the Structured Streaming migration guide (4.3 to 4.4).
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Added analyzer tests in `UnsupportedOperationsSuite` (range both-sides, right-only, unrelated attribute, constant and anti-monotonic bounds, multiple event-time columns, one-sided and both-sided equi keys, composite keys, kill-switch fallback) and runtime tests in `StreamingJoinSuite` for left outer and left semi. Ran locally: `UnsupportedOperationsSuite` (261) and `StreamingOuterJoinWithoutVCFSuite` + `StreamingLeftSemiJoinWithoutVCFSuite` (74), all pass.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
Generated-by: Claude Code (Opus 4.8)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58904][SS] Enforce correct watermark placement for stream-stream left semi and left outer joins [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #58238:
URL: https://github.com/apache/spark/pull/58238#discussion_r3850747934
##########
docs/streaming/apis-on-dataframes-and-datasets.md:
##########
@@ -1293,6 +1293,20 @@ joined <- join(
</div>
+For a **left outer** join, the surviving unmatched left rows are emitted when the left-side state is
Review Comment:
The documentation is user-facing. Shall we make this be concise and not too much on the detail? Good to skim through the doc and understand the mood of max complexity we push in the guide doc, and don't go beyond that.
This change "as it is" isn't consumable for most of the users.
##########
docs/streaming/apis-on-dataframes-and-datasets.md:
##########
@@ -1318,6 +1332,15 @@ constraints must be specified for semi join. This is to evict unmatched input ro
the engine must know when an input row on left side is not going to match with anything on right
side in future.
+As with a left outer join, the left state must be evicted so that never-matched left rows do not
Review Comment:
Same here, this is a user facing doc. Apply the same suggestion here, please.
##########
docs/streaming/apis-on-dataframes-and-datasets.md:
##########
@@ -1293,6 +1293,20 @@ joined <- join(
</div>
+For a **left outer** join, the surviving unmatched left rows are emitted when the left-side state is
+evicted, and an already-emitted `NULL`-extended row can be invalidated by a right row that arrives
+later. Correct results therefore require the watermark to be placed so that (1) the left state is
+actually evicted and (2) both sides are late-filtered on the dimension that bounds matching.
+Concretely: for an equality join on a watermarked event-time key, that key must be watermarked on
+**both** sides; for a time range condition, the range bound must relate watermarked event-time
+columns from both sides. The recommended, always-correct configuration -- watermarking both sides on
+the event-time column used by the join, as in the example above -- satisfies both. Configurations
+that leave the left state un-evicted (a watermark on only the right side of a range condition) or
+leave either side unfiltered on the eviction key (a watermark on only one equality join key) are
+rejected at analysis time. To restore the previous, looser behavior, set
Review Comment:
I wouldn't have that config in the guide doc. It's OK to keep it in migration guide doc only.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3857,6 +3857,22 @@ object SQLConf {
.checkValue(v => Set(1, 2, 3, 4).contains(v), "Valid versions are 1, 2, 3, and 4")
.createWithDefault(2)
+ val STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED =
+ buildConf("spark.sql.streaming.join.stricterWatermarkRequirements.enabled")
+ .doc("When true, the analyzer enforces stricter watermark-placement requirements for " +
Review Comment:
`.internal()`, please. This config must not be used beyond quick mitigation during migration.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3857,6 +3857,22 @@ object SQLConf {
.checkValue(v => Set(1, 2, 3, 4).contains(v), "Valid versions are 1, 2, 3, and 4")
.createWithDefault(2)
+ val STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED =
+ buildConf("spark.sql.streaming.join.stricterWatermarkRequirements.enabled")
+ .doc("When true, the analyzer enforces stricter watermark-placement requirements for " +
+ "stream-stream left semi and left outer joins, so that the left-side state which their " +
Review Comment:
Can we fix this for left/right/full outer and left semi while we are here? Now this is pretty much weird to say we only fix the issue in left outer and left semi.
##########
docs/streaming/ss-migration-guide.md:
##########
@@ -23,6 +23,10 @@ Note that this migration guide describes the items specific to Structured Stream
Many items of SQL migration can be applied when migrating Structured Streaming to higher versions.
Please refer [Migration Guide: SQL, Datasets and DataFrame](../sql-migration-guide.html).
+## Upgrading from Structured Streaming 4.3 to 4.4
+
+- Since Spark 4.4, stream-stream left semi and left outer joins enforce stricter watermark-placement requirements at analysis time, so that the left-side state their output (or bounded state size) depends on is actually evicted, and, for left outer, so that late rows cannot invalidate an already-emitted unmatched row. Configurations that previously ran but could silently produce incorrect results or unbounded state -- for example a range-condition join whose range bound is not between watermarked attributes on both sides, or a left outer equality join whose eviction key is not watermarked on both sides -- now fail with an `AnalysisException`. To restore the previous behavior, set `spark.sql.streaming.join.stricterWatermarkRequirements.enabled` to `false`. (See [SPARK-58904](https://issues.apache.org/jira/browse/SPARK-58904) for more details.)
Review Comment:
This is too detailed as a user-facing doc; I'd describe as following:
> Since Spark 4.4, stream-stream left outer and left semi joins require watermarks to be defined on the event-time columns used by time-range join conditions. Left outer equality joins require watermarks on the corresponding join keys from both inputs. Queries that do not meet these requirements now fail analysis instead of potentially producing incomplete results or retaining state indefinitely. To restore the previous behavior, set `spark.sql.streaming.join.stricterWatermarkRequirements.enabled` to `false`. (See [SPARK-58904](https://issues.apache.org/jira/browse/SPARK-58904) for more details.)
##########
docs/streaming/apis-on-dataframes-and-datasets.md:
##########
@@ -1293,6 +1293,20 @@ joined <- join(
</div>
+For a **left outer** join, the surviving unmatched left rows are emitted when the left-side state is
Review Comment:
Always push LLM to be concise and brief, and get the mood of content in the doc before writing it. You also need to be clear about whether the content is user-facing vs internal, to make a judgement on the result of LLM, not just blindly picking it up.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3857,6 +3857,22 @@ object SQLConf {
.checkValue(v => Set(1, 2, 3, 4).contains(v), "Valid versions are 1, 2, 3, and 4")
.createWithDefault(2)
+ val STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED =
+ buildConf("spark.sql.streaming.join.stricterWatermarkRequirements.enabled")
+ .doc("When true, the analyzer enforces stricter watermark-placement requirements for " +
+ "stream-stream left semi and left outer joins, so that the left-side state which their " +
Review Comment:
This is one of LLM's patterns - config doc does not really need to describe the implementation detail unless it is required to understand the impact of the config. Maybe similar level of content in migration guide can put here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58904][SS] Enforce correct watermark placement for stream-stream left semi and left outer joins [spark]
Posted by "ganeshashree (via GitHub)" <gi...@apache.org>.
ganeshashree commented on PR #58238:
URL: https://github.com/apache/spark/pull/58238#issuecomment-5409036421
> My apologizes. I changed my mind - the fix of watermark placement is much more complicated than it seems and it requires focused effort to get this done properly. I feel like the contribution is LLM driven and it's taking much more time to review than I could prompt, unfortunately. (I only partially reviewed the code and the characteristic of the doc change is purely what I have been seen from LLM. I'm not saying we should not generate the sentence from LLM but that's always unnecessary verbose where we can prompt further to be concise and brief - I suspect this change was self-reviewed.)
>
> I'm OK with simply adding LeftAnti join, leaving the existing watermark placement issue as it is (and even expand that problem to LeftAnti) and asking folks (or myself) who are more expert on this subject to revisit it.
>
> @ganeshashree Sorry for back and forth. I'd love to make the code change be more thoughtful and more time and effort is given to make it better since the complexity for StreamingJoinHelper is already outstanding. Let's step back a bit and just make a simpler change to introduce LeftAnti join. I wouldn't ask to solve this issue in your PR of LeftAnti join; just extend the issue to LeftAnti join as same as LeftSemi join.
Thanks @HeartSaVioR, that makes sense. I'd rather get this right than rush it in.
For the record, what this PR did: it added a shared analyzer check, `checkStreamStreamJoinWatermarkPlacement`, enforcing two requirements for left semi and left outer, left state eviction (the equi eviction key must be watermarked, or the range bound must sit between watermarked attributes on both sides), and for left outer only, both eviction ordinal join keys watermarked so a late row cannot invalidate an already emitted unmatched row. The helpers mirror the runtime's own attribute selection, and it was gated behind a kill switch. That logic can carry into the focused effort below.
I'll take the redirect:
- Drop the watermark placement fix from this line of work. I'll close this PR and keep the LeftAnti change in #57813 minimal, just extending the existing behavior to LeftAnti the same way it works for LeftSemi today, without touching placement.
- Park the placement fix as its own focused effort. Since it touches a critical area, this is best done by an SS expert.
On the docs: fair point, and taken. I'll keep user-facing docs to the level of detail already in the guide, move the config out of the guide (migration guide only), mark it `.internal()`, and use your migration-guide wording. I'll tighten the prose by hand rather than leaving it verbose.
Sorry for the extra review churn here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58904][SS] Enforce correct watermark placement for stream-stream left semi and left outer joins [spark]
Posted by "ganeshashree (via GitHub)" <gi...@apache.org>.
ganeshashree closed pull request #58238: [SPARK-58904][SS] Enforce correct watermark placement for stream-stream left semi and left outer joins
URL: https://github.com/apache/spark/pull/58238
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58904][SS] Enforce correct watermark placement for stream-stream left semi and left outer joins [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #58238:
URL: https://github.com/apache/spark/pull/58238#issuecomment-5410159824
Thank you @HeartSaVioR for the detailed review and help here! Sorry @ganeshashree I also thought this could go forward, but it seems more involved than I initially scoped
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58904][SS] Enforce correct watermark placement for stream-stream left semi and left outer joins [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #58238:
URL: https://github.com/apache/spark/pull/58238#issuecomment-5392850990
Thank you @ganeshashree! cc @HeartSaVioR for SS
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org