[SPARK-24634][SS][FOLLOWUP] Rename the variable from "numLateInputs" to "numRowsDroppedByWatermark"

docs/structured-streaming-programming-guide.md

@@ -1678,12 +1678,11 @@ emits late rows if the operator uses Append mode.
 Spark provides two ways to check the number of late rows on stateful operators which would help you identify the issue:
 
 1. On Spark UI: check the metrics in stateful operator nodes in query execution details page in SQL tab
-2. On Streaming Query Listener: check "numLateInputs" in "stateOperators" in QueryProcessEvent.
+2. On Streaming Query Listener: check "numRowsDroppedByWatermark" in "stateOperators" in QueryProcessEvent.
 
-Please note that the definition of "input" is relative: it doesn't always mean "input rows" for the operator.
-Streaming aggregation does pre-aggregate input rows and checks the late inputs against pre-aggregated inputs,
-hence the number is not same as the number of original input rows. You'd like to check the fact whether the value is zero
-or non-zero.
+Please note that "numRowsDroppedByWatermark" represents the number of "dropped" rows by watermark, which is not always same as the count of "late input rows" for the operator.
+It depends on the implementation of the operator - e.g. streaming aggregation does pre-aggregate input rows and checks the late inputs against pre-aggregated inputs,
+hence the number is not same as the number of original input rows. You'd like to just check the fact whether the value is zero or non-zero.
 
 There's a known workaround: split your streaming query into multiple queries per stateful operator, and ensure
 end-to-end exactly once per query. Ensuring end-to-end exactly once for the last query is optional.

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

@@ -77,8 +77,8 @@ trait StateStoreWriter extends StatefulOperator { self: SparkPlan =>
 
   override lazy val metrics = Map(
     "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows"),
-    "numLateInputs" -> SQLMetrics.createMetric(sparkContext,
-      "number of inputs which are later than watermark ('inputs' are relative to operators)"),
+    "numRowsDroppedByWatermark" -> SQLMetrics.createMetric(sparkContext,
+      "number of rows which are dropped by watermark"),
     "numTotalStateRows" -> SQLMetrics.createMetric(sparkContext, "number of total state rows"),
     "numUpdatedStateRows" -> SQLMetrics.createMetric(sparkContext, "number of updated state rows"),
     "allUpdatesTimeMs" -> SQLMetrics.createTimingMetric(sparkContext, "time to update"),
@@ -102,7 +102,7 @@ trait StateStoreWriter extends StatefulOperator { self: SparkPlan =>
       numRowsTotal = longMetric("numTotalStateRows").value,
       numRowsUpdated = longMetric("numUpdatedStateRows").value,
       memoryUsedBytes = longMetric("stateMemory").value,
-      numLateInputs = longMetric("numLateInputs").value,
+      numRowsDroppedByWatermark = longMetric("numRowsDroppedByWatermark").value,
       javaConvertedCustomMetrics
     )
   }
