[SPARK-35785][SS] Cleanup support for RocksDB instance #32933

xuanyuanking · 2021-06-16T14:01:59Z

What changes were proposed in this pull request?

Add the functionality of cleaning up files of old versions for the RocksDB instance and RocksDBFileManager.

Why are the changes needed?

Part of the implementation of RocksDB state store.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

New UT added.

SparkQA · 2021-06-16T14:10:48Z

Test build #139876 has finished for PR 32933 at commit f8f9d20.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-06-16T14:50:35Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44406/

SparkQA · 2021-06-16T15:23:03Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44406/

viirya · 2021-06-23T01:44:42Z

retest this please

viirya · 2021-06-23T01:45:57Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

Is this the change for review in this PR?

That's right.

viirya · 2021-06-23T01:48:28Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

Do you mean, if SST file F was last used in version V, then it won't be used in version V+2 or later?

In other words, a SST file F can be only used in continuous versions. It won't be used in V, not used in V+1, and then used again in V+2.

nit: won't

That's right. The file can not be shared with skipping versions. If a file used in V and not used in V+1, the checkpoint of V+1 should already create new files for all the KVs included in the original file.

viirya · 2021-06-23T01:50:13Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

What the second case for? When it will happen?

If I understand correctly, it would occur with reattempt of same micro-batch.

Yep, that's right.

viirya · 2021-06-23T02:55:19Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

maxVersionPresent -> maxUsedVersion?

Ah thanks, done in the next commit.

viirya · 2021-06-23T02:56:45Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

version -> versionFile.

Actually, maybe s"Error deleting version file $versionFile for version $version".

Make sense, done in the next commit.

viirya · 2021-06-23T02:57:36Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

Maybe we need to reduce the files failed to delete? It is possible some files are failed to delete, but seems we ignore such failure and continue.

Yeah may need to count successful ones and failed ones separately.

Agree, done in the next commit.

SparkQA · 2021-06-23T02:57:55Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44698/

viirya

If deleteOldVersions is the only change needed for review, then it looks okay. Just a few minor comments.

SparkQA · 2021-06-23T03:11:06Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44698/

SparkQA · 2021-06-23T04:41:42Z

Test build #140171 has finished for PR 32933 at commit f8f9d20.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2021-06-28T08:42:22Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

This will fail to pass scalastyle now.

HeartSaVioR

First round. Need to look into test cases a bit more.

HeartSaVioR · 2021-06-28T09:12:43Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

nit: It's

Thanks, done in the next commit.

HeartSaVioR · 2021-06-28T09:13:11Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

nit: won't

HeartSaVioR · 2021-06-28T09:36:58Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

If I understand correctly, it would occur with reattempt of same micro-batch.

HeartSaVioR · 2021-06-28T09:52:10Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

nit: space before }

Thanks, done in the next commit.

HeartSaVioR · 2021-06-28T09:56:35Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

Yeah may need to count successful ones and failed ones separately.

HeartSaVioR · 2021-06-28T09:58:22Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

Is localTempDir used only here? Just to make sure deleting the directory won't break anything.

Yes, the localTempDir is only used for unzip files.

HeartSaVioR · 2021-06-28T10:02:39Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala

This is interesting - we don't remove any keys but expect some SST files to be invalid. Would compaction chime in and compact several SSTs into bigger one?

Right. We'll do the RocksDB checkpoint for each commit operation, each checkpoint is a full snapshot and includes all data. In this UT we have 50 versions but only retain 10 versions, so the SST files for deleted versions(1 to 40) will be deleted.

HeartSaVioR · 2021-06-28T10:07:30Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala

nit: double empty lines

Thanks, done in the next commit.

SparkQA · 2021-06-29T18:39:13Z

Test build #140373 has finished for PR 32933 at commit 321c8b0.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-06-30T09:49:27Z

Test build #140438 has finished for PR 32933 at commit 56439b9.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-06-30T09:56:21Z

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44952/

