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

Author: HeartSaVioR

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

@@ -441,9 +441,21 @@ object UnsupportedOperationChecker extends Logging {
           }
 
         case j @ Join(left, right, joinType, condition, _) =>
-          if (left.isStreaming && right.isStreaming && outputMode != InternalOutputModes.Append) {
-            throwError("Join between two streaming DataFrames/Datasets is not supported" +
-              s" in ${outputMode} output mode, only in Append output mode")
+          if (left.isStreaming && right.isStreaming) {
+            joinType match {
+              case LeftOuter | RightOuter | FullOuter =>
+                if (outputMode != InternalOutputModes.Append) {
+                  throwError(s"$joinType join between two streaming DataFrames/Datasets" +
+                    s" is not supported in ${outputMode} output mode, only in Append output mode")
+                }
+              case _ =>
+                if (outputMode != InternalOutputModes.Append &&
+                    outputMode != InternalOutputModes.Update) {
+                  throwError(s"$joinType join between two streaming DataFrames/Datasets" +
+                    s" is not supported in ${outputMode} output mode, only in Append and Update " +
+                    "output modes")
+                }
+            }
           }
 
           joinType match {

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

@@ -370,9 +370,7 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper {
   testBinaryOperationInStreamingPlan(
     "inner join in update mode",
     _.join(_, joinType = Inner),
-    outputMode = Update,
-    streamStreamSupported = false,
-    expectedMsg = "is not supported in Update output mode")
+    outputMode = Update)
 
   // Full outer joins: stream-batch/batch-stream join are not allowed,
   // and stream-stream join is allowed 'conditionally' - see below check
@@ -403,16 +401,25 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper {
     streamStreamSupported = false,
     expectedMsg = "RightOuter join")
 
-  // 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
+  Seq(LeftOuter, RightOuter, FullOuter).foreach { joinType =>
     assertNotSupportedInStreamingPlan(
       s"$joinType join with stream-stream relations and update mode",
       streamRelation.join(streamRelation, joinType = joinType,
         condition = Some(attribute === attribute)),
       OutputMode.Update(),
       Seq("is not supported in Update output mode"))
+  }
+
+  // LeftSemi join: Update mode allowed (equivalent to Append mode for non-outer joins)
+  assertSupportedInStreamingPlan(
+    s"LeftSemi join with stream-stream relations and update mode",
+    streamRelation.join(streamRelation, joinType = LeftSemi,
+      condition = Some(attributeWithWatermark === attribute)),
+    OutputMode.Update())
 
+  // Left outer, right outer, full outer, left semi joins
+  Seq(LeftOuter, RightOuter, FullOuter, LeftSemi).foreach { joinType =>
     // Complete mode not allowed
     assertNotSupportedInStreamingPlan(
       s"$joinType join with stream-stream relations and complete mode",

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinExec.scala

@@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.analysis.StreamingJoinHelper
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Expression, GenericInternalRow, JoinedRow, Literal, Predicate, UnsafeProjection, UnsafeRow}
 import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.plans.physical._
+import org.apache.spark.sql.catalyst.streaming.InternalOutputModes
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
 import org.apache.spark.sql.execution.{BinaryExecNode, SparkPlan}
 import org.apache.spark.sql.execution.metric.SQLMetric
@@ -35,6 +36,7 @@ import org.apache.spark.sql.execution.streaming.operators.stateful.join.Streamin
 import org.apache.spark.sql.execution.streaming.operators.stateful.join.SymmetricHashJoinStateManager.KeyToValuePair
 import org.apache.spark.sql.execution.streaming.state._
 import org.apache.spark.sql.internal.{SessionState, SQLConf}
+import org.apache.spark.sql.streaming.OutputMode
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.util.{CompletionIterator, SerializableConfiguration}
 
@@ -142,7 +144,9 @@ case class StreamingSymmetricHashJoinExec(
     stateWatermarkPredicates: JoinStateWatermarkPredicates,
     stateFormatVersion: Int,
     left: SparkPlan,
-    right: SparkPlan) extends BinaryExecNode with StateStoreWriter with SchemaValidationUtils {
+    right: SparkPlan,
+    outputMode: Option[OutputMode] = None)
+    extends BinaryExecNode with StateStoreWriter with SchemaValidationUtils {
 
   def this(
       leftKeys: Seq[Expression],
@@ -184,6 +188,13 @@ case class StreamingSymmetricHashJoinExec(
     joinType == LeftSemi,
     errorMessageForJoinType)
 
+  outputMode.foreach { mode =>
+    if (mode == InternalOutputModes.Update) {
+      require(joinType == Inner || joinType == LeftSemi,
+        s"Update output mode is not supported for stream-stream $joinType join")
+    }
+  }
+
   // The assertion against join keys is same as hash join for batch query.
   require(leftKeys.length == rightKeys.length &&
     leftKeys.map(_.dataType)

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/IncrementalExecution.scala

@@ -411,7 +411,8 @@ class IncrementalExecution(
         j.copy(
           stateInfo = Some(nextStatefulOperationStateInfo()),
           eventTimeWatermarkForLateEvents = None,
-          eventTimeWatermarkForEviction = None
+          eventTimeWatermarkForEviction = None,
+          outputMode = Some(outputMode)
         )
 
       case l: StreamingGlobalLimitExec =>

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

@@ -30,7 +30,7 @@ import org.scalatest.{BeforeAndAfter, Tag}
 
 import org.apache.spark.SparkUnsupportedOperationException
 import org.apache.spark.scheduler.ExecutorCacheTaskLocation
-import org.apache.spark.sql.{DataFrame, Row, SparkSession}
+import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SparkSession}
 import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression}
 import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
 import org.apache.spark.sql.execution.datasources.v2.state.StateSourceOptions
@@ -352,6 +352,28 @@ abstract class StreamingInnerJoinBase extends StreamingJoinSuite {
     )
   }
 
+  // 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.
+  test("stream stream inner join with Update mode on non-time column") {
+    val input1 = MemoryStream[Int]
+    val input2 = MemoryStream[Int]
+
+    val df1 = input1.toDF().select($"value" as "key", ($"value" * 2) as "leftValue")
+    val df2 = input2.toDF().select($"value" as "key", ($"value" * 3) as "rightValue")
+    val joined = df1.join(df2, "key")
+
+    testStream(joined, OutputMode.Update())(
+      AddData(input1, 1),
+      CheckAnswer(),
+      AddData(input2, 1, 10),
+      CheckNewAnswer((1, 2, 3)),
+      AddData(input1, 10),
+      CheckNewAnswer((10, 20, 30)),
+      AddData(input2, 1),
+      CheckNewAnswer((1, 2, 3))
+    )
+  }
+
   test("stream stream inner join on windows - without watermark") {
     val input1 = MemoryStream[Int]
     val input2 = MemoryStream[Int]
@@ -669,7 +691,7 @@ abstract class StreamingInnerJoinBase extends StreamingJoinSuite {
         assert(query.lastExecution.executedPlan.collect {
           case j @ StreamingSymmetricHashJoinExec(_, _, _, _, _, _, _, _, _,
             ShuffleExchangeExec(opA: HashPartitioning, _, _, _),
-            ShuffleExchangeExec(opB: HashPartitioning, _, _, _))
+            ShuffleExchangeExec(opB: HashPartitioning, _, _, _), _)
               if partitionExpressionsColumns(opA.expressions) === Seq("a", "b")
                 && partitionExpressionsColumns(opB.expressions) === Seq("a", "b")
                 && opA.numPartitions == numPartitions && opB.numPartitions == numPartitions => j
@@ -1242,6 +1264,25 @@ abstract class StreamingOuterJoinBase extends StreamingJoinSuite {
   import testImplicits._
   import org.apache.spark.sql.functions._
 
+  Seq("left_outer", "right_outer").foreach { joinType =>
+    test(s"stream-stream $joinType join does not support Update mode") {
+      val input1 = MemoryStream[Int]
+      val input2 = MemoryStream[Int]
+
+      val df1 = input1.toDF().select($"value" as "key", ($"value" * 2) as "leftValue")
+      val df2 = input2.toDF().select($"value" as "key", ($"value" * 3) as "rightValue")
+      val joined = df1.join(df2, Seq("key"), joinType)
+
+      val e = intercept[AnalysisException] {
+        testStream(joined, OutputMode.Update())(
+          AddData(input1, 1),
+          CheckAnswer()
+        )
+      }
+      assert(e.getMessage.contains("is not supported in Update output mode"))
+    }
+  }
+
   test("left outer early state exclusion on left") {
     withTempDir { checkpointDir =>
       val (leftInput, rightInput, joined) = setupWindowedJoinWithLeftCondition("left_outer")
@@ -1954,6 +1995,25 @@ abstract class StreamingOuterJoinSuite extends StreamingOuterJoinBase {
 @SlowSQLTest
 abstract class StreamingFullOuterJoinBase extends StreamingJoinSuite {
 
+  import testImplicits._
+
+  test("stream-stream full outer join does not support Update mode") {
+    val input1 = MemoryStream[Int]
+    val input2 = MemoryStream[Int]
+
+    val df1 = input1.toDF().select($"value" as "key", ($"value" * 2) as "leftValue")
+    val df2 = input2.toDF().select($"value" as "key", ($"value" * 3) as "rightValue")
+    val joined = df1.join(df2, Seq("key"), "full_outer")
+
+    val e = intercept[AnalysisException] {
+      testStream(joined, OutputMode.Update())(
+        AddData(input1, 1),
+        CheckAnswer()
+      )
+    }
+    assert(e.getMessage.contains("is not supported in Update output mode"))
+  }
+
   test("windowed full outer join") {
     withTempDir { checkpointDir =>
       val (leftInput, rightInput, joined) = setupWindowedJoin("full_outer")
@@ -2176,6 +2236,22 @@ abstract class StreamingLeftSemiJoinBase extends StreamingJoinSuite {
 
   import testImplicits._
 
+  // 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.
+  test("windowed left semi join with Update mode") {
+    withTempDir { checkpointDir =>
+      val (leftInput, rightInput, joined) = setupWindowedJoin("left_semi")
+
+      testStream(joined, OutputMode.Update())(
+        StartStream(checkpointLocation = checkpointDir.getCanonicalPath),
+        MultiAddData(leftInput, 1, 2, 3, 4, 5)(rightInput, 3, 4, 5, 6, 7),
+        CheckNewAnswer(Row(3, 10, 6), Row(4, 10, 8), Row(5, 10, 10)),
+        MultiAddData(leftInput, 21)(rightInput, 22),
+        CheckNewAnswer()
+      )
+    }
+  }
+
   test("windowed left semi join") {
     withTempDir { checkpointDir =>
       val (leftInput, rightInput, joined) = setupWindowedJoin("left_semi")