[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS

[GitBox <gi...@apache.org>]

xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r644694974



##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala
##########
@@ -20,12 +20,125 @@ package org.apache.spark.sql.execution.streaming.state
 import java.io._
 import java.nio.charset.Charset
 
+import scala.language.implicitConversions
+
 import org.apache.commons.io.FileUtils
+import org.apache.hadoop.conf.Configuration
 
 import org.apache.spark._
+import org.apache.spark.sql.catalyst.util.quietly
+import org.apache.spark.sql.execution.streaming.CreateAtomicTestManager
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.util.Utils
 
 class RocksDBSuite extends SparkFunSuite {
 
+  test("RocksDBFileManager: upload only new immutable files") {
+    withTempDir { dir =>
+      val dfsRootDir = dir.getAbsolutePath
+      val verificationDir = Utils.createTempDir().getAbsolutePath // local dir to load checkpoints
+      val fileManager = new RocksDBFileManager(
+        dfsRootDir, Utils.createTempDir(), new Configuration)
+      val sstDir = s"$dfsRootDir/SSTs"
+      def numRemoteSSTFiles: Int = listFiles(sstDir).length
+      val logDir = s"$dfsRootDir/logs"
+      def numRemoteLogFiles: Int = listFiles(logDir).length
+
+      // Verify behavior before any saved checkpoints
+      assert(fileManager.getLatestVersion() === 0)
+
+      // Try to load incorrect versions
+      intercept[FileNotFoundException] {
+        fileManager.loadCheckpointFromDfs(1, Utils.createTempDir())
+      }
+
+      // Save a version of checkpoint files
+      val cpFiles1 = Seq(
+        "sst-file1.sst" -> 10,
+        "sst-file2.sst" -> 20,
+        "other-file1" -> 100,
+        "other-file2" -> 200,
+        "archive/00001.log" -> 1000,
+        "archive/00002.log" -> 2000
+      )
+      saveCheckpointFiles(fileManager, cpFiles1, version = 1, numKeys = 101)
+      assert(fileManager.getLatestVersion() === 1)
+      assert(numRemoteSSTFiles == 2) // 2 sst files copied
+      assert(numRemoteLogFiles == 2) // 2 log files copied
+
+      // Load back the checkpoint files into another local dir with existing files and verify
+      generateFiles(verificationDir, Seq(
+        "sst-file1.sst" -> 11, // files with same exist but different sizes, should get overwritten
+        "other-file1" -> 101,
+        "archive/00001.log" -> 1001,
+        "random-sst-file.sst" -> 100, // unnecessary files, should get deleted
+        "random-other-file" -> 9,
+        "00005.log" -> 101,
+        "archive/00007.log" -> 101
+      ))
+      loadAndVerifyCheckpointFiles(fileManager, verificationDir, version = 1, cpFiles1, 101)

Review comment:
       Per https://github.com/apache/spark/pull/32582#discussion_r642036286




-- 
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.

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