cloud-fan commented on code in PR #57945:
URL: https://github.com/apache/spark/pull/57945#discussion_r3784160446
##########
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:
**Nit:**
`recentProgress` is retention-bounded, so this helper does not total an
arbitrarily long query's full batch history. Please describe this as the total
over retained recent progress, and make the same wording change in
`totalStateRowsRemoved` below.
##########
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:
**Nit:**
This is an ordinary line comment, so Scaladoc will not resolve `[[...]]`.
Please use backticks for `EvictionIterator` here.
--
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]
