You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "anishshri-db (via GitHub)" <gi...@apache.org> on 2025/10/03 01:18:47 UTC
[PR] [SPARK-53794] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
anishshri-db opened a new pull request, #52511:
URL: https://github.com/apache/spark/pull/52511
### What changes were proposed in this pull request?
Add option to limit deletions per maintenance operation associated with rocksdb state provider
### Why are the changes needed?
We see some instances where the changelog deletion can take a really long time. This means that for that partition, we also cannot upload full snapshots which affects recovery/replay scenarios. This problem is much more apparent on resource constrained clusters. So, we add an option to allow for incremental cleanup per maintenance operation invocation.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Added unit tests
```
[info] Run completed in 17 seconds, 591 milliseconds.
[info] Total number of tests run: 8
[info] Suites: completed 1, aborted 0
[info] Tests: succeeded 8, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
```
### Was this patch authored or co-authored using generative AI tooling?
No
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #52511:
URL: https://github.com/apache/spark/pull/52511#discussion_r2403037701
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2664,6 +2664,14 @@ object SQLConf {
.doubleConf
.createWithDefault(0.3)
+ val MAX_VERSIONS_TO_DELETE = buildConf("spark.sql.streaming.stateStore.maxVersionsToDelete")
+ .internal()
+ .doc("The maximum number of versions to delete per maintenance operation. By default, " +
Review Comment:
Done - added the note for rocksdb
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on code in PR #52511:
URL: https://github.com/apache/spark/pull/52511#discussion_r2402922794
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2664,6 +2664,14 @@ object SQLConf {
.doubleConf
.createWithDefault(0.3)
+ val MAX_VERSIONS_TO_DELETE = buildConf("spark.sql.streaming.stateStore.maxVersionsToDelete")
+ .internal()
+ .doc("The maximum number of versions to delete per maintenance operation. By default, " +
Review Comment:
Should we specify that it's RocksDB only? Or should we also port the changes over to HDFS?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #52511:
URL: https://github.com/apache/spark/pull/52511#issuecomment-3375003325
Hi @dongjoon-hyun - thanks for the pointer. Sorry not intentional, was not sure whether we always need committer stamp for committer authored PRs as well. Could you please stamp this one for me ? Will make sure to get committer stamp henceforth before merging. Thanks !
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #52511:
URL: https://github.com/apache/spark/pull/52511#discussion_r2404237057
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -603,14 +606,21 @@ class RocksDBFileManager(
// Find the versions to delete
val maxSnapshotVersionPresent = sortedSnapshotVersionsAndUniqueIds.last._1
+ val minSnapshotVersionPresent = sortedSnapshotVersionsAndUniqueIds.head._1
// In order to reconstruct numVersionsToRetain version, retain the latest snapshot
// that satisfies (version <= maxSnapshotVersionPresent - numVersionsToRetain + 1).
+ // Also require minVersionToRetain <= minSnapshotVersionPresent + maxVersionsToDelete.
// If none of the snapshots satisfy the condition, minVersionToRetain will be 0 and
// no version gets deleted.
val minVersionToRetain = sortedSnapshotVersionsAndUniqueIds
.map(_._1)
.filter(_ <= maxSnapshotVersionPresent - numVersionsToRetain + 1)
+ .filter( v =>
+ if (maxVersionsToDelete != -1) {
Review Comment:
Prefer to keep this explicit actually
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db closed pull request #52511: [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider
URL: https://github.com/apache/spark/pull/52511
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "liviazhu (via GitHub)" <gi...@apache.org>.
liviazhu commented on code in PR #52511:
URL: https://github.com/apache/spark/pull/52511#discussion_r2408478653
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -580,8 +580,22 @@ class RocksDBFileManager(
* - Partially written SST files
* - SST files that were used in a version, but that version got overwritten with a different
* set of SST files.
+ *
+ * @param numVersionsToRetain the number of RocksDB versions to keep in object store after the
+ * deletion. Must be greater than 0 or -1 to retain all versions.
+ * @param maxVersionsToDeletePerMaintenance the max number of RocksDB versions
+ * to delete per maintenance operation.
+ * Must be greater than 0 or -1 to delete all stale versions.
Review Comment:
```suggestion
* Must be greater than 0, or -1 to delete all stale versions.
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #52511:
URL: https://github.com/apache/spark/pull/52511#discussion_r2408545021
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -580,8 +580,22 @@ class RocksDBFileManager(
* - Partially written SST files
* - SST files that were used in a version, but that version got overwritten with a different
* set of SST files.
+ *
+ * @param numVersionsToRetain the number of RocksDB versions to keep in object store after the
+ * deletion. Must be greater than 0 or -1 to retain all versions.
+ * @param maxVersionsToDeletePerMaintenance the max number of RocksDB versions
+ * to delete per maintenance operation.
+ * Must be greater than 0 or -1 to delete all stale versions.
Review Comment:
Done
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #52511:
URL: https://github.com/apache/spark/pull/52511#issuecomment-3374928755
Hi, @anishshri-db , just a question, do you have any reason not to wait an approval from the Apache Spark committers?
For some urgent cases, Apache Spark release managers can do the merge (and expects a late post-merge approval), but I'm just curious if you have those kind of exceptional cases.
BTW, according to the code, I can give my +1 to meet the Apache Spark community requirements (if you missed the approval mistakenly).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "liviazhu (via GitHub)" <gi...@apache.org>.
liviazhu commented on code in PR #52511:
URL: https://github.com/apache/spark/pull/52511#discussion_r2403604024
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -2144,7 +2146,8 @@ object RocksDBConf {
storeConf.compressionCodec,
getBooleanConf(ALLOW_FALLOCATE_CONF),
getStringConf(COMPRESSION_CONF),
- storeConf.reportSnapshotUploadLag)
+ storeConf.reportSnapshotUploadLag,
+ storeConf.maxVersionsToDelete)
Review Comment:
Can we check that minVersionToDelete <= maxVersionsToDelete and add a test?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -603,14 +606,21 @@ class RocksDBFileManager(
// Find the versions to delete
val maxSnapshotVersionPresent = sortedSnapshotVersionsAndUniqueIds.last._1
+ val minSnapshotVersionPresent = sortedSnapshotVersionsAndUniqueIds.head._1
// In order to reconstruct numVersionsToRetain version, retain the latest snapshot
// that satisfies (version <= maxSnapshotVersionPresent - numVersionsToRetain + 1).
+ // Also require minVersionToRetain <= minSnapshotVersionPresent + maxVersionsToDelete.
// If none of the snapshots satisfy the condition, minVersionToRetain will be 0 and
// no version gets deleted.
val minVersionToRetain = sortedSnapshotVersionsAndUniqueIds
.map(_._1)
.filter(_ <= maxSnapshotVersionPresent - numVersionsToRetain + 1)
+ .filter( v =>
+ if (maxVersionsToDelete != -1) {
Review Comment:
nit: you can probably do `(maxVersionsToDelete != -1) && (v <= minSnapshotVersionPresent + maxVersionsToDelete)`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "liviazhu (via GitHub)" <gi...@apache.org>.
liviazhu commented on code in PR #52511:
URL: https://github.com/apache/spark/pull/52511#discussion_r2407466125
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -2144,7 +2146,8 @@ object RocksDBConf {
storeConf.compressionCodec,
getBooleanConf(ALLOW_FALLOCATE_CONF),
getStringConf(COMPRESSION_CONF),
- storeConf.reportSnapshotUploadLag)
+ storeConf.reportSnapshotUploadLag,
+ storeConf.maxVersionsToDelete)
Review Comment:
Hm, the similar naming of minVersionToDelete and maxVersionsToDelete is very confusing. Can we rename the new parameter to be "maxVersionsDeletedByMaintenance" or something?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #52511:
URL: https://github.com/apache/spark/pull/52511#discussion_r2407976694
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -2144,7 +2146,8 @@ object RocksDBConf {
storeConf.compressionCodec,
getBooleanConf(ALLOW_FALLOCATE_CONF),
getStringConf(COMPRESSION_CONF),
- storeConf.reportSnapshotUploadLag)
+ storeConf.reportSnapshotUploadLag,
+ storeConf.maxVersionsToDelete)
Review Comment:
Done
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #52511:
URL: https://github.com/apache/spark/pull/52511#issuecomment-3363807473
cc - @HeartSaVioR - PTAL, thx !
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #52511:
URL: https://github.com/apache/spark/pull/52511#discussion_r2404236635
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -2144,7 +2146,8 @@ object RocksDBConf {
storeConf.compressionCodec,
getBooleanConf(ALLOW_FALLOCATE_CONF),
getStringConf(COMPRESSION_CONF),
- storeConf.reportSnapshotUploadLag)
+ storeConf.reportSnapshotUploadLag,
+ storeConf.maxVersionsToDelete)
Review Comment:
The naming here is a bit confusing - I added a comment in the file manager to explain what these values mean
```
* @param numVersionsToRetain the number of RocksDB versions to keep in object store after the deletion.
* Must be greater than 0 or -1 to retain all versions.
* @param maxVersionsToDelete the max number of RocksDB versions to delete if a deletion occurs.
* Must be greater than 0 or -1 to delete all non-retained versions.
* @param minVersionsToDelete the min number of stale versions required to trigger deletion.
* If its set to <= 0, then we will always perform list operations
* to determine deletion candidates. If set to a positive value, then
* we will skip deletion if the number of stale versions is less than
* this value.
```
Also added a test to verify for combination of both values
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #52511:
URL: https://github.com/apache/spark/pull/52511#discussion_r2404236635
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -2144,7 +2146,8 @@ object RocksDBConf {
storeConf.compressionCodec,
getBooleanConf(ALLOW_FALLOCATE_CONF),
getStringConf(COMPRESSION_CONF),
- storeConf.reportSnapshotUploadLag)
+ storeConf.reportSnapshotUploadLag,
+ storeConf.maxVersionsToDelete)
Review Comment:
The naming here is a bit confusing - I added a comment in the file manager to explain what these values mean
```
* @param numVersionsToRetain the number of RocksDB versions to keep in DBFS after the deletion.
* Must be greater than 0 or -1 to retain all versions.
* @param maxVersionsToDelete the max number of RocksDB versions to delete if a deletion occurs.
* Must be greater than 0 or -1 to delete all non-retained versions.
* @param minVersionsToDelete the min number of stale versions required to trigger deletion.
* If its set to <= 0, then we will always perform list operations
* to determine deletion candidates. If set to a positive value, then
* we will skip deletion if the number of stale versions is less than
* this value.
```
Also added a test to verify for combination of both values
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53794][SS] Add option to limit deletions per maintenance operation associated with rocksdb state provider [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #52511:
URL: https://github.com/apache/spark/pull/52511#issuecomment-3375127178
+1, LGTM.
No problem at all. I just want to remove any such an ASF issue from this PR . Thank you, @anishshri-db .
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org