[SPARK-56384][SS] Support stream-stream non-outer join in Update mode by HeartSaVioR · Pull Request #55249 · apache/spark

HeartSaVioR · 2026-04-07T22:20:07Z

What changes were proposed in this pull request?

This PR proposes to support stream-stream non-outer join (Inner/LeftSemi) in Update mode. The expected behavior of Inner/LeftSemi join in Update mode is simply the same with Append mode (since there is no point of early-firing), hence this is bound to the definition of Update mode, "if the query doesn't contain aggregations, it is equivalent to Append mode".

Why are the changes needed?

Previously stream-stream inner join cannot be used in update mode, where there could be some workload benefited by this. e.g. stream-stream inner join, followed by time window aggregation - this workload has different ability based on the output mode. On Update mode, we have early-firing result of time window aggregation from the input of stream-stream join. This PR unblocks the case.

Does this PR introduce any user-facing change?

Yes, stream-stream non-outer join will be supported with Update mode.

How was this patch tested?

New tests.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude 4.6 Opus

…th stream-stream inner join

funrollloops · 2026-04-08T21:28:47Z

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

+                  throwError(s"$joinType join between two streaming DataFrames/Datasets" +
+                    s" is not supported in ${outputMode} output mode, only in Append output mode")
+                }
+              case _ =>


Can we list the supported join types explicitly?

Sounds good.

Btw we have pattern match at the very next which will handle the failure based on join types. This pattern match is mostly about checking compatibility with output mode.

funrollloops · 2026-04-08T21:29:54Z

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

+      condition = Some(attributeWithWatermark === attribute)),
+    OutputMode.Update())
+
+  // Left outer, right outer, full outer, left semi joins


This comment seems a little redundant.

Let me see if this is the consistent pattern within file or we just added this only here. If it's consistent over the file, I guess we can just leave it as it is. If not we can remove it.

I guess the pattern is consistent in the file (that's what Claude said), let's leave it as it is.

The comment is just duplicating the list on the next line. Please remove it.

If you disagree, please say so, but I don't care what Claude thinks.

There is a misunderstanding - I think I have commented already about my rationale.

Let me see if this is the consistent pattern within file or we just added this only here. If it's consistent over the file, I guess we can just leave it as it is. If not we can remove it.

I just asked Claude to "check", not Claude to "judge". So it is my disagreement.

The test suite is large enough so people leverages code comment (not code) as index where the test is placed. With LLM it's going to be less and less necessary and maybe there are some ways to make it better, but I'd rather not break what existing code is trying to do - if we think we have better way, we should consider it as a refactor and review separately.

funrollloops · 2026-04-08T21:42:30Z

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

+      // dedup1: filters 1, 2 (already seen), passes only 4
+      // dedup2: filters 2, 3 (already seen), passes only 4
+      // join: only (4, 4) matches


nit: I feel like we don't need to have such verbose comments. The query is pretty short, and the test output should be verified to match the query anyway.

For stateful operators we tend to be very reader friendly, but for the case when watermark doesn't come into play, maybe it's very obvious to think through without guidance. We can remove comments for this test.

funrollloops · 2026-04-08T21:42:36Z

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

+      .withColumnRenamed("value", "value1")
+      .withColumn("eventTime1", timestamp_seconds($"value1"))


nit: select?

withColumn is a syntax sugar to append a new column - it has been a best practice.

I won't insist, but this really seems unnecessarily verbose. Typical Java style, I suppose.

Maybe I misunderstood your point? Did you suggest consolidating two lines into one select? Then it makes sense totally. If you just refer to the single withColumn then I'd say withColumn is better than select but a chain of withColumnRenamed -> withColumn isn't really necessary. I'll change.

funrollloops · 2026-04-08T21:45:03Z

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

  }

+  // Stream-stream non-outer join produces the same behavior between Append mode and Update mode.
+  // We only run a sanity test here rather than replicating the full Append mode test suite.


Can we run a loop over the existing append-mode test suite so all the test cases run for both append and update mode?

OK that works for me. I'll see whether it is running too long (so that I need to have separate suite).

funrollloops · 2026-04-08T21:46:15Z

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

    )
  }

