This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 0557668f405 [fix](cloud) Exclude covered rowsets from compaction
minimum timestamps (#67617)
0557668f405 is described below
commit 0557668f40516aa11458ce663006f63474412917
Author: Luwei <[email protected]>
AuthorDate: Wed Sep 9 20:42:12 2026 +0800
[fix](cloud) Exclude covered rowsets from compaction minimum timestamps
(#67617)
Problem Summary:
Versioned rowset reads add load metadata versions to the reader minimum
before compact rowsets remove covered loads from the returned set. With
loads L2[2,2]@80 and L3[3,3]@90 covered by A[2,3]@150, plus L4[4,4]@170,
reading [2,4] returns A and L4 but records a minimum of 80. Later
compaction logs can consequently retain successive compacted rowsets for
a snapshot that only needs L2/L3.
Keep each candidate's metadata Versionstamp alongside its rowset, then
merge only the final returned candidates into the existing minimum after
both scans succeed. The returned set, scan/coverage rules, snapshot
bounds and error propagation are unchanged. Previous stats and other
real dependencies remain tracked, and persisted historical logs are
unchanged.
The regression exercises two real MetaService compactions and the
Recycler reference checker using MemTxnKv: snapshot@100 still protects
the first log containing L2/L3, while the later log has
min_timestamp=150 and is no longer protected by that snapshot.
### Release note
Fix unnecessary retention of later compacted rowsets when snapshots
protect older load rowsets already covered by compaction.
---
cloud/src/meta-store/meta_reader.cpp | 13 ++-
cloud/test/clone_chain_reader_test.cpp | 52 +++++++++
cloud/test/meta_reader_test.cpp | 130 +++++++++++++++++++++
cloud/test/recycler_operation_log_test.cpp | 179 +++++++++++++++++++++++++++++
4 files changed, 368 insertions(+), 6 deletions(-)
diff --git a/cloud/src/meta-store/meta_reader.cpp
b/cloud/src/meta-store/meta_reader.cpp
index 91314c39b4e..fa1a01f0147 100644
--- a/cloud/src/meta-store/meta_reader.cpp
+++ b/cloud/src/meta-store/meta_reader.cpp
@@ -757,7 +757,7 @@ TxnErrorCode MetaReader::get_rowset_metas(Transaction* txn,
int64_t tablet_id,
int64_t start_version, int64_t
end_version,
std::vector<RowsetMetaCloudPB>*
rowset_metas,
bool snapshot) {
- std::map<int64_t, RowsetMetaCloudPB> rowset_graph;
+ std::map<int64_t, std::pair<RowsetMetaCloudPB, Versionstamp>> rowset_graph;
{
std::string start_key =
@@ -776,8 +776,7 @@ TxnErrorCode MetaReader::get_rowset_metas(Transaction* txn,
int64_t tablet_id,
versioned::document_get_range<RowsetMetaCloudPB>(txn,
start_key, end_key, options);
for (auto&& kvp = iter->next(); kvp.has_value(); kvp = iter->next()) {
auto&& [key, version, rowset_meta] = *kvp;
- rowset_graph.emplace(rowset_meta.end_version(),
std::move(rowset_meta));
- min_read_versionstamp_ = std::min(min_read_versionstamp_, version);
+ rowset_graph.try_emplace(rowset_meta.end_version(),
std::move(rowset_meta), version);
DCHECK(version < snapshot_version_)
<< "version: " << version.to_string()
<< ", snapshot_version: " << snapshot_version_.to_string();
@@ -822,12 +821,11 @@ TxnErrorCode MetaReader::get_rowset_metas(Transaction*
txn, int64_t tablet_id,
continue;
}
- min_read_versionstamp_ = std::min(min_read_versionstamp_, version);
last_start_version = start_version;
// erase the rowsets that are covered by this compact rowset
rowset_graph.erase(rowset_graph.lower_bound(start_version),
rowset_graph.upper_bound(end_version));
- rowset_graph.emplace(end_version, std::move(rowset_meta));
+ rowset_graph.try_emplace(end_version, std::move(rowset_meta),
version);
}
if (!iter->is_valid()) {
LOG_ERROR("failed to get compacted rowset metas")
@@ -842,7 +840,10 @@ TxnErrorCode MetaReader::get_rowset_metas(Transaction*
txn, int64_t tablet_id,
rowset_metas->clear();
rowset_metas->reserve(rowset_graph.size());
- for (auto&& [version, rowset_meta] : rowset_graph) {
+ // Only returned rowsets contribute dependencies after both scans succeed.
+ for (auto&& [end_version, entry] : rowset_graph) {
+ auto& [rowset_meta, versionstamp] = entry;
+ min_read_versionstamp_ = std::min(min_read_versionstamp_,
versionstamp);
rowset_metas->emplace_back(std::move(rowset_meta));
}
diff --git a/cloud/test/clone_chain_reader_test.cpp
b/cloud/test/clone_chain_reader_test.cpp
index f6ecf0abf42..a94e531b57e 100644
--- a/cloud/test/clone_chain_reader_test.cpp
+++ b/cloud/test/clone_chain_reader_test.cpp
@@ -1929,6 +1929,58 @@ TEST_F(CloneChainReaderTest, GetRowsetMeta) {
}
}
+TEST_F(CloneChainReaderTest, GetRowsetMetasMinReadVersionstamp) {
+ constexpr int64_t tablet_id = 15001;
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
+ auto put_rowset = [&](const std::string& instance_id, const std::string&
id, int64_t start,
+ int64_t end, Versionstamp version, bool compact) {
+ doris::RowsetMetaCloudPB rowset;
+ rowset.set_rowset_id(0);
+ rowset.set_rowset_id_v2(id);
+ rowset.set_tablet_id(tablet_id);
+ rowset.set_start_version(start);
+ rowset.set_end_version(end);
+ auto key = compact ? versioned::meta_rowset_compact_key({instance_id,
tablet_id, end})
+ : versioned::meta_rowset_load_key({instance_id,
tablet_id, end});
+ ASSERT_TRUE(versioned::document_put(txn.get(), key, version,
std::move(rowset)));
+ };
+ put_rowset("A", "L1", 1, 1, Versionstamp(70, 1), false);
+ put_rowset("A", "L2", 2, 2, Versionstamp(80, 1), false);
+ put_rowset("A", "L3", 3, 3, Versionstamp(90, 1), false);
+ put_rowset("A", "compact", 2, 3, Versionstamp(150, 2), true);
+ // This source rowset is newer than A's source snapshot (1000) and is
invisible to C.
+ put_rowset("A", "too_new", 1, 3, Versionstamp(1000, 1), true);
+ put_rowset("C", "L4", 4, 4, Versionstamp(2170, 1), false);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ CloneChainReader reader(instance_ids_[2], Versionstamp(3000),
txn_kv_.get(),
+ resource_mgr_.get());
+ std::vector<doris::RowsetMetaCloudPB> rowsets;
+ ASSERT_EQ(reader.get_rowset_metas(tablet_id, 2, 4, &rowsets),
TxnErrorCode::TXN_OK);
+ ASSERT_EQ(rowsets.size(), 2);
+ EXPECT_EQ(rowsets[0].rowset_id_v2(), "compact");
+ EXPECT_EQ(rowsets[0].start_version(), 2);
+ EXPECT_EQ(rowsets[0].end_version(), 3);
+ EXPECT_EQ(rowsets[0].reference_instance_id(), "A");
+ EXPECT_EQ(rowsets[1].rowset_id_v2(), "L4");
+ EXPECT_EQ(rowsets[1].start_version(), 4);
+ EXPECT_EQ(rowsets[1].end_version(), 4);
+ EXPECT_EQ(rowsets[1].reference_instance_id(), "C");
+ EXPECT_EQ(reader.min_read_versionstamp(), Versionstamp(150, 2));
+
+ ASSERT_EQ(reader.get_rowset_metas(tablet_id, 1, 1, &rowsets),
TxnErrorCode::TXN_OK);
+ ASSERT_EQ(rowsets.size(), 1);
+ EXPECT_EQ(rowsets[0].rowset_id_v2(), "L1");
+ EXPECT_EQ(rowsets[0].reference_instance_id(), "A");
+ EXPECT_EQ(reader.min_read_versionstamp(), Versionstamp(70, 1));
+ ASSERT_EQ(reader.get_rowset_metas(tablet_id, 2, 4, &rowsets),
TxnErrorCode::TXN_OK);
+ EXPECT_EQ(reader.min_read_versionstamp(), Versionstamp(70, 1));
+ ASSERT_EQ(reader.get_rowset_metas(tablet_id, 5, 6, &rowsets),
TxnErrorCode::TXN_OK);
+ EXPECT_TRUE(rowsets.empty());
+ EXPECT_EQ(reader.min_read_versionstamp(), Versionstamp(70, 1));
+}
+
TEST_F(CloneChainReaderTest, GetRowsetMetas) {
std::string instance_id = instance_ids_[2]; // C
Versionstamp snapshot_version = snapshot_versions_[2];
diff --git a/cloud/test/meta_reader_test.cpp b/cloud/test/meta_reader_test.cpp
index 9c90a4b42f0..d14ea18f1a8 100644
--- a/cloud/test/meta_reader_test.cpp
+++ b/cloud/test/meta_reader_test.cpp
@@ -23,6 +23,7 @@
#include <gtest/gtest.h>
#include <memory>
+#include <tuple>
#include "common/config.h"
#include "common/logging.h"
@@ -1687,6 +1688,135 @@ TEST(MetaReaderTest, GetRowsetMetas) {
}
}
+TEST(MetaReaderTest, GetRowsetMetasMinReadVersionstamp) {
+ auto txn_kv = std::make_shared<MemTxnKv>();
+ ASSERT_EQ(txn_kv->init(), 0);
+ const std::string instance_id = "rowset_min_version";
+ constexpr int64_t tablet_id = 4001;
+
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ auto put_rowset = [&](const std::string& id, int64_t start, int64_t end,
Versionstamp version,
+ bool compact) {
+ doris::RowsetMetaCloudPB rowset;
+ rowset.set_rowset_id(0);
+ rowset.set_rowset_id_v2(id);
+ rowset.set_tablet_id(tablet_id);
+ rowset.set_start_version(start);
+ rowset.set_end_version(end);
+ auto key = compact ? versioned::meta_rowset_compact_key({instance_id,
tablet_id, end})
+ : versioned::meta_rowset_load_key({instance_id,
tablet_id, end});
+ ASSERT_TRUE(versioned::document_put(txn.get(), key, version,
std::move(rowset)));
+ };
+ put_rowset("L1", 1, 1, Versionstamp(70, 1), false);
+ put_rowset("L2", 2, 2, Versionstamp(80, 1), false);
+ put_rowset("L3", 3, 3, Versionstamp(90, 1), false);
+ put_rowset("A", 2, 3, Versionstamp(150, 2), true);
+ put_rowset("L4", 4, 4, Versionstamp(170, 1), false);
+ // Compact keys are scanned by descending end version: A also hides this
older compact.
+ put_rowset("old_compact", 2, 2, Versionstamp(110, 1), true);
+ put_rowset("old_A", 2, 3, Versionstamp(120, 1), true);
+ TabletStatsPB stats;
+ stats.set_num_rowsets(1);
+ versioned_put(txn.get(), versioned::tablet_load_stats_key({instance_id,
tablet_id}),
+ Versionstamp(70, 2), stats.SerializeAsString());
+ versioned_put(txn.get(), versioned::tablet_compact_stats_key({instance_id,
tablet_id}),
+ Versionstamp(75, 1), stats.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ using Rowset = std::tuple<std::string, int64_t, int64_t>;
+ auto check_rowsets = [&](MetaReader& reader, int64_t start, int64_t end,
+ const std::vector<Rowset>& expected, Versionstamp
expected_min) {
+ SCOPED_TRACE(fmt::format("rowset range [{}, {}]", start, end));
+ std::vector<doris::RowsetMetaCloudPB> rowsets;
+ ASSERT_EQ(reader.get_rowset_metas(tablet_id, start, end, &rowsets),
TxnErrorCode::TXN_OK);
+ std::vector<Rowset> actual;
+ for (const auto& rowset : rowsets) {
+ EXPECT_EQ(rowset.tablet_id(), tablet_id);
+ actual.emplace_back(rowset.rowset_id_v2(), rowset.start_version(),
+ rowset.end_version());
+ }
+ EXPECT_EQ(actual, expected);
+ EXPECT_EQ(reader.min_read_versionstamp(), expected_min)
+ << "actual metadata version: " << reader.min_read_version()
+ << ", expected: " << expected_min.version();
+ };
+ {
+ MetaReader reader(instance_id, txn_kv.get());
+ // Covered L2@80/L3@90 must not lower the minimum of A@150/L4@170.
+ check_rowsets(reader, 2, 4, {{"A", 2, 3}, {"L4", 4, 4}},
Versionstamp(150, 2));
+ }
+ {
+ MetaReader reader(instance_id, txn_kv.get());
+ check_rowsets(reader, 2, 3, {{"A", 2, 3}}, Versionstamp(150, 2));
+ }
+ {
+ MetaReader reader(instance_id, txn_kv.get());
+ // A selected old load still contributes its full metadata
versionstamp.
+ check_rowsets(reader, 1, 4, {{"L1", 1, 1}, {"A", 2, 3}, {"L4", 4, 4}},
Versionstamp(70, 1));
+ }
+ {
+ MetaReader reader(instance_id, txn_kv.get(), Versionstamp(100, 0));
+ check_rowsets(reader, 2, 3, {{"L2", 2, 2}, {"L3", 3, 3}},
Versionstamp(80, 1));
+ }
+ for (bool load_stats : {true, false}) {
+ SCOPED_TRACE(load_stats ? "prior load stats" : "prior compact stats");
+ MetaReader reader(instance_id, txn_kv.get());
+ auto err = load_stats ? reader.get_tablet_load_stats(tablet_id,
&stats, nullptr)
+ : reader.get_tablet_compact_stats(tablet_id,
&stats, nullptr);
+ ASSERT_EQ(err, TxnErrorCode::TXN_OK);
+ auto previous_min = load_stats ? Versionstamp(70, 2) :
Versionstamp(75, 1);
+ EXPECT_EQ(reader.min_read_versionstamp(), previous_min);
+ check_rowsets(reader, 2, 4, {{"A", 2, 3}, {"L4", 4, 4}}, previous_min);
+ check_rowsets(reader, 5, 6, {}, previous_min);
+ }
+ {
+ MetaReader reader(instance_id, txn_kv.get());
+ check_rowsets(reader, 5, 6, {}, Versionstamp::max());
+ check_rowsets(reader, 2, 3, {{"A", 2, 3}}, Versionstamp(150, 2));
+ check_rowsets(reader, 4, 4, {{"L4", 4, 4}}, Versionstamp(150, 2));
+ check_rowsets(reader, 1, 1, {{"L1", 1, 1}}, Versionstamp(70, 1));
+ check_rowsets(reader, 2, 4, {{"A", 2, 3}, {"L4", 4, 4}},
Versionstamp(70, 1));
+ check_rowsets(reader, 5, 6, {}, Versionstamp(70, 1));
+ }
+ {
+ // Equal transaction versions must still compare the two-byte order
component.
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ put_rowset("L5", 5, 5, Versionstamp(150, 1), false);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ MetaReader reader(instance_id, txn_kv.get());
+ check_rowsets(reader, 2, 5, {{"A", 2, 3}, {"L4", 4, 4}, {"L5", 5, 5}},
+ Versionstamp(150, 1));
+ }
+}
+
+TEST(MetaReaderTest, GetRowsetMetasScanFailure) {
+ auto txn_kv = std::make_shared<MemTxnKv>();
+ ASSERT_EQ(txn_kv->init(), 0);
+ const std::string instance_id = "rowset_scan_failure";
+ constexpr int64_t tablet_id = 4001;
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ doris::RowsetMetaCloudPB load;
+ load.set_rowset_id(0);
+ load.set_start_version(2);
+ load.set_end_version(2);
+ ASSERT_TRUE(versioned::document_put(
+ txn.get(), versioned::meta_rowset_load_key({instance_id,
tablet_id, 2}),
+ Versionstamp(80, 1), std::move(load)));
+ // A malformed compact document fails after the load scan has succeeded.
+ versioned_put(txn.get(), versioned::meta_rowset_compact_key({instance_id,
tablet_id, 3}),
+ Versionstamp(150, 1), "invalid protobuf");
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ MetaReader reader(instance_id, txn_kv.get());
+ std::vector<doris::RowsetMetaCloudPB> rowsets(1);
+ rowsets[0].set_rowset_id_v2("unchanged");
+ ASSERT_EQ(reader.get_rowset_metas(tablet_id, 2, 3, &rowsets),
TxnErrorCode::TXN_INVALID_DATA);
+ ASSERT_EQ(rowsets.size(), 1);
+ EXPECT_EQ(rowsets[0].rowset_id_v2(), "unchanged");
+ EXPECT_EQ(reader.min_read_versionstamp(), Versionstamp::max());
+}
+
TEST(MetaReaderTest, GetPartitionPendingTxnId) {
auto txn_kv = std::make_shared<MemTxnKv>();
ASSERT_EQ(txn_kv->init(), 0);
diff --git a/cloud/test/recycler_operation_log_test.cpp
b/cloud/test/recycler_operation_log_test.cpp
index f23c74cb036..fcacb503882 100644
--- a/cloud/test/recycler_operation_log_test.cpp
+++ b/cloud/test/recycler_operation_log_test.cpp
@@ -1294,6 +1294,185 @@ doris::RowsetMetaCloudPB create_rowset(int64_t txn_id,
int64_t tablet_id, int pa
return rowset;
}
+TEST(RecycleOperationLogTest, CoveredLoadDoesNotProtectLaterCompaction) {
+ auto meta_service = get_meta_service(false);
+ const std::string test_instance_id = "covered_load_compaction";
+ auto txn_kv = std::dynamic_pointer_cast<MemTxnKv>(meta_service->txn_kv());
+ ASSERT_NE(txn_kv, nullptr);
+ auto* sp = SyncPoint::get_instance();
+ DORIS_CLOUD_DEFER {
+ sp->clear_all_call_backs();
+ sp->disable_processing();
+ };
+ sp->set_call_back("get_instance_id", [&](auto&& args) {
+ auto* ret = try_any_cast_ret<std::string>(args);
+ ret->first = test_instance_id;
+ ret->second = true;
+ });
+ sp->enable_processing();
+
+ constexpr int64_t table_id = 20001, index_id = 20002, partition_id =
20003, tablet_id = 20004;
+ InstanceInfoPB instance;
+ instance.set_instance_id(test_instance_id);
+ instance.set_multi_version_status(MULTI_VERSION_ENABLED);
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ txn->put(instance_key(test_instance_id), instance.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ meta_service->resource_mgr()->refresh_instance(test_instance_id);
+
ASSERT_TRUE(meta_service->resource_mgr()->is_version_read_enabled(test_instance_id));
+ create_tablet(meta_service.get(), table_id, index_id, partition_id,
tablet_id);
+
+ auto l2 = create_rowset(102, tablet_id, partition_id, 2, 10);
+ auto l3 = create_rowset(103, tablet_id, partition_id, 3, 10);
+ auto l4 = create_rowset(104, tablet_id, partition_id, 4, 10);
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ ASSERT_TRUE(versioned::document_put(
+ txn.get(), versioned::meta_rowset_load_key({test_instance_id,
tablet_id, 2}),
+ Versionstamp(80), doris::RowsetMetaCloudPB(l2)));
+ ASSERT_TRUE(versioned::document_put(
+ txn.get(), versioned::meta_rowset_load_key({test_instance_id,
tablet_id, 3}),
+ Versionstamp(90), doris::RowsetMetaCloudPB(l3)));
+ SnapshotPB snapshot;
+ snapshot.set_status(SNAPSHOT_NORMAL);
+ versioned_put(txn.get(), versioned::snapshot_full_key(test_instance_id),
Versionstamp(100),
+ snapshot.SerializeAsString());
+ // Keep both stats dependencies at/after 150 so only the selected inputs
decide protection.
+ TabletStatsPB compact_stats;
+ compact_stats.set_num_rows(20);
+ compact_stats.set_num_rowsets(2);
+ compact_stats.set_num_segments(2);
+ compact_stats.set_data_size(2200);
+ compact_stats.set_index_size(200);
+ compact_stats.set_segment_size(2000);
+ compact_stats.set_cumulative_point(2);
+ versioned_put(txn.get(),
versioned::tablet_compact_stats_key({test_instance_id, tablet_id}),
+ Versionstamp(150), compact_stats.SerializeAsString());
+ versioned_put(txn.get(),
versioned::tablet_load_stats_key({test_instance_id, tablet_id}),
+ Versionstamp(150), TabletStatsPB().SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ std::vector<std::string> output_rowset_ids;
+ auto compact = [&](int64_t end, int64_t commit_version, int base_count,
OperationLogPB* log,
+ Versionstamp* log_version) {
+ SCOPED_TRACE(commit_version);
+ const int64_t txn_id = 30000 + end;
+ const int64_t rows = (end - 1) * 10;
+ auto output = create_rowset(txn_id, tablet_id, partition_id, 2, rows);
+ output.set_end_version(end);
+ output_rowset_ids.push_back(output.rowset_id_v2());
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ txn->put(meta_rowset_tmp_key({test_instance_id, txn_id, tablet_id}),
+ output.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ StartTabletJobRequest start;
+ auto* idx = start.mutable_job()->mutable_idx();
+ idx->set_table_id(table_id);
+ idx->set_index_id(index_id);
+ idx->set_partition_id(partition_id);
+ idx->set_tablet_id(tablet_id);
+ auto* job = start.mutable_job()->add_compaction();
+ job->set_id(fmt::format("base_{}", base_count));
+ job->set_initiator("test_be");
+ job->set_type(TabletCompactionJobPB::BASE);
+ job->set_base_compaction_cnt(base_count);
+ job->set_cumulative_compaction_cnt(0);
+ job->add_input_versions(2);
+ job->add_input_versions(end);
+ job->set_expiration(time(nullptr) + 3600);
+ job->set_lease(time(nullptr) + 3600);
+ brpc::Controller cntl;
+ StartTabletJobResponse start_response;
+ meta_service->start_tablet_job(&cntl, &start, &start_response,
nullptr);
+ ASSERT_EQ(start_response.status().code(), MetaServiceCode::OK)
+ << start_response.status().msg();
+
+ FinishTabletJobRequest finish;
+ finish.set_action(FinishTabletJobRequest::COMMIT);
+ finish.mutable_job()->CopyFrom(start.job());
+ job = finish.mutable_job()->mutable_compaction(0);
+ job->add_txn_id(txn_id);
+ job->add_output_versions(end);
+ job->add_output_rowset_ids(output.rowset_id_v2());
+ job->set_output_cumulative_point(2);
+ job->set_num_input_rows(rows);
+ job->set_num_output_rows(rows);
+ job->set_num_input_rowsets(2);
+ job->set_num_output_rowsets(1);
+ job->set_num_input_segments(2);
+ job->set_num_output_segments(1);
+ job->set_size_input_rowsets(rows * 110);
+ job->set_size_output_rowsets(rows * 110);
+ job->set_index_size_input_rowsets(rows * 10);
+ job->set_index_size_output_rowsets(rows * 10);
+ job->set_segment_size_input_rowsets(rows * 100);
+ job->set_segment_size_output_rowsets(rows * 100);
+ txn_kv->update_commit_version(commit_version - 1);
+ FinishTabletJobResponse finish_response;
+ meta_service->finish_tablet_job(&cntl, &finish, &finish_response,
nullptr);
+ ASSERT_EQ(finish_response.status().code(), MetaServiceCode::OK)
+ << finish_response.status().msg();
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ ASSERT_EQ(read_operation_log(txn.get(),
versioned::log_key(test_instance_id), log_version,
+ log),
+ TxnErrorCode::TXN_OK);
+ ASSERT_EQ(log_version->version(), commit_version);
+ ASSERT_TRUE(log->has_compaction());
+ ASSERT_EQ(log->compaction().recycle_rowsets_size(), 2);
+ };
+
+ OperationLogPB first_log, second_log;
+ Versionstamp first_version, second_version;
+ ASSERT_NO_FATAL_FAILURE(compact(3, 150, 0, &first_log, &first_version));
+ EXPECT_EQ(first_log.min_timestamp(), 80);
+
EXPECT_EQ(first_log.compaction().recycle_rowsets(0).rowset_meta().rowset_id_v2(),
+ l2.rowset_id_v2());
+
EXPECT_EQ(first_log.compaction().recycle_rowsets(1).rowset_meta().rowset_id_v2(),
+ l3.rowset_id_v2());
+
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ ASSERT_TRUE(versioned::document_put(
+ txn.get(), versioned::meta_rowset_load_key({test_instance_id,
tablet_id, 4}),
+ Versionstamp(170), doris::RowsetMetaCloudPB(l4)));
+ TabletStatsPB load_stats;
+ load_stats.set_num_rows(10);
+ load_stats.set_num_rowsets(1);
+ load_stats.set_num_segments(1);
+ load_stats.set_data_size(1100);
+ load_stats.set_index_size(100);
+ load_stats.set_segment_size(1000);
+ versioned_put(txn.get(),
versioned::tablet_load_stats_key({test_instance_id, tablet_id}),
+ Versionstamp(170), load_stats.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ ASSERT_NO_FATAL_FAILURE(compact(4, 200, 1, &second_log, &second_version));
+ EXPECT_EQ(second_log.min_timestamp(), 150);
+ const auto& inputs = second_log.compaction().recycle_rowsets();
+ EXPECT_EQ(inputs[0].rowset_meta().start_version(), 2);
+ EXPECT_EQ(inputs[0].rowset_meta().end_version(), 3);
+ EXPECT_EQ(inputs[0].rowset_meta().rowset_id_v2(), output_rowset_ids[0]);
+ EXPECT_EQ(inputs[1].rowset_meta().rowset_id_v2(), l4.rowset_id_v2());
+
+ // Exercise the real Recycler snapshot-reference decision with the actual
MS logs.
+ OperationLogRecycleChecker checker(test_instance_id, txn_kv.get(),
instance);
+ ASSERT_EQ(checker.init(), 0);
+ OperationLogReferenceInfo first_reference, second_reference;
+ EXPECT_FALSE(checker.can_recycle(first_version, first_log.min_timestamp(),
&first_reference));
+ EXPECT_TRUE(first_reference.referenced_by_snapshot);
+ EXPECT_EQ(first_reference.referenced_snapshot_timestamp,
Versionstamp(100));
+ EXPECT_TRUE(checker.can_recycle(second_version,
second_log.min_timestamp(), &second_reference));
+ EXPECT_FALSE(second_reference.referenced_by_snapshot);
+
+ MetaReader snapshot_reader(test_instance_id, txn_kv.get(),
Versionstamp(100));
+ std::vector<doris::RowsetMetaCloudPB> snapshot_rowsets;
+ ASSERT_EQ(snapshot_reader.get_rowset_metas(tablet_id, 2, 3,
&snapshot_rowsets),
+ TxnErrorCode::TXN_OK);
+ ASSERT_EQ(snapshot_rowsets.size(), 2);
+ EXPECT_EQ(snapshot_rowsets[0].rowset_id_v2(), l2.rowset_id_v2());
+ EXPECT_EQ(snapshot_rowsets[1].rowset_id_v2(), l3.rowset_id_v2());
+ EXPECT_EQ(snapshot_reader.min_read_versionstamp(), Versionstamp(80));
+}
+
TEST(RecycleOperationLogTest, RecycleCompactionLog) {
// Ensure strip behavior is enabled for this test
auto old_flag = config::enable_recycle_rowset_strip_key_bounds;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
