reflect review comments

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

@@ -672,10 +672,10 @@ case class StreamingSymmetricHashJoinExec(
         // - Left side: input can be skipped to be added to the state store if it's already matched
         //   and the join type is left semi.
         //   For other cases, the input should be added, including the case it's going to be evicted
-        //   in this batch. It hasn't yet evaluated with inputs from right side "for this batch".
+        //   in this batch. It hasn't yet evaluated with inputs from right side for this batch.
         //   Refer to the classdoc of SteramingSymmetricHashJoinExec about how stream-stream join
         //   works.
-        // - Right side: for this side, the evaluation with inputs from left side "for this batch"
+        // - Right side: for this side, the evaluation with inputs from left side for this batch
         //   is done at this point. That said, input can be skipped to be added to the state store
         //   if input is going to be evicted in this batch. Though, input should be added to the
         //   state store if it's right outer join or full outer join, as unmatched output is

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

@@ -908,7 +908,7 @@ class MultiStatefulOperatorsSuite
     // does not delay the state watermark in stream-stream join).
     // Before SPARK-49829, left side does not add the input to state store if it's going to evict
     // in this batch, which breaks the match between input from left side and input from right
-    // side "for this batch".
+    // side for this batch.
     testStream(joined)(
       MultiAddData(
         (inputStream1, Seq(1L, 2L, 3L, 4L, 5L)),
@@ -921,6 +921,7 @@ class MultiStatefulOperatorsSuite
         (inputStream2, Seq(15L, 16L, 17L, 18L, 19L))
       ),
       // watermark: 15 - 5 = 10 (windows for [0, 10) are completed)
+      // Before SPARK-49829, the test fails because this row is not produced.
       CheckNewAnswer((10L, 15L, 35L)),
       MultiAddData(
         (inputStream1, Seq(100L)),

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

@@ -1630,7 +1630,9 @@ class StreamingOuterJoinSuite extends StreamingJoinSuite {
           // Before Spark introduces multiple stateful operator, WM for late record was same as
           // WM for eviction, hence ("d", 1) was treated as late record.
           // With the multiple state operator, ("d", 1) is added in batch 1 but also evicted in
-          // batch 1. Before SPARK-49829, this wasn't producing unmatched row, and it is fixed.
+          // batch 1. Note that the eviction is happening with state watermark: for this join,
+          // state watermark = state eviction under join condition. Before SPARK-49829, this
+          // wasn't producing unmatched row, and it is fixed.
           AddData(memoryStream1, ("d", 1)),
           CheckNewAnswer(("a", 1), ("d", 1)),
           assertLeftRows(Seq()),
@@ -2152,6 +2154,9 @@ class StreamingLeftSemiJoinSuite extends StreamingJoinSuite {
         // just trigger a new batch with arbitrary data as the original test relies on no-data
         // batch, and we need to check with remaining unmatched outputs
         AddData(memoryStream1, (100L, 6)),
+        // Before SPARK-49829, the test fails because (23, 4, null, null) wasn't produced.
+        // (The assertion of state for left inputs & right inputs weren't included on the test
+        // before SPARK-49829.)
         CheckNewAnswer(Row(22, 3, null, 3), Row(23, 4, null, null))
       )