Cleanup staging directory at start of checkpoint (#4035)

Summary:
- Attempt to clean the checkpoint staging directory before starting a checkpoint. It was already cleaned up at the end of checkpoint. But it wasn't cleaned up in the edge case where the process crashed while staging checkpoint files.
- Attempt to clean the checkpoint directory before calling `Checkpoint::Create` in `db_stress`. This handles the case where checkpoint directory was created by a previous `db_stress` run but the process crashed before cleaning it up.
- Use `DestroyDB` for cleaning checkpoint directory since a checkpoint is a DB.
Closes https://github.com/facebook/rocksdb/pull/4035

Reviewed By: yiwu-arbug

Differential Revision: D8580223

Pulled By: ajkr

fbshipit-source-id: 28c667400e249fad0fdedc664b349031b7b61599

tools/db_stress.cc

@@ -1783,6 +1783,7 @@ class StressTest {
           thread->rand.Uniform(FLAGS_checkpoint_one_in) == 0) {
         std::string checkpoint_dir =
             FLAGS_db + "/.checkpoint" + ToString(thread->tid);
+        DestroyDB(checkpoint_dir, Options());
         Checkpoint* checkpoint;
         Status s = Checkpoint::Create(db_, &checkpoint);
         if (s.ok()) {
@@ -1792,16 +1793,7 @@ class StressTest {
         if (s.ok()) {
           s = FLAGS_env->GetChildren(checkpoint_dir, &files);
         }
-        size_t file_idx = 0;
-        while (s.ok() && file_idx < files.size()) {
-          if (files[file_idx] != "." && files[file_idx] != "..") {
-            s = FLAGS_env->DeleteFile(checkpoint_dir + "/" + files[file_idx]);
-          }
-          ++file_idx;
-        }
-        if (s.ok()) {
-          s = FLAGS_env->DeleteDir(checkpoint_dir);
-        }
+        DestroyDB(checkpoint_dir, Options());
         if (!s.ok()) {
           printf("A checkpoint operation failed with: %s\n",
                  s.ToString().c_str());

utilities/checkpoint/checkpoint_impl.cc

@@ -42,6 +42,28 @@ Status Checkpoint::CreateCheckpoint(const std::string& /*checkpoint_dir*/,
   return Status::NotSupported("");
 }
 
+void CheckpointImpl::CleanStagingDirectory(
+    const std::string& full_private_path, Logger* info_log) {
+    std::vector<std::string> subchildren;
+  Status s = db_->GetEnv()->FileExists(full_private_path);
+  if (s.IsNotFound()) {
+    return;
+  }
+  ROCKS_LOG_INFO(info_log, "File exists %s -- %s",
+                 full_private_path.c_str(), s.ToString().c_str());
+  db_->GetEnv()->GetChildren(full_private_path, &subchildren);
+  for (auto& subchild : subchildren) {
+    std::string subchild_path = full_private_path + "/" + subchild;
+    s = db_->GetEnv()->DeleteFile(subchild_path);
+    ROCKS_LOG_INFO(info_log, "Delete file %s -- %s",
+                   subchild_path.c_str(), s.ToString().c_str());
+  }
+  // finally delete the private dir
+  s = db_->GetEnv()->DeleteDir(full_private_path);
+  ROCKS_LOG_INFO(info_log, "Delete dir %s -- %s",
+                 full_private_path.c_str(), s.ToString().c_str());
+}
+
 // Builds an openable snapshot of RocksDB
 Status CheckpointImpl::CreateCheckpoint(const std::string& checkpoint_dir,
                                         uint64_t log_size_for_flush) {
@@ -75,6 +97,7 @@ Status CheckpointImpl::CreateCheckpoint(const std::string& checkpoint_dir,
       db_options.info_log,
       "Snapshot process -- using temporary directory %s",
       full_private_path.c_str());
+  CleanStagingDirectory(full_private_path, db_options.info_log.get());
   // create snapshot directory
   s = db_->GetEnv()->CreateDir(full_private_path);
   uint64_t sequence_number = 0;
@@ -125,19 +148,7 @@ Status CheckpointImpl::CreateCheckpoint(const std::string& checkpoint_dir,
     // clean all the files we might have created
     ROCKS_LOG_INFO(db_options.info_log, "Snapshot failed -- %s",
                    s.ToString().c_str());
-    // we have to delete the dir and all its children
-    std::vector<std::string> subchildren;
-    db_->GetEnv()->GetChildren(full_private_path, &subchildren);
-    for (auto& subchild : subchildren) {
-      std::string subchild_path = full_private_path + "/" + subchild;
-      Status s1 = db_->GetEnv()->DeleteFile(subchild_path);
-      ROCKS_LOG_INFO(db_options.info_log, "Delete file %s -- %s",
-                     subchild_path.c_str(), s1.ToString().c_str());
-    }
-    // finally delete the private dir
-    Status s1 = db_->GetEnv()->DeleteDir(full_private_path);
-    ROCKS_LOG_INFO(db_options.info_log, "Delete dir %s -- %s",
-                   full_private_path.c_str(), s1.ToString().c_str());
+    CleanStagingDirectory(full_private_path, db_options.info_log.get());
   }
   return s;
 }

utilities/checkpoint/checkpoint_impl.h

@@ -47,6 +47,7 @@ class CheckpointImpl : public Checkpoint {
       uint64_t* sequence_number, uint64_t log_size_for_flush);
 
  private:
+  void CleanStagingDirectory(const std::string& path, Logger* info_log);
   DB* db_;
 };