@@ -137,10 +137,10 @@ trait StateStoreWriter extends StatefulOperator { self: SparkPlan =>
 
   protected def applyRemovingRowsOlderThanWatermark(
       iter: Iterator[InternalRow],
-      predicateFilterOutLateInput: BasePredicate): Iterator[InternalRow] = {
+      predicateDropRowByWatermark: BasePredicate): Iterator[InternalRow] = {
     iter.filterNot { row =>
-      val lateInput = predicateFilterOutLateInput.eval(row)
-      if (lateInput) longMetric("numLateInputs") += 1
+      val lateInput = predicateDropRowByWatermark.eval(row)
+      if (lateInput) longMetric("numRowsDroppedByWatermark") += 1
       lateInput
     }
   }

sql/core/src/main/scala/org/apache/spark/sql/streaming/progress.scala

@@ -43,7 +43,7 @@ class StateOperatorProgress private[sql](
     val numRowsTotal: Long,
     val numRowsUpdated: Long,
     val memoryUsedBytes: Long,
-    val numLateInputs: Long,
+    val numRowsDroppedByWatermark: Long,
     val customMetrics: ju.Map[String, JLong] = new ju.HashMap()
   ) extends Serializable {
 
@@ -63,7 +63,7 @@ class StateOperatorProgress private[sql](
     ("numRowsTotal" -> JInt(numRowsTotal)) ~
     ("numRowsUpdated" -> JInt(numRowsUpdated)) ~
     ("memoryUsedBytes" -> JInt(memoryUsedBytes)) ~
-    ("numLateInputs" -> JInt(numLateInputs)) ~
+    ("numRowsDroppedByWatermark" -> JInt(numRowsDroppedByWatermark)) ~
     ("customMetrics" -> {
       if (!customMetrics.isEmpty) {
         val keys = customMetrics.keySet.asScala.toSeq.sorted

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

@@ -298,11 +298,11 @@ class EventTimeWatermarkSuite extends StreamTest with BeforeAndAfter with Matche
       AddData(inputData, 25),   // Advance watermark to 15 seconds
       CheckNewAnswer((10, 5)),
       assertNumStateRows(2),
-      assertNumLateInputs(0),
+      assertNumRowsDroppedByWatermark(0),
       AddData(inputData, 10),   // Should not emit anything as data less than watermark
       CheckNewAnswer(),
       assertNumStateRows(2),
-      assertNumLateInputs(1)
+      assertNumRowsDroppedByWatermark(1)
     )
   }
 
@@ -323,15 +323,15 @@ class EventTimeWatermarkSuite extends StreamTest with BeforeAndAfter with Matche
       AddData(inputData, 25),     // Advance watermark to 15 seconds
       CheckNewAnswer((25, 1)),
       assertNumStateRows(2),
-      assertNumLateInputs(0),
+      assertNumRowsDroppedByWatermark(0),
       AddData(inputData, 10, 25), // Ignore 10 as its less than watermark
       CheckNewAnswer((25, 2)),
       assertNumStateRows(2),
-      assertNumLateInputs(1),
+      assertNumRowsDroppedByWatermark(1),
       AddData(inputData, 10),     // Should not emit anything as data less than watermark
       CheckNewAnswer(),
       assertNumStateRows(2),
-      assertNumLateInputs(1)
+      assertNumRowsDroppedByWatermark(1)
     )
   }
 
@@ -788,15 +788,17 @@ class EventTimeWatermarkSuite extends StreamTest with BeforeAndAfter with Matche
     true
   }
 
-  private def assertNumLateInputs(numLateInputs: Long): AssertOnQuery = AssertOnQuery { q =>
+  private def assertNumRowsDroppedByWatermark(
+      numRowsDroppedByWatermark: Long): AssertOnQuery = AssertOnQuery { q =>
     q.processAllAvailable()
     val progressWithData = q.recentProgress.filterNot { p =>
       // filter out batches which are falling into one of types:
       // 1) doesn't execute the batch run
       // 2) empty input batch
       p.inputRowsPerSecond == 0
     }.lastOption.get
-    assert(progressWithData.stateOperators(0).numLateInputs === numLateInputs)
+    assert(progressWithData.stateOperators(0).numRowsDroppedByWatermark
+      === numRowsDroppedByWatermark)
     true
   }
 

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

@@ -55,8 +55,8 @@ trait StateStoreMetricsTest extends StreamTest {
         val allNumUpdatedRowsSinceLastCheck =
           progressesSinceLastCheck.map(_.stateOperators.map(_.numRowsUpdated))
 
-        val allNumLateInputsSinceLastCheck =
-          progressesSinceLastCheck.map(_.stateOperators.map(_.numLateInputs))
+        val allNumRowsDroppedByWatermarkSinceLastCheck =
+          progressesSinceLastCheck.map(_.stateOperators.map(_.numRowsDroppedByWatermark))
 
         lazy val debugString = "recent progresses:\n" +
           progressesSinceLastCheck.map(_.prettyJson).mkString("\n\n")
@@ -67,8 +67,10 @@ trait StateStoreMetricsTest extends StreamTest {
         val numUpdatedRows = arraySum(allNumUpdatedRowsSinceLastCheck, numStateOperators)
         assert(numUpdatedRows === updated, s"incorrect updates rows, $debugString")
 
-        val numLateInputs = arraySum(allNumLateInputsSinceLastCheck, numStateOperators)
-        assert(numLateInputs === lateInputs, s"incorrect late inputs, $debugString")
+        val numRowsDroppedByWatermark = arraySum(allNumRowsDroppedByWatermarkSinceLastCheck,
+          numStateOperators)
+        assert(numRowsDroppedByWatermark === lateInputs,
+          s"incorrect dropped rows by watermark, $debugString")
 
         lastCheckedRecentProgressIndex = recentProgress.length - 1
       }

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

@@ -64,7 +64,7 @@ class StreamingQueryStatusAndProgressSuite extends StreamTest with Eventually {
         |    "numRowsTotal" : 0,
         |    "numRowsUpdated" : 1,
         |    "memoryUsedBytes" : 3,
-        |    "numLateInputs" : 0,
+        |    "numRowsDroppedByWatermark" : 0,
         |    "customMetrics" : {
         |      "loadedMapCacheHitCount" : 1,
         |      "loadedMapCacheMissCount" : 0,
@@ -115,7 +115,7 @@ class StreamingQueryStatusAndProgressSuite extends StreamTest with Eventually {
          |    "numRowsTotal" : 0,
          |    "numRowsUpdated" : 1,
          |    "memoryUsedBytes" : 2,
-         |    "numLateInputs" : 0
+         |    "numRowsDroppedByWatermark" : 0
          |  } ],
          |  "sources" : [ {
          |    "description" : "source",
@@ -323,7 +323,7 @@ object StreamingQueryStatusAndProgressSuite {
       "avg" -> "2016-12-05T20:54:20.827Z",
       "watermark" -> "2016-12-05T20:54:20.827Z").asJava),
     stateOperators = Array(new StateOperatorProgress(
-      numRowsTotal = 0, numRowsUpdated = 1, memoryUsedBytes = 3, numLateInputs = 0,
+      numRowsTotal = 0, numRowsUpdated = 1, memoryUsedBytes = 3, numRowsDroppedByWatermark = 0,
       customMetrics = new java.util.HashMap(Map("stateOnCurrentVersionSizeBytes" -> 2L,
         "loadedMapCacheHitCount" -> 1L, "loadedMapCacheMissCount" -> 0L)
         .mapValues(long2Long).asJava)
@@ -355,7 +355,7 @@ object StreamingQueryStatusAndProgressSuite {
     // empty maps should be handled correctly
     eventTime = new java.util.HashMap(Map.empty[String, String].asJava),
     stateOperators = Array(new StateOperatorProgress(
-      numRowsTotal = 0, numRowsUpdated = 1, memoryUsedBytes = 2, numLateInputs = 0)),
+      numRowsTotal = 0, numRowsUpdated = 1, memoryUsedBytes = 2, numRowsDroppedByWatermark = 0)),
     sources = Array(
       new SourceProgress(
         description = "source",