Address part of review comments

Author: HeartSaVioR

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

@@ -299,7 +299,7 @@ case class StreamingSymmetricHashJoinExec(
             case 1 => matchesWithRightSideState(
               new UnsafeRowPair(kvAndMatched.key, kvAndMatched.value))
             case 2 => kvAndMatched.matched
-            case _ => throw new IllegalStateException("Incorrect state format version! " +
+            case _ => throw new IllegalStateException("Unexpected state format version! " +
               s"version $stateFormatVersion")
           }
         }.map(pair => joinedRow.withLeft(pair.value).withRight(nullRight))
@@ -318,7 +318,7 @@ case class StreamingSymmetricHashJoinExec(
             case 1 => matchesWithLeftSideState(
               new UnsafeRowPair(kvAndMatched.key, kvAndMatched.value))
             case 2 => kvAndMatched.matched
-            case _ => throw new IllegalStateException("Incorrect state format version! " +
+            case _ => throw new IllegalStateException("Unexpected state format version! " +
               s"version $stateFormatVersion")
           }
         }.map(pair => joinedRow.withLeft(nullLeft).withRight(pair.value))
@@ -491,15 +491,7 @@ case class StreamingSymmetricHashJoinExec(
         thisRow: UnsafeRow,
         subIter: Iterator[JoinedRow])
       extends CompletionIterator[JoinedRow, Iterator[JoinedRow]](subIter) {
-      private var iteratorNotEmpty: Boolean = false
-
-      override def hasNext: Boolean = {
-        val ret = super.hasNext
-        if (ret && !iteratorNotEmpty) {
-          iteratorNotEmpty = true
-        }
-        ret
-      }
+      private val iteratorNotEmpty: Boolean = super.hasNext
 
       override def completion(): Unit = {
         val shouldAddToState = // add only if both removal predicates do not match

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

@@ -107,7 +107,6 @@ class SymmetricHashJoinStateManager(
       val joinedRow = generateJoinedRow(keyIdxToValue.value)
       if (predicate(joinedRow)) {
         if (!keyIdxToValue.matched) {
-          // only update when matched flag is false
           keyWithIndexToValue.put(key, keyIdxToValue.valueIndex, keyIdxToValue.value,
             matched = true)
         }
@@ -134,7 +133,7 @@ class SymmetricHashJoinStateManager(
       private val allKeyToNumValues = keyToNumValues.iterator
 
       private var currentKeyToNumValue: KeyAndNumValues = null
-      private var currentValues: Iterator[KeyWithIndexAndValueWithMatched] = null
+      private var currentValues: Iterator[KeyWithIndexAndValue] = null
 
       private def currentKey = currentKeyToNumValue.key
 
@@ -412,7 +411,7 @@ class SymmetricHashJoinStateManager(
    * Helper class for representing data returned by [[KeyWithIndexToValueStore]].
    * Designed for object reuse.
    */
-  private class KeyWithIndexAndValueWithMatched(
+  private class KeyWithIndexAndValue(
     var key: UnsafeRow = null,
     var valueIndex: Long = -1,
     var value: UnsafeRow = null,
@@ -520,11 +519,11 @@ class SymmetricHashJoinStateManager(
      * Get all values and indices for the provided key.
      * Should not return null.
      */
-    def getAll(key: UnsafeRow, numValues: Long): Iterator[KeyWithIndexAndValueWithMatched] = {
-      val keyWithIndexAndValueWithMatched = new KeyWithIndexAndValueWithMatched()
+    def getAll(key: UnsafeRow, numValues: Long): Iterator[KeyWithIndexAndValue] = {
+      val keyWithIndexAndValueWithMatched = new KeyWithIndexAndValue()
       var index = 0
-      new NextIterator[KeyWithIndexAndValueWithMatched] {
-        override protected def getNext(): KeyWithIndexAndValueWithMatched = {
+      new NextIterator[KeyWithIndexAndValue] {
+        override protected def getNext(): KeyWithIndexAndValue = {
           if (index >= numValues) {
             finished = true
             null
@@ -565,8 +564,8 @@ class SymmetricHashJoinStateManager(
       }
     }
 
-    def iterator: Iterator[KeyWithIndexAndValueWithMatched] = {
-      val keyWithIndexAndValueWithMatched = new KeyWithIndexAndValueWithMatched()
+    def iterator: Iterator[KeyWithIndexAndValue] = {
+      val keyWithIndexAndValueWithMatched = new KeyWithIndexAndValue()
       stateStore.getRange(None, None).map { pair =>
         val (value, matched) = valueRowConverter.convertValue(pair.value)
         keyWithIndexAndValueWithMatched.withNew(

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

@@ -800,8 +800,8 @@ class StreamingOuterJoinSuite extends StreamTest with StateStoreMetricsTest with
         expr("leftId = rightId AND leftTime >= rightTime AND " +
           "leftTime <= rightTime + interval 5 seconds"),
         joinType = "rightOuter")
-      .select(col("rightId"), col("rightTime").cast("int"),
-        col("leftId"), col("leftTime").cast("int"))
+      .select(col("leftId"), col("leftTime").cast("int"),
+        col("rightId"), col("rightTime").cast("int"))
 
     // we can just flip left and right in the explanation of left outer query test
     // to assume the status of right outer query, hence skip explaining here
@@ -817,7 +817,7 @@ class StreamingOuterJoinSuite extends StreamTest with StateStoreMetricsTest with
       AddData(inputStream, (11, 11L), (12, 12L), (13, 13L), (14, 14L), (15, 15L)),
       CheckNewAnswer(
         Row(12, 12L, 12, 12L), Row(14, 14L, 14, 14L),
-        Row(1, 1L, null, null), Row(3, 3L, null, null)),
+        Row(null, null, 1, 1L), Row(null, null, 3, 3L)),
       assertNumStateRows(15, 7)
     )
   }