review comments #2

tejasapatil · tejasapatil · commit 23acc3ff82bc · 2017-03-15T07:23:04.000-07:00
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -718,18 +718,21 @@ object SQLConf {
 
   val WINDOW_EXEC_BUFFER_SPILL_THRESHOLD =
     buildConf("spark.sql.windowExec.buffer.spill.threshold")
+      .internal()
       .doc("Threshold for number of rows buffered in window operator")
       .intConf
       .createWithDefault(4096)
 
   val SORT_MERGE_JOIN_EXEC_BUFFER_SPILL_THRESHOLD =
     buildConf("spark.sql.sortMergeJoinExec.buffer.spill.threshold")
+      .internal()
       .doc("Threshold for number of rows buffered in sort merge join operator")
       .intConf
       .createWithDefault(Int.MaxValue)
 
   val CARTESIAN_PRODUCT_EXEC_BUFFER_SPILL_THRESHOLD =
     buildConf("spark.sql.cartesianProductExec.buffer.spill.threshold")
+      .internal()
       .doc("Threshold for number of rows buffered in cartesian product operator")
       .intConf
       .createWithDefault(UnsafeExternalSorter.DEFAULT_NUM_ELEMENTS_FOR_SPILL_THRESHOLD.toInt)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowFunctionFrame.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowFunctionFrame.scala
@@ -44,6 +44,12 @@ private[window] abstract class WindowFunctionFrame {
   def write(index: Int, current: InternalRow): Unit
 }
 
+object WindowFunctionFrame {
+  def getNextOrNull(iterator: Iterator[UnsafeRow]): UnsafeRow = {
+    if (iterator.hasNext) iterator.next() else null
+  }
+}
+
 /**
  * The offset window frame calculates frames containing LEAD/LAG statements.
  *
@@ -123,7 +129,7 @@ private[window] final class OffsetWindowFunctionFrame(
 
   override def write(index: Int, current: InternalRow): Unit = {
     if (inputIndex >= 0 && inputIndex < input.length) {
-      val r = if (inputIterator.hasNext) inputIterator.next() else null
+      val r = WindowFunctionFrame.getNextOrNull(inputIterator)
       projection(r)
     } else {
       // Use default values since the offset row does not exist.
@@ -179,9 +185,7 @@ private[window] final class SlidingWindowFunctionFrame(
   override def prepare(rows: ExternalAppendOnlyUnsafeRowArray): Unit = {
     input = rows
     inputIterator = input.generateIterator()
-    if (inputIterator.hasNext) {
-      nextRow = inputIterator.next()
-    }
+    nextRow = WindowFunctionFrame.getNextOrNull(inputIterator)
     inputHighIndex = 0
     inputLowIndex = 0
     buffer.clear()
@@ -195,7 +199,7 @@ private[window] final class SlidingWindowFunctionFrame(
     // the output row upper bound.
     while (nextRow != null && ubound.compare(nextRow, inputHighIndex, current, index) <= 0) {
       buffer.add(nextRow.copy())
-      nextRow = if (inputIterator.hasNext) inputIterator.next() else null
+      nextRow = WindowFunctionFrame.getNextOrNull(inputIterator)
       inputHighIndex += 1
       bufferUpdated = true
     }
@@ -311,7 +315,7 @@ private[window] final class UnboundedPrecedingWindowFunctionFrame(
     // the output row upper bound.
     while (nextRow != null && ubound.compare(nextRow, inputIndex, current, index) <= 0) {
       processor.update(nextRow)
-      nextRow = if (inputIterator.hasNext) inputIterator.next() else null
+      nextRow = WindowFunctionFrame.getNextOrNull(inputIterator)
       inputIndex += 1
       bufferUpdated = true
     }
@@ -368,23 +372,21 @@ private[window] final class UnboundedFollowingWindowFunctionFrame(
     // the output row lower bound.
     val iterator = input.generateIterator(startIndex = inputIndex)
 
-    def getNextOrNull(iterator: Iterator[UnsafeRow]): UnsafeRow = {
-      if (iterator.hasNext) iterator.next() else null
-    }
-
-    var nextRow = getNextOrNull(iterator)
+    var nextRow = WindowFunctionFrame.getNextOrNull(iterator)
     while (nextRow != null && lbound.compare(nextRow, inputIndex, current, index) < 0) {
       inputIndex += 1
       bufferUpdated = true
-      nextRow = getNextOrNull(iterator)
+      nextRow = WindowFunctionFrame.getNextOrNull(iterator)
     }
 
     // Only recalculate and update when the buffer changes.
     if (bufferUpdated) {
       processor.initialize(input.length)
-      while (nextRow != null) {
+      if (nextRow != null) {
         processor.update(nextRow)
-        nextRow = getNextOrNull(iterator)
+      }
+      while (iterator.hasNext) {
+        processor.update(iterator.next())
       }
       processor.evaluate(target)
     }
