(doris) branch branch-3.1 updated: branch-3.1: pick 5 mow lock prs #49716 #49223 #50242 #50281 #50062 (#52038)

[morrysnow]

This is an automated email from the ASF dual-hosted git repository.

morrysnow pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/doris.git


The following commit(s) were added to refs/heads/branch-3.1 by this push:
     new 72eeccf01bc branch-3.1: pick 5 mow lock prs #49716 #49223 #50242 
#50281 #50062 (#52038)
72eeccf01bc is described below

commit 72eeccf01bc0c452be8b4d78fcbec313d85ee80e
Author: meiyi <me...@selectdb.com>
AuthorDate: Fri Jun 20 23:45:05 2025 +0800

    branch-3.1: pick 5 mow lock prs #49716 #49223 #50242 #50281 #50062 (#52038)
    
    pick
    #49716
    #49223
    #50242
    #50281
    #50062
    
    ---------
    
    Co-authored-by: huanghaibin <huanghai...@selectdb.com>
---
 cloud/src/common/config.h                   |   9 +
 cloud/src/meta-service/keys.cpp             |  20 +-
 cloud/src/meta-service/keys.h               |   1 +
 cloud/src/meta-service/meta_service.cpp     | 341 ++++++++++-
 cloud/src/meta-service/meta_service.h       |  20 +
 cloud/src/meta-service/meta_service_job.cpp | 187 +++---
 cloud/src/recycler/checker.cpp              |  78 +++
 cloud/src/recycler/checker.h                |   2 +
 cloud/test/meta_service_job_test.cpp        | 903 ++++++++++++++++++++++++++++
 cloud/test/recycler_test.cpp                | 124 ++++
 10 files changed, 1590 insertions(+), 95 deletions(-)

diff --git a/cloud/src/common/config.h b/cloud/src/common/config.h
index 84d889a059e..5963adea3aa 100644
--- a/cloud/src/common/config.h
+++ b/cloud/src/common/config.h
@@ -104,6 +104,9 @@ CONF_mInt64(recycle_task_threshold_seconds, "10800"); // 3h
 // **just for TEST**
 CONF_Bool(force_immediate_recycle, "false");
 
+CONF_mBool(enable_mow_compaction_key_check, "false");
+CONF_mInt64(compaction_key_check_expiration_diff_seconds, "600"); // 10min
+
 CONF_String(test_s3_ak, "");
 CONF_String(test_s3_sk, "");
 CONF_String(test_s3_endpoint, "");
@@ -273,6 +276,12 @@ CONF_Bool(enable_check_instance_id, "true");
 
 // Check if ip eq 127.0.0.1, ms/recycler exit
 CONF_Bool(enable_loopback_address_for_ms, "false");
+
+// delete_bitmap_lock version config
+CONF_mString(use_delete_bitmap_lock_version, "v1");
+// FOR DEBUGGING
+CONF_mBool(use_delete_bitmap_lock_random_version, "false");
+
 // Which vaults should be recycled. If empty, recycle all vaults.
 // Comma seprated list: recycler_storage_vault_white_list="aaa,bbb,ccc"
 CONF_Strings(recycler_storage_vault_white_list, "");
diff --git a/cloud/src/meta-service/keys.cpp b/cloud/src/meta-service/keys.cpp
index e089e0a1f0b..b00988b7a77 100644
--- a/cloud/src/meta-service/keys.cpp
+++ b/cloud/src/meta-service/keys.cpp
@@ -33,7 +33,6 @@ static const char* STATS_KEY_PREFIX    = "stats";
 static const char* JOB_KEY_PREFIX      = "job";
 static const char* COPY_KEY_PREFIX     = "copy";
 static const char* VAULT_KEY_PREFIX    = "storage_vault";
-static const char* MOW_KEY_PREFIX      = "mow";
 
 // Infix
 static const char* TXN_KEY_INFIX_LABEL                  = "txn_label";
