Apply sst file manager in myrocks clone to add slow-rm during checkpoints removal (#1386)

Summary:
This diff makes the following two changes:
1. Clone use rocksdb::DestroyDB() to clean the checkpoint. However, we didn't pass the sst file manager so didn't make use of the slow-rm feature provided by sst file manager. This will lead up to a discard spike when rolling checkpoints. This change pass the sst file manager when cleaning checkpoints.
2. During slow-rm sst files, the checkpoint directory may not be deleted immediately. In the rolling checkpoint function call, we will first delete the checkpoint directory, then create a new one with the same name. Since we are slow removing the sst files, it's possible that when we create the new checkpoints, the old directory hasn't been deleted yet. This will cause a 'directory exist' error when creating a new checkpoint directory. In this change, we add an additional suffix to differentiate rolling checkpoints directory to avoid this error.

Pull Request resolved: #1386

Test Plan:
Imported from GitHub, without a `Test Plan:` line.

Directory creation and deletion test is covered by `rolling_checkpoint.test`.

Did additional cp_clone to verify the correctness.

```
[root@udb18397.ftw5 /var/log]# less mysqld-3301.log | grep '.clone_checkpoint-1-'
2023-11-02T11:49:43.142720-07:00 1355 [Note] [MY-011071] [Server] Plugin rocksdb reported: 'creating checkpoint in directory: /data/mysql/3301/.rocksdb/.clone_checkpoint-1-1 '
2023-11-02T11:49:46.671314-07:00 1355 [Note] [MY-011071] [Server] Plugin rocksdb reported: 'created checkpoint in directory: /data/mysql/3301/.rocksdb/.clone_checkpoint-1-1 '
2023-11-02T11:58:20.464175-07:00 1355 [Note] [MY-011071] [Server] Plugin rocksdb reported: 'deleting temporary checkpoint in directory : /data/mysql/3301/.rocksdb/.clone_checkpoint-1-1 '
2023-11-02T11:58:20.753155-07:00 1355 [Note] [MY-011071] [Server] Plugin rocksdb reported: 'creating checkpoint in directory: /data/mysql/3301/.rocksdb/.clone_checkpoint-1-2 '
2023-11-02T11:58:22.045484-07:00 1355 [Note] [MY-011071] [Server] Plugin rocksdb reported: 'created checkpoint in directory: /data/mysql/3301/.rocksdb/.clone_checkpoint-1-2 '
2023-11-02T11:58:23.794134-07:00 1355 [Note] [MY-011071] [Server] Plugin rocksdb reported: 'deleting temporary checkpoint in directory : /data/mysql/3301/.rocksdb/.clone_checkpoint-1-2 '
```

Differential Revision: D50938261

fbshipit-source-id: d53058f53764335b556da9c6d18d0f4b1444b212

Author: sunshine-Chun

storage/rocksdb/clone/donor.cc

@@ -68,7 +68,9 @@ class [[nodiscard]] rdb_checkpoint final {
   // Returns MySQL error code
   [[nodiscard]] int init() {
     assert(!m_active);
-    const auto result = myrocks::rocksdb_create_checkpoint(m_dir.c_str());
+    m_full_dir = make_dir_full_name(
+        m_dir.c_str(), m_next_sub_id.fetch_add(1, std::memory_order_relaxed));
+    const auto result = myrocks::rocksdb_create_checkpoint(m_full_dir.c_str());
     m_active = (result == HA_EXIT_SUCCESS);
     return m_active ? 0 : ER_INTERNAL_ERROR;
   }
@@ -77,21 +79,22 @@ class [[nodiscard]] rdb_checkpoint final {
   [[nodiscard]] int cleanup() {
     if (!m_active) return 0;
     m_active = false;
-    return myrocks::rocksdb_remove_checkpoint(m_dir.c_str()) == HA_EXIT_SUCCESS
+    return myrocks::rocksdb_remove_checkpoint(m_full_dir.c_str()) ==
+                   HA_EXIT_SUCCESS
                ? 0
                : ER_INTERNAL_ERROR;
   }
 
   [[nodiscard]] constexpr const std::string &get_dir() const noexcept {
     assert(m_active);
-    return m_dir;
+    return m_full_dir;
   }
 
   [[nodiscard]] std::string path(const std::string &file_name) const {
     // We might be calling this for inactive checkpoint too, if the donor is in
     // the middle of a checkpoint roll. The caller will handle any ENOENTs as
     // needed.
-    return myrocks::rdb_concat_paths(m_dir, file_name);
+    return myrocks::rdb_concat_paths(m_full_dir, file_name);
   }
 
   rdb_checkpoint(const rdb_checkpoint &) = delete;
@@ -102,10 +105,14 @@ class [[nodiscard]] rdb_checkpoint final {
  private:
   const std::string m_dir;
 
+  std::string m_full_dir;
+
   bool m_active = false;
 
   static std::atomic<std::uint64_t> m_next_id;
 
+  std::atomic<std::uint64_t> m_next_sub_id{1};
+
   [[nodiscard]] static std::string make_dir_name(std::uint64_t id) {
     const auto base_str = myrocks::clone::checkpoint_base_dir();
     const auto id_str = std::to_string(id);
@@ -120,6 +127,19 @@ class [[nodiscard]] rdb_checkpoint final {
     result += id_str;
     return result;
   }
+
+  [[nodiscard]] static std::string make_dir_full_name(std::string base_dir,
+                                                      std::uint64_t id) {
+    const auto id_str = std::to_string(id);
+    std::string result;
+    result.reserve(base_dir.length() + id_str.length() +
+                   1);  // +1 for '-', the trailing
+                        // '\0' is accounted by the sizeof.
+    result = base_dir;
+    result += '-';
+    result += id_str;
+    return result;
+  }
 };
 
 std::atomic<std::uint64_t> rdb_checkpoint::m_next_id{1};

storage/rocksdb/ha_rocksdb.cc

@@ -476,20 +476,6 @@ int rocksdb_create_checkpoint(const char *checkpoint_dir_raw) {
   return HA_EXIT_FAILURE;
 }
 
-int rocksdb_remove_checkpoint(const char *checkpoint_dir_raw) {
-  const auto checkpoint_dir = rdb_normalize_dir(checkpoint_dir_raw);
-  LogPluginErrMsg(INFORMATION_LEVEL, ER_LOG_PRINTF_MSG,
-                  "deleting temporary checkpoint in directory : %s\n",
-                  checkpoint_dir.c_str());
-  const auto status = rocksdb::DestroyDB(checkpoint_dir, rocksdb::Options());
-  if (status.ok()) {
-    return HA_EXIT_SUCCESS;
-  }
-  my_error(ER_GET_ERRMSG, MYF(0), status.code(), status.ToString().c_str(),
-           rocksdb_hton_name);
-  return HA_EXIT_FAILURE;
-}
-
 static int rocksdb_create_checkpoint_validate(
     THD *const thd MY_ATTRIBUTE((__unused__)),
     struct SYS_VAR *const var MY_ATTRIBUTE((__unused__)),
@@ -1325,6 +1311,27 @@ static void rocksdb_set_io_write_timeout(
   RDB_MUTEX_UNLOCK_CHECK(rdb_sysvars_mutex);
 }
 
+int rocksdb_remove_checkpoint(const char *checkpoint_dir_raw) {
+  const auto checkpoint_dir = rdb_normalize_dir(checkpoint_dir_raw);
+  LogPluginErrMsg(INFORMATION_LEVEL, ER_LOG_PRINTF_MSG,
+                  "deleting temporary checkpoint in directory : %s\n",
+                  checkpoint_dir.c_str());
+
+  std::string trash_dir = std::string(rocksdb_datadir) + "/trash";
+  rocksdb::Options op = rocksdb::Options();
+  op.sst_file_manager.reset(NewSstFileManager(
+      rocksdb_db_options->env, rocksdb_db_options->info_log, trash_dir,
+      rocksdb_sst_mgr_rate_bytes_per_sec, true /* delete_existing_trash */));
+  const auto status = rocksdb::DestroyDB(checkpoint_dir, op);
+
+  if (status.ok()) {
+    return HA_EXIT_SUCCESS;
+  }
+  my_error(ER_GET_ERRMSG, MYF(0), status.code(), status.ToString().c_str(),
+           rocksdb_hton_name);
+  return HA_EXIT_FAILURE;
+}
+
 #endif  // !__APPLE__
 
 enum rocksdb_flush_log_at_trx_commit_type : unsigned int {