refactor: move func ingestion_files to rocksdb_wrapper

src/server/pegasus_write_service_impl.h

@@ -503,15 +503,10 @@ class pegasus_write_service::impl : public dsn::replication::replica_base
         }
 
         // ingest external files
-        rocksdb::IngestExternalFileOptions ifo;
-        rocksdb::Status s = _db->IngestExternalFile(sst_file_list, ifo);
-        if (!s.ok()) {
-            derror_rocksdb("IngestExternalFile", s.ToString(), "decree = {}", decree);
+        if (dsn_unlikely(_rocksdb_wrapper->ingestion_files(decree, sst_file_list) != 0)) {
             return dsn::ERR_INGESTION_FAILED;
-        } else {
-            ddebug_rocksdb("IngestExternalFile", "Ingest files succeed, decree = {}", decree);
-            return dsn::ERR_OK;
         }
+        return dsn::ERR_OK;
     }
 
     /// For batch write.

src/server/rocksdb_wrapper.cpp

@@ -181,6 +181,18 @@ int rocksdb_wrapper::write_batch_delete(int64_t decree, dsn::string_view raw_key
 
 void rocksdb_wrapper::clear_up_write_batch() { _write_batch->Clear(); }
 
+int rocksdb_wrapper::ingestion_files(int64_t decree, const std::vector<std::string> &sst_file_list)
+{
+    rocksdb::IngestExternalFileOptions ifo;
+    rocksdb::Status s = _db->IngestExternalFile(sst_file_list, ifo);
+    if (dsn_unlikely(!s.ok())) {
+        derror_rocksdb("IngestExternalFile", s.ToString(), "decree = {}", decree);
+    } else {
+        ddebug_rocksdb("IngestExternalFile", "Ingest files succeed, decree = {}", decree);
+    }
+    return s.code();
+}
+
 void rocksdb_wrapper::set_default_ttl(uint32_t ttl)
 {
     if (_default_ttl != ttl) {

src/server/rocksdb_wrapper.h

@@ -64,6 +64,7 @@ class rocksdb_wrapper : public dsn::replication::replica_base
     int write(int64_t decree);
     int write_batch_delete(int64_t decree, dsn::string_view raw_key);
     void clear_up_write_batch();
+    int ingestion_files(int64_t decree, const std::vector<std::string> &sst_file_list);
 
     void set_default_ttl(uint32_t ttl);