github-actions[bot] commented on code in PR #69868:
URL: https://github.com/apache/doris/pull/69868#discussion_r4420015348
##########
be/src/storage/compaction/compaction.cpp:
##########
@@ -388,15 +392,41 @@ void Compaction::filter_changelog_ttl_rowsets() {
if (rowset->num_rows() == 0) {
continue;
}
- if (rowset->end_version() > _changelog_ttl_visible_version ||
- !changelog_rowset_expired(*rowset->rowset_meta(),
_changelog_ttl_cutoff_tso)) {
+ if (!is_expired_changelog_rowset(rowset)) {
_data_input_rowsets.push_back(rowset);
Review Comment:
[P2] Publish the maximum success timestamp. Two successful cleanups can
interleave between UnixSeconds() and this unconditional set_value: an older
timestamp can overwrite a newer one, making
changelog_ttl_last_success_timestamp_seconds move backwards and trigger
stale-cleanup alerts. Update with an atomic max/CAS loop or serialize
publication so the gauge remains monotonic.
##########
be/src/storage/compaction/cumulative_compaction.cpp:
##########
@@ -121,6 +121,7 @@ Status CumulativeCompaction::execute_compact() {
SCOPED_ATTACH_TASK(_mem_tracker);
st = CompactionMixin::execute_compact();
+ record_changelog_ttl_cleanup(st.ok());
Review Comment:
[P2] Record TTL cleanup failure when the cumulative lock is unavailable.
After preparation has selected expired inputs, another task can hold this lock
before execute reacquires it; the TRY_LOCK_FAILED return at lines 115-118
bypasses record_changelog_ttl_cleanup, so the failed cleanup counter
undercounts exactly the execution failure this metric is meant to expose. Route
this early return through the same failure recording path while keeping the
no-expired-input no-op.
##########
be/test/storage/compaction/cumulative_compaction_test.cpp:
##########
@@ -492,4 +494,105 @@ TEST_F(CumulativeCompactionTest,
TestFilterExpiredChangelogRowsets) {
EXPECT_EQ(compaction._changelog_ttl_filtered_rows, 10);
}
+struct ChangelogTtlCleanupMetrics {
+ int64_t successes;
+ int64_t failures;
+ int64_t reclaimed_bytes;
+ int64_t last_success_s;
+};
+
+static ChangelogTtlCleanupMetrics changelog_ttl_cleanup_metrics() {
+ auto* metrics = DorisMetrics::instance();
+ return {metrics->changelog_ttl_cleanup_success_total->value(),
+ metrics->changelog_ttl_cleanup_failed_total->value(),
+ metrics->changelog_ttl_reclaimed_bytes_total->value(),
+ metrics->changelog_ttl_last_success_timestamp_seconds->value()};
+}
+
+TEST_F(CumulativeCompactionTest, TestRecordChangelogTtlCleanup) {
+ EngineOptions options;
+ StorageEngine storage_engine(options);
+ auto tablet_meta = std::make_shared<TabletMeta>(1, 2, 15673, 15674, 4, 5,
TTabletSchema(), 6,
+
std::unordered_map<uint32_t, uint32_t> {{7, 8}},
+ UniqueId(9, 10),
TTabletType::TABLET_TYPE_DISK,
+ TCompressionType::LZ4F);
+ tablet_meta->set_tablet_role(TabletRolePB::TABLET_ROLE_CHANGELOG);
+ BinlogConfig binlog_config(true, 1, 1024, 10, BinlogFormatPB::CHANGELOG,
false);
+ tablet_meta->set_binlog_config(binlog_config);
+ const int64_t reference_tso = 2000L << kTsoLogicalBits;
+ const int64_t cutoff = changelog_ttl_cutoff_tso(reference_tso, 1);
+ _cluster.advance_changelog_ttl_reference_tso(reference_tso);
+ auto tablet = std::make_shared<Tablet>(storage_engine, tablet_meta,
nullptr,
+ CUMULATIVE_SIZE_BASED_POLICY);
+ auto visible = std::make_shared<VersionWithTime>();
+ visible->version.store(4);
+ tablet->set_visible_version(visible);
+
+ auto empty = create_rowset({1, 1}, 0, false, 0);
+ auto expired = create_rowset({2, 2}, 1, false, 1024);
+ expired->rowset_meta()->set_num_rows(10);
+ expired->rowset_meta()->set_commit_tso(cutoff);
+ auto retained = create_rowset({3, 3}, 1, false, 2048);
+ retained->rowset_meta()->set_num_rows(20);
+ retained->rowset_meta()->set_commit_tso(cutoff + 1);
+ auto expired_again = create_rowset({4, 4}, 1, false, 4096);
+ expired_again->rowset_meta()->set_num_rows(30);
+ expired_again->rowset_meta()->set_commit_tso(cutoff);
+ // Expired by its TSO, but above the visible version the TTL snapshot was
taken at, so it is merged.
+ auto invisible = create_rowset({5, 5}, 1, false, 8192);
+ invisible->rowset_meta()->set_num_rows(40);
+ invisible->rowset_meta()->set_commit_tso(cutoff);
+
+ TestableCumulativeCompactionMixin compaction(storage_engine, tablet);
+ compaction.snapshot_changelog_ttl();
+ compaction.set_input_rowsets({empty, expired, retained, expired_again,
invisible});
+ // What a cleanup reclaims is what its merge leaves out: the inputs with
data that it does not read.
+ compaction.filter_changelog_ttl_rowsets();
+ const auto& merged = compaction._data_input_rowsets;
+ int64_t dropped_bytes = 0;
+ for (const auto& rowset : compaction._input_rowsets) {
+ if (rowset->num_rows() != 0 &&
+ std::find(merged.begin(), merged.end(), rowset) == merged.end()) {
+ dropped_bytes += rowset->total_disk_size();
Review Comment:
[P2] Restore the global TTL metrics after this test.
DorisMetrics::instance() is shared by the monolithic doris_be_test binary, but
this test leaves the timestamp gauge at 1 and the cloud counterpart leaves all
four new values incremented. Snapshot and restore the new counters/gauge with
Defer so later tests do not depend on test order.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
