Apply sst file manager in myrocks clone to add slow-rm during checkpoints removal (facebook#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: facebook#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

Author: sunshine-Chun

storage/rocksdb/clone/donor.cc

@@ -56,8 +56,8 @@ namespace {
 class [[nodiscard]] rdb_checkpoint final {
  public:
   rdb_checkpoint()
-      : m_dir{
-            make_dir_name(m_next_id.fetch_add(1, std::memory_order_relaxed))} {}
+      : m_prefix_dir{make_dir_prefix_name(
+            m_next_id.fetch_add(1, std::memory_order_relaxed))} {}
 
   ~rdb_checkpoint() {
     // Ignore the return value - at this point the clone operation is completing
@@ -68,6 +68,7 @@ class [[nodiscard]] rdb_checkpoint final {
   // Returns MySQL error code
   [[nodiscard]] int init() {
     assert(!m_active);
+    m_dir = make_dir_name(m_prefix_dir, m_next_sub_id++);
     const auto result = myrocks::rocksdb_create_checkpoint(m_dir.c_str());
     m_active = (result == HA_EXIT_SUCCESS);
     return m_active ? 0 : ER_INTERNAL_ERROR;
@@ -100,13 +101,17 @@ class [[nodiscard]] rdb_checkpoint final {
   rdb_checkpoint &operator=(rdb_checkpoint &&) = delete;
 
  private:
-  const std::string m_dir;
+  const std::string m_prefix_dir;
+
+  std::string m_dir;
 
   bool m_active = false;
 
   static std::atomic<std::uint64_t> m_next_id;
 
-  [[nodiscard]] static std::string make_dir_name(std::uint64_t id) {
+  std::uint64_t m_next_sub_id = 1;
+
+  [[nodiscard]] static std::string make_dir_prefix_name(std::uint64_t id) {
     const auto base_str = myrocks::clone::checkpoint_base_dir();
     const auto id_str = std::to_string(id);
     std::string result;
@@ -120,6 +125,19 @@ class [[nodiscard]] rdb_checkpoint final {
     result += id_str;
     return result;
   }
+
+  [[nodiscard]] static std::string make_dir_name(
+      const std::string &dir_name_prefix, std::uint64_t id) {
+    const auto id_str = std::to_string(id);
+    std::string result;
+    result.reserve(dir_name_prefix.length() + id_str.length() +
+                   1);  // +1 for '-', the trailing
+                        // '\0' is accounted by the sizeof.
+    result = dir_name_prefix;
+    result += '-';
+    result += id_str;
+    return result;
+  }
 };
 
 std::atomic<std::uint64_t> rdb_checkpoint::m_next_id{1};

storage/rocksdb/ha_rocksdb.cc

@@ -475,20 +475,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__)),
@@ -1301,6 +1287,26 @@ static void rocksdb_set_reset_stats(
   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());
+
+  auto op = rocksdb::Options();
+  op.sst_file_manager.reset(NewSstFileManager(
+      rocksdb_db_options->env, rocksdb_db_options->info_log, "",
+      rocksdb_sst_mgr_rate_bytes_per_sec, false /* 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;
+}
+
 #ifndef __APPLE__
 
 static void rocksdb_set_io_write_timeout(