shrirangmhalgi commented on code in PR #56598:
URL: https://github.com/apache/spark/pull/56598#discussion_r3439729958
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -1583,6 +1583,18 @@ class RocksDB(
originalEndKey
}
+ // Count the keys that are about to be deleted (gated on
trackTotalNumberOfRows).
+ if (conf.trackTotalNumberOfRows) {
+ val (deletedKeys, deletedInternalKeys) = countKeysInRange(
+ beginKeyWithPrefix,
+ endKeyWithPrefix,
+ cfName,
+ includesPrefix
+ )
+ numKeysOnWritingVersion -= deletedKeys
Review Comment:
**Correctness concern**: The counters are decremented before
`db.deleteRange()` executes. If `deleteRange` throws (RocksDB write stall, disk
full, or a closed-DB exception), the keys still exist but the counters have
already been reduced - permanently under-counting for the rest of this version.
Move the decrement after the successful `db.deleteRange()` call.
--
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]