@@ -137,7 +136,8 @@ static void encode_prefix(const T& t, std::string* key) {
                       || std::is_same_v<T, MetaSchemaPBDictionaryInfo>
                       || std::is_same_v<T, MetaDeleteBitmapInfo>
                       || std::is_same_v<T, MetaDeleteBitmapUpdateLockInfo>
-                      || std::is_same_v<T, MetaPendingDeleteBitmapInfo>) {
+                      || std::is_same_v<T, MetaPendingDeleteBitmapInfo>
+                      || std::is_same_v<T, MowTabletCompactionInfo>) {
         encode_bytes(META_KEY_PREFIX, key);
     } else if constexpr (std::is_same_v<T, PartitionVersionKeyInfo>
                       || std::is_same_v<T, TableVersionKeyInfo>) {
@@ -159,8 +159,6 @@ static void encode_prefix(const T& t, std::string* key) {
         encode_bytes(COPY_KEY_PREFIX, key);
     } else if constexpr (std::is_same_v<T, StorageVaultKeyInfo>) {
         encode_bytes(VAULT_KEY_PREFIX, key);
-    } else if constexpr(std::is_same_v<T, MowTabletCompactionInfo>) {
-        encode_bytes(MOW_KEY_PREFIX, key);
     } else {
         // This branch mean to be unreachable, add an assert(false) here to
         // prevent missing branch match.
@@ -304,6 +302,13 @@ void meta_delete_bitmap_update_lock_key(const 
MetaDeleteBitmapUpdateLockInfo& in
     encode_int64(std::get<2>(in), out);                   // partition_id
 }
 
+void mow_tablet_compaction_key(const MowTabletCompactionInfo& in, std::string* 
out) {
+    encode_prefix(in, out);                                  // 0x01 "meta" 
${instance_id}
+    encode_bytes(META_KEY_INFIX_MOW_TABLET_COMPACTION, out); // 
"mow_tablet_comp"
+    encode_int64(std::get<1>(in), out);                      // table_id
+    encode_int64(std::get<2>(in), out);                      // initiator
+}
+
 void meta_pending_delete_bitmap_key(const MetaPendingDeleteBitmapInfo& in, 
std::string* out) {
     encode_prefix(in, out);                                  // 0x01 "meta" 
${instance_id}
     encode_bytes(META_KEY_INFIX_DELETE_BITMAP_PENDING, out); // 
"delete_bitmap_pending"
@@ -502,13 +507,6 @@ std::string system_meta_service_encryption_key_info_key() {
 // Other keys
 
//==============================================================================
 
-void mow_tablet_compaction_key(const MowTabletCompactionInfo& in, std::string* 
out) {
-    encode_prefix(in, out);                                  // 0x01 "mow" 
${instance_id}
-    encode_bytes(META_KEY_INFIX_MOW_TABLET_COMPACTION, out); // 
"mow_tablet_comp"
-    encode_int64(std::get<1>(in), out);                      // table_id
-    encode_int64(std::get<2>(in), out);                      // initiator
-}
-
 
//==============================================================================
 // Decode keys
 
//==============================================================================
diff --git a/cloud/src/meta-service/keys.h b/cloud/src/meta-service/keys.h
index 9ff51374927..b819d8e2608 100644
--- a/cloud/src/meta-service/keys.h
+++ b/cloud/src/meta-service/keys.h
@@ -45,6 +45,7 @@
 // 0x01 "meta" ${instance_id} "delete_bitmap_pending" ${table_id}              
                  -> PendingDeleteBitmapPB 
 // 0x01 "meta" ${instance_id} "delete_bitmap" ${tablet_id} ${rowset_id} 
${version} ${segment_id} -> roaringbitmap
 // 0x01 "meta" ${instance_id} "tablet_schema_pb_dict" ${index_id}              
                  -> SchemaCloudDictionary
+// 0x01 "meta" ${instance_id} "mow_tablet_comp" ${table_id} ${initiator_id}    
                  -> MowTabletCompactionPB
 //
 // 0x01 "stats" ${instance_id} "tablet" ${table_id} ${index_id} 
${partition_id} ${tablet_id}               -> TabletStatsPB
 // 0x01 "stats" ${instance_id} "tablet" ${table_id} ${index_id} 
${partition_id} ${tablet_id} "data_size"   -> int64
diff --git a/cloud/src/meta-service/meta_service.cpp 
b/cloud/src/meta-service/meta_service.cpp
index a280bc53303..bac1b90dea0 100644
--- a/cloud/src/meta-service/meta_service.cpp
+++ b/cloud/src/meta-service/meta_service.cpp
@@ -1845,10 +1845,11 @@ static bool check_delete_bitmap_lock(MetaServiceCode& 
code, std::string& msg, st
                                      std::unique_ptr<Transaction>& txn, 
std::string& instance_id,
                                      int64_t table_id, int64_t lock_id, 
int64_t lock_initiator,
                                      std::string& lock_key, 
DeleteBitmapUpdateLockPB& lock_info,
-                                     std::string log = "") {
+                                     std::string use_version, std::string log 
= "") {
     std::string lock_val;
     LOG(INFO) << "check_delete_bitmap_lock, table_id=" << table_id << " 
lock_id=" << lock_id
-              << " initiator=" << lock_initiator << " key=" << hex(lock_key) 
<< log;
+              << " initiator=" << lock_initiator << " key=" << hex(lock_key) 
<< log
+              << " use_version=" << use_version;
     auto err = txn->get(lock_key, &lock_val);
     
TEST_SYNC_POINT_CALLBACK("check_delete_bitmap_lock.inject_get_lock_key_err", 
&err);
     if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
@@ -1874,7 +1875,7 @@ static bool check_delete_bitmap_lock(MetaServiceCode& 
code, std::string& msg, st
         code = MetaServiceCode::LOCK_EXPIRED;
         return false;
     }
-    if (lock_id == COMPACTION_DELETE_BITMAP_LOCK_ID) {
+    if (use_version == "v2" && lock_id == COMPACTION_DELETE_BITMAP_LOCK_ID) {
         std::string tablet_compaction_key =
                 mow_tablet_compaction_key({instance_id, table_id, 
lock_initiator});
         std::string tablet_compaction_val;
@@ -1910,6 +1911,15 @@ static bool check_delete_bitmap_lock(MetaServiceCode& 
code, std::string& msg, st
     }
     return true;
 }
+static bool use_new_version_random() {
+    std::mt19937 gen {std::random_device {}()};
+    auto p = 0.5;
+    std::bernoulli_distribution inject_fault {p};
+    if (inject_fault(gen)) {
+        return true;
+    }
+    return false;
+}
 
 static bool process_pending_delete_bitmap(MetaServiceCode& code, std::string& 
msg,
                                           std::stringstream& ss, 
std::unique_ptr<Transaction>& txn,
@@ -2052,6 +2062,10 @@ void 
MetaServiceImpl::update_delete_bitmap(google::protobuf::RpcController* cont
                                            const UpdateDeleteBitmapRequest* 
request,
                                            UpdateDeleteBitmapResponse* 
response,
                                            ::google::protobuf::Closure* done) {
+    std::string use_version = config::use_delete_bitmap_lock_version;
+    if (config::use_delete_bitmap_lock_random_version && 
!use_new_version_random()) {
+        use_version = "v1";
+    }
     RPC_PREPROCESS(update_delete_bitmap);
     std::string cloud_unique_id = request->has_cloud_unique_id() ? 
request->cloud_unique_id() : "";
     if (cloud_unique_id.empty()) {
@@ -2088,7 +2102,8 @@ void 
MetaServiceImpl::update_delete_bitmap(google::protobuf::RpcController* cont
         std::string lock_key = 
meta_delete_bitmap_update_lock_key({instance_id, table_id, -1});
         DeleteBitmapUpdateLockPB lock_info;
         if (!check_delete_bitmap_lock(code, msg, ss, txn, instance_id, 
table_id, request->lock_id(),
-                                      request->initiator(), lock_key, 
lock_info, log)) {
+                                      request->initiator(), lock_key, 
lock_info, use_version,
+                                      log)) {
             LOG(WARNING) << "failed to check delete bitmap lock, table_id=" << 
table_id
                          << " request lock_id=" << request->lock_id()
                          << " request initiator=" << request->initiator() << " 
msg " << msg;
@@ -2200,7 +2215,7 @@ void 
MetaServiceImpl::update_delete_bitmap(google::protobuf::RpcController* cont
                 DeleteBitmapUpdateLockPB lock_info;
                 if (!check_delete_bitmap_lock(code, msg, ss, txn, instance_id, 
table_id,
                                               request->lock_id(), 
request->initiator(), lock_key,
-                                              lock_info, log)) {
+                                              lock_info, use_version, log)) {
                     LOG(WARNING) << "failed to check delete bitmap lock, 
table_id=" << table_id
                                  << " request lock_id=" << request->lock_id()
                                  << " request initiator=" << 
request->initiator() << " msg " << msg;
@@ -2473,10 +2488,12 @@ static bool 
put_delete_bitmap_update_lock_key(MetaServiceCode& code, std::string
     return true;
 }
 
-void 
MetaServiceImpl::get_delete_bitmap_update_lock(google::protobuf::RpcController* 
controller,
-                                                    const 
GetDeleteBitmapUpdateLockRequest* request,
-                                                    
GetDeleteBitmapUpdateLockResponse* response,
-                                                    
::google::protobuf::Closure* done) {
+void MetaServiceImpl::get_delete_bitmap_update_lock_v2(
+        google::protobuf::RpcController* controller,
+        const GetDeleteBitmapUpdateLockRequest* request,
+        GetDeleteBitmapUpdateLockResponse* response, 
::google::protobuf::Closure* done) {
+    VLOG_DEBUG << "get delete bitmap update lock in v2 for table=" << 
request->table_id()
+               << ",lock id=" << request->lock_id() << ",initiator=" << 
request->initiator();
     RPC_PREPROCESS(get_delete_bitmap_update_lock);
     std::string cloud_unique_id = request->has_cloud_unique_id() ? 
request->cloud_unique_id() : "";
     if (cloud_unique_id.empty()) {
@@ -2804,7 +2821,7 @@ void 
MetaServiceImpl::get_delete_bitmap_update_lock(google::protobuf::RpcControl
 
     DeleteBitmapUpdateLockPB lock_info_tmp;
     if (!check_delete_bitmap_lock(code, msg, ss, txn, instance_id, table_id, 
request->lock_id(),
-                                  request->initiator(), lock_key, 
lock_info_tmp)) {
+                                  request->initiator(), lock_key, 
lock_info_tmp, "v2")) {
         LOG(WARNING) << "failed to check delete bitmap lock after get tablet 
stats and tablet "
                         "states, table_id="
                      << table_id << " request lock_id=" << request->lock_id()
@@ -2813,10 +2830,200 @@ void 
MetaServiceImpl::get_delete_bitmap_update_lock(google::protobuf::RpcControl
     }
 }
 
-void MetaServiceImpl::remove_delete_bitmap_update_lock(
+void MetaServiceImpl::get_delete_bitmap_update_lock_v1(
+        google::protobuf::RpcController* controller,
+        const GetDeleteBitmapUpdateLockRequest* request,
+        GetDeleteBitmapUpdateLockResponse* response, 
::google::protobuf::Closure* done) {
+    VLOG_DEBUG << "get delete bitmap update lock in v1 for table=" << 
request->table_id()
+               << ",lock id=" << request->lock_id() << ",initiator=" << 
request->initiator();
+    RPC_PREPROCESS(get_delete_bitmap_update_lock);
+    std::string cloud_unique_id = request->has_cloud_unique_id() ? 
request->cloud_unique_id() : "";
+    if (cloud_unique_id.empty()) {
+        code = MetaServiceCode::INVALID_ARGUMENT;
+        msg = "cloud unique id not set";
+        return;
+    }
+    instance_id = get_instance_id(resource_mgr_, cloud_unique_id);
+    if (instance_id.empty()) {
+        code = MetaServiceCode::INVALID_ARGUMENT;
+        msg = "empty instance_id";
+        LOG(INFO) << msg << ", cloud_unique_id=" << cloud_unique_id;
+        return;
+    }
+
+    RPC_RATE_LIMIT(get_delete_bitmap_update_lock)
+    std::unique_ptr<Transaction> txn;
+    TxnErrorCode err = txn_kv_->create_txn(&txn);
+    if (err != TxnErrorCode::TXN_OK) {
+        code = cast_as<ErrCategory::CREATE>(err);
+        msg = "failed to init txn";
+        return;
+    }
+    auto table_id = request->table_id();
+    std::string lock_key = meta_delete_bitmap_update_lock_key({instance_id, 
table_id, -1});
+    std::string lock_val;
+    DeleteBitmapUpdateLockPB lock_info;
+    err = txn->get(lock_key, &lock_val);
+    if (err != TxnErrorCode::TXN_OK && err != TxnErrorCode::TXN_KEY_NOT_FOUND) 
{
+        ss << "failed to get delete bitmap update lock, instance_id=" << 
instance_id
+           << " table_id=" << table_id << " key=" << hex(lock_key) << " err=" 
<< err;
+        msg = ss.str();
+        code = MetaServiceCode::KV_TXN_GET_ERR;
+        return;
+    }
+    using namespace std::chrono;
+    int64_t now = 
duration_cast<seconds>(system_clock::now().time_since_epoch()).count();
+    if (err == TxnErrorCode::TXN_OK) {
+        if (!lock_info.ParseFromString(lock_val)) [[unlikely]] {
+            code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+            msg = "failed to parse DeleteBitmapUpdateLockPB";
+            return;
+        }
+        if (lock_info.expiration() > 0 && lock_info.expiration() < now) {
+            LOG(INFO) << "delete bitmap lock expired, continue to process. 
lock_id="
+                      << lock_info.lock_id() << " table_id=" << table_id << " 
now=" << now;
+            lock_info.clear_initiators();
+        } else if (lock_info.lock_id() != request->lock_id()) {
+            ss << "already be locked. request lock_id=" << request->lock_id()
+               << " locked by lock_id=" << lock_info.lock_id() << " table_id=" 
<< table_id
+               << " now=" << now << " expiration=" << lock_info.expiration();
+            msg = ss.str();
+            code = MetaServiceCode::LOCK_CONFLICT;
+            return;
+        }
+    }
+
+    lock_info.set_lock_id(request->lock_id());
+    lock_info.set_expiration(now + request->expiration());
+    bool found = false;
+    for (auto initiator : lock_info.initiators()) {
+        if (request->initiator() == initiator) {
+            found = true;
+            break;
+        }
+    }
+    if (!found) {
+        lock_info.add_initiators(request->initiator());
+    }
+    lock_info.SerializeToString(&lock_val);
+    if (lock_val.empty()) {
+        code = MetaServiceCode::PROTOBUF_SERIALIZE_ERR;
+        msg = "pb serialization error";
+        return;
+    }
+    txn->put(lock_key, lock_val);
+    LOG(INFO) << "xxx put lock_key=" << hex(lock_key) << " table_id=" << 
table_id
+              << " lock_id=" << request->lock_id() << " initiator=" << 
request->initiator()
+              << " initiators_size=" << lock_info.initiators_size();
+
+    err = txn->commit();
+    if (err != TxnErrorCode::TXN_OK) {
+        code = cast_as<ErrCategory::COMMIT>(err);
+        ss << "failed to get_delete_bitmap_update_lock, err=" << err;
+        msg = ss.str();
+        return;
+    }
+
+    bool require_tablet_stats =
+            request->has_require_compaction_stats() ? 
request->require_compaction_stats() : false;
+    if (!require_tablet_stats) return;
+    // this request is from fe when it commits txn for MOW table, we send the 
compaction stats
+    // along with the GetDeleteBitmapUpdateLockResponse which will be sent to 
BE later to let
+    // BE eliminate unnecessary sync_rowsets() calls if possible
+    // 1. hold the delete bitmap update lock in MS(update lock_info.lock_id to 
current load's txn id)
+    // 2. read tablets' stats
+    // 3. check whether we still hold the delete bitmap update lock
+    // these steps can be done in different fdb txns
+
+    StopWatch read_stats_sw;
+    err = txn_kv_->create_txn(&txn);
+    if (err != TxnErrorCode::TXN_OK) {
+        code = cast_as<ErrCategory::CREATE>(err);
+        msg = "failed to init txn";
+        return;
+    }
+    for (const auto& tablet_idx : request->tablet_indexes()) {
+        // 1. get compaction cnts
+        TabletStatsPB tablet_stat;
+        std::string stats_key =
+                stats_tablet_key({instance_id, tablet_idx.table_id(), 
tablet_idx.index_id(),
+                                  tablet_idx.partition_id(), 
tablet_idx.tablet_id()});
+        std::string stats_val;
+        TxnErrorCode err = txn->get(stats_key, &stats_val);
+        
TEST_SYNC_POINT_CALLBACK("get_delete_bitmap_update_lock.get_compaction_cnts_inject_error",
+                                 &err);
+        if (err == TxnErrorCode::TXN_TOO_OLD) {
+            code = MetaServiceCode::OK;
+            err = txn_kv_->create_txn(&txn);
+            if (err != TxnErrorCode::TXN_OK) {
+                code = cast_as<ErrCategory::CREATE>(err);
+                ss << "failed to init txn when get tablet stats";
+                msg = ss.str();
+                return;
+            }
+            err = txn->get(stats_key, &stats_val);
+        }
+        if (err != TxnErrorCode::TXN_OK) {
+            code = cast_as<ErrCategory::READ>(err);
+            msg = fmt::format("failed to get tablet stats, err={} 
tablet_id={}", err,
+                              tablet_idx.tablet_id());
+            return;
+        }
+        if (!tablet_stat.ParseFromArray(stats_val.data(), stats_val.size())) {
+            code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+            msg = fmt::format("marformed tablet stats value, key={}", 
hex(stats_key));
+            return;
+        }
+        response->add_base_compaction_cnts(tablet_stat.base_compaction_cnt());
+        
response->add_cumulative_compaction_cnts(tablet_stat.cumulative_compaction_cnt());
+        response->add_cumulative_points(tablet_stat.cumulative_point());
+        // 2. get tablet states
+        std::string tablet_meta_key =
+                meta_tablet_key({instance_id, tablet_idx.table_id(), 
tablet_idx.index_id(),
+                                 tablet_idx.partition_id(), 
tablet_idx.tablet_id()});
+        std::string tablet_meta_val;
+        err = txn->get(tablet_meta_key, &tablet_meta_val);
+        if (err != TxnErrorCode::TXN_OK) {
+            ss << "failed to get tablet meta"
+               << (err == TxnErrorCode::TXN_KEY_NOT_FOUND ? " (not found)" : 
"")
+               << " instance_id=" << instance_id << " tablet_id=" << 
tablet_idx.tablet_id()
+               << " key=" << hex(tablet_meta_key) << " err=" << err;
+            msg = ss.str();
+            code = err == TxnErrorCode::TXN_KEY_NOT_FOUND ? 
MetaServiceCode::TABLET_NOT_FOUND
+                                                          : 
cast_as<ErrCategory::READ>(err);
+            return;
+        }
+        doris::TabletMetaCloudPB tablet_meta;
+        if (!tablet_meta.ParseFromString(tablet_meta_val)) {
+            code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+            msg = "malformed tablet meta";
+            return;
+        }
+        response->add_tablet_states(
+                
static_cast<std::underlying_type_t<TabletStatePB>>(tablet_meta.tablet_state()));
+    }
+    read_stats_sw.pause();
+    LOG(INFO) << fmt::format(
+            "tablet_idxes.size()={}, read tablet compaction cnts and tablet 
states cost={} ms",
+            request->tablet_indexes().size(), read_stats_sw.elapsed_us() / 
1000);
+
+    DeleteBitmapUpdateLockPB lock_info_tmp;
+    if (!check_delete_bitmap_lock(code, msg, ss, txn, instance_id, table_id, 
request->lock_id(),
+                                  request->initiator(), lock_key, 
lock_info_tmp, "v1")) {
+        LOG(WARNING) << "failed to check delete bitmap lock after get tablet 
stats and tablet "
+                        "states, table_id="
+                     << table_id << " request lock_id=" << request->lock_id()
+                     << " request initiator=" << request->initiator() << " 
code=" << code
+                     << " msg=" << msg;
+    }
+}
+
+void MetaServiceImpl::remove_delete_bitmap_update_lock_v2(
         google::protobuf::RpcController* controller,
         const RemoveDeleteBitmapUpdateLockRequest* request,
         RemoveDeleteBitmapUpdateLockResponse* response, 
::google::protobuf::Closure* done) {
+    VLOG_DEBUG << "remove delete bitmap update lock in v2 for table=" << 
request->table_id()
+               << ",lock id=" << request->lock_id() << ",initiator=" << 
request->initiator();
     RPC_PREPROCESS(remove_delete_bitmap_update_lock);
     std::string cloud_unique_id = request->has_cloud_unique_id() ? 
request->cloud_unique_id() : "";
     if (cloud_unique_id.empty()) {
@@ -2855,7 +3062,7 @@ void MetaServiceImpl::remove_delete_bitmap_update_lock(
         DeleteBitmapUpdateLockPB lock_info;
         if (!check_delete_bitmap_lock(code, msg, ss, txn, instance_id, 
request->table_id(),
                                       request->lock_id(), 
request->initiator(), lock_key, lock_info,
-                                      ", remove lock")) {
+                                      "v2", ", remove lock")) {
             LOG(WARNING) << "failed to check delete bitmap tablet lock"
                          << " table_id=" << request->table_id()
                          << " tablet_id=" << request->tablet_id()
@@ -2903,6 +3110,116 @@ void MetaServiceImpl::remove_delete_bitmap_update_lock(
     }
 }
 
+void MetaServiceImpl::remove_delete_bitmap_update_lock_v1(
+        google::protobuf::RpcController* controller,
+        const RemoveDeleteBitmapUpdateLockRequest* request,
+        RemoveDeleteBitmapUpdateLockResponse* response, 
::google::protobuf::Closure* done) {
+    VLOG_DEBUG << "remove delete bitmap update lock in v1 for table=" << 
request->table_id()
+               << ",lock id=" << request->lock_id() << ",initiator=" << 
request->initiator();
+    RPC_PREPROCESS(remove_delete_bitmap_update_lock);
+    std::string cloud_unique_id = request->has_cloud_unique_id() ? 
request->cloud_unique_id() : "";
+    if (cloud_unique_id.empty()) {
+        code = MetaServiceCode::INVALID_ARGUMENT;
+        msg = "cloud unique id not set";
+        return;
+    }
+    instance_id = get_instance_id(resource_mgr_, cloud_unique_id);
+    if (instance_id.empty()) {
+        code = MetaServiceCode::INVALID_ARGUMENT;
+        msg = "empty instance_id";
+        LOG(INFO) << msg << ", cloud_unique_id=" << cloud_unique_id;
+        return;
+    }
+    RPC_RATE_LIMIT(remove_delete_bitmap_update_lock)
+    std::unique_ptr<Transaction> txn;
+    TxnErrorCode err = txn_kv_->create_txn(&txn);
+    if (err != TxnErrorCode::TXN_OK) {
+        code = cast_as<ErrCategory::CREATE>(err);
+        msg = "failed to init txn";
+        return;
+    }
+    std::string lock_key =
+            meta_delete_bitmap_update_lock_key({instance_id, 
request->table_id(), -1});
+    std::string lock_val;
+    DeleteBitmapUpdateLockPB lock_info;
+    if (!check_delete_bitmap_lock(code, msg, ss, txn, instance_id, 
request->table_id(),
+                                  request->lock_id(), request->initiator(), 
lock_key, lock_info,
+                                  "v1")) {
+        LOG(WARNING) << "failed to check delete bitmap tablet lock"
+                     << " table_id=" << request->table_id() << " tablet_id=" 
<< request->tablet_id()
+                     << " request lock_id=" << request->lock_id()
+                     << " request initiator=" << request->initiator() << " msg 
" << msg;
+        return;
+    }
+    bool modify_initiators = false;
+    auto initiators = lock_info.mutable_initiators();
+    for (auto iter = initiators->begin(); iter != initiators->end(); iter++) {
+        if (*iter == request->initiator()) {
+            initiators->erase(iter);
+            modify_initiators = true;
+            break;
+        }
+    }
+    if (!modify_initiators) {
+        LOG(INFO) << "initiators don't have initiator=" << request->initiator()
+                  << ",initiators_size=" << lock_info.initiators_size() << 
",just return";
+        return;
+    } else if (initiators->empty()) {
+        LOG(INFO) << "remove delete bitmap lock, table_id=" << 
request->table_id()
+                  << " lock_id=" << request->lock_id() << " key=" << 
hex(lock_key);
+        txn->remove(lock_key);
+    } else {
+        lock_info.SerializeToString(&lock_val);
+        if (lock_val.empty()) {
+            LOG(WARNING) << "failed to seiralize lock_info, table_id=" << 
request->table_id()
+                         << " key=" << hex(lock_key);
+            return;
+        }
+        LOG(INFO) << "remove delete bitmap lock initiator, table_id=" << 
request->table_id()
+                  << ", key=" << hex(lock_key) << " lock_id=" << 
request->lock_id()
+                  << " initiator=" << request->initiator()
+                  << " initiators_size=" << lock_info.initiators_size();
+        txn->put(lock_key, lock_val);
+    }
+    err = txn->commit();
+    if (err != TxnErrorCode::TXN_OK) {
+        code = cast_as<ErrCategory::COMMIT>(err);
+        ss << "failed to remove delete bitmap tablet lock , err=" << err;
+        msg = ss.str();
+        return;
+    }
+}
+
+void 
MetaServiceImpl::get_delete_bitmap_update_lock(google::protobuf::RpcController* 
controller,
+                                                    const 
GetDeleteBitmapUpdateLockRequest* request,
+                                                    
GetDeleteBitmapUpdateLockResponse* response,
+                                                    
::google::protobuf::Closure* done) {
+    std::string use_version = config::use_delete_bitmap_lock_version;
+    if (config::use_delete_bitmap_lock_random_version && 
!use_new_version_random()) {
+        use_version = "v1";
+    }
+    if (use_version == "v2") {
+        get_delete_bitmap_update_lock_v2(controller, request, response, done);
+    } else {
+        get_delete_bitmap_update_lock_v1(controller, request, response, done);
+    }
+}
+
+void MetaServiceImpl::remove_delete_bitmap_update_lock(
+        google::protobuf::RpcController* controller,
+        const RemoveDeleteBitmapUpdateLockRequest* request,
+        RemoveDeleteBitmapUpdateLockResponse* response, 
::google::protobuf::Closure* done) {
+    std::string use_version = config::use_delete_bitmap_lock_version;
+    if (config::use_delete_bitmap_lock_random_version && 
!use_new_version_random()) {
+        use_version = "v1";
+    }
+    if (use_version == "v2") {
+        remove_delete_bitmap_update_lock_v2(controller, request, response, 
done);
+    } else {
+        remove_delete_bitmap_update_lock_v1(controller, request, response, 
done);
+    }
+}
+
 void MetaServiceImpl::remove_delete_bitmap(google::protobuf::RpcController* 
controller,
                                            const RemoveDeleteBitmapRequest* 
request,
                                            RemoveDeleteBitmapResponse* 
response,
diff --git a/cloud/src/meta-service/meta_service.h 
b/cloud/src/meta-service/meta_service.h
index 57f88d51dfe..ae2dc2997ad 100644
--- a/cloud/src/meta-service/meta_service.h
+++ b/cloud/src/meta-service/meta_service.h
@@ -321,6 +321,26 @@ private:
             const AlterInstanceRequest* request,
             std::function<std::pair<MetaServiceCode, 
std::string>(InstanceInfoPB*)> action);
 
+    void get_delete_bitmap_update_lock_v2(google::protobuf::RpcController* 
controller,
+                                          const 
GetDeleteBitmapUpdateLockRequest* request,
+                                          GetDeleteBitmapUpdateLockResponse* 
response,
+                                          ::google::protobuf::Closure* done);
+
+    void get_delete_bitmap_update_lock_v1(google::protobuf::RpcController* 
controller,
+                                          const 
GetDeleteBitmapUpdateLockRequest* request,
+                                          GetDeleteBitmapUpdateLockResponse* 
response,
+                                          ::google::protobuf::Closure* done);
+
+    void remove_delete_bitmap_update_lock_v2(google::protobuf::RpcController* 
controller,
+                                             const 
RemoveDeleteBitmapUpdateLockRequest* request,
+                                             
RemoveDeleteBitmapUpdateLockResponse* response,
+                                             ::google::protobuf::Closure* 
done);
+
+    void remove_delete_bitmap_update_lock_v1(google::protobuf::RpcController* 
controller,
+                                             const 
RemoveDeleteBitmapUpdateLockRequest* request,
+                                             
RemoveDeleteBitmapUpdateLockResponse* response,
+                                             ::google::protobuf::Closure* 
done);
+
     std::shared_ptr<TxnKv> txn_kv_;
     std::shared_ptr<ResourceManager> resource_mgr_;
     std::shared_ptr<RateLimiter> rate_limiter_;
diff --git a/cloud/src/meta-service/meta_service_job.cpp 
b/cloud/src/meta-service/meta_service_job.cpp
index 627479bc19a..f4897b6ed44 100644
--- a/cloud/src/meta-service/meta_service_job.cpp
+++ b/cloud/src/meta-service/meta_service_job.cpp
@@ -476,12 +476,13 @@ void 
MetaServiceImpl::start_tablet_job(::google::protobuf::RpcController* contro
     }
 }
 
-static bool check_and_remove_delete_bitmap_update_lock(MetaServiceCode& code, 
std::string& msg,
-                                                       std::stringstream& ss,
-                                                       
std::unique_ptr<Transaction>& txn,
-                                                       std::string& 
instance_id, int64_t table_id,
-                                                       int64_t tablet_id, 
int64_t lock_id,
-                                                       int64_t lock_initiator) 
{
+static bool check_and_remove_delete_bitmap_update_lock(
+        MetaServiceCode& code, std::string& msg, std::stringstream& ss,
+        std::unique_ptr<Transaction>& txn, std::string& instance_id, int64_t 
table_id,
+        int64_t tablet_id, int64_t lock_id, int64_t lock_initiator, 
std::string use_version) {
+    VLOG_DEBUG << "check_and_remove_delete_bitmap_update_lock table_id=" << 
table_id
+               << " tablet_id=" << tablet_id << " lock_id=" << lock_id
+               << " initiator=" << lock_initiator << " use_version=" << 
use_version;
     std::string lock_key = meta_delete_bitmap_update_lock_key({instance_id, 
table_id, -1});
     std::string lock_val;
     TxnErrorCode err = txn->get(lock_key, &lock_val);
@@ -507,7 +508,7 @@ static bool 
check_and_remove_delete_bitmap_update_lock(MetaServiceCode& code, st
         code = MetaServiceCode::LOCK_EXPIRED;
         return false;
     }
-    if (lock_id == COMPACTION_DELETE_BITMAP_LOCK_ID) {
+    if (use_version == "v2" && lock_id == COMPACTION_DELETE_BITMAP_LOCK_ID) {
         // when upgrade ms, prevent old ms get delete bitmap update lock
         if (lock_info.initiators_size() > 0) {
             ss << "compaction lock has " << lock_info.initiators_size() << " 
initiators";
@@ -579,79 +580,107 @@ static bool 
check_and_remove_delete_bitmap_update_lock(MetaServiceCode& code, st
     return true;
 }
 
+static void remove_delete_bitmap_update_lock_v1(std::unique_ptr<Transaction>& 
txn,
+                                                const std::string& 
instance_id, int64_t table_id,
+                                                int64_t tablet_id, int64_t 
lock_id,
+                                                int64_t lock_initiator) {
+    std::string lock_key = meta_delete_bitmap_update_lock_key({instance_id, 
table_id, -1});
+    std::string lock_val;
+    TxnErrorCode err = txn->get(lock_key, &lock_val);
+    LOG(INFO) << "get remove delete bitmap update lock info, table_id=" << 
table_id
+              << " key=" << hex(lock_key) << " err=" << err;
+    if (err != TxnErrorCode::TXN_OK) {
+        LOG(WARNING) << "failed to get delete bitmap update lock key, 
instance_id=" << instance_id
+                     << " table_id=" << table_id << " key=" << hex(lock_key) 
<< " err=" << err;
+        return;
+    }
+    DeleteBitmapUpdateLockPB lock_info;
+    if (!lock_info.ParseFromString(lock_val)) [[unlikely]] {
+        LOG(WARNING) << "failed to parse DeleteBitmapUpdateLockPB, 
instance_id=" << instance_id
+                     << " table_id=" << table_id << " key=" << hex(lock_key);
+        return;
+    }
+    if (lock_info.lock_id() != lock_id) {
+        return;
+    }
+    bool found = false;
+    auto initiators = lock_info.mutable_initiators();
+    for (auto iter = initiators->begin(); iter != initiators->end(); iter++) {
+        if (*iter == lock_initiator) {
+            initiators->erase(iter);
+            found = true;
+            break;
+        }
+    }
+    if (!found) {
+        INSTANCE_LOG(WARNING) << "failed to find lock_initiator, table_id=" << 
table_id
+                              << " tablet_id=" << tablet_id << " lock_id=" << 
lock_id
+                              << " initiator=" << lock_initiator << " key=" << 
hex(lock_key);
+        return;
+    }
+    if (initiators->empty()) {
+        INSTANCE_LOG(INFO) << "remove delete bitmap lock, table_id=" << 
table_id
+                           << " tablet_id=" << tablet_id << " lock_id=" << 
lock_id
+                           << " initiator=" << lock_initiator << " key=" << 
hex(lock_key);
+        txn->remove(lock_key);
+        return;
+    }
+    lock_info.SerializeToString(&lock_val);
+    if (lock_val.empty()) {
+        INSTANCE_LOG(WARNING) << "failed to seiralize lock_info, table_id=" << 
table_id
+                              << " key=" << hex(lock_key);
+        return;
+    }
+    INSTANCE_LOG(INFO) << "remove delete bitmap lock initiator, table_id=" << 
table_id
+                       << " tablet_id=" << tablet_id << " key=" << 
hex(lock_key)
+                       << " lock_id=" << lock_id << " initiator=" << 
lock_initiator
+                       << " initiators_size=" << lock_info.initiators_size();
+    txn->put(lock_key, lock_val);
+}
+
 static void remove_delete_bitmap_update_lock(std::unique_ptr<Transaction>& txn,
                                              const std::string& instance_id, 
int64_t table_id,
                                              int64_t tablet_id, int64_t 
lock_id,
-                                             int64_t lock_initiator) {
-    if (lock_id == COMPACTION_DELETE_BITMAP_LOCK_ID) {
+                                             int64_t lock_initiator, 
std::string use_version) {
+    VLOG_DEBUG << "remove_delete_bitmap_update_lock table_id=" << table_id
+               << " initiator=" << lock_initiator << " tablet_id=" << tablet_id
+               << " lock_id=" << lock_id << " use_version=" << use_version;
+    if (use_version == "v2" && lock_id == COMPACTION_DELETE_BITMAP_LOCK_ID) {
         std::string tablet_compaction_key =
                 mow_tablet_compaction_key({instance_id, table_id, 
lock_initiator});
         std::string tablet_compaction_val;
         TxnErrorCode err = txn->get(tablet_compaction_key, 
&tablet_compaction_val);
-        if (err != TxnErrorCode::TXN_OK) {
-            LOG(WARNING) << "failed to get tablet compaction key, 
instance_id=" << instance_id
-                         << " table_id=" << table_id << " initiator=" << 
lock_initiator
-                         << " key=" << hex(tablet_compaction_key) << " err=" 
<< err;
+        if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
+            remove_delete_bitmap_update_lock_v1(txn, instance_id, table_id, 
tablet_id, lock_id,
+                                                lock_initiator);
+        } else if (err != TxnErrorCode::TXN_OK) {
+            INSTANCE_LOG(WARNING) << "failed to get tablet compaction key, 
instance_id="
+                                  << instance_id << " table_id=" << table_id
+                                  << " initiator=" << lock_initiator
+                                  << " key=" << hex(tablet_compaction_key) << 
" err=" << err;
             return;
+        } else {
+            txn->remove(tablet_compaction_key);
+            INSTANCE_LOG(INFO) << "remove tablet compaction key, table_id=" << 
table_id
+                               << ", key=" << hex(tablet_compaction_key)
+                               << " initiator=" << lock_initiator;
         }
-        txn->remove(tablet_compaction_key);
-        INSTANCE_LOG(INFO) << "remove tablet compaction key, table_id=" << 
table_id
-                           << ", key=" << hex(tablet_compaction_key)
-                           << " initiator=" << lock_initiator;
     } else {
-        std::string lock_key = 
meta_delete_bitmap_update_lock_key({instance_id, table_id, -1});
-        std::string lock_val;
-        TxnErrorCode err = txn->get(lock_key, &lock_val);
-        LOG(INFO) << "get remove delete bitmap update lock info, table_id=" << 
table_id
-                  << " key=" << hex(lock_key) << " err=" << err;
-        if (err != TxnErrorCode::TXN_OK) {
-            LOG(WARNING) << "failed to get delete bitmap update lock key, 
instance_id="
-                         << instance_id << " table_id=" << table_id << " key=" 
<< hex(lock_key)
-                         << " err=" << err;
-            return;
-        }
-        DeleteBitmapUpdateLockPB lock_info;
-        if (!lock_info.ParseFromString(lock_val)) [[unlikely]] {
-            LOG(WARNING) << "failed to parse DeleteBitmapUpdateLockPB, 
instance_id=" << instance_id
-                         << " table_id=" << table_id << " key=" << 
hex(lock_key);
-            return;
-        }
-        if (lock_info.lock_id() != lock_id) {
-            return;
-        }
-        bool found = false;
-        auto initiators = lock_info.mutable_initiators();
-        for (auto iter = initiators->begin(); iter != initiators->end(); 
iter++) {
-            if (*iter == lock_initiator) {
-                initiators->erase(iter);
-                found = true;
-                break;
-            }
-        }
-        if (!found) {
-            return;
-        }
-        if (initiators->empty()) {
-            INSTANCE_LOG(INFO) << "remove delete bitmap lock, table_id=" << 
table_id
-                               << " tablet_id=" << tablet_id << " lock_id=" << 
lock_id
-                               << " initiator=" << lock_initiator << " key=" 
<< hex(lock_key);
-            txn->remove(lock_key);
-            return;
-        }
-        lock_info.SerializeToString(&lock_val);
-        if (lock_val.empty()) {
-            INSTANCE_LOG(WARNING) << "failed to seiralize lock_info, 
table_id=" << table_id
-                                  << " key=" << hex(lock_key);
-            return;
-        }
-        INSTANCE_LOG(INFO) << "remove delete bitmap lock initiator, table_id=" 
<< table_id
-                           << " tablet_id=" << tablet_id << " key=" << 
hex(lock_key)
-                           << " lock_id=" << lock_id << " initiator=" << 
lock_initiator
-                           << " initiators_size=" << 
lock_info.initiators_size();
-        txn->put(lock_key, lock_val);
+        remove_delete_bitmap_update_lock_v1(txn, instance_id, table_id, 
tablet_id, lock_id,
+                                            lock_initiator);
     }
 }
 
+static bool use_new_version_random() {
+    std::mt19937 gen {std::random_device {}()};
+    auto p = 0.5;
+    std::bernoulli_distribution inject_fault {p};
+    if (inject_fault(gen)) {
+        return true;
+    }
+    return false;
+}
+
 void process_compaction_job(MetaServiceCode& code, std::string& msg, 
std::stringstream& ss,
                             std::unique_ptr<Transaction>& txn,
                             const FinishTabletJobRequest* request,
@@ -727,9 +756,13 @@ void process_compaction_job(MetaServiceCode& code, 
std::string& msg, std::string
         INSTANCE_LOG(INFO) << "abort tablet compaction job, tablet_id=" << 
tablet_id
                            << " key=" << hex(job_key);
         if (compaction.has_delete_bitmap_lock_initiator()) {
-            remove_delete_bitmap_update_lock(txn, instance_id, table_id, 
tablet_id,
-                                             COMPACTION_DELETE_BITMAP_LOCK_ID,
-                                             
compaction.delete_bitmap_lock_initiator());
+            std::string use_version = config::use_delete_bitmap_lock_version;
+            if (config::use_delete_bitmap_lock_random_version && 
!use_new_version_random()) {
+                use_version = "v1";
+            }
+            remove_delete_bitmap_update_lock(
+                    txn, instance_id, table_id, tablet_id, 
COMPACTION_DELETE_BITMAP_LOCK_ID,
+                    compaction.delete_bitmap_lock_initiator(), use_version);
         }
         need_commit = true;
         return;
@@ -879,9 +912,14 @@ void process_compaction_job(MetaServiceCode& code, 
std::string& msg, std::string
 
     // remove delete bitmap update lock for MoW table
     if (compaction.has_delete_bitmap_lock_initiator()) {
+        std::string use_version = config::use_delete_bitmap_lock_version;
+        if (config::use_delete_bitmap_lock_random_version && 
!use_new_version_random()) {
+            use_version = "v1";
+        }
         bool success = check_and_remove_delete_bitmap_update_lock(
                 code, msg, ss, txn, instance_id, table_id, tablet_id,
-                COMPACTION_DELETE_BITMAP_LOCK_ID, 
compaction.delete_bitmap_lock_initiator());
+                COMPACTION_DELETE_BITMAP_LOCK_ID, 
compaction.delete_bitmap_lock_initiator(),
+                use_version);
         if (!success) {
             return;
         }
@@ -1378,9 +1416,14 @@ void process_schema_change_job(MetaServiceCode& code, 
std::string& msg, std::str
 
     // process mow table, check lock
     if (new_tablet_meta.enable_unique_key_merge_on_write()) {
+        std::string use_version = config::use_delete_bitmap_lock_version;
+        if (config::use_delete_bitmap_lock_random_version && 
!use_new_version_random()) {
+            use_version = "v1";
+        }
         bool success = check_and_remove_delete_bitmap_update_lock(
                 code, msg, ss, txn, instance_id, new_table_id, new_tablet_id,
-                SCHEMA_CHANGE_DELETE_BITMAP_LOCK_ID, 
schema_change.delete_bitmap_lock_initiator());
+                SCHEMA_CHANGE_DELETE_BITMAP_LOCK_ID, 
schema_change.delete_bitmap_lock_initiator(),
+                use_version);
         if (!success) {
             return;
         }
diff --git a/cloud/src/recycler/checker.cpp b/cloud/src/recycler/checker.cpp
index 60b6b7fc5ee..513d1685b2f 100644
--- a/cloud/src/recycler/checker.cpp
+++ b/cloud/src/recycler/checker.cpp
@@ -193,6 +193,12 @@ int Checker::start() {
                 }
             }
 
+            if (config::enable_mow_compaction_key_check) {
+                if (int ret = checker->do_mow_compaction_key_check(); ret != 
0) {
+                    success = false;
+                }
+            }
+
             // If instance checker has been aborted, don't finish this job
             if (!checker->stopped()) {
                 finish_instance_recycle_job(txn_kv_.get(), check_job_key, 
instance.instance_id(),
@@ -1189,4 +1195,76 @@ int 
InstanceChecker::do_delete_bitmap_storage_optimize_check() {
     return (failed_tablets_num > 0) ? 1 : 0;
 }
 
+int InstanceChecker::do_mow_compaction_key_check() {
+    std::unique_ptr<RangeGetIterator> it;
+    std::string begin = mow_tablet_compaction_key({instance_id_, 0, 0});
+    std::string end = mow_tablet_compaction_key({instance_id_, INT64_MAX, 0});
+    MowTabletCompactionPB mow_tablet_compaction;
+    do {
+        std::unique_ptr<Transaction> txn;
+        TxnErrorCode err = txn_kv_->create_txn(&txn);
+        if (err != TxnErrorCode::TXN_OK) {
+            LOG(WARNING) << "failed to create txn";
+            return -1;
+        }
+        err = txn->get(begin, end, &it);
+        if (err != TxnErrorCode::TXN_OK) {
+            LOG(WARNING) << "failed to get mow tablet compaction key, err=" << 
err;
+            return -1;
+        }
+        int64_t now = duration_cast<std::chrono::seconds>(
+                              
std::chrono::system_clock::now().time_since_epoch())
+                              .count();
+        while (it->has_next() && !stopped()) {
+            auto [k, v] = it->next();
+            std::string_view k1 = k;
+            k1.remove_prefix(1);
+            std::vector<std::tuple<std::variant<int64_t, std::string>, int, 
int>> out;
+            decode_key(&k1, &out);
+            // 0x01 "meta" ${instance_id} "mow_tablet_comp" ${table_id} 
${initiator}
+            auto table_id = std::get<int64_t>(std::get<0>(out[3]));
+            auto initiator = std::get<int64_t>(std::get<0>(out[4]));
+            if (!mow_tablet_compaction.ParseFromArray(v.data(), v.size())) 
[[unlikely]] {
+                LOG(WARNING) << "failed to parse MowTabletCompactionPB";
+                return -1;
+            }
+            int64_t expiration = mow_tablet_compaction.expiration();
+            //check compaction key failed should meet both following two 
condition:
+            //1.compaction key is expired
+            //2.table lock key is not found or key is not expired
+            if (expiration < now - 
config::compaction_key_check_expiration_diff_seconds) {
+                std::string lock_key =
+                        meta_delete_bitmap_update_lock_key({instance_id_, 
table_id, -1});
+                std::string lock_val;
+                err = txn->get(lock_key, &lock_val);
+                std::string reason = "";
+                if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
+                    reason = "table lock key not found";
+
+                } else {
+                    DeleteBitmapUpdateLockPB lock_info;
+                    if (!lock_info.ParseFromString(lock_val)) [[unlikely]] {
+                        LOG(WARNING) << "failed to parse 
DeleteBitmapUpdateLockPB";
+                        return -1;
+                    }
+                    if (lock_info.expiration() > now || lock_info.lock_id() != 
-1) {
+                        reason = "table lock is not expired,lock_id=" +
+                                 std::to_string(lock_info.lock_id());
+                    }
+                }
+                if (reason != "") {
+                    LOG(WARNING) << fmt::format(
+                            "[compaction key check fails] compaction key check 
fail for "
+                            "instance_id={}, table_id={}, initiator={}, 
expiration={}, now={}, "
+                            "reason={}",
+                            instance_id_, table_id, initiator, expiration, 
now, reason);
+                    return -1;
+                }
+            }
+        }
+        begin = it->next_begin_key(); // Update to next smallest key for 
iteration
+    } while (it->more() && !stopped());
+    return 0;
+}
+
 } // namespace doris::cloud
diff --git a/cloud/src/recycler/checker.h b/cloud/src/recycler/checker.h
index 7f87e90f7cb..7dd7d06fe6c 100644
--- a/cloud/src/recycler/checker.h
+++ b/cloud/src/recycler/checker.h
@@ -103,6 +103,8 @@ public:
     // which will not be cleared.
     int do_delete_bitmap_storage_optimize_check();
 
+    int do_mow_compaction_key_check();
+
     // If there are multiple buckets, return the minimum lifecycle; if there 
are no buckets (i.e.
     // all accessors are HdfsAccessor), return INT64_MAX.
     // Return 0 if success, otherwise error
diff --git a/cloud/test/meta_service_job_test.cpp 
b/cloud/test/meta_service_job_test.cpp
index 0eb43660c29..048295d8643 100644
--- a/cloud/test/meta_service_job_test.cpp
+++ b/cloud/test/meta_service_job_test.cpp
@@ -1115,6 +1115,909 @@ TEST(MetaServiceJobTest, CompactionJobTest) {
     ASSERT_NO_FATAL_FAILURE(test_abort_compaction_job(1, 2, 3, 7));
 }
 
+TEST(MetaServiceJobTest, DeleteBitmapUpdateLockCompatibilityTest) {
+    auto meta_service = get_meta_service();
+    auto sp = SyncPoint::get_instance();
+    std::unique_ptr<int, std::function<void(int*)>> defer(
+            (int*)0x01, [](int*) { 
SyncPoint::get_instance()->clear_all_call_backs(); });
+    sp->set_call_back("get_instance_id", [&](auto&& args) {
+        auto* ret = try_any_cast_ret<std::string>(args);
+        ret->first = instance_id;
+        ret->second = true;
+    });
+    sp->enable_processing();
+
+    brpc::Controller cntl;
+
+    auto test_start_compaction_job = [&](int64_t table_id, int64_t index_id, 
int64_t partition_id,
+                                         int64_t tablet_id,
+                                         TabletCompactionJobPB::CompactionType 
type,
+                                         std::string job_id = "job_id123") {
+        StartTabletJobResponse res;
+
+        auto index_key = meta_tablet_idx_key({instance_id, tablet_id});
+        TabletIndexPB idx_pb;
+        idx_pb.set_table_id(table_id);
+        idx_pb.set_index_id(index_id);
+        idx_pb.set_partition_id(partition_id);
+        idx_pb.set_tablet_id(tablet_id);
+        std::string idx_val = idx_pb.SerializeAsString();
+        std::unique_ptr<Transaction> txn;
+        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), 
TxnErrorCode::TXN_OK);
+        txn->put(index_key, idx_val);
+        std::string stats_key =
+                stats_tablet_key({instance_id, table_id, index_id, 
partition_id, tablet_id});
+        TabletStatsPB stats;
+        stats.set_base_compaction_cnt(9);
+        stats.set_cumulative_compaction_cnt(19);
+        txn->put(stats_key, stats.SerializeAsString());
+        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+        start_compaction_job(meta_service.get(), tablet_id, job_id, "ip:port", 
9, 19, type, res);
+        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+    };
+    FinishTabletJobResponse res;
+    auto test_commit_compaction_job = [&](int64_t table_id, int64_t index_id, 
int64_t partition_id,
+                                          int64_t tablet_id,
+                                          
TabletCompactionJobPB::CompactionType type,
+                                          int64_t initiator = 12345,
+                                          std::string job_id = "job_id123") {
+        FinishTabletJobRequest req;
+
+        auto compaction = req.mutable_job()->add_compaction();
+        compaction->set_id(job_id);
+        compaction->set_initiator("ip:port");
+        compaction->set_base_compaction_cnt(10);
+        compaction->set_cumulative_compaction_cnt(20);
+        compaction->set_delete_bitmap_lock_initiator(initiator);
+        req.mutable_job()->mutable_idx()->set_table_id(table_id);
+        req.mutable_job()->mutable_idx()->set_index_id(index_id);
+        req.mutable_job()->mutable_idx()->set_partition_id(partition_id);
+        req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id);
+        req.set_action(FinishTabletJobRequest::COMMIT);
+        req.set_cloud_unique_id("test_cloud_unique_id");
+
+        auto tablet_meta_key =
+                meta_tablet_key({instance_id, table_id, index_id, 
partition_id, tablet_id});
+        std::unique_ptr<Transaction> txn;
+        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), 
TxnErrorCode::TXN_OK);
+        doris::TabletMetaCloudPB tablet_meta_pb;
+        tablet_meta_pb.set_table_id(table_id);
+        tablet_meta_pb.set_index_id(index_id);
+        tablet_meta_pb.set_partition_id(partition_id);
+        tablet_meta_pb.set_tablet_id(tablet_id);
+        tablet_meta_pb.set_cumulative_layer_point(50);
+        std::string tablet_meta_val = tablet_meta_pb.SerializeAsString();
+        ASSERT_FALSE(tablet_meta_val.empty());
+        txn->put(tablet_meta_key, tablet_meta_val);
+        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+        // Create tablet stats, compaction job will update stats
+        auto tablet_stats_key =
+                stats_tablet_key({instance_id, table_id, index_id, 
partition_id, tablet_id});
+        TabletStatsPB tablet_stats_pb;
+        tablet_stats_pb.mutable_idx()->set_table_id(table_id);
+        tablet_stats_pb.mutable_idx()->set_index_id(index_id);
+        tablet_stats_pb.mutable_idx()->set_partition_id(partition_id);
+        tablet_stats_pb.mutable_idx()->set_tablet_id(tablet_id);
+
+        std::mt19937 
rng(std::chrono::system_clock::now().time_since_epoch().count());
+        std::uniform_int_distribution<int> dist(1, 10000); // Positive numbers
+
+        
compaction->set_output_cumulative_point(tablet_stats_pb.cumulative_point() + 
dist(rng));
+        compaction->set_num_output_rows(dist(rng));
+        compaction->set_num_output_rowsets(dist(rng));
+        compaction->set_num_output_segments(dist(rng));
+        compaction->set_num_input_rows(dist(rng));
+        compaction->set_num_input_rowsets(dist(rng));
+        compaction->set_num_input_segments(dist(rng));
+        compaction->set_size_input_rowsets(dist(rng));
+        compaction->set_size_output_rowsets(dist(rng));
+        compaction->set_type(type);
+
+        tablet_stats_pb.set_cumulative_compaction_cnt(dist(rng));
+        tablet_stats_pb.set_base_compaction_cnt(dist(rng));
+        
tablet_stats_pb.set_cumulative_point(tablet_meta_pb.cumulative_layer_point());
+        // MUST let data stats be larger than input data size
+        tablet_stats_pb.set_num_rows(dist(rng) + compaction->num_input_rows());
+        tablet_stats_pb.set_data_size(dist(rng) + 
compaction->size_input_rowsets());
+        tablet_stats_pb.set_num_rowsets(dist(rng) + 
compaction->num_input_rowsets());
+        tablet_stats_pb.set_num_segments(dist(rng) + 
compaction->num_input_segments());
+        tablet_stats_pb.set_index_size(dist(rng) + 
compaction->index_size_input_rowsets());
+        tablet_stats_pb.set_segment_size(dist(rng) + 
compaction->segment_size_input_rowsets());
+
+        std::string tablet_stats_val = tablet_stats_pb.SerializeAsString();
+        ASSERT_FALSE(tablet_stats_val.empty());
+        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), 
TxnErrorCode::TXN_OK);
+        txn->put(tablet_stats_key, tablet_stats_val);
+        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+        // Provide input and output rowset info
+        int64_t input_version_start = dist(rng);
+        int64_t input_version_end = input_version_start + 100;
+        compaction->add_input_versions(input_version_start);
+        compaction->add_input_versions(input_version_end);
+        compaction->add_output_versions(input_version_end);
+        compaction->add_output_rowset_ids("output rowset id");
+
+        // Provide input rowset KVs, boundary test, 5 input rowsets
+        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), 
TxnErrorCode::TXN_OK);
+        // clang-format off
+        std::vector<std::string> input_rowset_keys = {
+                meta_rowset_key({instance_id, tablet_id, input_version_start - 
1}),
+                meta_rowset_key({instance_id, tablet_id, input_version_start}),
+                meta_rowset_key({instance_id, tablet_id, input_version_start + 
1}),
+                meta_rowset_key({instance_id, tablet_id, (input_version_start 
+ input_version_end) / 2}),
+                meta_rowset_key({instance_id, tablet_id, input_version_end - 
1}),
+                meta_rowset_key({instance_id, tablet_id, input_version_end}),
+                meta_rowset_key({instance_id, tablet_id, input_version_end + 
1}),
+        };
+        // clang-format on
+        std::vector<std::unique_ptr<std::string>> input_rowset_vals;
+        for (auto& i : input_rowset_keys) {
+            doris::RowsetMetaCloudPB rs_pb;
+            rs_pb.set_rowset_id(0);
+            rs_pb.set_rowset_id_v2(hex(i));
+            input_rowset_vals.emplace_back(new 
std::string(rs_pb.SerializeAsString()));
+            txn->put(i, *input_rowset_vals.back());
+        }
+        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+        int64_t txn_id = dist(rng);
+        compaction->add_txn_id(txn_id);
+
+        // Provide output rowset meta
+        auto tmp_rowset_key = meta_rowset_tmp_key({instance_id, txn_id, 
tablet_id});
+        doris::RowsetMetaCloudPB tmp_rs_pb;
+        tmp_rs_pb.set_rowset_id(0);
+        tmp_rs_pb.set_txn_id(10086);
+        auto tmp_rowset_val = tmp_rs_pb.SerializeAsString();
+        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), 
TxnErrorCode::TXN_OK);
+        txn->put(tmp_rowset_key, tmp_rowset_val);
+        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+        
meta_service->finish_tablet_job(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
+                                        &req, &res, nullptr);
+    };
+
+    auto test_abort_compaction_job = [&](int64_t table_id, int64_t index_id, 
int64_t partition_id,
+                                         int64_t tablet_id, int64_t initiator 
= 12345,
+                                         std::string job_id = "job_id123") {
+        FinishTabletJobRequest req;
+        FinishTabletJobResponse res;
+
+        auto compaction = req.mutable_job()->add_compaction();
+        compaction->set_id(job_id);
+        compaction->set_initiator("ip:port");
+        compaction->set_delete_bitmap_lock_initiator(initiator);
+        req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id);
+        req.set_action(FinishTabletJobRequest::ABORT);
+        
meta_service->finish_tablet_job(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
+                                        &req, &res, nullptr);
+
+        std::unique_ptr<Transaction> txn;
+        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), 
TxnErrorCode::TXN_OK);
+        auto job_key = job_tablet_key({instance_id, table_id, index_id, 
partition_id, tablet_id});
+        std::string job_val;
+        ASSERT_EQ(txn->get(job_key, &job_val), TxnErrorCode::TXN_OK);
+        TabletJobInfoPB job_pb;
+        ASSERT_TRUE(job_pb.ParseFromString(job_val));
+        ASSERT_TRUE(job_pb.compaction().empty());
+    };
+
+    auto clear_rowsets = [&](int64_t tablet_id) {
+        std::unique_ptr<Transaction> txn;
+        ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), 
TxnErrorCode::TXN_OK);
+        std::string key1 = meta_rowset_key({instance_id, tablet_id, 1});
+        std::string key2 = meta_rowset_key({instance_id, tablet_id, 10001});
+        txn->remove(key1, key2);
+        ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+    };
+
+    config::use_delete_bitmap_lock_random_version = false;
+    int64_t table_id = 111;
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+
+    // case 1: lock key does not exist, get and remove load lock in new way, 
success
+    config::use_delete_bitmap_lock_version = "v2";
+    auto res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 123, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 123, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+
+    // case 2: lock key does not exist, get and remove load lock in old way, 
success
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 123, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 123, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+
+    // case 3: lock key does not exist, get and remove compaction lock in new 
way, success
+    config::use_delete_bitmap_lock_version = "v2";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 12345);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
12345);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+    clear_rowsets(5);
+
+    // case 4: lock key does not exist, get and remove compaction lock in old 
way, success
+    config::use_delete_bitmap_lock_version = "v1";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 12345);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
12345);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+    clear_rowsets(5);
+
+    // case 5:
+    // 5.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 5.2 load get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 222, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 5.3 compaction get lock in new way, failed
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 111);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_CONFLICT);
+    // 5.4 load remove lock in old way, success
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 222, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+
+    // case 6:
+    // 6.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 6.2 load get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 222, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 6.3 compaction get lock in old way, failed
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 111);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_CONFLICT);
+    // 6.4 load remove lock in new way, success
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 222, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+
+    // case 7:
+    // 7.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 7.2 compaction get lock in new way
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 777);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 7.3 load get lock in new way
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 123, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_CONFLICT);
+    // 7.4 compaction update delete bitmap in old way failed, abort compaction 
job
+    config::use_delete_bitmap_lock_version = "v1";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
777);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 777);
+    clear_rowsets(5);
+
+    // case 8:
+    // 8.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 8.2 compaction get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 888);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 8.3 load get lock in old way
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 124, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_CONFLICT);
+    // 8.4 compaction update delete bitmap in new way failed, abort compaction 
job
+    config::use_delete_bitmap_lock_version = "v2";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
888);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 888);
+    clear_rowsets(5);
+
+    // case 9:
+    // 9.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 9.2 compaction1 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 901);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 9.3 compaction2 get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 902);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
902);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE, 
902);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+    clear_rowsets(5);
+    // 9.4 load get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 199, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 199, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 9.5 compaction3 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 903);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 9.6 compaction1 update delete bitmap in new way failed, abort job
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
901);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 901);
+    clear_rowsets(5);
+
+    // case 10:
+    // 10.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 10.2 compaction1 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1001);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 10.3 compaction2 get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1002);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1002);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE, 
1002);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+    clear_rowsets(5);
+    // 10.4 load get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 1910, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 1910, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 10.5 compaction3 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1003);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 10.6 compaction1 update delete bitmap in new way success, commit job 
failed
+    config::use_delete_bitmap_lock_version = "v2";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1001);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE, 
1001);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::LOCK_EXPIRED);
+    clear_rowsets(5);
+
+    // case 11:
+    // 11.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 11.2 compaction1 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1101);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 11.3 compaction2 get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1102);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1102);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE, 
1102);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+    clear_rowsets(5);
+    // 11.4 sc get and remove lock in old way
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 11.5 compaction3 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1103);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 11.6 compaction1 update delete bitmap in new way failed, abort job
+    config::use_delete_bitmap_lock_version = "v2";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1101);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 1101);
+    clear_rowsets(5);
+
+    // case 12:
+    // 12.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 12.2 compaction1 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1201);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 12.3 compaction2 get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1202);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1202);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE, 
1202);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+    clear_rowsets(5);
+    // 12.4 sc get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 12.5 compaction3 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1203);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 12.6 compaction1 update delete bitmap in new way success, commit job 
failed
+    config::use_delete_bitmap_lock_version = "v2";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1201);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE, 
1201);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::LOCK_EXPIRED);
+    clear_rowsets(5);
+
+    // case 13:
+    // 13.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 13.2 compaction1 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1301);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 13.3 compaction2 get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1302);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1302);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE, 
1302);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+    clear_rowsets(5);
+    // 13.4 load get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 1390, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 1390, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 13.5 compaction3 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1303);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 13.6 compaction1 update delete bitmap in new way failed, abort job
+    config::use_delete_bitmap_lock_version = "v2";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1301);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 1301);
+    clear_rowsets(5);
+
+    // case 14:
+    // 14.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 14.2 compaction1 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1401);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 14.3 compaction2 get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1402);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1402);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE, 
1402);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+    clear_rowsets(5);
+    // 14.4 load get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 1490, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 1490, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 14.5 compaction3 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1403);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 14.6 compaction1 update delete bitmap in old way failed, abort job
+    config::use_delete_bitmap_lock_version = "v1";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1401);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 1401);
+    clear_rowsets(5);
+
+    // case 15:
+    // 15.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 15.2 compaction1 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1501);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 15.3 compaction2 get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1502);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1502);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE, 
1502);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+    clear_rowsets(5);
+    // 15.4 sc get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 15.5 compaction3 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1503);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 15.6 compaction1 update delete bitmap in old way failed, abort job
+    config::use_delete_bitmap_lock_version = "v1";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1501);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 1501);
+    clear_rowsets(5);
+
+    // case 16:
+    // 16.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 16.2 compaction1 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1601);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 16.3 compaction2 get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1602);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1602);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    test_commit_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE, 
1602);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
+    clear_rowsets(5);
+    // 16.4 sc get and remove lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 16.5 compaction3 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1603);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 16.6 compaction1 update delete bitmap in old way failed, abort job
+    config::use_delete_bitmap_lock_version = "v1";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1601);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 1601);
+    clear_rowsets(5);
+
+    // case 17:
+    // 17.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 17.2 compaction1 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1701);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 17.3 load get and remove lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 1793, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 1793, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 17.4 compaction2 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1702);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 17.5 compaction1 update delete bitmap in new way failed, abort job
+    config::use_delete_bitmap_lock_version = "v2";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1701);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 1701);
+    clear_rowsets(5);
+
+    // case 18:
+    // 18.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 18.2 compaction1 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1801);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 18.3 load get and remove lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 1893, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 1893, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 18.4 compaction2 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1802);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 18.5 compaction1 update delete bitmap in new way failed, abort job
+    config::use_delete_bitmap_lock_version = "v2";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1801);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 1801);
+    clear_rowsets(5);
+
+    // case 19:
+    // 19.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 19.2 compaction1 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1901);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 19.3 sc get and remove lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 19.4 compaction2 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 1902);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 19.5 compaction1 update delete bitmap in new way failed, abort job
+    config::use_delete_bitmap_lock_version = "v2";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
1901);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 1901);
+    clear_rowsets(5);
+
+    // case 20:
+    // 20.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 20.2 compaction1 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 2001);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 20.3 sc get and remove lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 20.4 compaction2 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 2002);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 20.5 compaction1 update delete bitmap in new way failed, abort job
+    config::use_delete_bitmap_lock_version = "v2";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
2001);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 2001);
+    clear_rowsets(5);
+
+    // case 21:
+    // 21.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 21.2 compaction1 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 2101);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 21.3 load get and remove lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 2190, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 2190, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 21.4 compaction2 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 2102);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 21.5 compaction1 update delete bitmap in old way failed, abort job
+    config::use_delete_bitmap_lock_version = "v1";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
2101);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 2101);
+    clear_rowsets(5);
+
+    // case 22:
+    // 22.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 22.2 compaction1 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 2201);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 22.3 load get and remove lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 2290, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 2290, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 22.4 compaction2 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 2202);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 22.5 compaction1 update delete bitmap in old way failed, abort job
+    config::use_delete_bitmap_lock_version = "v1";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
2201);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 2201);
+    clear_rowsets(5);
+
+    // case 23:
+    // 23.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 23.2 compaction1 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 2301);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 23.3 sc get and remove lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 2390, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 2390, 
-1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 23.4 compaction2 get lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 2302);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 23.5 compaction1 update delete bitmap in old way failed, abort job
+    config::use_delete_bitmap_lock_version = "v1";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
2301);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 2301);
+    clear_rowsets(5);
+
+    // case 24:
+    // 24.1 lock key does not exist
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    clear_rowsets(table_id);
+    // 24.2 compaction1 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 2401);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 24.3 sc get and remove lock in new way
+    config::use_delete_bitmap_lock_version = "v2";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, -2, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 24.4 compaction2 get lock in old way
+    config::use_delete_bitmap_lock_version = "v1";
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 2402);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    // 24.5 compaction1 update delete bitmap in old way failed, abort job
+    config::use_delete_bitmap_lock_version = "v1";
+    test_start_compaction_job(table_id, 2, 3, 5, TabletCompactionJobPB::BASE);
+    res_code = update_delete_bitmap(meta_service.get(), table_id, 3, 5, -1, 
2401);
+    ASSERT_EQ(res_code, MetaServiceCode::LOCK_EXPIRED);
+    test_abort_compaction_job(table_id, 2, 3, 5, 2401);
+    clear_rowsets(5);
+
+    // fuzzy case 1:
+    // 1.1 lock key does not exist
+    // 1.2 compaction1 get lock in old way
+    // 1.3 sc、load、compaction get and remove lock in new or old way
+    // 1.4 compaction2 get lock in old way
+    // 1.5 if sc or load succeed，compaction1 remove lock in old way failed
+
+    // fuzzy case 2:
+    // 2.1 lock key does not exist
+    // 2.2 compaction1 get lock in new way
+    // 2.3 sc、load、compaction get and remove lock in new or old way
+    // 2.4 compaction2 get lock in new way
+    // 2.5 if sc or load succeed，compaction1 remove lock in new way failed
+    table_id = 222;
+    for (int i = 1; i <= 2; i++) {
+        std::string lock_version = "v1";
+        if (i == 1) {
+            lock_version = "v1";
+        } else {
+            lock_version = "v2";
+        }
+        // 1 lock key does not exist
+        remove_delete_bitmap_lock(meta_service.get(), table_id);
+        clear_rowsets(table_id);
+        // 2 compaction1 get lock in old/new way
+        config::use_delete_bitmap_lock_version = lock_version;
+        res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 
2501);
+        ASSERT_EQ(res_code, MetaServiceCode::OK);
+        // 3 sc、load、compaction get and remove lock in new or old way
+        bool load_or_sc_succeed = false;
+        bool compaction_succeed = false;
+        std::srand(std::time(0));
+        for (int i = 0; i < 10; i++) {
+            int num = std::rand() % 3;
+            std::string use_version = (std::rand() % 2 == 0 ? "v2" : "v1");
+            config::use_delete_bitmap_lock_version = use_version;
+            LOG(INFO) << "i=" << i << ",num=" << num << ",use_version=" << 
use_version;
+            switch (num) {
+            case 0: {
+                res_code = get_delete_bitmap_lock(meta_service.get(), 
table_id, 2590, -1);
+                if (res_code == MetaServiceCode::OK) {
+                    res_code = remove_delete_bitmap_lock(meta_service.get(), 
table_id, 2590, -1);
+                    if (res_code == MetaServiceCode::OK) {
+                        load_or_sc_succeed = true;
+                    }
+                }
+                LOG(INFO) << "i=" << i << ",load_or_sc_succeed=" << 
load_or_sc_succeed
+                          << ",compaction_succeed=" << compaction_succeed;
+                break;
+            }
+            case 1: {
+                res_code = get_delete_bitmap_lock(meta_service.get(), 
table_id, -2, -1);
+                if (res_code == MetaServiceCode::OK) {
+                    res_code = remove_delete_bitmap_lock(meta_service.get(), 
table_id, -2, -1);
+                    if (res_code == MetaServiceCode::OK) {
+                        load_or_sc_succeed = true;
+                    }
+                }
+                LOG(INFO) << "i=" << i << ",load_or_sc_succeed=" << 
load_or_sc_succeed
+                          << ",compaction_succeed=" << compaction_succeed;
+                break;
+            }
+            case 2: {
+                res_code = get_delete_bitmap_lock(meta_service.get(), 
table_id, -1, 6600 + i);
+                if (res_code == MetaServiceCode::OK) {
+                    test_start_compaction_job(table_id, 2, 3, 5, 
TabletCompactionJobPB::BASE);
+                    res_code =
+                            update_delete_bitmap(meta_service.get(), table_id, 
3, 5, -1, 6600 + i);
+                    if (res.status().code() == MetaServiceCode::OK) {
+                        test_commit_compaction_job(table_id, 2, 3, 5, 
TabletCompactionJobPB::BASE,
+                                                   6600 + i);
+                        if (res.status().code() == MetaServiceCode::OK) {
+                            compaction_succeed = true;
+                        }
+                    }
+                    clear_rowsets(5);
+                }
+                LOG(INFO) << "i=" << i << ",load_or_sc_succeed=" << 
load_or_sc_succeed
+                          << ",compaction_succeed=" << compaction_succeed;
+                break;
+            }
+            }
+        }
+        // 4 compaction2 get lock in old/new way
+        config::use_delete_bitmap_lock_version = lock_version;
+        res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 
2502);
+        ASSERT_EQ(res_code, MetaServiceCode::OK);
+        // 5 if sc or load succeed，compaction1 remove lock in old/new way 
failed
+        config::use_delete_bitmap_lock_version = lock_version;
+        test_start_compaction_job(table_id, 2, 3, 5, 
TabletCompactionJobPB::BASE);
+        test_commit_compaction_job(table_id, 2, 3, 5, 
TabletCompactionJobPB::BASE, 2501);
+        if (load_or_sc_succeed) {
+            ASSERT_EQ(res.status().code(), MetaServiceCode::LOCK_EXPIRED);
+        }
+        clear_rowsets(5);
+    }
+}
+
 TEST(MetaServiceJobTest, CompactionJobWithMoWTest) {
     auto meta_service = get_meta_service();
     auto sp = SyncPoint::get_instance();
diff --git a/cloud/test/recycler_test.cpp b/cloud/test/recycler_test.cpp
index ac93e48e9be..81da3646772 100644
--- a/cloud/test/recycler_test.cpp
+++ b/cloud/test/recycler_test.cpp
@@ -3109,6 +3109,130 @@ TEST(CheckerTest, 
delete_bitmap_storage_optimize_check_abnormal) {
     ASSERT_EQ(expected_abnormal_rowsets, real_abnormal_rowsets);
 }
 
+std::unique_ptr<MetaServiceProxy> get_meta_service() {
+    int ret = 0;
+    // MemKv
+    auto txn_kv = 
std::dynamic_pointer_cast<TxnKv>(std::make_shared<MemTxnKv>());
+    if (txn_kv != nullptr) {
+        ret = txn_kv->init();
+        [&] { ASSERT_EQ(ret, 0); }();
+    }
+    [&] { ASSERT_NE(txn_kv.get(), nullptr); }();
+
+    std::unique_ptr<Transaction> txn;
+    EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+    txn->remove("\x00", "\xfe"); // This is dangerous if the fdb is not 
correctly set
+    EXPECT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+    auto rs = std::make_shared<MockResourceManager>(txn_kv);
+    auto rl = std::make_shared<RateLimiter>();
+    auto meta_service = std::make_unique<MetaServiceImpl>(txn_kv, rs, rl);
+    return std::make_unique<MetaServiceProxy>(std::move(meta_service));
+}
+
+MetaServiceCode get_delete_bitmap_lock(MetaServiceProxy* meta_service, int64_t 
table_id,
+                                       int64_t lock_id, int64_t initiator, 
int64_t expiration = 5) {
+    brpc::Controller cntl;
+    GetDeleteBitmapUpdateLockRequest req;
+    GetDeleteBitmapUpdateLockResponse res;
+    req.set_cloud_unique_id("test_cloud_unique_id");
+    req.set_table_id(table_id);
+    req.set_expiration(expiration);
+    req.set_lock_id(lock_id);
+    req.set_initiator(initiator);
+    meta_service->get_delete_bitmap_update_lock(
+            reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, 
&res, nullptr);
+    return res.status().code();
+}
+
+MetaServiceCode remove_delete_bitmap_lock(MetaServiceProxy* meta_service, 
int64_t table_id,
+                                          int64_t lock_id, int64_t initiator) {
+    brpc::Controller cntl;
+    RemoveDeleteBitmapUpdateLockRequest req;
+    RemoveDeleteBitmapUpdateLockResponse res;
+    req.set_cloud_unique_id("test_cloud_unique_id");
+    req.set_table_id(table_id);
+    req.set_lock_id(lock_id);
+    req.set_initiator(initiator);
+    meta_service->remove_delete_bitmap_update_lock(
+            reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, 
&res, nullptr);
+    return res.status().code();
+}
+
+void remove_delete_bitmap_lock(MetaServiceProxy* meta_service, int64_t 
table_id) {
+    std::string lock_key =
+            meta_delete_bitmap_update_lock_key({"test_check_compaction_key", 
table_id, -1});
+    std::unique_ptr<Transaction> txn;
+    ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+    txn->remove(lock_key);
+    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+}
+
+TEST(CheckerTest, check_compaction_key) {
+    config::enable_mow_compaction_key_check = true;
+    config::compaction_key_check_expiration_diff_seconds = 0;
+    config::use_delete_bitmap_lock_version = "v2";
+    std::string instance_id = "test_check_compaction_key";
+    [[maybe_unused]] auto sp = SyncPoint::get_instance();
+    std::unique_ptr<int, std::function<void(int*)>> defer(
+            (int*)0x01, [](int*) { 
SyncPoint::get_instance()->clear_all_call_backs(); });
+    sp->set_call_back("get_instance_id", [&](auto&& args) {
+        auto* ret = try_any_cast_ret<std::string>(args);
+        ret->first = instance_id;
+        ret->second = true;
+    });
+    sp->enable_processing();
+    int64_t table_id = 2;
+
+    //test 1:
+    //1.compaction get lock and write compaction key, but not release lock
+    //2.after lock expired, load get lock, compaction key is removed
+    auto meta_service = get_meta_service();
+    auto res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 
123);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    std::this_thread::sleep_for(std::chrono::seconds(6));
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 100, -1);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+
+    auto txn_kv = std::make_shared<MemTxnKv>();
+    ASSERT_EQ(txn_kv->init(), 0);
+
+    InstanceInfoPB instance;
+    instance.set_instance_id(instance_id);
+    auto obj_info = instance.add_obj_info();
+    obj_info->set_id("1");
+    InstanceChecker checker(meta_service->txn_kv(), instance_id);
+    ASSERT_EQ(checker.init(instance), 0);
+    ASSERT_EQ(checker.do_mow_compaction_key_check(), 0);
+    res_code = remove_delete_bitmap_lock(meta_service.get(), table_id, 100, 
-1);
+
+    //test 2:
+    //1.compaction a get lock and write compaction key, but not release lock
+    //2.compaction key is expired
+    //3.remove delete bitmap lock accidentally（it should not happen）
+    //4.load get lock
+    //5.checker found residual compaction key
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 124, 
0);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    std::this_thread::sleep_for(std::chrono::seconds(1));
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, 101, -1);
+    ASSERT_EQ(checker.do_mow_compaction_key_check(), -1);
+    std::this_thread::sleep_for(std::chrono::seconds(6));
+
+    //test 3:
+    //1.compaction a get lock and write compaction key, but not release lock
+    //2.compaction key is expired
+    //3.remove delete bitmap lock accidentally（it should not happen）
+    //4.checker found residual compaction key
+    table_id = 3;
+    res_code = get_delete_bitmap_lock(meta_service.get(), table_id, -1, 126, 
0);
+    ASSERT_EQ(res_code, MetaServiceCode::OK);
+    std::this_thread::sleep_for(std::chrono::seconds(1));
+    remove_delete_bitmap_lock(meta_service.get(), table_id);
+    ASSERT_EQ(checker.do_mow_compaction_key_check(), -1);
+}
+
 TEST(RecyclerTest, delete_rowset_data) {
     auto txn_kv = std::make_shared<MemTxnKv>();
     ASSERT_EQ(txn_kv->init(), 0);


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org