anishshri-db commented on code in PR #45937:
URL: https://github.com/apache/spark/pull/45937#discussion_r1561386255
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TTLState.scala:
##########
@@ -105,17 +109,21 @@ abstract class SingleKeyTTLStateImpl(
/**
* Clears any state which has ttl older than [[ttlExpirationMs]].
*/
- override def clearExpiredState(): Unit = {
+ override def clearExpiredState(): Long = {
val iterator = store.iterator(ttlColumnFamilyName)
+ var numValuesExpired = 0L
iterator.takeWhile { kv =>
val expirationMs = kv.key.getLong(0)
StateTTL.isExpired(expirationMs, ttlExpirationMs)
}.foreach { kv =>
val groupingKey = kv.key.getBinary(1)
- clearIfExpired(groupingKey)
+ if (clearIfExpired(groupingKey)) {
+ numValuesExpired += 1
+ }
Review Comment:
Discussed offline - but we will keep this as it is currently to avoid
passing the metrics object to the TTL related classes. Can refactor once the
list/map state ttl changes are merged
--
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]
