feedback

ericm-db · Jan 5, 2024 · dc8380c · dc8380c
1 parent 149058c
commit dc8380c
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 8 deletions.
diff --git a/...core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala b/...core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala
@@ -113,7 +113,7 @@ case class TransformWithStateExec(
           case ProcessingTime =>
             val mi = statefulProcessor.handleProcessingTimeTimers(tsWithKey.key,
               tsWithKey.expiryTimestampMs,
-              new TimerValuesImpl(batchTimestampMs, eventTimeWatermarkForLateEvents)).map {
+              new TimerValuesImpl(batchTimestampMs, eventTimeWatermarkForEviction)).map {
               obj =>
                 getOutputRow(obj)
             }
@@ -123,8 +123,7 @@ case class TransformWithStateExec(
           case EventTime =>
             val mi = statefulProcessor.handleEventTimeTimers(tsWithKey.key,
               tsWithKey.expiryTimestampMs,
-              new TimerValuesImpl(eventTimeWatermarkForEviction,
-                eventTimeWatermarkForLateEvents)).map {
+              new TimerValuesImpl(batchTimestampMs, eventTimeWatermarkForEviction)).map {
               obj =>
                 getOutputRow(obj)
             }

diff --git a/...t/scala/org/apache/spark/sql/execution/streaming/state/StatefulProcessorHandleSuite.scala b/...t/scala/org/apache/spark/sql/execution/streaming/state/StatefulProcessorHandleSuite.scala
@@ -60,10 +60,10 @@ class StatefulProcessorHandleSuite extends SharedSparkSession
   val schemaForValueRow: StructType = new StructType().add("value", BinaryType)
 
   private def newStoreProviderWithHandle(useColumnFamilies: Boolean):
-  RocksDBStateStoreProvider = {
-    newStoreProviderWithHandle(StateStoreId(newDir(), Random.nextInt(), 0),
-      numColsPrefixKey = 0,
-      useColumnFamilies = useColumnFamilies)
+    RocksDBStateStoreProvider = {
+      newStoreProviderWithHandle(StateStoreId(newDir(), Random.nextInt(), 0),
+        numColsPrefixKey = 0,
+        useColumnFamilies = useColumnFamilies)
   }
 
   private def newStoreProviderWithHandle(

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala
@@ -433,7 +433,6 @@ class TransformWithStateSuite extends StateStoreMetricsTest
           TimeoutMode.EventTime(),
           OutputMode.Update())
 
-
       testStream(result, OutputMode.Update())(
         StartStream(),
         AddData(inputData, ("a", Timestamp.valueOf("2023-08-01 00:00:00"))),