Apply sst file manager in myrocks clone to add slow-rm during checkpoints removal

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

@@ -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__)),
@@ -1305,6 +1291,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(