xuanyuanking · 2021-06-30T14:07:05Z

Rebased and addressed all the comments. Thanks for your heads-up! cc @viirya @HeartSaVioR

SparkQA · 2021-06-30T15:26:07Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44958/

SparkQA · 2021-06-30T16:01:14Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44958/

xuanyuanking · 2021-07-01T09:22:52Z

Thanks again for the help of @HeartSaVioR and @viirya.
The test failure is not related to the changes. Let me retrigger.

SparkQA · 2021-07-01T09:42:56Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45009/

SparkQA · 2021-07-01T10:19:14Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45009/

SparkQA · 2021-07-01T12:31:49Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45024/

SparkQA · 2021-07-01T12:50:32Z

Test build #140511 has finished for PR 32933 at commit ea94983.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-07-01T12:57:45Z

Test build #140498 has finished for PR 32933 at commit 9c99cdf.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-07-01T13:04:48Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45024/

HeartSaVioR · 2021-07-01T13:09:15Z

retest this, please

SparkQA · 2021-07-01T14:31:50Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45034/

SparkQA · 2021-07-01T15:07:19Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45034/

SparkQA · 2021-07-01T17:55:06Z

Test build #140521 has finished for PR 32933 at commit ea94983.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2021-07-01T20:12:51Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

    logInfo(s"Rolled back to $loadedVersion")
  }

+  def cleanup(): Unit = {


When will we call this method?

It will be called in the RocksDBStateStoreProvider.doMaintenace. I'll submit the state store provider PR (the last one) today.

viirya · 2021-07-01T20:14:19Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala

    }
  }