+  test("dedup on both sides -> stream-stream inner join, update mode") {


same as below: if the behavior is the same, could we run a loop over test case with different output modes?

Yes let's do that.

funrollloops · 2026-04-08T21:50:25Z

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

-  // Left outer, right outer, full outer, left semi joins
-  Seq(LeftOuter, RightOuter, FullOuter, LeftSemi).foreach { joinType =>
-    // Update mode not allowed
+  // Left outer, right outer, full outer joins: Update mode not allowed


Suggested change

// Left outer, right outer, full outer joins: Update mode not allowed

// The behavior for unmatched rows in outer joins with update mode hasn't been defined yet.

That's good for explaining rationale, but this is a test. If we want to have this comment I'd say we should have it in the place where we block the operation. I'll do that in UnsupportedOperationChecker and keep this comment as it is.

I added that comment in UnsupportedOperationChecker.

HeartSaVioR · 2026-04-09T01:59:13Z

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

  import testImplicits._

-  test("windowed left semi join") {
+  testWithAppendAndUpdate("windowed left semi join") { outputMode =>


Not to be confused: we do not apply time window aggregation to come up with time window. It's just a projection. That's how this works in Update mode.

HeartSaVioR · 2026-04-09T02:59:12Z

@funrollloops Would you mind taking a second look? Thanks!

funrollloops · 2026-04-09T18:21:30Z

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

+      condition = Some(attributeWithWatermark === attribute)),
+    OutputMode.Update())
+
+  // Left outer, right outer, full outer, left semi joins


The comment is just duplicating the list on the next line. Please remove it.

If you disagree, please say so, but I don't care what Claude thinks.

funrollloops · 2026-04-09T18:23:06Z

...e/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala

    left: SparkPlan,
-    right: SparkPlan) extends BinaryExecNode with StateStoreWriter with SchemaValidationUtils {
+    right: SparkPlan,
+    outputMode: Option[OutputMode] = None)


Why optional? The outputMode is actually required right, it just defaults to append? We should only default in one place IMO.

That doesn't really need to be optional. Good point.

This should actually be optional as we have two phases of constructing the physical node.

When we convert logical plan to physical plan, that "general" rule does not know about output mode. We handle the semantic of "streaming specific concept" in scope to IncrementalExecution, where we assign watermark value to the physical node (via copying). In short, we actually take the same approach how we assign watermark value here.

Though it could slightly open more chance to miss to assign when we have default value, hence I removed the default value.

funrollloops · 2026-04-09T18:24:45Z

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

+      .withColumnRenamed("value", "value1")
+      .withColumn("eventTime1", timestamp_seconds($"value1"))


I won't insist, but this really seems unnecessarily verbose. Typical Java style, I suppose.

HeartSaVioR mentioned this pull request Apr 7, 2026

[WIP][REVIEW-DONE] Support stream-stream non-outer join in Update mode HeartSaVioR/spark#27

Closed

[SPARK-XXXXX] Support stream-stream non-outer join in Update mode

f772b2c

HeartSaVioR force-pushed the SPARK-56384 branch from bc454fe to f772b2c Compare April 8, 2026 08:01

Add tests for ability of multiple stateful operator in update mode wi…

fab4a3a

…th stream-stream inner join

funrollloops reviewed Apr 8, 2026

View reviewed changes

HeartSaVioR added 2 commits April 9, 2026 10:34

Review comments

7160f2b

smaller code diff

808550e

HeartSaVioR commented Apr 9, 2026

View reviewed changes

HeartSaVioR requested a review from funrollloops April 9, 2026 01:59

funrollloops reviewed Apr 9, 2026

View reviewed changes

Review comments

25390da

HeartSaVioR requested a review from funrollloops April 10, 2026 03:25

		.withColumnRenamed("value", "value1")
		.withColumn("eventTime1", timestamp_seconds($"value1"))

	// Left outer, right outer, full outer joins: Update mode not allowed
	// The behavior for unmatched rows in outer joins with update mode hasn't been defined yet.

Conversation

HeartSaVioR commented Apr 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR commented Apr 9, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Apr 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

HeartSaVioR commented Apr 7, 2026 •

edited

Loading

HeartSaVioR Apr 9, 2026 •

edited

Loading