Address backward compatibility with old state

Author: HeartSaVioR

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

@@ -266,15 +266,43 @@ case class StreamingSymmetricHashJoinExec(
       case Inner =>
         innerOutputIter
       case LeftOuter =>
+        // We generate the outer join input by:
+        // * Getting an iterator over the rows that have aged out on the left side. These rows are
+        //   candidates for being null joined. Note that to avoid doing two passes, this iterator
+        //   removes the rows from the state manager as they're processed.
+        // * Checking whether the current row matches a key in the right side state, and that key
+        //   has any value which satisfies the filter function when joined. If it doesn't,
+        //   we know we can join with null, since there was never (including this batch) a match
+        //   within the watermark period. If it does, there must have been a match at some point, so
+        //   we know we can't join with null.
+        def matchesWithRightSideState(leftKeyValue: UnsafeRowPair) = {
+          rightSideJoiner.get(leftKeyValue.key).exists { rightValue =>
+            postJoinFilter(joinedRow.withLeft(leftKeyValue.value).withRight(rightValue))
+          }
+        }
+
         val removedRowIter = leftSideJoiner.removeOldState()
-        val outerOutputIter = removedRowIter.filterNot(_.matched)
-          .map(pair => joinedRow.withLeft(pair.value).withRight(nullRight))
+        val outerOutputIter = removedRowIter.filterNot { kvAndMatched =>
+          kvAndMatched.matched.getOrElse(
+            // fail-back for previous state on SPARK-26154
+            matchesWithRightSideState(new UnsafeRowPair(kvAndMatched.key, kvAndMatched.value)))
+        }.map(pair => joinedRow.withLeft(pair.value).withRight(nullRight))
 
         innerOutputIter ++ outerOutputIter
       case RightOuter =>
+        // See comments for left outer case.
+        def matchesWithLeftSideState(rightKeyValue: UnsafeRowPair) = {
+          leftSideJoiner.get(rightKeyValue.key).exists { leftValue =>
+            postJoinFilter(joinedRow.withLeft(leftValue).withRight(rightKeyValue.value))
+          }
+        }
+
         val removedRowIter = rightSideJoiner.removeOldState()
-        val outerOutputIter = removedRowIter.filterNot(_.matched)
-          .map(pair => joinedRow.withLeft(nullLeft).withRight(pair.value))
+        val outerOutputIter = removedRowIter.filterNot { kvAndMatched =>
+          kvAndMatched.matched.getOrElse(
+            // fail-back for previous state on SPARK-26154
+            matchesWithLeftSideState(new UnsafeRowPair(kvAndMatched.key, kvAndMatched.value)))
+        }.map(pair => joinedRow.withLeft(nullLeft).withRight(pair.value))
 
         innerOutputIter ++ outerOutputIter
       case _ => throwBadJoinTypeException()
@@ -465,6 +493,15 @@ case class StreamingSymmetricHashJoinExec(
       }
     }
 
+    /**
+     * Get an iterator over the values stored in this joiner's state manager for the given key.
+     *
+     * Should not be interleaved with mutations.
+     */
+    def get(key: UnsafeRow): Iterator[UnsafeRow] = {
+      joinStateManager.get(key)
+    }
+
     /**
      * Builds an iterator over old state key-value pairs, removing them lazily as they're produced.
      *

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

@@ -105,13 +105,6 @@ class SymmetricHashJoinStateManager(
     }.filter(_ != null)
   }
 
-  /** Append a new value to the key */
-  def append(key: UnsafeRow, value: UnsafeRow): Unit = {
-    val numExistingValues = keyToNumValues.get(key)
-    keyWithIndexToValue.put(key, numExistingValues, value)
-    keyToNumValues.put(key, numExistingValues + 1)
-  }
-
   /** Append a new value to the key, with marking matched */
   def append(key: UnsafeRow, value: UnsafeRow, matched: Boolean): Unit = {
     val numExistingValues = keyToNumValues.get(key)
@@ -223,7 +216,7 @@ class SymmetricHashJoinStateManager(
 
       // Find the next value satisfying the condition, updating `currentKey` and `numValues` if
       // needed. Returns null when no value can be found.
-      private def findNextValueForIndex(): (UnsafeRow, Boolean) = {
+      private def findNextValueForIndex(): (UnsafeRow, Option[Boolean]) = {
         // Loop across all values for the current key, and then all other keys, until we find a
         // value satisfying the removal condition.
         def hasMoreValuesForCurrentKey = currentKey != null && index < numValues
@@ -273,9 +266,13 @@ class SymmetricHashJoinStateManager(
           keyWithIndexToValue.put(currentKey, index, valueAtMaxIndex)
           keyWithIndexToValue.remove(currentKey, numValues - 1)
 
-          val matchedAtMaxIndex = keyWithIndexToMatched.get(currentKey, numValues - 1)
-          keyWithIndexToMatched.put(currentKey, index, matchedAtMaxIndex)
-          keyWithIndexToMatched.remove(currentKey, numValues - 1)
+          keyWithIndexToMatched.get(currentKey, numValues - 1) match {
+            case Some(matchedAtMaxIndex) =>
+              keyWithIndexToMatched.put(currentKey, index, matchedAtMaxIndex)
+              keyWithIndexToMatched.remove(currentKey, numValues - 1)
+
+            case None =>
+          }
         } else {
           keyWithIndexToValue.remove(currentKey, 0)
           keyWithIndexToMatched.remove(currentKey, 0)
@@ -556,9 +553,9 @@ class SymmetricHashJoinStateManager(
 
     protected val stateStore = getStateStore(keyWithIndexSchema, booleanValueSchema)
 
-    def get(key: UnsafeRow, valueIndex: Long): Boolean = {
+    def get(key: UnsafeRow, valueIndex: Long): Option[Boolean] = {
       val row = stateStore.get(keyWithIndexRow(key, valueIndex))
-      if (row != null) row.getBoolean(0) else false
+      if (row != null) Some(row.getBoolean(0)) else None
     }
 
     /** Put matched for key at the given index */
@@ -601,8 +598,10 @@ object SymmetricHashJoinStateManager {
    * Designed for object reuse.
    */
   case class KeyToValueAndMatched(
-      var key: UnsafeRow = null, var value: UnsafeRow = null, var matched: Boolean = false) {
-    def withNew(newKey: UnsafeRow, newValue: UnsafeRow, newMatched: Boolean): this.type = {
+      var key: UnsafeRow = null,
+      var value: UnsafeRow = null,
+      var matched: Option[Boolean] = None) {
+    def withNew(newKey: UnsafeRow, newValue: UnsafeRow, newMatched: Option[Boolean]): this.type = {
       this.key = newKey
       this.value = newValue
       this.matched = newMatched

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

@@ -123,7 +123,7 @@ class SymmetricHashJoinStateManagerSuite extends StreamTest with BeforeAndAfter
   def toValueInt(inputValueRow: UnsafeRow): Int = inputValueRow.getInt(0)
 
   def append(key: Int, value: Int)(implicit manager: SymmetricHashJoinStateManager): Unit = {
-    manager.append(toJoinKeyRow(key), toInputValue(value))
+    manager.append(toJoinKeyRow(key), toInputValue(value), matched = false)
   }
 
   def get(key: Int)(implicit manager: SymmetricHashJoinStateManager): Seq[Int] = {