[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.

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.

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!

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

		.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!

HeartSaVioR commented Apr 9, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

HeartSaVioR commented Apr 7, 2026 •

edited

Loading