jerrypeng commented on code in PR #57945:
URL: https://github.com/apache/spark/pull/57945#discussion_r3787437168
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationSuite.scala:
##########
@@ -684,6 +684,111 @@ class StreamingDeduplicationSuite extends
StateStoreMetricsTest
sqlConf = spark.sessionState.conf
)
}
+
+ // Total incremental removals reported across all batches, read from the
operator's
Review Comment:
fixed
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/statefulOperators.scala:
##########
@@ -1571,18 +1687,29 @@ case class StreamingDeduplicateWithinWatermarkExec(
store.put(key, timeoutRow)
}
- protected def evictDupInfoFromState(store: StateStore): Unit = {
+ override protected def iteratorForEviction(store: StateStore): Iterator[Any]
= {
val numRemovedStateRows = longMetric("numRemovedStateRows")
+ val numRowsReadDuringEviction = longMetric("numRowsReadDuringEviction")
- // Convert watermark value to micros.
+ // Incremental cleanup is not enabled for this operator (this class does
not override the base
+ // incrementalCleanupFactor, which defaults to 0 -- see the class-level
comment for why), so
+ // eviction always runs once at batch end against the eviction watermark.
val watermarkForEviction =
DateTimeUtils.millisToMicros(eventTimeWatermarkForEviction.get)
- store.iterator().foreach { rowPair =>
- val valueRow = rowPair.value
- val expiresAt = valueRow.getLong(0)
+ // We cannot reuse [[EvictionIterator]] here because it reads the eviction
timestamp from the
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
