fix

Author: HeartSaVioR

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

@@ -854,11 +854,16 @@ class SymmetricHashJoinStateManagerV4(
     def scanEvictedKeys(
         endTimestamp: Long,
         startTimestamp: Option[Long] = None): Iterator[EvictedKeysResult] = {
+      // If startTimestamp == Long.MaxValue, everything has already been evicted;
+      // nothing can match, so return immediately.
+      if (startTimestamp.contains(Long.MaxValue)) {
+        return Iterator.empty
+      }
+
       // rangeScanWithMultiValues: startKey is inclusive, endKey is exclusive.
       // startTimestamp is exclusive (already evicted), so we seek from st + 1.
-      val startKeyRow = startTimestamp.flatMap { st =>
-        if (st < Long.MaxValue) Some(createScanBoundaryRow(st + 1))
-        else None
+      val startKeyRow = startTimestamp.map { st =>
+        createScanBoundaryRow(st + 1)
       }
       // endTimestamp is inclusive, so we use endTimestamp + 1 as the exclusive upper bound.
       // When endTimestamp == Long.MaxValue we cannot add 1, so endKeyRow is None. This is

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManagerSuite.scala

@@ -1178,10 +1178,9 @@ class SymmetricHashJoinStateManagerEventTimeInValueSuite
       // --- overflow boundary: startTimestamp = Some(Long.MaxValue) ---
       Seq(10, 20).foreach(append(40, _))
       // startTimestamp=Long.MaxValue (exclusive) means everything <= Long.MaxValue was already
-      // evicted. Since startKeyRow falls back to None, endTimestamp=50 bounds the scan.
-      // All remaining entries (10, 20, 30) have timestamps <= 50, so they are evicted.
-      assert(evictByTs.evictByTimestamp(50, Some(Long.MaxValue)) === 3)
-      assert(get(40) === Seq.empty)
+      // evicted. Nothing can remain, so the scan returns an empty iterator immediately.
+      assert(evictByTs.evictByTimestamp(50, Some(Long.MaxValue)) === 0)
+      assert(get(40) === Seq(10, 20, 30))
     }
   }