You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/12/02 09:47:19 UTC
[GitHub] [spark] HeartSaVioR opened a new pull request, #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
HeartSaVioR opened a new pull request, #38880:
URL: https://github.com/apache/spark/pull/38880
### What changes were proposed in this pull request?
This PR proposes to clear the write batch (and also corresponding prefix iterators) after commit has succeeded on RocksDB state store. This PR also fixes the test case as side effect, as it had been relying on the "sort of bug" that we didn't clean up write batch till either rollback or load has been called.
### Why are the changes needed?
Without this, the memory usage of WriteBatch for RocksDB state store is "accumulated" over the partitions in the same executor. Say, 10 partitions in stateful operator are assigned to an executor and run sequentially. Given that we didn't clear write batch after commit, when the executor processes the last partition assigned to it, 10 WriteBatch instances contain all writes being performed in this microbatch.
### Does this PR introduce _any_ user-facing change?
No. This is a sort of bugfix.
### How was this patch tested?
Existing tests, with fixing the test case.
--
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
[GitHub] [spark] Kimahriman commented on pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
Kimahriman commented on PR #38880:
URL: https://github.com/apache/spark/pull/38880#issuecomment-1335177296
> Also cc. @Kimahriman I'm sorry if you wanted to take this up.
No problem! I had a very similar PR I was working on before my previous one, and then tested it in production and realized it didn't fix the problem so I never submitted it.
The only other thing I had done was save off the write batch memory size before clearing it, and used that for the metrics. I believe the metrics are gathered post-commit (not sure if this is always the case but definitely true some of the cases), so the write batch mem size will always be 12 after creating a new one.
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1041595635
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
Sounds good, thanks for the suggestion!
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1037968475
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
This was something we could indicate the bug earlier...
The test was incorrect. The test expects that compaction will happen, but if we don't "overwrite" the key and just put the new keys here, compaction does not happen, at least with Spark's default config. The reason compaction had happened is, writebatch wasn't cleaned up during loop, hence writes are accumulated into the same writebatch and it led to overwrites. If we fix the code but leave this test as it is, then there are 50 SST files meaning no compaction happens, and test fails.
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #38880:
URL: https://github.com/apache/spark/pull/38880#issuecomment-1335826574
Btw I think there is still an argument of whether we can do better for writebatch. We defer the writes and flush till commit happens, to ease of rollback, but we expect there would be only a few of failures, even in many cases the query crashes which the optimization for rollback does not work. Maybe, using writebatch as "a buffer" rather than "semi transaction" one, would give more opportunity on performance, and then we will no longer require all writes being stored into writebatch.
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1041871281
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -310,6 +311,9 @@ class RocksDB(
"checkpoint" -> checkpointTimeMs,
"fileSync" -> fileSyncTimeMs
)
+ // reset resources as we already pushed the changes and it has been committed
Review Comment:
Actually I intended to put this here, as I thought we have to clean up only when it is successfully committed. But looks like it's also safe to place this to `finally`, as we invalidate the loaded version in any way if any exception happens. Will do.
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #38880:
URL: https://github.com/apache/spark/pull/38880#issuecomment-1335001543
cc. @zsxwing @xuanyuanking @viirya Please take a look. Thanks in advance!
Also cc. @Kimahriman I'm sorry if you wanted to take this up. I just went with creating a PR as I realized I asked @Myasuka to break down PR which contained this fix (and both of us forgot), hence I wanted to give credit to him.
--
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
[GitHub] [spark] Kimahriman commented on pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
Kimahriman commented on PR #38880:
URL: https://github.com/apache/spark/pull/38880#issuecomment-1335320875
> I'd say the calculation of total memory usage is correct even the write batch mem size is always 12. That reflects the right "snapshot" of memory usage when metrics() is called, mostly after commit().
Yeah I wasn't sure if kind of "peak memory usage" was more relevant to report, since the write batch may be a vast majority of what ends up requiring native memory.
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1041686643
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
I just changed it to remove the existing key. It may be more obvious.
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #38880:
URL: https://github.com/apache/spark/pull/38880#issuecomment-1335247831
> The only other thing I had done was save off the write batch memory size before clearing it, and used that for the metrics. I believe the metrics are gathered post-commit (not sure if this is always the case but definitely true some of the cases), so the write batch mem size will always be 12 after creating a new one.
I'd say the calculation of total memory usage is correct even the write batch mem size is always 12. That reflects the right "snapshot" of memory usage when metrics() is called, mostly after commit().
If we feel beneficial to report the memory usage of writebatch for the execution of partition, saving off the size of write batch memory would be a good way to do. Maybe one of custom metrics, rather than adding up to the total memory usage. We seem to store the memory usage of writebatch separately as well, but we don't report it separately to the custom metric.
--
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
[GitHub] [spark] pranavanand commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
pranavanand commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1041594455
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
I see, could you add this detail to the comment? it was really helpful for me to understand how we are ensuring compaction
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1040611003
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
I'm not an expert of RocksDB, but I could explain this according to the high level architecture.
https://github.com/facebook/rocksdb/wiki/RocksDB-Overview#3-high-level-architecture
During RocksDB.commit(), we write synchronously and flush synchronously, meaning that we will produce a new SST file for new writes. (As SST file is immutable, you can't append any new write against existing SST files.)
That said, for each loop, we write a bunch of overwritten keys, and produce a new SST file containing overwritten keys. L0 allows SST files to have overlapping key ranges, but it is no longer allowed in upper level, which is ensured via "compaction". We trigger manual compaction based on the config, `compactOnCommit = true`.
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1040611003
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
I'm not an expert of RocksDB, but looks like I could explain this according to the high level architecture.
https://github.com/facebook/rocksdb/wiki/RocksDB-Overview#3-high-level-architecture
During RocksDB.commit(), we write synchronously and flush synchronously, meaning that we will produce a new SST file for new writes. (As SST file is immutable, you can't append any new write against existing SST files.)
That said, for each loop, we write a bunch of overwritten keys, and produce a new SST file containing overwritten keys. L0 allows SST files to have overlapping key ranges, but it is no longer allowed in upper level, which is ensured via "compaction". We trigger manual compaction based on the config, `compactOnCommit = true`.
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1041654592
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
No, the functionality is not here. We don't leverage TTL for the general state store operations. But remove would work as well. If you think it's easier to reason about, I can update the test.
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #38880:
URL: https://github.com/apache/spark/pull/38880#issuecomment-1342007531
cc. @zsxwing @xuanyuanking @viirya Friendly reminder.
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1038137730
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
Yeah, I don't know much on the details of RocksDB so can't tell compaction would never happen with pure addition (It would be a smart move if RocksDB performs compaction if there are 10000s of SST files.) but at least the test was not sufficient to trigger the compaction.
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1040611003
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
I'm not an expert of RocksDB, but looks like we could expect compaction according to the high level architecture.
https://github.com/facebook/rocksdb/wiki/RocksDB-Overview#3-high-level-architecture
During RocksDB.commit(), we write synchronously and flush synchronously, meaning that we will produce a new SST file for new writes. (As SST file is immutable, you can't append any new write against existing SST files.)
That said, for each loop, we write a bunch of overwritten keys, and produce a new SST file containing overwritten keys. L0 allows SST files to have overlapping key ranges, but it is no longer allowed in upper level, which is ensured via "compaction".
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1041654592
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
No, the functionality is not here. We don't leverage TTL for the general state store operations.
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #38880:
URL: https://github.com/apache/spark/pull/38880#issuecomment-1335818435
Yeah maybe reporting the memory usage of write batch in this operator (they will get accumulated across partitions) would be also useful at such point. It's not an accurate peak memory usage if we simply add it to the total memory usage at the "snapshot" of when we call commit()/metrics(). We don't know about the "peak".
--
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
[GitHub] [spark] jerrypeng commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1041868304
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -310,6 +311,9 @@ class RocksDB(
"checkpoint" -> checkpointTimeMs,
"fileSync" -> fileSyncTimeMs
)
+ // reset resources as we already pushed the changes and it has been committed
Review Comment:
Should we do this in the finally clause to prevent any exception/error causing this not to run?
--
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
[GitHub] [spark] HeartSaVioR closed pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR closed pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
URL: https://github.com/apache/spark/pull/38880
--
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
[GitHub] [spark] pranavanand commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
pranavanand commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1040580290
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
how do we know this will definitely cause compaction?
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1037968475
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
This was something we could indicate the bug earlier...
The test was incorrect. The test expects that compaction will happen, but if we don't "overwrite" the key and just put the new keys here, compaction does not happen, at least with Spark's default config. The reason compaction had happened is, writebatch wasn't cleaned up during loop, hence writes are accumulated into the same writebatch and it led to overwrites.
--
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
[GitHub] [spark] Kimahriman commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
Kimahriman commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1038109358
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
I also had the same realization for this test 😂
I _think_ this is due to what I learned was "trivial moves" during compaction, where if the keys don't overlap you don't need to rewrite the SST files.
--
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
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1040611003
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
I'm not an expert of RocksDB, but looks like I could explain this according to the high level architecture.
https://github.com/facebook/rocksdb/wiki/RocksDB-Overview#3-high-level-architecture
During RocksDB.commit(), we write synchronously and flush synchronously, meaning that we will produce a new SST file for new writes. (As SST file is immutable, you can't append any new write against existing SST files.)
That said, for each loop, we write a bunch of overwritten keys, and produce a new SST file containing overwritten keys. L0 allows SST files to have overlapping key ranges, but it is no longer allowed in upper level, which is ensured via "compaction".
--
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
[GitHub] [spark] pranavanand commented on a diff in pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
pranavanand commented on code in PR #38880:
URL: https://github.com/apache/spark/pull/38880#discussion_r1041599019
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -116,7 +116,9 @@ class RocksDBSuite extends SparkFunSuite {
withDB(remoteDir, conf = conf) { db =>
// Generate versions without cleaning up
for (version <- 1 to 50) {
- db.put(version.toString, version.toString) // update "1" -> "1", "2" -> "2", ...
Review Comment:
Or can we do this maybe
`db.putWithTtl(key.getBytes(), value.getBytes(), ttl)` and ensure the TTL has been exceeded by that time?
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #38880: [SPARK-38277][SS] Clear write batch after RocksDB state store's commit
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #38880:
URL: https://github.com/apache/spark/pull/38880#issuecomment-1342158178
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