You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "riyaverm-db (via GitHub)" <gi...@apache.org> on 2024/07/17 21:49:06 UTC
[PR] [SPARK-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
riyaverm-db opened a new pull request, #47393:
URL: https://github.com/apache/spark/pull/47393
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Currently, during the state store maintenance process, we find which old version files of the RocksDB state store to delete by listing all existing snapshotted version files in the checkpoint directory every 1 minute by default. The frequent list calls in the cloud can result in high costs. To address this concern and reduce the cost associated with state store maintenance, we should aim to minimize the frequency of listing object stores inside the maintenance task. To minimize the frequency, we will try to accumulate versions to delete and only call list when the number of versions to delete reaches a configured threshold.
The changes include:
1. Adding new configuration variable `minVersionsToDelete`
2. Using estimated values of `minVersion` and `maxVersion` variables in **RocksDBFileManager** to ensure number of files to be deleted in a call to **deleteOldVersions** is at least the value of `minVersionsToDelete`
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Currently, maintenance operations like snapshotting, purging, deletion, and file management is done asynchronously for each data partition. We want to shift away periodic deletion and instead rely on number of files in the checkpoint directory to reduce list calls and introduce batch deletion.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Added unit tests.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "chaoqin-li1123 (via GitHub)" <gi...@apache.org>.
chaoqin-li1123 commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1681866435
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -185,6 +185,8 @@ class RocksDB(
val latestSnapshotVersion = fileManager.getLatestSnapshotVersion(version)
val metadata = fileManager.loadCheckpointFromDfs(latestSnapshotVersion, workingDir)
loadedVersion = latestSnapshotVersion
+ // Initialize maxVersion upon successful load from DFS
+ fileManager.initializeMaxVersion()
Review Comment:
setMaxVersion will handle the case where the query load a previous version locally.
maxVersion is an approximate anyway, it doesn't need to be accurate.
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "chaoqin-li1123 (via GitHub)" <gi...@apache.org>.
chaoqin-li1123 commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1681834923
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreConf.scala:
##########
@@ -41,6 +41,9 @@ class StateStoreConf(
/** Minimum versions a State Store implementation should retain to allow rollbacks */
val minVersionsToRetain: Int = sqlConf.minBatchesToRetain
+ /** Minimum number of stale versions to delete when maintenance is invoked */
Review Comment:
Please add more comment here to explain why we need this config.
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #47393:
URL: https://github.com/apache/spark/pull/47393#issuecomment-2246884836
Thanks! Merging to master.
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1683718327
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2129,6 +2129,13 @@ object SQLConf {
.intConf
.createWithDefault(100)
+ val MIN_VERSIONS_TO_DELETE = buildConf("spark.sql.streaming.minVersionsToDelete")
+ .internal()
+ .doc("The minimum number of stale versions to delete when maintenance is invoked.")
+ .version("2.1.1")
+ .intConf
+ .createWithDefault(30)
Review Comment:
minBatchesToRetain is a guaranteed lower bound of the number of retained batches, but it has been one side guarantee. This config is to leverage the other side which we do not guarantee to loose more, so that we make trade-off between number of files vs API cost.
I think this default value suggested here (30) is in line with the default value of minBatchesToRetain (100). If they don't mind retaining 100 batches they also wouldn't mind retaining 130-140 batches to reduce down the overhead and cost. For someone who really cares to minimize the number of files, they should have changed the config for minBatchesToRetain, which they could also change the config on this.
(Btw, I agree that minBatchesToRetain to be 100 by default hadn't been making sense, but with state data source reader, it figures out at least one of usages. I don't have strong opinion whether the default value has to be enforced to support this use case though.)
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1684960192
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -426,7 +441,20 @@ class RocksDBFileManager(
* - SST files that were used in a version, but that version got overwritten with a different
* set of SST files.
*/
- def deleteOldVersions(numVersionsToRetain: Int): Unit = {
+ def deleteOldVersions(numVersionsToRetain: Int, minVersionsToDelete: Int = 0): Unit = {
+ // If minVersionsToDelete <= 0, we call list every time maintenance is invoked
+ // This is the original behaviour without list api call optimization
+ if (minVersionsToDelete > 0) {
+ // When maxSeenVersion is defined, we check the if number of stale version files
+ // are at least the value of minVersionsToDelete for batch deletion of files
+ // We still proceed with deletion if maxSeenVersion isn't set to ensure the fallback
+ // is to clean up files if maxSeenVersion fails to be initialized
+ if (maxSeenVersion.isDefined) {
+ val versionsToDelete = maxSeenVersion.get - minSeenVersion + 1 - numVersionsToRetain
+ if (versionsToDelete < minVersionsToDelete) return
Review Comment:
Can we add a log line here indicating we are skipping the delete? Better to included these variables above
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -398,6 +401,18 @@ class RocksDBFileManager(
}
}
+ /**
+ * Set maxSeenVersion to max of itself and version we are uploading.
+ * This is to ensure accuracy in the case the query has restarted from a particular version.
+ */
+ def setMaxSeenVersion(version: Long): Unit = {
+ if (maxSeenVersion.isDefined) {
+ maxSeenVersion = Some(Math.max(maxSeenVersion.get, version))
+ } else {
+ maxSeenVersion = Some(version)
+ }
+ }
+
/**
* Delete old versions by deleting the associated version and SST files.
Review Comment:
Can we update the doc for this method to reflect the new strategy we use?
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "chaoqin-li1123 (via GitHub)" <gi...@apache.org>.
chaoqin-li1123 commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1681834138
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -451,6 +492,8 @@ class RocksDBFileManager(
.filter(_ <= maxSnapshotVersionPresent - numVersionsToRetain + 1)
.foldLeft(0L)(math.max)
+ minVersion = minVersionToRetain
Review Comment:
Better to update this after the delete has succeeded.
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "chaoqin-li1123 (via GitHub)" <gi...@apache.org>.
chaoqin-li1123 commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1681857191
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2129,6 +2129,13 @@ object SQLConf {
.intConf
.createWithDefault(100)
+ val MIN_VERSIONS_TO_DELETE = buildConf("spark.sql.streaming.minVersionsToDelete")
+ .internal()
+ .doc("The minimum number of stale versions to delete when maintenance is invoked.")
+ .version("2.1.1")
+ .intConf
+ .createWithDefault(30)
Review Comment:
This only apply to rocksdb state store.
You can change minBatchesToRetain to have the same size of checkpoint.
Actually, we can consider having a smaller minBatchesToRetain default.
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "Kimahriman (via GitHub)" <gi...@apache.org>.
Kimahriman commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1681869279
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2129,6 +2129,13 @@ object SQLConf {
.intConf
.createWithDefault(100)
+ val MIN_VERSIONS_TO_DELETE = buildConf("spark.sql.streaming.minVersionsToDelete")
+ .internal()
+ .doc("The minimum number of stale versions to delete when maintenance is invoked.")
+ .version("2.1.1")
+ .intConf
+ .createWithDefault(30)
Review Comment:
Yeah I'm not really sure why the default minBatchesToRetain is so high, we set it to 3 in our jobs hah. I would likely set this new config to 1 if that's not the default, I'm just not sure how many other rocksdb state users would be impacted by effectively maintaining more state store versions by default.
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #47393:
URL: https://github.com/apache/spark/pull/47393#issuecomment-2246902489
@riyaverm-db
Maybe we could add this into migration guide, as we already observe one contributor who will be impacted by this change?
https://github.com/apache/spark/blob/master/docs/ss-migration-guide.md
Please feel free to either file a new ticket or use the same ticket with adding `[FOLLOWUP]` into PR title. 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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #47393: [SPARK-48931][SS] Reduce Cloud Store List API cost for state store maintenance task
URL: https://github.com/apache/spark/pull/47393
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "chaoqin-li1123 (via GitHub)" <gi...@apache.org>.
chaoqin-li1123 commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1687060460
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -426,7 +441,29 @@ class RocksDBFileManager(
* - SST files that were used in a version, but that version got overwritten with a different
* set of SST files.
*/
- def deleteOldVersions(numVersionsToRetain: Int): Unit = {
+ def deleteOldVersions(numVersionsToRetain: Int, minVersionsToDelete: Long = 0): Unit = {
+ // If minVersionsToDelete <= 0, we call list every time maintenance is invoked
+ // This is the original behaviour without list api call optimization
+ if (minVersionsToDelete > 0) {
Review Comment:
Can we move this part to a helper function? deleteOldVersions is already a too long function
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "chaoqin-li1123 (via GitHub)" <gi...@apache.org>.
chaoqin-li1123 commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1681920047
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2129,6 +2129,13 @@ object SQLConf {
.intConf
.createWithDefault(100)
+ val MIN_VERSIONS_TO_DELETE = buildConf("spark.sql.streaming.minVersionsToDelete")
+ .internal()
+ .doc("The minimum number of stale versions to delete when maintenance is invoked.")
+ .version("2.1.1")
+ .intConf
+ .createWithDefault(30)
Review Comment:
Yes, setting min retained versions to 100 is to be super conservative :)
There is concern about increasing storage cost because of the new change. WDYT? @HeartSaVioR
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "chaoqin-li1123 (via GitHub)" <gi...@apache.org>.
chaoqin-li1123 commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1681833016
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -185,6 +185,8 @@ class RocksDB(
val latestSnapshotVersion = fileManager.getLatestSnapshotVersion(version)
val metadata = fileManager.loadCheckpointFromDfs(latestSnapshotVersion, workingDir)
loadedVersion = latestSnapshotVersion
+ // Initialize maxVersion upon successful load from DFS
+ fileManager.initializeMaxVersion()
Review Comment:
Always do a listing to discover the max version in load() is a waste of resource, you can simply call
_fileManager.setMaxVersion(version)_
Because usually the latest version is loaded.
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "Kimahriman (via GitHub)" <gi...@apache.org>.
Kimahriman commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1684338484
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2129,6 +2129,13 @@ object SQLConf {
.intConf
.createWithDefault(100)
+ val MIN_VERSIONS_TO_DELETE = buildConf("spark.sql.streaming.minVersionsToDelete")
+ .internal()
+ .doc("The minimum number of stale versions to delete when maintenance is invoked.")
+ .version("2.1.1")
+ .intConf
+ .createWithDefault(30)
Review Comment:
I agree the default of 30 is inline with the default 100 batches to retain. I guess the state reader will provide some use case to maintaining so many batches, don't understand what previous use case would have made sense to store that many batches, since you would have to do some manual checkpoint surgery to try to rollback.
I don't have a strong preference either way, just not sure how many other people would be in my boat or how unique my use case is (large aggregations and deduping with batches every few hours). Either way would just be good to document in case others are surprised at an average increase in state store size
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "riyaverm-db (via GitHub)" <gi...@apache.org>.
riyaverm-db commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1681860880
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -185,6 +185,8 @@ class RocksDB(
val latestSnapshotVersion = fileManager.getLatestSnapshotVersion(version)
val metadata = fileManager.loadCheckpointFromDfs(latestSnapshotVersion, workingDir)
loadedVersion = latestSnapshotVersion
+ // Initialize maxVersion upon successful load from DFS
+ fileManager.initializeMaxVersion()
Review Comment:
Listing is only done if maxVersion is not initialized based on the base condition of the call. If query loads a previous version, maxVersion will not be the max in the directory when using setMaxVersion. This could make maxVersion value inconsistent with the actual maximum version present, is that correct?
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "chaoqin-li1123 (via GitHub)" <gi...@apache.org>.
chaoqin-li1123 commented on PR #47393:
URL: https://github.com/apache/spark/pull/47393#issuecomment-2243936187
Can you also update the pr description about the new conf?
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "riyaverm-db (via GitHub)" <gi...@apache.org>.
riyaverm-db commented on PR #47393:
URL: https://github.com/apache/spark/pull/47393#issuecomment-2253457933
@HeartSaVioR Updated the migration doc here. https://github.com/apache/spark/pull/47507
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "Kimahriman (via GitHub)" <gi...@apache.org>.
Kimahriman commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1681854029
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2129,6 +2129,13 @@ object SQLConf {
.intConf
.createWithDefault(100)
+ val MIN_VERSIONS_TO_DELETE = buildConf("spark.sql.streaming.minVersionsToDelete")
+ .internal()
+ .doc("The minimum number of stale versions to delete when maintenance is invoked.")
+ .version("2.1.1")
+ .intConf
+ .createWithDefault(30)
Review Comment:
As an HDFS user where I care more about number of files than number of list calls, this would have been a little surprising and confusing to be the new default behavior when I upgraded. Should this default to 1? That way by default you at least don't waste time looking for files to delete if you haven't even written a new version since the last time you checked, but still respects `minBatchesToRetain` as a tighter bound
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "chaoqin-li1123 (via GitHub)" <gi...@apache.org>.
chaoqin-li1123 commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1681918789
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreConf.scala:
##########
@@ -41,6 +41,13 @@ class StateStoreConf(
/** Minimum versions a State Store implementation should retain to allow rollbacks */
val minVersionsToRetain: Int = sqlConf.minBatchesToRetain
+ /**
+ * Minimum number of stale snapshot version files that need to be present
Review Comment:
Not necessarily a stale snapshot version because changelog version can also be cleaned up.
_Minimum number of stale checkpoint versions that need to be present in the DFS checkpoint directory for old state checkpoint version deletion to be invoked. This is to amortize the cost of discovering and deleting old checkpoint versions._
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -146,6 +146,9 @@ class RocksDBFileManager(
private def codec = CompressionCodec.createCodec(sparkConf, codecName)
+ private var maxVersion: Option[Long] = None
Review Comment:
Can we rename to maxSeenVersion and minSeenVersion to indicates that it is only an approximate? Also change the method name accordingly.
--
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-48931][SS] Reduce Cloud Store List API cost for state store maintenance task [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #47393:
URL: https://github.com/apache/spark/pull/47393#discussion_r1683721948
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2129,6 +2129,13 @@ object SQLConf {
.intConf
.createWithDefault(100)
+ val MIN_VERSIONS_TO_DELETE = buildConf("spark.sql.streaming.minVersionsToDelete")
+ .internal()
+ .doc("The minimum number of stale versions to delete when maintenance is invoked.")
+ .version("2.1.1")
+ .intConf
+ .createWithDefault(30)
Review Comment:
But if we consider this as important behavioral change, we could set the default value to be effectively no-op and document how to config this instead.
--
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