[feat](snapshot) calculate snapshot data size (#59168)

Author: mymeiyi

cloud/src/meta-store/keys.cpp

@@ -1113,6 +1113,40 @@ bool decode_meta_tablet_key(std::string_view* in, int64_t* tablet_id, Versionsta
  return true;
 }
 
+// Decode tablet inverted index key
+// Return true if decode successfully, otherwise false
+bool decode_tablet_inverted_index_key(std::string_view* in, int64_t* db_id, int64_t* table_id,
+ int64_t* index_id, int64_t* partition_id,
+ int64_t* tablet_id) {
+ // 0x03 "index" ${instance_id} "tablet_inverted" ${db_id} ${table_id} ${index_id} ${partition} ${tablet}
+ if (in->empty() || static_cast<uint8_t>((*in)[0]) != CLOUD_VERSIONED_KEY_SPACE03) {
+ return false;
+ }
+ in->remove_prefix(1);
+
+ std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>> out;
+ auto res = decode_key(in, &out);
+ if (res != 0 || out.size() != 8) {
+ return false;
+ }
+
+ try {
+ if (std::get<std::string>(std::get<0>(out[0])) != INDEX_KEY_PREFIX ||
+ std::get<std::string>(std::get<0>(out[2])) != TABLET_INVERTED_INDEX_KEY_INFIX) {
+ return false;
+ }
+ *db_id = std::get<int64_t>(std::get<0>(out[3]));
+ *table_id = std::get<int64_t>(std::get<0>(out[4]));
+ *index_id = std::get<int64_t>(std::get<0>(out[5]));
+ *partition_id = std::get<int64_t>(std::get<0>(out[6]));
+ *tablet_id = std::get<int64_t>(std::get<0>(out[7]));
+ } catch (const std::bad_variant_access& e) {
+ return false;
+ }
+
+ return true;
+}
+
 bool decode_snapshot_ref_key(std::string_view* in, std::string* instance_id,
  Versionstamp* timestamp, std::string* ref_instance_id) {
  // Key format: 0x03 + encode_bytes("snapshot") + encode_bytes(instance_id) +

cloud/src/meta-store/keys.h

@@ -597,6 +597,11 @@ bool decode_meta_schema_key(std::string_view* in, int64_t* index_id, int64_t* sc
 // Return true if decode successfully, otherwise false
 bool decode_meta_tablet_key(std::string_view* in, int64_t* tablet_id, Versionstamp* timestamp);
 
+// Decode tablet inverted index key
+// Return true if decode successfully, otherwise false
+bool decode_tablet_inverted_index_key(std::string_view* in, int64_t* db_id, int64_t* table_id,
+ int64_t* index_id, int64_t* partition_id, int64_t* tablet_id);
+
 // Decode snapshot reference key
 // Return true if decode successfully, otherwise false
 bool decode_snapshot_ref_key(std::string_view* in, std::string* instance_id,

cloud/src/recycler/recycler.h

@@ -552,6 +552,12 @@ class InstanceRecycler {
  SegmentRecyclerMetricsContext segment_metrics_context_;
 };
 
+struct OperationLogReferenceInfo {
+ bool referenced_by_instance = false;
+ bool referenced_by_snapshot = false;
+ Versionstamp referenced_snapshot_timestamp;
+};
+
 // Helper class to check if operation logs can be recycled based on snapshots and versionstamps
 class OperationLogRecycleChecker {
 public:
@@ -563,10 +569,15 @@ class OperationLogRecycleChecker {
  int init();
 
  // Check if an operation log can be recycled
- bool can_recycle(const Versionstamp& log_versionstamp, int64_t log_min_timestamp) const;
+ bool can_recycle(const Versionstamp& log_versionstamp, int64_t log_min_timestamp,
+ OperationLogReferenceInfo* reference_info) const;
 
  Versionstamp max_versionstamp() const { return max_versionstamp_; }
 
+ const std::vector<std::pair<SnapshotPB, Versionstamp>>& get_snapshots() const {
+ return snapshots_;
+ }
+
 private:
  std::string_view instance_id_;
  TxnKv* txn_kv_;
@@ -577,4 +588,33 @@ class OperationLogRecycleChecker {
  std::vector<std::pair<SnapshotPB, Versionstamp>> snapshots_;
 };
 
+class SnapshotDataSizeCalculator {
+public:
+ SnapshotDataSizeCalculator(std::string_view instance_id, std::shared_ptr<TxnKv> txn_kv)
+ : instance_id_(instance_id), txn_kv_(std::move(txn_kv)) {}
+
+ void init(const std::vector<std::pair<SnapshotPB, Versionstamp>>& snapshots);
+
+ int calculate_operation_log_data_size(const std::string_view& log_key,
+ OperationLogPB& operation_log,
+ OperationLogReferenceInfo& reference_info);
+
+ int save_snapshot_data_size_with_retry();
+
+private:
+ int get_all_index_partitions(int64_t db_id, int64_t table_id, int64_t index_id,
+ std::vector<int64_t>* partition_ids);
+ int get_index_partition_data_size(int64_t db_id, int64_t table_id, int64_t index_id,
+ int64_t partition_id, int64_t* data_size);
+ int save_operation_log(const std::string_view& log_key, OperationLogPB& operation_log);
+ int save_snapshot_data_size();
+
+ std::string_view instance_id_;
+ std::shared_ptr<TxnKv> txn_kv_;
+
+ int64_t instance_retained_data_size_ = 0;
+ std::map<Versionstamp, int64_t> retained_data_size_;
+ std::set<std::string> calculated_partitions_;
+};
+
 } // namespace doris::cloud

cloud/src/recycler/recycler_operation_log.cpp

@@ -82,7 +82,8 @@ int OperationLogRecycleChecker::init() {
  snapshots_.clear();
  snapshot_indexes_.clear();
  MetaReader reader(instance_id_);
- err = reader.get_snapshots(txn.get(), &snapshots_);
+ std::vector<std::pair<SnapshotPB, Versionstamp>> snapshots;
+ err = reader.get_snapshots(txn.get(), &snapshots);
  if (err != TxnErrorCode::TXN_OK) {
  LOG_WARNING("failed to get snapshots").tag("err", err);
  return -1;
@@ -96,16 +97,22 @@ int OperationLogRecycleChecker::init() {
  }
 
  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));
+ for (size_t i = 0; i < snapshots.size(); ++i) {
+ auto&& [snapshot, versionstamp] = snapshots[i];
+ if (snapshot.status() == SnapshotStatus::SNAPSHOT_ABORTED ||
+ snapshot.status() == SnapshotStatus::SNAPSHOT_RECYCLED) {
+ continue;
+ }
+ snapshot_indexes_.insert(std::make_pair(versionstamp, snapshots_.size()));
+ snapshots_.push_back(std::make_pair(std::move(snapshot), versionstamp));
  }
 
  return 0;
 }
 
 bool OperationLogRecycleChecker::can_recycle(const Versionstamp& log_versionstamp,
- int64_t log_min_timestamp) const {
+ int64_t log_min_timestamp,
+ OperationLogReferenceInfo* reference_info) const {
  Versionstamp log_min_read_timestamp(log_min_timestamp, 0);
  if (log_versionstamp > max_versionstamp_) {
  // Not recycleable.
@@ -114,12 +121,15 @@ bool OperationLogRecycleChecker::can_recycle(const Versionstamp& log_versionstam
 
  // Do not recycle operation logs referenced by active snapshots.
  if (log_min_read_timestamp < source_snapshot_versionstamp_) {
+ reference_info->referenced_by_instance = true;
  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)
+ reference_info->referenced_by_snapshot = true;
+ reference_info->referenced_snapshot_timestamp = snapshots_[it->second].second;
  return false;
  }
 
@@ -677,6 +687,8 @@ int InstanceRecycler::recycle_operation_logs() {
  LOG_WARNING("failed to initialize recycle checker").tag("error_code", init_res);
  return init_res;
  }
+ SnapshotDataSizeCalculator calculator(instance_id_, txn_kv_);
+ calculator.init(recycle_checker.get_snapshots());
 
  auto scan_and_recycle_operation_log = [&](const std::string_view& key,
  const std::vector<std::string>& raw_keys,
@@ -690,7 +702,9 @@ int InstanceRecycler::recycle_operation_logs() {
  }
 
  size_t value_size = operation_log.ByteSizeLong();
- if (recycle_checker.can_recycle(log_versionstamp, operation_log.min_timestamp())) {
+ OperationLogReferenceInfo reference_info;
+ if (recycle_checker.can_recycle(log_versionstamp, operation_log.min_timestamp(),
+ &reference_info)) {
  AnnotateTag tag("log_key", hex(key));
  int res = recycle_operation_log(log_versionstamp, raw_keys, std::move(operation_log));
  if (res != 0) {
@@ -700,6 +714,13 @@ int InstanceRecycler::recycle_operation_logs() {
 
  recycled_operation_logs++;
  recycled_operation_log_data_size += value_size;
+ } else {
+ int res = calculator.calculate_operation_log_data_size(key, operation_log,
+ reference_info);
+ if (res != 0) {
+ LOG_WARNING("failed to calculate operation log data size").tag("error_code", res);
+ return res;
+ }
  }
 
  total_operation_logs++;
@@ -769,6 +790,11 @@ int InstanceRecycler::recycle_operation_logs() {
  .tag("error_code", iter->error_code());
  return -1;
  }
+ int res = calculator.save_snapshot_data_size_with_retry();
+ if (res != 0) {
+ LOG_WARNING("failed to save snapshot data size").tag("error_code", res);
+ return res;
+ }
  return 0;
 }