+  test("disallow concurrent updates to the same RocksDB instance") {


This test seems not related to clean up change here? Looks like more related to RocksDB instance PR.

Ah yea, this is the test for rollback.
Actually the original plan is expose rollback and cleanup in this PR. It should be a mistake for the last PR, I introduced the rollback without tests.

viirya · 2021-07-01T20:21:30Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala

+          override def run(): Unit = {
+            try {
+              for (version <- 0 to numUpdatesInEachThread) {
+                withDB(


Hm, what this test is used for? Each RocksDB in each thread uses the same remote root dir, won't they conflict?

This is to simulate the multi-thread scenario of updating and cleaning old versions. It will not conflict since we call commit for each update thread and the version get updated for each commits.

Hmm, will it happens? I think RocksDB is not thread-safe, and each state task only has one RocksDB instance. They should update and clean old versions individually as they are for different state store.

This looks to be more likely simulating the case multiple streaming queries with same checkpoint run concurrently.

SST files shouldn't conflict as we make the file name be unique, and for metadata files we use overwriteIfPossible = true, so won't throw error if the file already exists.

Yea, I think the purpose of this test is to make sure no error thrown and the result is correct in the end.
After taking a further look, there's a small issue is that exception never used. I'll confirm it separately.

@xuanyuanking and me discussed this test offline. Seems there is something wrong with exception usage. It doesn't look completely correct. @xuanyuanking will address it by fixing it or deleting the test later in a follow-up.

viirya

lgtm

viirya · 2021-07-02T07:46:09Z

Thanks @xuanyuanking for working on this and @HeartSaVioR for the review! Merging to master.

xuanyuanking · 2021-07-02T07:47:02Z

Great thanks for the help! @viirya @HeartSaVioR
I'll update the rest PR and submit the last one of RocksDBStateStoreProvider today.

viirya · 2021-07-02T07:49:51Z

Ah, sorry, I forgot branch-3.2 was cut and this should be in branch-3.2 too. @xuanyuanking Can you submit a PR for 3.2?

xuanyuanking · 2021-07-02T07:51:41Z

Sure. Let me do it now.

### What changes were proposed in this pull request? Add the functionality of cleaning up files of old versions for the RocksDB instance and RocksDBFileManager. ### Why are the changes needed? Part of the implementation of RocksDB state store. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? New UT added. Closes apache#32933 from xuanyuanking/SPARK-35785. Authored-by: Yuanjian Li <yuanjian.li@databricks.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>

dongjoon-hyun · 2021-07-02T17:00:11Z

Hi, All.
After this PR, Jenkins seems to fail twice consecutively. Could you take a look at that please, @xuanyuanking , @HeartSaVioR , @viirya ?

https://amplab.cs.berkeley.edu/jenkins/view/Spark%20QA%20Test%20(Dashboard)/job/spark-master-test-sbt-hadoop-3.2/2711/testReport/org.apache.spark.sql.execution.streaming.state/RocksDBSuite/ensure_that_concurrent_update_and_cleanup_consistent_versions/history/

cc @gengliangwang

dongjoon-hyun · 2021-07-02T17:06:16Z

It seems that https://issues.apache.org/jira/browse/SPARK-35993 is filed already by @attilapiros

viirya · 2021-07-02T17:27:28Z

Thanks @dongjoon-hyun! Let me ignore the test first to unblock others. @xuanyuanking will address (fix or delete) the test later.

### What changes were proposed in this pull request? This patch ignores the test "ensure that concurrent update and cleanup consistent versions" in #32933. The test is currently flaky and we will address it later. ### Why are the changes needed? Unblock other developments. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests. Closes #33195 from viirya/ignore-rocksdb-test. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

### What changes were proposed in this pull request? This patch ignores the test "ensure that concurrent update and cleanup consistent versions" in #32933. The test is currently flaky and we will address it later. ### Why are the changes needed? Unblock other developments. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests. Closes #33195 from viirya/ignore-rocksdb-test. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org> (cherry picked from commit a6e00ee) Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

github-actions bot added BUILD CORE SQL STRUCTURED STREAMING labels Jun 16, 2021

xuanyuanking mentioned this pull request Jun 16, 2021

[SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS #32767

Closed

viirya reviewed Jun 23, 2021

View reviewed changes

HeartSaVioR reviewed Jun 28, 2021

View reviewed changes

xuanyuanking force-pushed the SPARK-35785 branch from f8f9d20 to a45af9c Compare June 28, 2021 12:33

xuanyuanking force-pushed the SPARK-35785 branch from 321c8b0 to 56439b9 Compare June 30, 2021 09:37

xuanyuanking changed the title ~~[WIP][SPARK-35785][SS] Cleanup support for RocksDB instance~~ [SPARK-35785][SS] Cleanup support for RocksDB instance Jun 30, 2021

trigger test

ea94983

viirya reviewed Jul 1, 2021

View reviewed changes

viirya approved these changes Jul 2, 2021

View reviewed changes

viirya closed this in ca6acf0 Jul 2, 2021

xuanyuanking deleted the SPARK-35785 branch July 2, 2021 07:49

HeartSaVioR mentioned this pull request Jul 2, 2021

[SPARK-35785][SS][3.2] Cleanup support for RocksDB instance #33184

Closed

viirya mentioned this pull request Jul 2, 2021

[SPARK-35785][SS][FOLLOWUP] Ignore concurrent update and cleanup test #33195

Closed

[SPARK-35785][SS] Cleanup support for RocksDB instance #32933

[SPARK-35785][SS] Cleanup support for RocksDB instance #32933

Uh oh!

Conversation

xuanyuanking commented Jun 16, 2021

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

SparkQA commented Jun 16, 2021

Uh oh!

SparkQA commented Jun 16, 2021

Uh oh!

SparkQA commented Jun 16, 2021

Uh oh!

viirya commented Jun 23, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya Jun 23, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jun 23, 2021

Uh oh!

viirya left a comment

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jun 23, 2021

Uh oh!

SparkQA commented Jun 23, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya Jun 23, 2021 •

edited

Loading