Skip to content

Commit fe92389

Browse files
liaoxin01liaoxin01
authored
[fix](cloud) refresh tablet meta of continuously ingested tablets (apache#67972)
Problem Summary: `CloudTabletMgr::sync_tablets()` selected the tablets it would work on by `last_sync_time_s`, then did two things to each one: `sync_meta()` followed by `sync_rowsets()`. But that clock only tracks how stale a tablet's **rowsets** are. It is advanced inside `sync_tablet_rowsets()`, and only when that actually issues its RPC. A tablet under continuous ingest therefore keeps it permanently fresh, never falls below the staleness bound, and never has `sync_meta()` called on it at all. Its `TabletMeta` then stays at whatever it was built with for the lifetime of the object, so properties that only arrive through the tablet meta are never picked up. `ttl_seconds` is the one that shows: `CloudTablet::sync_meta()` is the only thing that refreshes it, and it feeds the file cache expiration computed on the write path (`CloudRowsetBuilder`, compaction output), the read path (`TabletReader`, `OlapScanner`) and warm-up. An `ALTER TABLE ... SET ("file_cache_ttl_seconds" = ...)` on a table under load therefore has no effect on those blocks, which keep landing in the wrong queue with an expiration derived from the stale value. ### Release note Fixed tablet metadata never being refreshed for tablets under continuous ingest, which left properties such as `file_cache_ttl_seconds` stale on those tablets after an ALTER. ### What is changed and how it works? **1. Give meta staleness its own clock.** `last_sync_tablet_meta_time_s` is advanced only by `sync_meta()`. **2. Rename `last_sync_time_s` to `last_sync_rowsets_time_s`** (separate commit, mechanical). The old name says "sync" while the clock only ever tracked rowsets, and reading it as "when this tablet was last synced" is exactly the mistake that let meta work be gated on it. With a second clock alongside it the old name would be actively misleading. This also keeps the field named the same as on branch-3.1. **3. `sync_tablets()` decides per tablet which of the two RPCs it is due for**, rather than sorting tablets into a single bucket: - rowsets stale -> sync rowsets, and the meta too. Pulling rowsets implies pulling the meta, which is the relationship the single pass had: the rowsets are only as trustworthy as the meta they belong to. - rowsets fresh but meta stale -> sync the meta only, one RPC instead of the two a full sync costs. This is the case that used to be skipped entirely. - both fresh -> skip. Work is still ordered by the older of the two clocks, so a mid-run stop has already served the tablets that waited longest. New bvars `sync_tablets_meta_num` and `sync_tablets_rowsets_num` split what `num_sync` used to lump together, and the finish log reports both.
1 parent 59de1a7 commit fe92389

12 files changed

Lines changed: 196 additions & 42 deletions

‎be/src/cloud/cloud_base_compaction.cpp‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ Status CloudBaseCompaction::prepare_compact() {
6767
// synchronized with meta-service.
6868
if (_tablet->tablet_meta()->all_rs_metas().size() >=
6969
cloud_tablet()->fetch_add_approximate_num_rowsets(0) &&
70-
cloud_tablet()->last_sync_time_s > 0) {
70+
cloud_tablet()->last_sync_rowsets_time_s > 0) {
7171
need_sync_tablet = false;
7272
}
7373
}
@@ -136,7 +136,7 @@ Status CloudBaseCompaction::request_global_lock() {
136136
cloud_tablet()->set_last_base_compaction_failure_time(UnixMillis());
137137
if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
138138
// set last_sync_time to 0 to force sync tablet next time
139-
cloud_tablet()->last_sync_time_s = 0;
139+
cloud_tablet()->last_sync_rowsets_time_s = 0;
140140
} else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
141141
// tablet not found
142142
cloud_tablet()->clear_cache();
@@ -440,7 +440,7 @@ Status CloudBaseCompaction::modify_rowsets() {
440440
// the tablet to be unable to synchronize the rowset meta changes generated by cumu compaction.
441441
cloud_tablet()->set_base_compaction_cnt(stats.base_compaction_cnt());
442442
if (stats.cumulative_point() > cloud_tablet()->cumulative_layer_point()) {
443-
cloud_tablet()->last_sync_time_s = 0;
443+
cloud_tablet()->last_sync_rowsets_time_s = 0;
444444
}
445445
if (output_rowset_delete_bitmap) {
446446
_tablet->tablet_meta()->delete_bitmap().merge(*output_rowset_delete_bitmap);

‎be/src/cloud/cloud_cumulative_compaction.cpp‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ Status CloudCumulativeCompaction::prepare_compact() {
157157
// synchronized with meta-service.
158158
if (_tablet->tablet_meta()->all_rs_metas().size() >=
159159
cloud_tablet()->fetch_add_approximate_num_rowsets(0) &&
160-
cloud_tablet()->last_sync_time_s > 0) {
160+
cloud_tablet()->last_sync_rowsets_time_s > 0) {
161161
need_sync_tablet = false;
162162
}
163163
}
@@ -238,7 +238,7 @@ Status CloudCumulativeCompaction::request_global_lock() {
238238
if (!st.ok()) {
239239
if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
240240
// set last_sync_time to 0 to force sync tablet next time
241-
cloud_tablet()->last_sync_time_s = 0;
241+
cloud_tablet()->last_sync_rowsets_time_s = 0;
242242
} else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
243243
// tablet not found
244244
cloud_tablet()->clear_cache();
@@ -925,7 +925,7 @@ void CloudCumulativeCompaction::update_cumulative_point(int64_t input_cumulative
925925
if (!st.ok()) {
926926
if (start_resp.status().code() == cloud::STALE_TABLET_CACHE) {
927927
// set last_sync_time to 0 to force sync tablet next time
928-
cloud_tablet()->last_sync_time_s = 0;
928+
cloud_tablet()->last_sync_rowsets_time_s = 0;
929929
} else if (start_resp.status().code() == cloud::TABLET_NOT_FOUND) {
930930
// tablet not found
931931
cloud_tablet()->clear_cache();

‎be/src/cloud/cloud_full_compaction.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -113,7 +113,7 @@ Status CloudFullCompaction::request_global_lock() {
113113
if (!st.ok()) {
114114
if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
115115
// set last_sync_time to 0 to force sync tablet next time
116-
cloud_tablet()->last_sync_time_s = 0;
116+
cloud_tablet()->last_sync_rowsets_time_s = 0;
117117
} else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
118118
// tablet not found
119119
cloud_tablet()->clear_cache();

‎be/src/cloud/cloud_index_change_compaction.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -170,7 +170,7 @@ Status CloudIndexChangeCompaction::request_global_lock(bool& should_skip_err) {
170170
if (!st.ok()) {
171171
if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
172172
// set last_sync_time to 0 to force sync tablet next time
173-
cloud_tablet()->last_sync_time_s = 0;
173+
cloud_tablet()->last_sync_rowsets_time_s = 0;
174174
should_skip_err = true;
175175
} else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
176176
// tablet not found
@@ -380,7 +380,7 @@ void CloudIndexChangeCompaction::_update_tablet_for_base_compaction(
380380
// the tablet to be unable to synchronize the rowset meta changes generated by cumu compaction.
381381
cloud_tablet()->set_base_compaction_cnt(stats.base_compaction_cnt());
382382
if (stats.cumulative_point() > cloud_tablet()->cumulative_layer_point()) {
383-
cloud_tablet()->last_sync_time_s = 0;
383+
cloud_tablet()->last_sync_rowsets_time_s = 0;
384384
}
385385
if (output_rowset_delete_bitmap) {
386386
_tablet->tablet_meta()->delete_bitmap().merge(*output_rowset_delete_bitmap);

‎be/src/cloud/cloud_meta_mgr.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -885,7 +885,7 @@ Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet,
885885
}
886886

887887
int64_t now = duration_cast<seconds>(system_clock::now().time_since_epoch()).count();
888-
tablet->last_sync_time_s = now;
888+
tablet->last_sync_rowsets_time_s = now;
889889

890890
if (sync_stats) {
891891
sync_stats->get_remote_rowsets_rpc_ns +=

‎be/src/cloud/cloud_tablet.cpp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1623,6 +1623,7 @@ Status CloudTablet::sync_meta() {
16231623
}
16241624
}
16251625

1626+
last_sync_tablet_meta_time_s = ::time(nullptr);
16261627
return Status::OK();
16271628
}
16281629

‎be/src/cloud/cloud_tablet.h‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -348,7 +348,14 @@ class CloudTablet final : public BaseTablet {
348348

349349
const auto& rowset_map() const { return _rs_version_map; }
350350

351-
int64_t last_sync_time_s = 0;
351+
// How long since this tablet's ROWSETS were pulled from MS. Only sync_rowsets() advances
352+
// it, and only when it actually issues the RPC.
353+
int64_t last_sync_rowsets_time_s = 0;
354+
// How long since this tablet's META was pulled from MS, which is what carries properties
355+
// such as the file cache TTL. Only sync_meta() advances it. Tracked separately on
356+
// purpose: a tablet under continuous ingest keeps the rowsets clock permanently fresh,
357+
// so gating meta work on that one starves the meta refresh entirely.
358+
int64_t last_sync_tablet_meta_time_s = 0;
352359
int64_t last_load_time_ms = 0;
353360
int64_t last_base_compaction_success_time_ms = 0;
354361
int64_t last_cumu_compaction_success_time_ms = 0;

‎be/src/cloud/cloud_tablet_mgr.cpp‎

Lines changed: 65 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,10 @@
1919

2020
#include <bthread/countdown_event.h>
2121

22+
#include <algorithm>
2223
#include <chrono>
24+
#include <set>
25+
#include <utility>
2326

2427
#include "cloud/cloud_cluster_info.h"
2528
#include "cloud/cloud_meta_mgr.h"
@@ -40,6 +43,8 @@ bvar::Adder<uint64_t> g_base_compaction_not_frozen_tablet_num(
4043
"base_compaction_not_frozen_tablet_num");
4144
bvar::Adder<uint64_t> g_cumu_compaction_not_frozen_tablet_num(
4245
"cumu_compaction_not_frozen_tablet_num");
46+
bvar::Adder<uint64_t> g_sync_tablets_meta_num("sync_tablets_meta_num");
47+
bvar::Adder<uint64_t> g_sync_tablets_rowsets_num("sync_tablets_rowsets_num");
4348
namespace {
4449

4550
// port from
@@ -386,53 +391,93 @@ std::vector<std::weak_ptr<CloudTablet>> CloudTabletMgr::get_weak_tablets() {
386391

387392
void CloudTabletMgr::sync_tablets(const CountDownLatch& stop_latch) {
388393
LOG_INFO("begin to sync tablets");
389-
int64_t last_sync_time_bound = ::time(nullptr) - config::tablet_sync_interval_s;
390394

391-
auto weak_tablets = get_weak_tablets();
395+
// A tablet carries two staleness clocks and each one gates a different RPC:
396+
//
397+
// last_sync_rowsets_time_s
398+
// how long since we pulled this tablet's ROWSETS from MS. Only sync_rowsets()
399+
// advances it, and only when it actually issues the RPC -- a query whose requested
400+
// version we already hold returns early and leaves the clock untouched.
401+
//
402+
// last_sync_tablet_meta_time_s
403+
// how long since we pulled this tablet's META from MS, which is what carries
404+
// properties such as the file cache TTL. Only sync_meta() advances it.
405+
//
406+
// They have to be read separately. A tablet under continuous ingest keeps the rowsets
407+
// clock permanently fresh, so selecting meta work by it -- as this used to -- means such a
408+
// tablet never has its meta refreshed at all, and it keeps serving whatever TTL it was
409+
// built with.
410+
const int64_t stale_before = ::time(nullptr) - config::tablet_sync_interval_s;
411+
412+
struct Work {
413+
std::weak_ptr<CloudTablet> tablet;
414+
bool needs_meta = false;
415+
bool needs_rowsets = false;
416+
};
392417

393-
// sort by last_sync_time
418+
// Ordered by the older of the two clocks, so that if we are told to stop half way, the
419+
// tablets that have been waiting longest have already been served.
394420
static auto cmp = [](const auto& a, const auto& b) { return a.first < b.first; };
395-
std::multiset<std::pair<int64_t, std::weak_ptr<CloudTablet>>, decltype(cmp)>
396-
sync_time_tablet_set(cmp);
421+
std::multiset<std::pair<int64_t, Work>, decltype(cmp)> due(cmp);
397422

398-
for (auto& weak_tablet : weak_tablets) {
399-
if (auto tablet = weak_tablet.lock()) {
400-
int64_t last_sync_time = tablet->last_sync_time_s;
401-
if (last_sync_time <= last_sync_time_bound) {
402-
sync_time_tablet_set.emplace(last_sync_time, weak_tablet);
403-
}
423+
for (auto& weak_tablet : get_weak_tablets()) {
424+
auto tablet = weak_tablet.lock();
425+
if (!tablet) {
426+
continue;
427+
}
428+
const bool needs_rowsets = tablet->last_sync_rowsets_time_s <= stale_before;
429+
Work work {
430+
.tablet = weak_tablet,
431+
// Pulling rowsets implies pulling the tablet meta: the rowsets we are about
432+
// to take are only as trustworthy as the meta they belong to, and this is
433+
// the relationship the previous single pass had.
434+
.needs_meta = needs_rowsets || tablet->last_sync_tablet_meta_time_s <= stale_before,
435+
.needs_rowsets = needs_rowsets};
436+
if (!work.needs_meta && !work.needs_rowsets) {
437+
continue;
404438
}
439+
due.emplace(
440+
std::min(tablet->last_sync_tablet_meta_time_s, tablet->last_sync_rowsets_time_s),
441+
std::move(work));
405442
}
406443

407444
int num_sync = 0;
408-
for (auto&& [_, weak_tablet] : sync_time_tablet_set) {
445+
int num_sync_meta = 0;
446+
for (auto&& [_, work] : due) {
409447
if (stop_latch.count() <= 0) {
410448
break;
411449
}
450+
auto tablet = work.tablet.lock();
451+
if (!tablet) {
452+
continue;
453+
}
412454

413-
if (auto tablet = weak_tablet.lock()) {
414-
if (tablet->last_sync_time_s > last_sync_time_bound) {
415-
continue;
416-
}
417-
418-
++num_sync;
455+
if (work.needs_meta) {
456+
++num_sync_meta;
457+
g_sync_tablets_meta_num << 1;
419458
auto st = tablet->sync_meta();
420459
if (!st) {
421460
LOG_WARNING("failed to sync tablet meta {}", tablet->tablet_id()).error(st);
422461
if (st.is<ErrorCode::NOT_FOUND>()) {
462+
// the tablet is gone from MS, there is nothing left to sync
423463
continue;
424464
}
425465
}
466+
}
467+
468+
if (work.needs_rowsets) {
469+
++num_sync;
470+
g_sync_tablets_rowsets_num << 1;
426471
SyncOptions options;
427472
options.query_version = -1;
428473
options.merge_schema = true;
429-
st = tablet->sync_rowsets(options);
474+
auto st = tablet->sync_rowsets(options);
430475
if (!st) {
431476
LOG_WARNING("failed to sync tablet rowsets {}", tablet->tablet_id()).error(st);
432477
}
433478
}
434479
}
435-
LOG_INFO("finish sync tablets").tag("num_sync", num_sync);
480+
LOG_INFO("finish sync tablets").tag("num_sync", num_sync).tag("num_sync_meta", num_sync_meta);
436481
}
437482

438483
Status CloudTabletMgr::get_topn_tablets_to_compact(

‎be/src/storage/compaction/compaction.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2112,7 +2112,7 @@ bool CloudCompactionMixin::should_apply_cumulative_compaction_result(
21122112
}
21132113
if (response_cumulative_compaction_cnt != local_cumulative_compaction_cnt + 1) {
21142114
// Only the current task's output is available locally. Sync all missing outputs instead.
2115-
cloud_tablet()->last_sync_time_s = 0;
2115+
cloud_tablet()->last_sync_rowsets_time_s = 0;
21162116
LOG_INFO("defer applying cumulative compaction result until tablet sync")
21172117
.tag("tablet_id", _tablet->tablet_id())
21182118
.tag("job_id", _uuid)

‎be/test/cloud/cloud_compaction_test.cpp‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -456,7 +456,7 @@ TEST_F(CloudCompactionTest, cumulative_global_lock_failure_keeps_thread_count_ba
456456
std::unique_lock lock(tablet->get_header_lock());
457457
tablet->add_rowsets(std::move(rowsets), false, lock);
458458
}
459-
tablet->last_sync_time_s = 1;
459+
tablet->last_sync_rowsets_time_s = 1;
460460
tablet->_approximate_num_rowsets = 0;
461461

462462
ASSERT_TRUE(ThreadPoolBuilder("CumuCompactionTaskThreadPoolTest")
@@ -579,16 +579,16 @@ class TestableCloudCompaction : public CloudCompactionMixin {
579579
TEST_F(CloudCompactionTest, cumulative_result_requires_next_counter) {
580580
auto tablet = std::make_shared<CloudTablet>(_engine, _tablet_meta);
581581
tablet->set_cumulative_compaction_cnt(1);
582-
tablet->last_sync_time_s = 1;
582+
tablet->last_sync_rowsets_time_s = 1;
583583
TestableCloudCompaction compaction(_engine, tablet);
584584

585585
std::unique_lock lock(tablet->get_header_lock());
586586
EXPECT_FALSE(compaction.test_should_apply_cumulative_compaction_result(1));
587-
EXPECT_EQ(tablet->last_sync_time_s, 1);
587+
EXPECT_EQ(tablet->last_sync_rowsets_time_s, 1);
588588
EXPECT_TRUE(compaction.test_should_apply_cumulative_compaction_result(2));
589-
EXPECT_EQ(tablet->last_sync_time_s, 1);
589+
EXPECT_EQ(tablet->last_sync_rowsets_time_s, 1);
590590
EXPECT_FALSE(compaction.test_should_apply_cumulative_compaction_result(3));
591-
EXPECT_EQ(tablet->last_sync_time_s, 0);
591+
EXPECT_EQ(tablet->last_sync_rowsets_time_s, 0);
592592
}
593593

594594
class TestableCloudCumulativeCompaction : public CloudCumulativeCompaction {
@@ -627,7 +627,7 @@ static CloudTabletSPtr create_cloud_tablet_with_rowsets(CloudStorageEngine& engi
627627
tablet->set_cumulative_layer_point(cumulative_point);
628628
tablet->fetch_add_approximate_num_rowsets(static_cast<int64_t>(num_rowsets) -
629629
tablet->fetch_add_approximate_num_rowsets(0));
630-
tablet->last_sync_time_s = 1;
630+
tablet->last_sync_rowsets_time_s = 1;
631631
return tablet;
632632
}
633633

@@ -694,7 +694,7 @@ TEST_F(CloudCompactionTest, base_result_with_newer_cumulative_point_forces_sync)
694694

695695
ASSERT_TRUE(compaction.modify_rowsets().ok());
696696
EXPECT_EQ(tablet->cumulative_layer_point(), 6);
697-
EXPECT_EQ(tablet->last_sync_time_s, expected_sync_time);
697+
EXPECT_EQ(tablet->last_sync_rowsets_time_s, expected_sync_time);
698698
};
699699

700700
run_case(10008, 8, 0);
@@ -903,7 +903,7 @@ TEST_F(CloudCompactionTest, parallel_pick_keeps_mode_after_dynamic_config_change
903903
EXPECT_TRUE(commit_called);
904904
EXPECT_EQ(tablet->cumulative_compaction_cnt(), 0);
905905
EXPECT_EQ(tablet->cumulative_layer_point(), 2);
906-
EXPECT_EQ(tablet->last_sync_time_s, 0);
906+
EXPECT_EQ(tablet->last_sync_rowsets_time_s, 0);
907907
}
908908

909909
TEST_F(CloudCompactionTest, parallel_pick_advances_continuous_low_prefix_through_delete) {

0 commit comments

Comments
 (0)