[feat](cloud) Recycle operation logs according the snapshot keys (#55678)

w41ter · web-flow · commit a56419bf8980 · 2025-09-08T14:46:37.000+08:00
diff --git a/cloud/src/meta-service/meta_service_job.cpp b/cloud/src/meta-service/meta_service_job.cpp
@@ -1706,6 +1706,7 @@ void process_schema_change_job(MetaServiceCode& code, std::string& msg, std::str
         index_size_remove_rowsets += rs.index_disk_size();
         segment_size_remove_rowsets += rs.data_disk_size();
 
+        int64_t start_version = rs.start_version(), end_version = rs.end_version();
         auto recycle_key = recycle_rowset_key({instance_id, new_tablet_id, rs.rowset_id_v2()});
         RecycleRowsetPB recycle_rowset;
         recycle_rowset.set_creation_time(now);
@@ -1717,8 +1718,8 @@ void process_schema_change_job(MetaServiceCode& code, std::string& msg, std::str
             auto recycle_val = recycle_rowset.SerializeAsString();
             txn->put(recycle_key, recycle_val);
         }
-        INSTANCE_LOG(INFO) << "put recycle rowset, new_tablet_id=" << new_tablet_id
-                           << " key=" << hex(recycle_key);
+        INSTANCE_LOG(INFO) << "put recycle rowset, new_tablet_id=" << new_tablet_id << " version=["
+                           << start_version << "-" << end_version << "] key=" << hex(recycle_key);
     };
 
     if (!is_versioned_read) {
diff --git a/cloud/src/meta-store/mem_txn_kv.h b/cloud/src/meta-store/mem_txn_kv.h
@@ -64,6 +64,12 @@ class MemTxnKv : public TxnKv, public std::enable_shared_from_this<MemTxnKv> {
         return mem_kv_.size();
     }
 
+    void update_commit_version(int64_t version) {
+        std::lock_guard<std::mutex> l(lock_);
+        committed_version_ = std::max(committed_version_, version);
+        read_version_ = std::max(committed_version_, read_version_);
+    }
+
     int64_t get_bytes_ {};
     int64_t put_bytes_ {};
     int64_t del_bytes_ {};
diff --git a/cloud/src/meta-store/meta_reader.cpp b/cloud/src/meta-store/meta_reader.cpp
@@ -24,6 +24,7 @@
 
 #include "common/logging.h"
 #include "common/util.h"
+#include "meta-store/codec.h"
 #include "meta-store/document_message.h"
 #include "meta-store/document_message_get_range.h"
 #include "meta-store/keys.h"
@@ -904,4 +905,57 @@ TxnErrorCode MetaReader::is_partition_exists(Transaction* txn, int64_t partition
     return TxnErrorCode::TXN_OK;
 }
 
+TxnErrorCode MetaReader::get_snapshots(
+        Transaction* txn, std::vector<std::pair<SnapshotPB, Versionstamp>>* snapshots) {
+    std::string snapshot_key = versioned::snapshot_full_key({instance_id_});
+    std::string snapshot_start_key = encode_versioned_key(snapshot_key, Versionstamp::min());
+    std::string snapshot_end_key = encode_versioned_key(snapshot_key, Versionstamp::max());
+
+    FullRangeGetOptions range_options;
+    range_options.prefetch = true;
+    auto it = txn->full_range_get(snapshot_start_key, snapshot_end_key, range_options);
+    for (auto&& kvp = it->next(); kvp.has_value(); kvp = it->next()) {
+        auto&& [key, snapshot_value] = *kvp;
+
+        Versionstamp version;
+        std::string_view key_view(key);
+        if (decode_tailing_versionstamp_end(&key_view) ||
+            decode_tailing_versionstamp(&key_view, &version)) {
+            LOG_WARNING("failed to decode versionstamp from snapshot full key")
+                    .tag("instance_id", instance_id_)
+                    .tag("key", hex(key));
+            return TxnErrorCode::TXN_INVALID_DATA;
+        }
+
+        SnapshotPB snapshot;
+        if (!snapshot.ParseFromArray(snapshot_value.data(), snapshot_value.size())) {
+            LOG_ERROR("Failed to parse SnapshotPB")
+                    .tag("instance_id", instance_id_)
+                    .tag("key", hex(key));
+            return TxnErrorCode::TXN_INVALID_DATA;
+        }
+
+        snapshots->emplace_back(std::move(snapshot), version);
+    }
+
+    if (!it->is_valid()) {
+        LOG_ERROR("failed to get snapshots")
+                .tag("instance_id", instance_id_)
+                .tag("error_code", it->error_code());
+        return it->error_code();
+    }
+
+    return TxnErrorCode::TXN_OK;
+}
+
+TxnErrorCode MetaReader::get_snapshots(
+        std::vector<std::pair<SnapshotPB, Versionstamp>>* snapshots) {
+    std::unique_ptr<Transaction> txn;
+    TxnErrorCode err = txn_kv_->create_txn(&txn);
+    if (err != TxnErrorCode::TXN_OK) {
+        return err;
+    }
+    return get_snapshots(txn.get(), snapshots);
+}
+
 } // namespace doris::cloud
diff --git a/cloud/src/meta-store/meta_reader.h b/cloud/src/meta-store/meta_reader.h
@@ -224,6 +224,11 @@ class MetaReader {
     TxnErrorCode is_partition_exists(int64_t partition_id, bool snapshot = false);
     TxnErrorCode is_partition_exists(Transaction* txn, int64_t partition_id, bool snapshot = false);
 
+    // Get the snapshots.
+    TxnErrorCode get_snapshots(Transaction* txn,
+                               std::vector<std::pair<SnapshotPB, Versionstamp>>* snapshots);
+    TxnErrorCode get_snapshots(std::vector<std::pair<SnapshotPB, Versionstamp>>* snapshots);
+
 private:
     const std::string_view instance_id_;
     const Versionstamp snapshot_version_;
diff --git a/cloud/src/recycler/recycler.h b/cloud/src/recycler/recycler.h
@@ -439,4 +439,26 @@ class InstanceRecycler {
     SegmentRecyclerMetricsContext segment_metrics_context_;
 };
 
+// Helper class to check if operation logs can be recycled based on snapshots and versionstamps
+class OperationLogRecycleChecker {
+public:
+    OperationLogRecycleChecker(std::string_view instance_id, TxnKv* txn_kv)
+            : instance_id_(instance_id), txn_kv_(txn_kv) {}
+
+    // Initialize the checker by loading snapshots and setting max version stamp
+    int init();
+
+    // Check if an operation log can be recycled
+    bool can_recycle(const Versionstamp& log_versionstamp, int64_t log_min_timestamp) const;
+
+    Versionstamp max_versionstamp() const { return max_versionstamp_; }
+
+private:
+    std::string_view instance_id_;
+    TxnKv* txn_kv_;
+    Versionstamp max_versionstamp_;
+    std::map<Versionstamp, size_t> snapshot_indexes_;
+    std::vector<std::pair<SnapshotPB, Versionstamp>> snapshots_;
+};
+
 } // namespace doris::cloud
diff --git a/cloud/src/recycler/recycler_operation_log.cpp b/cloud/src/recycler/recycler_operation_log.cpp
@@ -26,6 +26,7 @@
 #include <algorithm>
 #include <cstddef>
 #include <cstdint>
+#include <ranges>
 #include <string>
 #include <string_view>
 #include <utility>
@@ -53,6 +54,56 @@ namespace doris::cloud {
 
 using namespace std::chrono;
 
+int OperationLogRecycleChecker::init() {
+    std::unique_ptr<Transaction> txn;
+    TxnErrorCode err = txn_kv_->create_txn(&txn);
+    if (err != TxnErrorCode::TXN_OK) {
+        LOG_WARNING("failed to create txn").tag("err", err);
+        return -1;
+    }
+
+    snapshots_.clear();
+    snapshot_indexes_.clear();
+    MetaReader reader(instance_id_);
+    err = reader.get_snapshots(txn.get(), &snapshots_);
+    if (err != TxnErrorCode::TXN_OK) {
+        LOG_WARNING("failed to get snapshots").tag("err", err);
+        return -1;
+    }
+
+    int64_t read_version = -1;
+    err = txn->get_read_version(&read_version);
+    if (err != TxnErrorCode::TXN_OK) {
+        LOG_WARNING("failed to get the read version").tag("err", err);
+        return -1;
+    }
+
+    max_versionstamp_ = Versionstamp(read_version, 0);
+    for (size_t i = 0; i < snapshots_.size(); ++i) {
+        auto&& [snapshot, versionstamp] = snapshots_[i];
+        snapshot_indexes_.insert(std::make_pair(versionstamp, i));
+    }
+
+    return 0;
+}
+
+bool OperationLogRecycleChecker::can_recycle(const Versionstamp& log_versionstamp,
+                                             int64_t log_min_timestamp) const {
+    Versionstamp log_min_read_timestamp(log_min_timestamp, 0);
+    if (log_versionstamp > max_versionstamp_) {
+        // Not recycleable.
+        return false;
+    }
+
+    auto it = snapshot_indexes_.lower_bound(log_min_read_timestamp);
+    if (it != snapshot_indexes_.end() && snapshots_[it->second].second < log_versionstamp) {
+        // in [log_min_read_timestmap, log_versionstamp)
+        return false;
+    }
+
+    return true;
+}
+
 // A recycler for operation logs.
 class OperationLogRecycler {
 public:
@@ -144,6 +195,9 @@ int OperationLogRecycler::recycle_drop_index_log(const DropIndexLogPB& drop_inde
             return -1;
         }
         std::string recycle_key = recycle_index_key({instance_id_, index_id});
+        LOG_INFO("put recycle index key")
+                .tag("recycle_key", hex(recycle_key))
+                .tag("index_id", index_id);
         kvs_.emplace_back(std::move(recycle_key), std::move(recycle_index_value));
     }
     return 0;
@@ -557,11 +611,18 @@ int InstanceRecycler::recycle_operation_logs() {
                 .tag("recycled_operation_log_data_size", recycled_operation_log_data_size);
     };
 
+    OperationLogRecycleChecker recycle_checker(instance_id_, txn_kv_.get());
+    int init_res = recycle_checker.init();
+    if (init_res != 0) {
+        LOG_WARNING("failed to initialize recycle checker").tag("error_code", init_res);
+        return init_res;
+    }
+
     auto scan_and_recycle_operation_log = [&](const std::string_view& key,
                                               const std::string_view& value) {
         std::string_view log_key(key);
-        Versionstamp versionstamp;
-        if (!decode_versioned_key(&log_key, &versionstamp)) {
+        Versionstamp log_versionstamp;
+        if (!decode_versioned_key(&log_key, &log_versionstamp)) {
             LOG_WARNING("failed to decode versionstamp from operation log key")
                     .tag("key", hex(key));
             return -1;
@@ -577,15 +638,15 @@ int InstanceRecycler::recycle_operation_logs() {
         if (!operation_log.has_min_timestamp()) {
             LOG_WARNING("operation log has not set the min_timestamp")
                     .tag("key", hex(key))
-                    .tag("version", versionstamp.version())
-                    .tag("order", versionstamp.order())
+                    .tag("version", log_versionstamp.version())
+                    .tag("order", log_versionstamp.order())
                     .tag("log", operation_log.ShortDebugString());
+            return 0;
         }
 
-        bool need_recycle = true; // Always recycle operation logs for now
-        if (need_recycle) {
-            AnnotateTag tag("log_key", hex(log_key));
-            int res = recycle_operation_log(versionstamp, std::move(operation_log));
+        if (recycle_checker.can_recycle(log_versionstamp, operation_log.min_timestamp())) {
+            AnnotateTag tag("log_key", hex(key));
+            int res = recycle_operation_log(log_versionstamp, std::move(operation_log));
             if (res != 0) {
                 LOG_WARNING("failed to recycle operation log").tag("error_code", res);
                 return res;
diff --git a/cloud/test/meta_reader_test.cpp b/cloud/test/meta_reader_test.cpp
@@ -2050,3 +2050,58 @@ TEST(MetaReaderTest, BatchGetTabletCompactStats) {
         ASSERT_EQ(versionstamps.size(), tablet_ids.size());
     }
 }
+
+TEST(MetaReaderTest, GetSnapshots) {
+    auto txn_kv = std::make_shared<MemTxnKv>();
+    ASSERT_EQ(txn_kv->init(), 0);
+
+    std::string instance_id = "test_instance";
+
+    {
+        // Test empty result when no snapshots exist
+        MetaReader meta_reader(instance_id, txn_kv.get());
+        std::vector<std::pair<SnapshotPB, Versionstamp>> snapshots;
+        TxnErrorCode err = meta_reader.get_snapshots(&snapshots);
+        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
+        ASSERT_TRUE(snapshots.empty());
+    }
+
+    // Create some snapshots
+    std::vector<Versionstamp> expected_versionstamps;
+    {
+        // Create multiple snapshots with different timestamps
+        for (int i = 1; i <= 3; ++i) {
+            std::unique_ptr<Transaction> txn;
+            ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+
+            std::string snapshot_key = versioned::snapshot_full_key({instance_id});
+            SnapshotPB snapshot_pb;
+            snapshot_pb.set_label(fmt::format("snapshot_{}", i));
+            snapshot_pb.set_instance_id(instance_id);
+
+            std::string snapshot_value = snapshot_pb.SerializeAsString();
+            versioned_put(txn.get(), snapshot_key, snapshot_value);
+            ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+        }
+    }
+
+    {
+        // Test getting snapshots with created transaction
+        MetaReader meta_reader(instance_id, txn_kv.get());
+        std::vector<std::pair<SnapshotPB, Versionstamp>> snapshots;
+        TxnErrorCode err = meta_reader.get_snapshots(&snapshots);
+        ASSERT_EQ(err, TxnErrorCode::TXN_OK);
+        ASSERT_EQ(snapshots.size(), 3);
+
+        // Verify snapshots are returned (order may vary due to versionstamp ordering)
+        std::set<std::string> snapshot_ids;
+        for (const auto& [snapshot_pb, versionstamp] : snapshots) {
+            snapshot_ids.insert(snapshot_pb.label());
+            ASSERT_EQ(snapshot_pb.instance_id(), instance_id);
+        }
+        ASSERT_EQ(snapshot_ids.size(), 3);
+        ASSERT_TRUE(snapshot_ids.count("snapshot_1"));
+        ASSERT_TRUE(snapshot_ids.count("snapshot_2"));
+        ASSERT_TRUE(snapshot_ids.count("snapshot_3"));
+    }
+}
diff --git a/cloud/test/recycler_operation_log_test.cpp b/cloud/test/recycler_operation_log_test.cpp
