[fix](checker) Avoid false-positive leaked delete bitmaps for unexpired job tmp rowsets

cloud/src/recycler/checker.cpp

@@ -1227,6 +1227,11 @@ int InstanceChecker::do_delete_bitmap_inverted_check() {
  std::unordered_set<std::string> pending_delete_bitmaps {};
  } tablet_rowsets_cache {};
 
+ std::unordered_map<int64_t, std::unordered_set<std::string>> unexpired_tmp_rowsets;
+ if (int ret = collect_unexpired_job_tmp_rowsets(unexpired_tmp_rowsets); ret < 0) {
+ return ret;
+ }
+
  std::unique_ptr<RangeGetIterator> it;
  auto begin = meta_delete_bitmap_key({instance_id_, 0, "", 0, 0});
  auto end =
@@ -1267,6 +1272,9 @@ int InstanceChecker::do_delete_bitmap_inverted_check() {
 
  if (tablet_rowsets_cache.tablet_id == -1 ||
  tablet_rowsets_cache.tablet_id != tablet_id) {
+ if (tablet_rowsets_cache.tablet_id != -1) {
+ unexpired_tmp_rowsets.erase(tablet_rowsets_cache.tablet_id);
+ }
  TabletMetaCloudPB tablet_meta;
  int ret = get_tablet_meta(txn_kv_.get(), instance_id_, tablet_id, tablet_meta);
  if (ret < 0) {
@@ -1319,8 +1327,15 @@ int InstanceChecker::do_delete_bitmap_inverted_check() {
  continue;
  }
 
+ bool belongs_to_unexpired_tmp_rowset = false;
+ auto tmp_rowsets_it = unexpired_tmp_rowsets.find(tablet_id);
+ if (tmp_rowsets_it != unexpired_tmp_rowsets.end()) {
+ belongs_to_unexpired_tmp_rowset = tmp_rowsets_it->second.contains(rowset_id);
+ }
+
  if (!tablet_rowsets_cache.rowsets.contains(rowset_id) &&
- !tablet_rowsets_cache.pending_delete_bitmaps.contains(std::string(k))) {
+ !tablet_rowsets_cache.pending_delete_bitmaps.contains(std::string(k)) &&
+ !belongs_to_unexpired_tmp_rowset) {
  TEST_SYNC_POINT_CALLBACK(
  "InstanceChecker::do_delete_bitmap_inverted_check.get_leaked_delete_bitmap",
  &tablet_id, &rowset_id, &version, &segment_id);
@@ -1338,6 +1353,116 @@ int InstanceChecker::do_delete_bitmap_inverted_check() {
  return (leaked_delete_bitmaps > 0 || abnormal_delete_bitmaps > 0) ? 1 : 0;
 }
 
+int InstanceChecker::collect_unexpired_job_tmp_rowsets(
+ std::unordered_map<int64_t, std::unordered_set<std::string>>& tmp_rowsets) {
+ static constexpr int64_t max_unexpired_tmp_rowsets = 1000;
+ auto begin = meta_rowset_tmp_key({instance_id_, 0, 0});
+ auto end = meta_rowset_tmp_key({instance_id_, INT64_MAX, 0});
+ std::unique_ptr<RangeGetIterator> it;
+ int64_t num_scanned = 0;
+ int64_t num_non_job = 0;
+ int64_t num_skipped_non_job_txns = 0;
+ int64_t num_unexpired = 0;
+ int64_t num_expired = 0;
+ int64_t last_txn_id = -1;
+ int64_t current_time = duration_cast<seconds>(system_clock::now().time_since_epoch()).count();
+
+ while (it == nullptr /* may be not init */ || (it->more() && !stopped())) {
+ 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 tmp rowset kv, err=" << err;
+ return -1;
+ }
+ if (!it->has_next()) {
+ break;
+ }
+ while (it->has_next() && !stopped()) {
+ auto [k, v] = it->next();
+ ++num_scanned;
+
+ std::string_view k1 = k;
+ k1.remove_prefix(1);
+ std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>> out;
+ if (decode_key(&k1, &out) != 0 || out.size() < 5) {
+ LOG(WARNING) << "malformed tmp rowset key, key=" << hex(k);
+ return -1;
+ }
+ // 0x01 "meta" ${instance_id} "rowset_tmp" ${txn_id} ${tablet_id} -> RowsetMetaCloudPB
+ auto txn_id = std::get<int64_t>(std::get<0>(out[3]));
+ bool is_first_rowset_of_txn = last_txn_id != txn_id;
+ last_txn_id = txn_id;
+
+ doris::RowsetMetaCloudPB rowset;
+ if (!rowset.ParseFromArray(v.data(), v.size())) {
+ LOG(WARNING) << "malformed tmp rowset meta, key=" << hex(k);
+ return -1;
+ }
+ if (!rowset.has_job_id() || rowset.job_id().empty()) {
+ ++num_non_job;
+ if (is_first_rowset_of_txn) {
+ ++num_skipped_non_job_txns;
+ if (txn_id == INT64_MAX) {
+ begin = end;
+ } else {
+ begin = meta_rowset_tmp_key({instance_id_, txn_id + 1, 0});
+ }
+ it.reset();
+ break;
+ }
+ if (!it->has_next()) {
+ begin = k;
+ begin.push_back('\x00');
+ }
+ continue;
+ }
+
+ // Must use the same threshold as the recycler so that a delete bitmap is never
+ // reported as leaked while its tmp rowset is still alive from the recycler's view.
+ // `earlest_ts` is a local sentinel initialized to 0 on purpose: it keeps the value
+ // below any real expiration so the helper never updates the recycler's
+ // earliest-ts bvar (the checker must not touch the recycler's metrics).
+ int64_t earlest_ts = 0;
+ int64_t expiration =
+ calculate_tmp_rowset_expired_time(instance_id_, rowset, &earlest_ts);
+ if (current_time < expiration) {
+ tmp_rowsets[rowset.tablet_id()].insert(rowset.rowset_id_v2());
+ ++num_unexpired;
+ if (num_unexpired >= max_unexpired_tmp_rowsets) {
+ LOG(WARNING)
+ << "collect unexpired tmp rowsets for delete bitmap checker reached "
+ << "limit, remaining tmp rowsets will not be considered and may cause "
+ << "false positives, instance_id=" << instance_id_
+ << ", num_scanned=" << num_scanned << ", num_non_job=" << num_non_job
+ << ", num_skipped_non_job_txns=" << num_skipped_non_job_txns
+ << ", num_unexpired=" << num_unexpired
+ << ", num_expired=" << num_expired
+ << ", limit=" << max_unexpired_tmp_rowsets;
+ return 0;
+ }
+ } else {
+ ++num_expired;
+ }
+
+ if (!it->has_next()) {
+ begin = k;
+ begin.push_back('\x00');
+ }
+ }
+ }
+
+ LOG(INFO) << "collect unexpired tmp rowsets for delete bitmap checker finished, instance_id="
+ << instance_id_ << ", num_scanned=" << num_scanned << ", num_non_job=" << num_non_job
+ << ", num_skipped_non_job_txns=" << num_skipped_non_job_txns
+ << ", num_unexpired=" << num_unexpired << ", num_expired=" << num_expired;
+ return 0;
+}
+
 int InstanceChecker::get_pending_delete_bitmap_keys(
  int64_t tablet_id, std::unordered_set<std::string>& pending_delete_bitmaps) {
  std::unique_ptr<Transaction> txn;

cloud/src/recycler/checker.h

@@ -185,6 +185,8 @@ class InstanceChecker {
  int collect_tablet_rowsets(
  int64_t tablet_id,
  const std::function<void(const doris::RowsetMetaCloudPB&)>& collect_cb);
+ int collect_unexpired_job_tmp_rowsets(
+ std::unordered_map<int64_t, std::unordered_set<std::string>>& tmp_rowsets);
  int get_pending_delete_bitmap_keys(int64_t tablet_id,
  std::unordered_set<std::string>& pending_delete_bitmaps);
  int check_delete_bitmap_storage_optimize_v2(int64_t tablet_id, bool has_sequence_col,

cloud/src/recycler/recycler.h

@@ -57,6 +57,11 @@ class SimpleThreadPool;
 class RecyclerMetricsContext;
 class TabletRecyclerMetricsContext;
 class SegmentRecyclerMetricsContext;
+
+int64_t calculate_tmp_rowset_expired_time(
+ const std::string& instance_id_, const doris::RowsetMetaCloudPB& tmp_rowset_meta_pb,
+ int64_t* earlest_ts /* tmp_rowset earliest expiration ts */);
+
 struct RecyclerThreadPoolGroup {
  RecyclerThreadPoolGroup() = default;
  RecyclerThreadPoolGroup(std::shared_ptr<SimpleThreadPool> s3_producer_pool,

cloud/test/recycler_test.cpp

@@ -4850,6 +4850,131 @@ TEST(CheckerTest, delete_bitmap_inverted_check_normal) {
  ASSERT_EQ(checker.do_delete_bitmap_inverted_check(), 0);
 }
 
+TEST(CheckerTest, delete_bitmap_inverted_check_unexpired_tmp_rowset) {
+ auto retention_seconds = config::retention_seconds;
+ auto force_immediate_recycle = config::force_immediate_recycle;
+ DORIS_CLOUD_DEFER {
+ config::retention_seconds = retention_seconds;
+ config::force_immediate_recycle = force_immediate_recycle;
+ };
+ config::retention_seconds = 3600;
+ config::force_immediate_recycle = false;
+
+ 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(txn_kv, instance_id);
+ ASSERT_EQ(checker.init(instance), 0);
+ auto accessor = checker.accessor_map_.begin()->second;
+
+ constexpr int64_t table_id = 10000;
+ constexpr int64_t index_id = 10001;
+ constexpr int64_t partition_id = 10002;
+ constexpr int64_t tablet_id = 600011;
+ ASSERT_EQ(0, create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id, true));
+
+ doris::TabletSchemaCloudPB schema;
+ schema.set_schema_version(1);
+ auto tmp_rowset = create_rowset("1", tablet_id, index_id, 1, schema, 100001);
+ tmp_rowset.set_creation_time(current_time);
+ tmp_rowset.set_txn_expiration(current_time);
+ tmp_rowset.set_start_version(10);
+ tmp_rowset.set_end_version(10);
+ tmp_rowset.set_job_id("compaction-job-1");
+ ASSERT_EQ(0, create_tmp_rowset(txn_kv.get(), accessor.get(), tmp_rowset, false));
+ ASSERT_EQ(0, create_delete_bitmaps_v1(txn_kv.get(), tablet_id, tmp_rowset.rowset_id_v2()));
+
+ ASSERT_EQ(checker.do_delete_bitmap_inverted_check(), 0);
+}
+
+TEST(CheckerTest, delete_bitmap_inverted_check_unexpired_non_job_tmp_rowset) {
+ auto retention_seconds = config::retention_seconds;
+ auto force_immediate_recycle = config::force_immediate_recycle;
+ DORIS_CLOUD_DEFER {
+ config::retention_seconds = retention_seconds;
+ config::force_immediate_recycle = force_immediate_recycle;
+ };
+ config::retention_seconds = 3600;
+ config::force_immediate_recycle = false;
+
+ 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(txn_kv, instance_id);
+ ASSERT_EQ(checker.init(instance), 0);
+ auto accessor = checker.accessor_map_.begin()->second;
+
+ constexpr int64_t table_id = 10000;
+ constexpr int64_t index_id = 10001;
+ constexpr int64_t partition_id = 10002;
+ constexpr int64_t tablet_id = 600013;
+ ASSERT_EQ(0, create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id, true));
+
+ doris::TabletSchemaCloudPB schema;
+ schema.set_schema_version(1);
+ auto tmp_rowset = create_rowset("1", tablet_id, index_id, 1, schema, 100003);
+ tmp_rowset.set_creation_time(current_time);
+ tmp_rowset.set_txn_expiration(current_time);
+ tmp_rowset.set_start_version(10);
+ tmp_rowset.set_end_version(10);
+ ASSERT_EQ(0, create_tmp_rowset(txn_kv.get(), accessor.get(), tmp_rowset, false));
+ ASSERT_EQ(0, create_delete_bitmaps_v1(txn_kv.get(), tablet_id, tmp_rowset.rowset_id_v2()));
+
+ ASSERT_EQ(checker.do_delete_bitmap_inverted_check(), 1);
+}
+
+TEST(CheckerTest, delete_bitmap_inverted_check_expired_tmp_rowset) {
+ auto retention_seconds = config::retention_seconds;
+ auto force_immediate_recycle = config::force_immediate_recycle;
+ DORIS_CLOUD_DEFER {
+ config::retention_seconds = retention_seconds;
+ config::force_immediate_recycle = force_immediate_recycle;
+ };
+ config::retention_seconds = 3600;
+ config::force_immediate_recycle = false;
+
+ 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(txn_kv, instance_id);
+ ASSERT_EQ(checker.init(instance), 0);
+ auto accessor = checker.accessor_map_.begin()->second;
+
+ constexpr int64_t table_id = 10000;
+ constexpr int64_t index_id = 10001;
+ constexpr int64_t partition_id = 10002;
+ constexpr int64_t tablet_id = 600012;
+ ASSERT_EQ(0, create_tablet(txn_kv.get(), table_id, index_id, partition_id, tablet_id, true));
+
+ doris::TabletSchemaCloudPB schema;
+ schema.set_schema_version(1);
+ auto tmp_rowset = create_rowset("1", tablet_id, index_id, 1, schema, 100002);
+ tmp_rowset.set_creation_time(current_time - config::retention_seconds - 10);
+ tmp_rowset.set_txn_expiration(current_time - config::retention_seconds - 10);
+ tmp_rowset.set_start_version(10);
+ tmp_rowset.set_end_version(10);
+ tmp_rowset.set_job_id("compaction-job-2");
+ ASSERT_EQ(0, create_tmp_rowset(txn_kv.get(), accessor.get(), tmp_rowset, false));
+ ASSERT_EQ(0, create_delete_bitmaps_v1(txn_kv.get(), tablet_id, tmp_rowset.rowset_id_v2()));
+
+ ASSERT_EQ(checker.do_delete_bitmap_inverted_check(), 1);
+}
+
 TEST(CheckerTest, delete_bitmap_inverted_check_abnormal) {
  // abnormal case, some delete bitmaps arem leaked
  auto txn_kv = std::make_shared<MemTxnKv>();