branch-4.1: [fix](cloud) Skip versioned delete bitmap cleanup for non-MoW tablets #67084

cloud/src/recycler/recycler.cpp

@@ -2503,7 +2503,8 @@ void InstanceRecycler::submit_recycle_prepare_rowsets_job(
                 LOG(WARNING) << "failed to delete rowset data, key=" << hex(key);
                 continue;
             }
-            if (delete_versioned_delete_bitmap_kvs(current_meta.tablet_id(),
+            if (delete_versioned_delete_bitmap_kvs(current_meta.partition_id(),
+                                                   current_meta.tablet_id(),
                                                    current_meta.rowset_id_v2()) != 0) {
                 continue;
             }
@@ -2564,8 +2565,8 @@ void InstanceRecycler::submit_recycle_tmp_rowsets_job(SimpleThreadPool& worker_p
             return;
         }
         for (const auto& [_, rowset] : rowsets_to_delete) {
-            if (delete_versioned_delete_bitmap_kvs(rowset.tablet_id(), rowset.rowset_id_v2()) !=
-                0) {
+            if (delete_versioned_delete_bitmap_kvs(rowset.partition_id(), rowset.tablet_id(),
+                                                   rowset.rowset_id_v2()) != 0) {
                 return;
             }
             if (delete_delete_bitmap_kvs(rowset.tablet_id(), rowset.rowset_id_v2()) != 0) {
@@ -4253,8 +4254,7 @@ int InstanceRecycler::delete_rowset_data(
         // we have to treat tmp rowset as "orphans" that may not related to any existing tablets
         // due to aborted schema change.
         if (is_formal_rowset) {
-            std::lock_guard lock(recycled_tablets_mtx_);
-            if (recycled_tablets_.count(rs.tablet_id()) && rs.packed_slice_locations_size() == 0) {
+            if (is_tablet_recycled(rs.tablet_id()) && rs.packed_slice_locations_size() == 0) {
                 // Tablet has been recycled and this rowset has no packed slices, so file data
                 // should already be gone; skip to avoid redundant deletes. Rowsets with packed
                 // slice info must still run to decrement packed file ref counts.
@@ -4504,16 +4504,89 @@ int InstanceRecycler::delete_rowset_data(const std::string& resource_id, int64_t
     return accessor->delete_prefix(rowset_path_prefix(tablet_id, rowset_id));
 }
 
-int InstanceRecycler::delete_versioned_delete_bitmap_kvs(int64_t tablet_id,
+bool InstanceRecycler::is_tablet_recycled(int64_t tablet_id) {
+    std::lock_guard lock(recycled_tablets_mtx_);
+    return recycled_tablets_.contains(tablet_id);
+}
+
+int InstanceRecycler::should_delete_versioned_delete_bitmap_kvs(int64_t partition_id,
+                                                                int64_t tablet_id) {
+    bool is_mow = true;
+    if (partition_id != -1) {
+        std::lock_guard lock(partition_mow_cache_mutex);
+        if (auto it = partition_mow_cache.find(partition_id); it != partition_mow_cache.end()) {
+            // cache hit
+            is_mow = it->second;
+            return is_mow ? 1 : 0;
+        }
+    }
+
+    TabletIndexPB tablet_index;
+    int ret = get_tablet_idx(txn_kv_.get(), instance_id_, tablet_id, tablet_index);
+    if (ret == 1) {
+        // maybe recycled
+        return is_tablet_recycled(tablet_id) ? 0 : 1;
+    }
+    if (ret != 0) {
+        LOG(WARNING) << "failed to get tablet index, instance_id=" << instance_id_
+                     << ", tablet_id=" << tablet_id;
+        return ret;
+    }
+
+    // Legacy RecycleRowsetPB without type does not carry a partition ID. Check the cache after
+    // obtaining its partition ID from the tablet index.
+    if (partition_id == -1) {
+        partition_id = tablet_index.partition_id();
+        std::lock_guard lock(partition_mow_cache_mutex);
+        if (auto it = partition_mow_cache.find(partition_id); it != partition_mow_cache.end()) {
+            is_mow = it->second;
+            return is_mow ? 1 : 0;
+        }
+    }
+
+    std::string tablet_meta_key =
+            meta_tablet_key({instance_id_, tablet_index.table_id(), tablet_index.index_id(),
+                             partition_id, tablet_id});
+    std::string tablet_meta_value;
+    ret = txn_get(txn_kv_.get(), tablet_meta_key, tablet_meta_value);
+    if (ret == 1) {
+        // maybe recycled
+        return is_tablet_recycled(tablet_id) ? 0 : 1;
+    }
+    if (ret != 0) {
+        LOG(WARNING) << "failed to get tablet meta, instance_id=" << instance_id_
+                     << ", tablet_id=" << tablet_id;
+        return ret;
+    }
+
+    TabletMetaCloudPB tablet_meta;
+    if (!tablet_meta.ParseFromString(tablet_meta_value)) {
+        LOG(WARNING) << "failed to parse tablet meta, instance_id=" << instance_id_
+                     << ", tablet_id=" << tablet_id;
+        return -1;
+    }
+    bool tablet_is_mow = tablet_meta.enable_unique_key_merge_on_write();
+    std::lock_guard lock(partition_mow_cache_mutex);
+    auto [it, _] = partition_mow_cache.emplace(partition_id, tablet_is_mow);
+    return it->second ? 1 : 0;
+}
+
+int InstanceRecycler::delete_versioned_delete_bitmap_kvs(int64_t partition_id, int64_t tablet_id,
                                                          const std::string& rowset_id) {
+    int ret = should_delete_versioned_delete_bitmap_kvs(partition_id, tablet_id);
+    if (ret <= 0) {
+        return ret;
+    }
+
     std::string dbm_start_key =
             versioned::meta_delete_bitmap_key({instance_id_, tablet_id, rowset_id});
     std::string dbm_end_key = dbm_start_key;
     encode_int64(INT64_MAX, &dbm_end_key);
-    int ret = txn_remove(txn_kv_.get(), dbm_start_key, dbm_end_key);
+    ret = txn_remove(txn_kv_.get(), dbm_start_key, dbm_end_key);
     if (ret != 0) {
         LOG(WARNING) << "failed to delete versioned delete bitmap kv, instance_id=" << instance_id_
-                     << " tablet_id=" << tablet_id << " rowset_id=" << rowset_id;
+                     << " partition_id=" << partition_id << " tablet_id=" << tablet_id
+                     << " rowset_id=" << rowset_id;
     }
     return ret;
 }
@@ -5436,14 +5509,19 @@ int InstanceRecycler::recycle_rowsets() {
             config::instance_recycler_worker_pool_size, "recycle_rowsets");
     worker_pool->start();
     auto delete_rowset_data_by_prefix = [&](std::string key, const std::string& resource_id,
-                                            int64_t tablet_id, const std::string& rowset_id) {
+                                            int64_t partition_id, int64_t tablet_id,
+                                            const std::string& rowset_id) {
         // Try to delete rowset data in background thread
         int ret = worker_pool->submit_with_timeout(
-                [&, resource_id, tablet_id, rowset_id, key]() mutable {
+                [&, resource_id, partition_id, tablet_id, rowset_id, key]() mutable {
                     if (delete_rowset_data(resource_id, tablet_id, rowset_id) != 0) {
                         LOG(WARNING) << "failed to delete rowset data, key=" << hex(key);
                         return;
                     }
+                    if (delete_versioned_delete_bitmap_kvs(partition_id, tablet_id, rowset_id) !=
+                        0) {
+                        return;
+                    }
                     std::vector<std::string> keys;
                     {
                         std::lock_guard lock(async_recycled_rowset_keys_mutex);
@@ -5452,7 +5530,6 @@ int InstanceRecycler::recycle_rowsets() {
                             keys.swap(async_recycled_rowset_keys);
                         }
                     }
-                    delete_versioned_delete_bitmap_kvs(tablet_id, rowset_id);
                     if (keys.empty()) return;
                     if (txn_remove(txn_kv_.get(), keys) != 0) {
                         LOG(WARNING) << "failed to delete recycle rowset kv, instance_id="
@@ -5470,7 +5547,7 @@ int InstanceRecycler::recycle_rowsets() {
             LOG(WARNING) << "failed to delete rowset data, key=" << hex(key);
             return -1;
         }
-        if (delete_versioned_delete_bitmap_kvs(tablet_id, rowset_id) != 0) {
+        if (delete_versioned_delete_bitmap_kvs(partition_id, tablet_id, rowset_id) != 0) {
             return -1;
         }
         rowset_keys.push_back(std::move(key));
@@ -5524,7 +5601,7 @@ int InstanceRecycler::recycle_rowsets() {
             LOG(INFO) << "delete rowset data, instance_id=" << instance_id_
                       << " tablet_id=" << rowset.tablet_id() << " rowset_id=" << rowset_id
                       << " task_type=" << metrics_context.operation_type;
-            if (delete_rowset_data_by_prefix(std::string(k), rowset.resource_id(),
+            if (delete_rowset_data_by_prefix(std::string(k), rowset.resource_id(), -1,
                                              rowset.tablet_id(), rowset_id) != 0) {
                 return -1;
             }
@@ -5584,7 +5661,7 @@ int InstanceRecycler::recycle_rowsets() {
                 }
             }
             if (delete_rowset_data_by_prefix(std::string(k), rowset_meta->resource_id(),
-                                             rowset_meta->tablet_id(),
+                                             rowset_meta->partition_id(), rowset_meta->tablet_id(),
                                              rowset_meta->rowset_id_v2()) != 0) {
                 return -1;
             }
@@ -5626,7 +5703,8 @@ int InstanceRecycler::recycle_rowsets() {
                 return;
             }
             for (const auto& [_, rs] : rowsets_to_delete) {
-                if (delete_versioned_delete_bitmap_kvs(rs.tablet_id(), rs.rowset_id_v2()) != 0) {
+                if (delete_versioned_delete_bitmap_kvs(rs.partition_id(), rs.tablet_id(),
+                                                       rs.rowset_id_v2()) != 0) {
                     return;
                 }
             }
@@ -6366,7 +6444,8 @@ int InstanceRecycler::recycle_tmp_rowsets() {
                 return;
             }
             for (const auto& [_, rs] : tmp_rowsets_to_delete) {
-                if (delete_versioned_delete_bitmap_kvs(rs.tablet_id(), rs.rowset_id_v2()) != 0) {
+                if (delete_versioned_delete_bitmap_kvs(rs.partition_id(), rs.tablet_id(),
+                                                       rs.rowset_id_v2()) != 0) {
                     LOG(WARNING) << "failed to delete versioned delete bitmap kv, rs="
                                  << rs.ShortDebugString();
                     return;

cloud/src/recycler/recycler.h

@@ -486,7 +486,15 @@ class InstanceRecycler {
     int delete_rowset_data(const std::string& resource_id, int64_t tablet_id,
                            const std::string& rowset_id);
 
-    int delete_versioned_delete_bitmap_kvs(int64_t tablet_id, const std::string& rowset_id);
+    bool is_tablet_recycled(int64_t tablet_id);
+
+    // Return 1 if the versioned delete bitmap should be deleted,
+    // Return 0 if it can be skipped,
+    // negative on error.
+    int should_delete_versioned_delete_bitmap_kvs(int64_t partition_id, int64_t tablet_id);
+
+    int delete_versioned_delete_bitmap_kvs(int64_t partition_id, int64_t tablet_id,
+                                           const std::string& rowset_id);
 
     int delete_delete_bitmap_kvs(int64_t tablet_id, const std::string& rowset_id);
 
@@ -672,6 +680,12 @@ class InstanceRecycler {
 
     TabletRecyclerMetricsContext tablet_metrics_context_;
     SegmentRecyclerMetricsContext segment_metrics_context_;
+
+    // Data tablets in the same partition have the same MoW setting. Cache both true and false so
+    // subsequent rowsets can avoid reading the tablet index and tablet meta. Row-binlog tablets
+    // must not populate this cache because their MoW flag is deliberately false.
+    std::mutex partition_mow_cache_mutex;
+    std::map<int64_t, bool> partition_mow_cache;
 };
 
 struct OperationLogReferenceInfo {

cloud/test/recycler_test.cpp

@@ -815,7 +815,7 @@ static void create_delete_bitmaps(Transaction* txn, int64_t tablet_id, std::stri
 }
 
 static int create_tablet(TxnKv* txn_kv, int64_t table_id, int64_t index_id, int64_t partition_id,
-                         int64_t tablet_id, bool is_mow = false, bool has_sequence_col = false) {
+                         int64_t tablet_id, bool is_mow = false) {
     std::unique_ptr<Transaction> txn;
     if (txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) {
         return -1;
@@ -1612,7 +1612,7 @@ TEST(RecyclerTest, recycle_prepare_rowset_aborts_before_delete) {
     constexpr int64_t index_id = 10001;
     constexpr int64_t tablet_id = 10002;
     constexpr int64_t partition_id = 10003;
-    ASSERT_EQ(create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id), 0);
+    ASSERT_EQ(create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id, true), 0);
     ASSERT_EQ(create_prepared_txn(txn_kv.get(), txn_db_id, tablet_id, txn_id), 0);
 
     doris::TabletSchemaCloudPB schema;
@@ -1782,7 +1782,7 @@ TEST(RecyclerTest, recycle_prepare_compaction_job_aborts_before_delete) {
     constexpr int64_t index_id = 11001;
     constexpr int64_t partition_id = 11002;
     constexpr int64_t tablet_id = 11003;
-    ASSERT_EQ(create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id), 0);
+    ASSERT_EQ(create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id, true), 0);
 
     TabletIndexPB tablet_idx;
     ASSERT_EQ(get_tablet_idx(txn_kv.get(), instance_id, tablet_id, tablet_idx), 0);
@@ -1864,9 +1864,12 @@ TEST(RecyclerTest, recycle_prepare_schema_change_job_aborts_before_delete) {
     constexpr int64_t partition_id = 12003;
     constexpr int64_t base_tablet_id = 12004;
     constexpr int64_t new_tablet_id = 12005;
-    ASSERT_EQ(create_tablet(txn_kv.get(), table_id, base_index_id, partition_id, base_tablet_id),
+    ASSERT_EQ(create_tablet(txn_kv.get(), table_id, base_index_id, partition_id, base_tablet_id,
+                            true),
               0);
-    ASSERT_EQ(create_tablet(txn_kv.get(), table_id, new_index_id, partition_id, new_tablet_id), 0);
+    ASSERT_EQ(
+            create_tablet(txn_kv.get(), table_id, new_index_id, partition_id, new_tablet_id, true),
+            0);
 
     TabletIndexPB base_tablet_idx;
     TabletIndexPB new_tablet_idx;
@@ -1886,6 +1889,7 @@ TEST(RecyclerTest, recycle_prepare_schema_change_job_aborts_before_delete) {
     doris::TabletMetaCloudPB new_tablet_meta;
     new_tablet_meta.set_tablet_id(new_tablet_id);
     new_tablet_meta.set_tablet_state(doris::TabletStatePB::PB_NOTREADY);
+    new_tablet_meta.set_enable_unique_key_merge_on_write(true);
     std::unique_ptr<Transaction> txn;
     ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
     txn->put(meta_tablet_key({instance_id, table_id, new_index_id, partition_id, new_tablet_id}),
@@ -7323,6 +7327,75 @@ TEST(RecyclerTest, delete_rowset_data_without_delete_bitmap_meta) {
     EXPECT_EQ(deleted_paths[0], segment_path(rowset.tablet_id(), rowset.rowset_id_v2(), 0));
 }
 
+TEST(RecyclerTest, delete_versioned_delete_bitmap_kvs_caches_partition_mow_state) {
+    auto txn_kv = std::make_shared<MemTxnKv>();
+    ASSERT_EQ(txn_kv->init(), 0);
+
+    constexpr int64_t table_id = 30001;
+    constexpr int64_t index_id = 30002;
+    constexpr int64_t partition_id = 30003;
+    constexpr int64_t tablet_id = 30004;
+    InstanceRecycler recycler(txn_kv, create_recycler_test_instance("versioned_dbm_cache"),
+                              thread_group, std::make_shared<TxnLazyCommitter>(txn_kv));
+    ASSERT_EQ(recycler.init(), 0);
+    auto accessor = recycler.accessor_map_.begin()->second;
+    ASSERT_EQ(create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id, true), 0);
+
+    ASSERT_EQ(create_delete_bitmaps_v2(txn_kv.get(), accessor.get(), tablet_id, "cache_rowset_1"),
+              0);
+    const int64_t get_count_before_first_delete = txn_kv->get_count_;
+    ASSERT_EQ(
+            recycler.delete_versioned_delete_bitmap_kvs(partition_id, tablet_id, "cache_rowset_1"),
+            0);
+    EXPECT_EQ(txn_kv->get_count_, get_count_before_first_delete + 2);
+    check_delete_bitmap_keys_size(txn_kv.get(), tablet_id, 0);
+
+    ASSERT_EQ(create_delete_bitmaps_v2(txn_kv.get(), accessor.get(), tablet_id, "cache_rowset_2"),
+              0);
+    const int64_t get_count_before_cached_delete = txn_kv->get_count_;
+    ASSERT_EQ(
+            recycler.delete_versioned_delete_bitmap_kvs(partition_id, tablet_id, "cache_rowset_2"),
+            0);
+    EXPECT_EQ(txn_kv->get_count_, get_count_before_cached_delete);
+    check_delete_bitmap_keys_size(txn_kv.get(), tablet_id, 0);
+}
+
+TEST(RecyclerTest, delete_versioned_delete_bitmap_kvs_only_deletes_mow_tablet) {
+    auto txn_kv = std::make_shared<MemTxnKv>();
+    ASSERT_EQ(txn_kv->init(), 0);
+
+    constexpr int64_t table_id = 30101;
+    constexpr int64_t index_id = 30102;
+    constexpr int64_t mow_partition_id = 30103;
+    constexpr int64_t mow_tablet_id = 30104;
+    constexpr int64_t non_mow_partition_id = 30105;
+    constexpr int64_t non_mow_tablet_id = 30106;
+    InstanceRecycler recycler(txn_kv, create_recycler_test_instance("versioned_dbm_mow"),
+                              thread_group, std::make_shared<TxnLazyCommitter>(txn_kv));
+    ASSERT_EQ(recycler.init(), 0);
+    auto accessor = recycler.accessor_map_.begin()->second;
+    ASSERT_EQ(
+            create_tablet(txn_kv.get(), table_id, index_id, mow_partition_id, mow_tablet_id, true),
+            0);
+    ASSERT_EQ(create_tablet(txn_kv.get(), table_id, index_id, non_mow_partition_id,
+                            non_mow_tablet_id, false),
+              0);
+    ASSERT_EQ(create_delete_bitmaps_v2(txn_kv.get(), accessor.get(), mow_tablet_id, "mow_rowset"),
+              0);
+    ASSERT_EQ(create_delete_bitmaps_v2(txn_kv.get(), accessor.get(), non_mow_tablet_id,
+                                       "non_mow_rowset"),
+              0);
+
+    ASSERT_EQ(recycler.delete_versioned_delete_bitmap_kvs(mow_partition_id, mow_tablet_id,
+                                                          "mow_rowset"),
+              0);
+    ASSERT_EQ(recycler.delete_versioned_delete_bitmap_kvs(non_mow_partition_id, non_mow_tablet_id,
+                                                          "non_mow_rowset"),
+              0);
+
+    check_delete_bitmap_keys_size(txn_kv.get(), mow_tablet_id, 0);
+    check_delete_bitmap_keys_size(txn_kv.get(), non_mow_tablet_id, 1);
+}
 TEST(RecyclerTest, delete_rowset_data_packed_file_single_rowset) {
     auto txn_kv = std::make_shared<MemTxnKv>();
     ASSERT_EQ(txn_kv->init(), 0);