From f6a7ab1bb4e83d28447d45e46a748f50b4ee9fd7 Mon Sep 17 00:00:00 2001 From: Yixuan Wang Date: Thu, 9 Jul 2026 22:37:20 +0800 Subject: [PATCH] 1 --- cloud/src/recycler/recycler.cpp | 370 ++++++++++++++++++++++++++++++-- cloud/src/recycler/recycler.h | 25 ++- cloud/test/recycler_test.cpp | 210 ++++++++++++++++++ 3 files changed, 583 insertions(+), 22 deletions(-) diff --git a/cloud/src/recycler/recycler.cpp b/cloud/src/recycler/recycler.cpp index bd1b9bff878cee..3950618da7d64d 100644 --- a/cloud/src/recycler/recycler.cpp +++ b/cloud/src/recycler/recycler.cpp @@ -42,6 +42,7 @@ #include #include #include +#include #include #include @@ -3223,6 +3224,231 @@ int InstanceRecycler::delete_rowset_data(const RowsetMetaCloudPB& rs_meta_pb) { return accessor->delete_files(file_paths); } +void InstanceRecycler::collect_packed_file_refs( + const doris::RowsetMetaCloudPB& rs_meta_pb, + std::unordered_map>* packed_file_refs) { + DCHECK(packed_file_refs != nullptr); + std::unordered_map expected_slice_counts; + expected_slice_counts.reserve(rs_meta_pb.packed_slice_locations_size()); + for (const auto& [_, index_pb] : rs_meta_pb.packed_slice_locations()) { + if (!index_pb.has_packed_file_path() || index_pb.packed_file_path().empty()) { + continue; + } + ++expected_slice_counts[index_pb.packed_file_path()]; + } + + for (auto& [packed_file_path, expected_slice_count] : expected_slice_counts) { + (*packed_file_refs)[packed_file_path].push_back( + PackedRowsetRef {.tablet_id = rs_meta_pb.tablet_id(), + .rowset_id = rs_meta_pb.rowset_id_v2(), + .expected_slice_count = expected_slice_count}); + } +} + +int InstanceRecycler::flush_packed_file_refs( + const std::unordered_map>& packed_file_refs) { + const int max_retry_times = std::max(1, config::decrement_packed_file_ref_counts_retry_times); + int ret = 0; + for (const auto& [packed_file_path, rowset_refs] : packed_file_refs) { + if (rowset_refs.empty()) { + continue; + } + + bool success = false; + for (int attempt = 1; attempt <= max_retry_times; ++attempt) { + std::unique_ptr txn; + TxnErrorCode err = txn_kv_->create_txn(&txn); + if (err != TxnErrorCode::TXN_OK) { + LOG_WARNING("failed to create txn when flushing packed file refs") + .tag("instance_id", instance_id_) + .tag("packed_file_path", packed_file_path) + .tag("attempt", attempt) + .tag("err", err); + ret = -1; + break; + } + + std::string packed_key = packed_file_key({instance_id_, packed_file_path}); + std::string packed_val; + err = txn->get(packed_key, &packed_val); + if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) { + LOG_WARNING("packed file info not found when flushing packed file refs") + .tag("instance_id", instance_id_) + .tag("packed_file_path", packed_file_path) + .tag("rowset_ref_count", rowset_refs.size()) + .tag("key", hex(packed_key)); + success = true; + break; + } + if (err != TxnErrorCode::TXN_OK) { + LOG_WARNING("failed to get packed file info when flushing packed file refs") + .tag("instance_id", instance_id_) + .tag("packed_file_path", packed_file_path) + .tag("attempt", attempt) + .tag("err", err); + ret = -1; + break; + } + + cloud::PackedFileInfoPB packed_info; + if (!packed_info.ParseFromString(packed_val)) { + LOG_WARNING("failed to parse packed file info when flushing packed file refs") + .tag("instance_id", instance_id_) + .tag("packed_file_path", packed_file_path) + .tag("attempt", attempt); + ret = -1; + break; + } + + struct RowsetRefKey { + int64_t tablet_id = 0; + std::string rowset_id; + + bool operator==(const RowsetRefKey& other) const { + return tablet_id == other.tablet_id && rowset_id == other.rowset_id; + } + }; + struct RowsetRefKeyHash { + size_t operator()(const RowsetRefKey& key) const { + return std::hash()(key.tablet_id) ^ + (std::hash()(key.rowset_id) << 1); + } + }; + std::unordered_map expected_slice_counts; + expected_slice_counts.reserve(rowset_refs.size()); + for (const auto& rowset_ref : rowset_refs) { + expected_slice_counts[RowsetRefKey {.tablet_id = rowset_ref.tablet_id, + .rowset_id = rowset_ref.rowset_id}] += + rowset_ref.expected_slice_count; + } + std::unordered_map matched_slice_counts; + + int64_t changed_slices = 0; + int64_t already_deleted_slices = 0; + auto* slice_entries = packed_info.mutable_slices(); + for (auto& slice_entry : *slice_entries) { + if (!slice_entry.has_tablet_id() || !slice_entry.has_rowset_id()) { + continue; + } + RowsetRefKey key {.tablet_id = slice_entry.tablet_id(), + .rowset_id = slice_entry.rowset_id()}; + if (!expected_slice_counts.contains(key)) { + continue; + } + ++matched_slice_counts[key]; + if (!slice_entry.deleted()) { + slice_entry.set_deleted(true); + if (!slice_entry.corrected()) { + slice_entry.set_corrected(true); + } + ++changed_slices; + } else { + ++already_deleted_slices; + } + } + + for (const auto& [key, expected_slice_count] : expected_slice_counts) { + const int64_t matched_slice_count = matched_slice_counts[key]; + if (matched_slice_count != expected_slice_count) { + LOG_WARNING("packed file rowset slice count mismatch when flushing refs") + .tag("instance_id", instance_id_) + .tag("packed_file_path", packed_file_path) + .tag("tablet_id", key.tablet_id) + .tag("rowset_id", key.rowset_id) + .tag("expected_slice_count", expected_slice_count) + .tag("matched_slice_count", matched_slice_count); + } + } + + if (changed_slices == 0) { + LOG_INFO("skip packed file ref flush: no slices changed") + .tag("instance_id", instance_id_) + .tag("packed_file_path", packed_file_path) + .tag("rowset_ref_count", rowset_refs.size()) + .tag("already_deleted_slices", already_deleted_slices) + .tag("packed_slices", packed_info.slices_size()); + success = true; + break; + } + + int64_t left_file_count = 0; + int64_t left_file_bytes = 0; + for (const auto& slice_entry : packed_info.slices()) { + if (!slice_entry.deleted()) { + ++left_file_count; + left_file_bytes += slice_entry.size(); + } + } + packed_info.set_remaining_slice_bytes(left_file_bytes); + packed_info.set_ref_cnt(left_file_count); + if (left_file_count == 0) { + packed_info.set_state(cloud::PackedFileInfoPB::RECYCLING); + } + + std::string updated_val; + if (!packed_info.SerializeToString(&updated_val)) { + LOG_WARNING("failed to serialize packed file info when flushing refs") + .tag("instance_id", instance_id_) + .tag("packed_file_path", packed_file_path) + .tag("attempt", attempt); + ret = -1; + break; + } + + txn->put(packed_key, updated_val); + err = txn->commit(); + if (err == TxnErrorCode::TXN_OK) { + LOG_INFO("flushed packed file refs") + .tag("instance_id", instance_id_) + .tag("packed_file_path", packed_file_path) + .tag("rowset_ref_count", rowset_refs.size()) + .tag("changed_slices", changed_slices) + .tag("already_deleted_slices", already_deleted_slices) + .tag("ref_cnt", left_file_count) + .tag("remaining_slice_bytes", left_file_bytes); + if (left_file_count == 0 && + delete_packed_file_and_kv(packed_file_path, packed_key, packed_info) != 0) { + ret = -1; + } + success = true; + break; + } + if (err == TxnErrorCode::TXN_CONFLICT) { + if (attempt >= max_retry_times) { + LOG_WARNING("packed file ref flush conflict after max retry") + .tag("instance_id", instance_id_) + .tag("packed_file_path", packed_file_path) + .tag("attempt", attempt) + .tag("changed_slices", changed_slices); + ret = -1; + break; + } + LOG_WARNING("packed file ref flush conflict, retrying") + .tag("instance_id", instance_id_) + .tag("packed_file_path", packed_file_path) + .tag("attempt", attempt) + .tag("changed_slices", changed_slices); + sleep_for_packed_file_retry(); + continue; + } + + LOG_WARNING("failed to commit packed file ref flush") + .tag("instance_id", instance_id_) + .tag("packed_file_path", packed_file_path) + .tag("attempt", attempt) + .tag("changed_slices", changed_slices) + .tag("err", err); + ret = -1; + break; + } + + if (!success) { + ret = -1; + } + } + return ret; +} + int InstanceRecycler::decrement_packed_file_ref_counts(const doris::RowsetMetaCloudPB& rs_meta_pb) { LOG_INFO("begin process_packed_file_location_index") .tag("instance_id", instance_id_) @@ -3462,12 +3688,16 @@ int InstanceRecycler::decrement_packed_file_ref_counts(const doris::RowsetMetaCl return ret; } -int InstanceRecycler::decrement_delete_bitmap_packed_file_ref_counts( - int64_t tablet_id, const std::string& rowset_id, - DeleteBitmapStorageType* out_storage_type) { +int InstanceRecycler::get_delete_bitmap_storage_type(int64_t tablet_id, + const std::string& rowset_id, + DeleteBitmapStorageType* out_storage_type, + std::string* packed_file_path) { if (out_storage_type) { *out_storage_type = DeleteBitmapStorageType::NOT_FOUND; } + if (packed_file_path) { + packed_file_path->clear(); + } // Get delete bitmap storage info from FDB std::string dbm_key = versioned::meta_delete_bitmap_key({instance_id_, tablet_id, rowset_id}); @@ -3530,8 +3760,24 @@ int InstanceRecycler::decrement_delete_bitmap_packed_file_ref_counts( *out_storage_type = DeleteBitmapStorageType::PACKED_FILE; } - const auto& packed_loc = storage.packed_slice_location(); - const std::string& packed_file_path = packed_loc.packed_file_path(); + if (packed_file_path) { + *packed_file_path = storage.packed_slice_location().packed_file_path(); + } + + return 0; +} + +int InstanceRecycler::decrement_delete_bitmap_packed_file_ref_counts( + int64_t tablet_id, const std::string& rowset_id, + DeleteBitmapStorageType* out_storage_type) { + std::string packed_file_path; + if (get_delete_bitmap_storage_type(tablet_id, rowset_id, out_storage_type, &packed_file_path) != + 0) { + return -1; + } + if (!out_storage_type || *out_storage_type != DeleteBitmapStorageType::PACKED_FILE) { + return 0; + } LOG_INFO("decrementing delete bitmap packed file ref count") .tag("instance_id", instance_id_) @@ -3542,7 +3788,7 @@ int InstanceRecycler::decrement_delete_bitmap_packed_file_ref_counts( const int max_retry_times = std::max(1, config::decrement_packed_file_ref_counts_retry_times); for (int attempt = 1; attempt <= max_retry_times; ++attempt) { std::unique_ptr update_txn; - err = txn_kv_->create_txn(&update_txn); + TxnErrorCode err = txn_kv_->create_txn(&update_txn); if (err != TxnErrorCode::TXN_OK) { LOG_WARNING("failed to create txn for delete bitmap packed file update") .tag("instance_id", instance_id_) @@ -3810,13 +4056,16 @@ int InstanceRecycler::delete_packed_file_and_kv(const std::string& packed_file_p int InstanceRecycler::delete_rowset_data( const std::map& rowsets, RowsetRecyclingState type, - RecyclerMetricsContext& metrics_context) { + RecyclerMetricsContext& metrics_context, + std::unordered_map>* packed_file_refs) { int ret = 0; // resource_id -> file_paths std::map> resource_file_paths; // (resource_id, tablet_id, rowset_id) std::vector> rowsets_delete_by_prefix; + std::unordered_map> pending_packed_file_refs; bool is_formal_rowset = (type == RowsetRecyclingState::FORMAL_ROWSET); + bool batch_packed_file_update = packed_file_refs != nullptr; for (const auto& [_, rs] : rowsets) { // we have to treat tmp rowset as "orphans" that may not related to any existing tablets @@ -3861,15 +4110,22 @@ int InstanceRecycler::delete_rowset_data( .tag("tablet_id", tablet_id) .tag("rowset_id", rowset_id) .tag("merge_index_size", rs.packed_slice_locations_size()); - if (decrement_packed_file_ref_counts(rs) != 0) { + if (!batch_packed_file_update && decrement_packed_file_ref_counts(rs) != 0) { ret = -1; continue; } // Process delete bitmap - check where it's stored. DeleteBitmapStorageType delete_bitmap_storage_type = DeleteBitmapStorageType::NOT_FOUND; - if (decrement_delete_bitmap_packed_file_ref_counts(tablet_id, rowset_id, - &delete_bitmap_storage_type) != 0) { + std::string delete_bitmap_packed_file_path; + int delete_bitmap_ret = + batch_packed_file_update + ? get_delete_bitmap_storage_type(tablet_id, rowset_id, + &delete_bitmap_storage_type, + &delete_bitmap_packed_file_path) + : decrement_delete_bitmap_packed_file_ref_counts( + tablet_id, rowset_id, &delete_bitmap_storage_type); + if (delete_bitmap_ret != 0) { LOG_WARNING("failed to decrement delete bitmap packed file ref count") .tag("instance_id", instance_id_) .tag("tablet_id", tablet_id) @@ -3879,6 +4135,12 @@ int InstanceRecycler::delete_rowset_data( } if (delete_bitmap_storage_type == DeleteBitmapStorageType::STANDALONE_FILE) { file_paths.push_back(delete_bitmap_path(tablet_id, rowset_id)); + } else if (delete_bitmap_storage_type == DeleteBitmapStorageType::PACKED_FILE) { + DCHECK(batch_packed_file_update || delete_bitmap_packed_file_path.empty()); + if (batch_packed_file_update) { + pending_packed_file_refs[delete_bitmap_packed_file_path].push_back(PackedRowsetRef { + .tablet_id = tablet_id, .rowset_id = rowset_id, .expected_slice_count = 1}); + } } // Process inverted indexes @@ -3943,6 +4205,9 @@ int InstanceRecycler::delete_rowset_data( // if rowset state is RowsetStatePB::BEGIN_PARTIAL_UPDATE, the number of segments data // may be larger than num_segments field in RowsetMeta, so we need to delete the rowset's data by prefix rowsets_delete_by_prefix.emplace_back(rs.resource_id(), tablet_id, rs.rowset_id_v2()); + if (batch_packed_file_update) { + collect_packed_file_refs(rs, &pending_packed_file_refs); + } continue; } for (int64_t i = 0; i < num_segments; ++i) { @@ -3970,12 +4235,18 @@ int InstanceRecycler::delete_rowset_data( rs, inverted_index_path_v2(tablet_id, rowset_id, i), &file_paths); } } + if (batch_packed_file_update) { + collect_packed_file_refs(rs, &pending_packed_file_refs); + } } SyncExecutor concurrent_delete_executor(_thread_pool_group.s3_producer_pool, "delete_rowset_data", [](const int& ret) { return ret != 0; }); for (auto& [resource_id, file_paths] : resource_file_paths) { + if (file_paths.empty()) { + continue; + } concurrent_delete_executor.add([&, rid = &resource_id, paths = &file_paths]() -> int { DCHECK(accessor_map_.count(*rid)) << "uninitilized accessor, instance_id=" << instance_id_ @@ -4055,6 +4326,13 @@ int InstanceRecycler::delete_rowset_data( } } ret = finished ? ret : -1; + if (ret == 0 && batch_packed_file_update) { + for (auto& [packed_file_path, refs] : pending_packed_file_refs) { + auto& target_refs = (*packed_file_refs)[packed_file_path]; + target_refs.insert(target_refs.end(), std::make_move_iterator(refs.begin()), + std::make_move_iterator(refs.end())); + } + } return ret; } @@ -5006,6 +5284,12 @@ int InstanceRecycler::recycle_rowsets() { // Store keys of rowset recycled by background workers std::mutex async_recycled_rowset_keys_mutex; std::vector async_recycled_rowset_keys; + std::mutex rowset_keys_to_remove_after_packed_flush_mutex; + std::vector rowset_keys_to_remove_after_packed_flush; + std::mutex delete_bitmaps_to_remove_after_packed_flush_mutex; + std::vector> delete_bitmaps_to_remove_after_packed_flush; + std::mutex packed_file_refs_mutex; + std::unordered_map> packed_file_refs; std::vector rowset_keys_without_data; auto worker_pool = std::make_unique( config::instance_recycler_worker_pool_size, "recycle_rowsets"); @@ -5201,23 +5485,50 @@ int InstanceRecycler::recycle_rowsets() { std::map rowsets) { worker_pool->submit([&, rowset_keys_to_delete = std::move(rowset_keys), rowsets_to_delete = std::move(rowsets)]() { + std::unordered_map> local_packed_file_refs; if (!rowsets_to_delete.empty() && delete_rowset_data(rowsets_to_delete, RowsetRecyclingState::FORMAL_ROWSET, - metrics_context) != 0) { + metrics_context, &local_packed_file_refs) != 0) { LOG(WARNING) << "failed to delete rowset data, instance_id=" << instance_id_; return; } - for (const auto& [_, rs] : rowsets_to_delete) { - if (delete_versioned_delete_bitmap_kvs(rs.tablet_id(), rs.rowset_id_v2()) != 0) { + if (!local_packed_file_refs.empty()) { + { + std::lock_guard lock(packed_file_refs_mutex); + for (auto& [packed_file_path, refs] : local_packed_file_refs) { + auto& target_refs = packed_file_refs[packed_file_path]; + target_refs.insert(target_refs.end(), std::make_move_iterator(refs.begin()), + std::make_move_iterator(refs.end())); + } + } + { + std::lock_guard lock(rowset_keys_to_remove_after_packed_flush_mutex); + rowset_keys_to_remove_after_packed_flush.insert( + rowset_keys_to_remove_after_packed_flush.end(), + std::make_move_iterator(rowset_keys_to_delete.begin()), + std::make_move_iterator(rowset_keys_to_delete.end())); + } + { + std::lock_guard lock(delete_bitmaps_to_remove_after_packed_flush_mutex); + for (const auto& [_, rs] : rowsets_to_delete) { + delete_bitmaps_to_remove_after_packed_flush.emplace_back(rs.tablet_id(), + rs.rowset_id_v2()); + } + } + } else { + for (const auto& [_, rs] : rowsets_to_delete) { + if (delete_versioned_delete_bitmap_kvs(rs.tablet_id(), rs.rowset_id_v2()) != + 0) { + return; + } + } + if (txn_remove(txn_kv_.get(), rowset_keys_to_delete) != 0) { + LOG(WARNING) << "failed to delete recycle rowset kv, instance_id=" + << instance_id_; return; } + num_recycled.fetch_add(rowset_keys_to_delete.size(), std::memory_order_relaxed); } - if (txn_remove(txn_kv_.get(), rowset_keys_to_delete) != 0) { - LOG(WARNING) << "failed to delete recycle rowset kv, instance_id=" << instance_id_; - return; - } - - num_recycled.fetch_add(rowset_keys_to_delete.size(), std::memory_order_relaxed); }); }; @@ -5335,6 +5646,27 @@ int InstanceRecycler::recycle_rowsets() { mark_abort_worker_pool->stop(); worker_pool->stop(); + if (!packed_file_refs.empty()) { + if (flush_packed_file_refs(packed_file_refs) != 0) { + LOG(WARNING) << "failed to flush packed file refs, instance_id=" << instance_id_; + ret = -1; + } else if (!rowset_keys_to_remove_after_packed_flush.empty()) { + for (const auto& [tablet_id, rowset_id] : delete_bitmaps_to_remove_after_packed_flush) { + if (delete_versioned_delete_bitmap_kvs(tablet_id, rowset_id) != 0) { + return -1; + } + } + if (txn_remove(txn_kv_.get(), rowset_keys_to_remove_after_packed_flush) != 0) { + LOG(WARNING) << "failed to delete recycle rowset kv after packed file ref flush, " + "instance_id=" + << instance_id_; + return -1; + } + num_recycled.fetch_add(rowset_keys_to_remove_after_packed_flush.size(), + std::memory_order_relaxed); + } + } + if (!async_recycled_rowset_keys.empty()) { if (txn_remove(txn_kv_.get(), async_recycled_rowset_keys) != 0) { LOG(WARNING) << "failed to delete recycle rowset kv, instance_id=" << instance_id_; diff --git a/cloud/src/recycler/recycler.h b/cloud/src/recycler/recycler.h index 2e8b6d57f7f3ae..2d499444723d43 100644 --- a/cloud/src/recycler/recycler.h +++ b/cloud/src/recycler/recycler.h @@ -469,9 +469,24 @@ class InstanceRecycler { int delete_rowset_data(const std::string& resource_id, int64_t tablet_id, const std::string& rowset_id); + struct PackedRowsetRef { + int64_t tablet_id = 0; + std::string rowset_id; + int64_t expected_slice_count = 0; + }; + // return 0 for success otherwise error int delete_rowset_data(const std::map& rowsets, - RowsetRecyclingState type, RecyclerMetricsContext& metrics_context); + RowsetRecyclingState type, RecyclerMetricsContext& metrics_context, + std::unordered_map>* + packed_file_refs = nullptr); + + void collect_packed_file_refs( + const doris::RowsetMetaCloudPB& rs_meta_pb, + std::unordered_map>* packed_file_refs); + + int flush_packed_file_refs( + const std::unordered_map>& packed_file_refs); // Decrement packed file ref counts for rowset segments. // Returns 0 for success, -1 for error. @@ -488,8 +503,12 @@ class InstanceRecycler { // Returns 0 for success, -1 for error. // out_storage_type: if not null, will be set to the delete bitmap storage type. int decrement_delete_bitmap_packed_file_ref_counts(int64_t tablet_id, - const std::string& rowset_id, - DeleteBitmapStorageType* out_storage_type); + const std::string& rowset_id, + DeleteBitmapStorageType* out_storage_type); + + int get_delete_bitmap_storage_type(int64_t tablet_id, const std::string& rowset_id, + DeleteBitmapStorageType* out_storage_type, + std::string* packed_file_path = nullptr); int delete_packed_file_and_kv(const std::string& packed_file_path, const std::string& packed_key, diff --git a/cloud/test/recycler_test.cpp b/cloud/test/recycler_test.cpp index ad8075cec9c55f..13e924dba57ae1 100644 --- a/cloud/test/recycler_test.cpp +++ b/cloud/test/recycler_test.cpp @@ -163,6 +163,17 @@ int main(int argc, char** argv) { namespace doris::cloud { +static std::unique_ptr create_mock_recycler(std::shared_ptr txn_kv, + const std::string& resource_id); +static void put_packed_file_info(TxnKv* txn_kv, const std::string& packed_file_path, + const PackedFileInfoPB& packed_info); +static bool get_packed_file_info(TxnKv* txn_kv, const std::string& packed_file_path, + PackedFileInfoPB* packed_info); +static PackedSlicePB create_packed_slice(int64_t tablet_id, const std::string& rowset_id, + int64_t size, bool deleted = false); +static PackedFileInfoPB create_packed_file_info(const std::string& resource_id, + std::initializer_list slices); + TEST(RecyclerTest, WhiteBlackList) { WhiteBlackList filter; EXPECT_FALSE(filter.filter_out("instance1")); @@ -180,6 +191,124 @@ TEST(RecyclerTest, WhiteBlackList) { EXPECT_TRUE(filter.filter_out("instance2")); } +TEST(RecyclerTest, FlushPackedFileRefsMarksMultipleRowsetsInOneRmw) { + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + const std::string resource_id = "packed_resource_1"; + const std::string packed_file_path = "data/packed_file/flush_multi"; + auto recycler = create_mock_recycler(txn_kv, resource_id); + + PackedFileInfoPB packed_info = create_packed_file_info( + resource_id, + {create_packed_slice(10001, "rowset_a", 10), + create_packed_slice(10001, "rowset_a", 20), + create_packed_slice(10002, "rowset_b", 30), + create_packed_slice(10003, "rowset_c", 40)}); + put_packed_file_info(txn_kv.get(), packed_file_path, packed_info); + + std::unordered_map> refs; + refs[packed_file_path].push_back({10001, "rowset_a", 2}); + refs[packed_file_path].push_back({10002, "rowset_b", 1}); + + ASSERT_EQ(recycler->flush_packed_file_refs(refs), 0); + + PackedFileInfoPB updated; + ASSERT_TRUE(get_packed_file_info(txn_kv.get(), packed_file_path, &updated)); + ASSERT_EQ(updated.slices_size(), 4); + EXPECT_TRUE(updated.slices(0).deleted()); + EXPECT_TRUE(updated.slices(0).corrected()); + EXPECT_TRUE(updated.slices(1).deleted()); + EXPECT_TRUE(updated.slices(1).corrected()); + EXPECT_TRUE(updated.slices(2).deleted()); + EXPECT_TRUE(updated.slices(2).corrected()); + EXPECT_FALSE(updated.slices(3).deleted()); + EXPECT_EQ(updated.ref_cnt(), 1); + EXPECT_EQ(updated.remaining_slice_bytes(), 40); + EXPECT_EQ(updated.state(), PackedFileInfoPB::NORMAL); +} + +TEST(RecyclerTest, FlushPackedFileRefsTreatsAlreadyDeletedSlicesAsIdempotent) { + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + const std::string resource_id = "packed_resource_2"; + const std::string packed_file_path = "data/packed_file/flush_idempotent"; + auto recycler = create_mock_recycler(txn_kv, resource_id); + + PackedFileInfoPB packed_info = create_packed_file_info( + resource_id, + {create_packed_slice(20001, "rowset_a", 10, true), + create_packed_slice(20001, "rowset_a", 20, true), + create_packed_slice(20002, "rowset_b", 30)}); + put_packed_file_info(txn_kv.get(), packed_file_path, packed_info); + + std::unordered_map> refs; + refs[packed_file_path].push_back({20001, "rowset_a", 2}); + + ASSERT_EQ(recycler->flush_packed_file_refs(refs), 0); + + PackedFileInfoPB updated; + ASSERT_TRUE(get_packed_file_info(txn_kv.get(), packed_file_path, &updated)); + EXPECT_TRUE(updated.slices(0).deleted()); + EXPECT_TRUE(updated.slices(1).deleted()); + EXPECT_FALSE(updated.slices(2).deleted()); + EXPECT_EQ(updated.ref_cnt(), 1); + EXPECT_EQ(updated.remaining_slice_bytes(), 30); + EXPECT_EQ(updated.state(), PackedFileInfoPB::NORMAL); +} + +TEST(RecyclerTest, FlushPackedFileRefsRecomputesRefCountAndDeletesEmptyPackedFile) { + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + const std::string resource_id = "packed_resource_3"; + const std::string packed_file_path = "data/packed_file/flush_empty"; + auto recycler = create_mock_recycler(txn_kv, resource_id); + auto accessor = std::dynamic_pointer_cast(recycler->accessor_map_[resource_id]); + ASSERT_NE(accessor, nullptr); + ASSERT_EQ(accessor->put_file(packed_file_path, "packed"), 0); + + PackedFileInfoPB packed_info = create_packed_file_info( + resource_id, + {create_packed_slice(30001, "rowset_a", 10), + create_packed_slice(30001, "rowset_a", 20)}); + put_packed_file_info(txn_kv.get(), packed_file_path, packed_info); + + std::unordered_map> refs; + refs[packed_file_path].push_back({30001, "rowset_a", 2}); + + ASSERT_EQ(recycler->flush_packed_file_refs(refs), 0); + + PackedFileInfoPB updated; + EXPECT_FALSE(get_packed_file_info(txn_kv.get(), packed_file_path, &updated)); + EXPECT_EQ(accessor->exists(packed_file_path), 1); +} + +TEST(RecyclerTest, FlushPackedFileRefsKeepsConsistentCountersWhenExpectedSliceCountDiffers) { + auto txn_kv = std::make_shared(); + ASSERT_EQ(txn_kv->init(), 0); + const std::string resource_id = "packed_resource_4"; + const std::string packed_file_path = "data/packed_file/flush_mismatch"; + auto recycler = create_mock_recycler(txn_kv, resource_id); + + PackedFileInfoPB packed_info = create_packed_file_info( + resource_id, + {create_packed_slice(40001, "rowset_a", 10), + create_packed_slice(40002, "rowset_b", 20)}); + put_packed_file_info(txn_kv.get(), packed_file_path, packed_info); + + std::unordered_map> refs; + refs[packed_file_path].push_back({40001, "rowset_a", 2}); + + ASSERT_EQ(recycler->flush_packed_file_refs(refs), 0); + + PackedFileInfoPB updated; + ASSERT_TRUE(get_packed_file_info(txn_kv.get(), packed_file_path, &updated)); + EXPECT_TRUE(updated.slices(0).deleted()); + EXPECT_TRUE(updated.slices(0).corrected()); + EXPECT_FALSE(updated.slices(1).deleted()); + EXPECT_EQ(updated.ref_cnt(), 1); + EXPECT_EQ(updated.remaining_slice_bytes(), 20); +} + static std::string next_rowset_id() { static int64_t cnt = 0; return std::to_string(++cnt); @@ -402,6 +531,87 @@ static int create_tmp_rowset(TxnKv* txn_kv, StorageVaultAccessor* accessor, return 0; } +static InstanceInfoPB create_mock_instance_info(const std::string& resource_id) { + InstanceInfoPB instance; + instance.set_instance_id(instance_id); + auto* obj_info = instance.add_obj_info(); + obj_info->set_id(resource_id); + obj_info->set_ak(config::test_s3_ak); + obj_info->set_sk(config::test_s3_sk); + obj_info->set_endpoint(config::test_s3_endpoint); + obj_info->set_region(config::test_s3_region); + obj_info->set_bucket(config::test_s3_bucket); + obj_info->set_prefix("recycler_test"); + return instance; +} + +static void put_packed_file_info(TxnKv* txn_kv, const std::string& packed_file_path, + const PackedFileInfoPB& packed_info) { + std::unique_ptr txn; + ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + std::string val; + ASSERT_TRUE(packed_info.SerializeToString(&val)); + txn->put(packed_file_key({instance_id, packed_file_path}), val); + ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); +} + +static bool get_packed_file_info(TxnKv* txn_kv, const std::string& packed_file_path, + PackedFileInfoPB* packed_info) { + std::unique_ptr txn; + EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); + std::string val; + TxnErrorCode err = txn->get(packed_file_key({instance_id, packed_file_path}), &val); + if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) { + return false; + } + EXPECT_EQ(err, TxnErrorCode::TXN_OK); + EXPECT_TRUE(packed_info->ParseFromString(val)); + return true; +} + +static PackedSlicePB create_packed_slice(int64_t tablet_id, const std::string& rowset_id, + int64_t size, bool deleted) { + PackedSlicePB slice; + slice.set_tablet_id(tablet_id); + slice.set_rowset_id(rowset_id); + slice.set_txn_id(next_small_file_txn_id()); + slice.set_path(fmt::format("data/{}/{}_{}.dat", tablet_id, rowset_id, size)); + slice.set_size(size); + slice.set_deleted(deleted); + slice.set_corrected(deleted); + return slice; +} + +static PackedFileInfoPB create_packed_file_info(const std::string& resource_id, + std::initializer_list slices) { + PackedFileInfoPB packed_info; + packed_info.set_resource_id(resource_id); + packed_info.set_state(PackedFileInfoPB::NORMAL); + packed_info.set_created_at_sec(current_time); + int64_t ref_cnt = 0; + int64_t remaining_bytes = 0; + for (const auto& slice : slices) { + *packed_info.add_slices() = slice; + if (!slice.deleted()) { + ++ref_cnt; + remaining_bytes += slice.size(); + } + } + packed_info.set_ref_cnt(ref_cnt); + packed_info.set_remaining_slice_bytes(remaining_bytes); + return packed_info; +} + +static std::unique_ptr create_mock_recycler(std::shared_ptr txn_kv, + const std::string& resource_id) { + auto instance = create_mock_instance_info(resource_id); + auto recycler = std::make_unique( + txn_kv, instance, thread_group, + std::make_shared(txn_kv, std::make_shared(txn_kv))); + EXPECT_EQ(recycler->init(), 0); + return recycler; +} + static int create_committed_rowset_by_real_index_v2_file(TxnKv* txn_kv, StorageVaultAccessor* accessor, const std::string& resource_id,