morningman commented on code in PR #69868:
URL: https://github.com/apache/doris/pull/69868#discussion_r4432078212
##########
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:
Thanks. The window is real: the lock is released between `prepare_compact`
and `execute_compact`, and a manual compaction through `/api/compaction/run`, a
clone or a disk migration can take it in between. But a compaction that loses
the lock there has not started its cleanup, and counting it would make the
series disagree with itself and with cloud mode.
- It is the same condition `prepare_compact` checks, found later: both
return `TRY_LOCK_FAILED` with "The tablet is under cumulative compaction", and
a compaction that meets it in `prepare_compact` is not counted. Counting it
only in the execution would count the same contention or not depending on
whether it began while the task was queued. A lock held for long is met in the
preparation of the later rounds, and would not show either way.
- Cloud mode has the same gate and does not count it: the compaction job is
requested from the meta service at the head of the pool task
(`_request_tablet_global_compaction_lock`), and when it is refused,
`JOB_TABLET_BUSY` for example, the task returns before `execute_compact`, so
`record_compaction_failure` is not reached. In both modes, counting starts once
the compaction holds the cumulative compaction of the tablet: the lock taken
again in shared-nothing mode, the job in cloud mode.
- When the holder is another cumulative compaction, the manual one for
example, it picks the expired rowsets first and counts the cleanup itself as it
ends. Counting the one that lost the lock as failed would report one cleanup as
a success and a failure.
- The compaction that loses the lock sets
`last_cumu_compaction_failure_time`, and the TTL scanner submits the tablet
again after `min_compaction_failure_interval_ms`; the expired rowsets are still
there. It is a retry, not a cleanup that cannot make progress.
The description said "does not get past its preparation", which leaves out
the lock taken again in the execution. I reworded that sentence to name it.
So I'm keeping it as it is.
##########
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:
Thanks. Two successful cleanups can interleave there, but what that costs is
bounded, and a max would cost more.
- The window is the two steps of one expression: reading the clock and the
atomic store. `UnixSeconds()` has a resolution of one second, so two threads
read the same or adjacent seconds, and the gauge can go back by about a second
until the next success sets it again. A stale-cleanup alert, `time() -
doris_be_changelog_ttl_last_success_timestamp_seconds > threshold`, needs a
threshold of several rounds of scheduling and compaction; a second does not
change it.
- The other last-time series of the BE that many threads publish,
`fragment_last_active_time` and
`file_cache_warm_up_rowset_last_finish_unix_ts`, are stored the same way, and
so is `set_to_current_time()` in the Prometheus clients.
- A max would pin the gauge to the future once a cleanup succeeds while the
clock is ahead, a clock set a day forward and then corrected for example:
`time() - gauge` stays negative until real time catches up, and no
stale-cleanup alert can fire in the meantime, even with every cleanup stopped.
That is worse than a second of jitter. Serializing the publication avoids that,
but it adds a lock for a second that no alert can observe, which the other
series do without.
So I'm keeping it as it is.
##########
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:
Thanks. They do stay changed, by
`CompactionMetricsTest.TestChangelogTtlCleanupCountedAsCompactionFinishes` too,
but no test depends on them, so the order does not matter.
- The only readers of these four series in `be/test` are the three new
tests, and each compares against values it takes right before: the counters by
their difference, the timestamp against the time the test started after a
success, and against the value it set or read itself after a failure.
`DorisMetricsTest.Normal` does not read them.
- What another suite reads as an absolute value is restored: recording a
cloud success also adds to `cumulative_compaction_deltas_total` and
`cumulative_compaction_bytes_total`, which `DorisMetricsTest.Normal` compares
as absolute values in the same process, so the cloud test restores those two
with a `Defer` (see the comment there).
- A test that reads these series later should compare differences too,
rather than count on the tests before it to restore them.
So I'm keeping it as it is.
--
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]
