[SPARK-55973][SS] LeftSemi optimization for stream-stream join

### What changes were proposed in this pull request?

This PR proposes to apply a couple optimizations which stream-stream join with LeftSemi join type currently misses, mostly for state store.

The current implementation has below issues:

1. The operator bypasses storing the left side of the row when there is a match in the right side. But since the operator processes left side -> right side, we can't leverage this optimization when there is a match in the same batch.
2. The operator keeps the rows in the left side till watermark passes by, even though they were already matched and they won't be used as output ever again.

This PR proposes the below changes based on the observation:

1. The operator processes right side first, and then left side for LeftSemi join type. It actually doesn't matter for other join types to change the order, but they don't get the benefit so we don't introduce that change.
2. The operator aggressively removes the rows in the state on left side as long as it realizes they are matched with new row in the right side, rather than waiting for watermark to pass by.

The code change for the verification of the metrics from the tests for LeftSemi shows the impact of this optimization.

### Why are the changes needed?

Explained the above section.

### Does this PR introduce _any_ user-facing change?

It's mostly an internal change. There is user-facing change if they ever change the join type during the restart, but that is an area of "undocumented" and we do not define any behavior with it.

### How was this patch tested?

Existing tests and new tests.

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

Generated-by: Claude 4.6 opus (co-authored)

Closes #54769 from HeartSaVioR/SPARK-55973.

Authored-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

Author: HeartSaVioR

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

@@ -394,6 +394,10 @@ case class StreamingSymmetricHashJoinExec(
     //    - Left Semi Join: generates all stored left input rows, from matching new right input
     //      with stored left input, and also stores all the right input. Note only first-time
     //      matched left input rows will be generated, this is to guarantee left semi semantics.
+    //
+    //  For Left Semi Join, we process the right side first so that new right input is stored
+    //  before left input probes against it. This way, new left rows that match new right rows
+    //  in the same microbatch are emitted immediately without being buffered in state.
     val leftOutputIter =
       joinerManager.leftSideJoiner.storeAndJoinWithOtherSide(joinerManager.rightSideJoiner) {
         (input: InternalRow, matched: InternalRow) => joinedRow.withLeft(input).withRight(matched)
@@ -412,8 +416,13 @@ case class StreamingSymmetricHashJoinExec(
     // This is the iterator which produces the inner and left semi join rows. For other joins,
     // this will be prepended to a second iterator producing other rows; for inner and left semi
     // joins, this is the full output.
+    //
+    // For left semi join, right side is processed first so new right rows are available in
+    // state when left rows probe, avoiding unnecessary left-side state buffering.
     val hashJoinOutputIter = CompletionIterator[InternalRow, Iterator[InternalRow]](
-      leftOutputIter ++ rightOutputIter, onHashJoinOutputCompletion())
+      if (joinType == LeftSemi) rightOutputIter ++ leftOutputIter
+      else leftOutputIter ++ rightOutputIter,
+      onHashJoinOutputCompletion())
 
     val outputIter: Iterator[InternalRow] = joinType match {
       case Inner | LeftSemi =>
@@ -535,6 +544,11 @@ case class StreamingSymmetricHashJoinExec(
           math.max(NANOSECONDS.toMillis(System.nanoTime - hashJoinOutputCompletionTimeNs), 0)
       }
 
+      // Count rows removed from the other side's state during the join phase
+      // (e.g., left semi join optimization removes matched left-side rows while processing
+      // right-side input, instead of a separate eviction pass).
+      numRemovedStateRows += joinerManager.numRemovedFromOtherSideDuringJoin
+
       allRemovalsTimeMs += timeTakenMs {
         // Remove any remaining state rows which aren't needed because they're below the watermark.
         //
@@ -664,6 +678,7 @@ case class StreamingSymmetricHashJoinExec(
     }
 
     private[this] var updatedStateRowsCount = 0
+    private[this] var numRemovedFromOtherSideDuringJoinCount = 0
     private[this] val allowMultipleStatefulOperators: Boolean =
       conf.getConf(SQLConf.STATEFUL_OPERATOR_ALLOW_MULTIPLE)
 
@@ -699,8 +714,6 @@ case class StreamingSymmetricHashJoinExec(
         case _ => (_: InternalRow) => Iterator.empty
       }
 
-      val excludeRowsAlreadyMatched = joinType == LeftSemi && joinSide == RightSide
-
       val generateOutputIter: (InternalRow, Iterator[JoinedRow]) => Iterator[InternalRow] =
         joinSide match {
           case LeftSide if joinType == LeftSemi =>
@@ -719,6 +732,9 @@ case class StreamingSymmetricHashJoinExec(
           case _ => (_: InternalRow, joinedRowIter: Iterator[JoinedRow]) => joinedRowIter
         }
 
+      val removeMatchedFromOtherSideState =
+        joinType == LeftSemi && joinSide == RightSide
+
       nonLateRows.flatMap { row =>
         val thisRow = row.asInstanceOf[UnsafeRow]
         // If this row fails the pre join filter, that means it can never satisfy the full join
@@ -727,11 +743,23 @@ case class StreamingSymmetricHashJoinExec(
         // the case of inner join).
         if (preJoinFilter(thisRow)) {
           val key = keyGenerator(thisRow)
-          val joinedRowIter: Iterator[JoinedRow] = otherSideJoiner.joinStateManager.getJoinedRows(
-            key,
-            thatRow => generateJoinedRow(thisRow, thatRow),
-            postJoinFilter,
-            excludeRowsAlreadyMatched)
+          // If the join type is Left Semi and this is the right side, we can remove the matched
+          // row from the other (left) side's state, since the row won't be produced anymore for
+          // the following input rows.
+          val joinedRowIter: Iterator[JoinedRow] = if (removeMatchedFromOtherSideState) {
+            otherSideJoiner.joinStateManager.getJoinedRowsAndRemoveMatched(
+              key,
+              thatRow => generateJoinedRow(thisRow, thatRow),
+              postJoinFilter).map { row =>
+              numRemovedFromOtherSideDuringJoinCount += 1
+              row
+            }
+          } else {
+            otherSideJoiner.joinStateManager.getJoinedRows(
+              key,
+              thatRow => generateJoinedRow(thisRow, thatRow),
+              postJoinFilter)
+          }
           val outputIter = generateOutputIter(thisRow, joinedRowIter)
           new AddingProcessedRowToStateCompletionIterator(key, thisRow, outputIter)
         } else {
@@ -877,6 +905,8 @@ case class StreamingSymmetricHashJoinExec(
     }
 
     def numUpdatedStateRows: Long = updatedStateRowsCount
+
+    def numRemovedFromOtherSideDuringJoin: Long = numRemovedFromOtherSideDuringJoinCount
   }
 
   /**
@@ -886,6 +916,11 @@ case class StreamingSymmetricHashJoinExec(
   private case class OneSideHashJoinerManager(
       leftSideJoiner: OneSideHashJoiner, rightSideJoiner: OneSideHashJoiner) {
 
+    def numRemovedFromOtherSideDuringJoin: Long = {
+      leftSideJoiner.numRemovedFromOtherSideDuringJoin +
+        rightSideJoiner.numRemovedFromOtherSideDuringJoin
+    }
+
     def removeOldState(): Long = {
       leftSideJoiner.removeOldState() + rightSideJoiner.removeOldState()
     }