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 fe923893522 [fix](cloud) refresh tablet meta of continuously ingested
tablets (#67972)
fe923893522 is described below
commit fe923893522b884bef31e2972ed00a2339e59c60
Author: Xin Liao <[email protected]>
AuthorDate: Wed Sep 16 10:01:54 2026 +0800
[fix](cloud) refresh tablet meta of continuously ingested tablets (#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.
---
be/src/cloud/cloud_base_compaction.cpp | 6 +-
be/src/cloud/cloud_cumulative_compaction.cpp | 6 +-
be/src/cloud/cloud_full_compaction.cpp | 2 +-
be/src/cloud/cloud_index_change_compaction.cpp | 4 +-
be/src/cloud/cloud_meta_mgr.cpp | 2 +-
be/src/cloud/cloud_tablet.cpp | 1 +
be/src/cloud/cloud_tablet.h | 9 +-
be/src/cloud/cloud_tablet_mgr.cpp | 85 +++++++++++++----
be/src/storage/compaction/compaction.cpp | 2 +-
be/test/cloud/cloud_compaction_test.cpp | 16 ++--
be/test/cloud/cloud_tablet_mgr_test.cpp | 101 +++++++++++++++++++++
.../cloud_index_change_compaction_test.cpp | 4 +-
12 files changed, 196 insertions(+), 42 deletions(-)
diff --git a/be/src/cloud/cloud_base_compaction.cpp
b/be/src/cloud/cloud_base_compaction.cpp
index 87e2378fdba..3570abe492f 100644
--- a/be/src/cloud/cloud_base_compaction.cpp
+++ b/be/src/cloud/cloud_base_compaction.cpp
@@ -67,7 +67,7 @@ Status CloudBaseCompaction::prepare_compact() {
// synchronized with meta-service.
if (_tablet->tablet_meta()->all_rs_metas().size() >=
cloud_tablet()->fetch_add_approximate_num_rowsets(0) &&
- cloud_tablet()->last_sync_time_s > 0) {
+ cloud_tablet()->last_sync_rowsets_time_s > 0) {
need_sync_tablet = false;
}
}
@@ -136,7 +136,7 @@ Status CloudBaseCompaction::request_global_lock() {
cloud_tablet()->set_last_base_compaction_failure_time(UnixMillis());
if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
// set last_sync_time to 0 to force sync tablet next time
- cloud_tablet()->last_sync_time_s = 0;
+ cloud_tablet()->last_sync_rowsets_time_s = 0;
} else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
// tablet not found
cloud_tablet()->clear_cache();
@@ -440,7 +440,7 @@ Status CloudBaseCompaction::modify_rowsets() {
// the tablet to be unable to synchronize the rowset meta changes
generated by cumu compaction.
cloud_tablet()->set_base_compaction_cnt(stats.base_compaction_cnt());
if (stats.cumulative_point() >
cloud_tablet()->cumulative_layer_point()) {
- cloud_tablet()->last_sync_time_s = 0;
+ cloud_tablet()->last_sync_rowsets_time_s = 0;
}
if (output_rowset_delete_bitmap) {
_tablet->tablet_meta()->delete_bitmap().merge(*output_rowset_delete_bitmap);
diff --git a/be/src/cloud/cloud_cumulative_compaction.cpp
b/be/src/cloud/cloud_cumulative_compaction.cpp
index 42c97932c0b..10c00bebae1 100644
--- a/be/src/cloud/cloud_cumulative_compaction.cpp
+++ b/be/src/cloud/cloud_cumulative_compaction.cpp
@@ -157,7 +157,7 @@ Status CloudCumulativeCompaction::prepare_compact() {
// synchronized with meta-service.
if (_tablet->tablet_meta()->all_rs_metas().size() >=
cloud_tablet()->fetch_add_approximate_num_rowsets(0) &&
- cloud_tablet()->last_sync_time_s > 0) {
+ cloud_tablet()->last_sync_rowsets_time_s > 0) {
need_sync_tablet = false;
}
}
@@ -238,7 +238,7 @@ Status CloudCumulativeCompaction::request_global_lock() {
if (!st.ok()) {
if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
// set last_sync_time to 0 to force sync tablet next time
- cloud_tablet()->last_sync_time_s = 0;
+ cloud_tablet()->last_sync_rowsets_time_s = 0;
} else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
// tablet not found
cloud_tablet()->clear_cache();
@@ -925,7 +925,7 @@ void
CloudCumulativeCompaction::update_cumulative_point(int64_t input_cumulative
if (!st.ok()) {
if (start_resp.status().code() == cloud::STALE_TABLET_CACHE) {
// set last_sync_time to 0 to force sync tablet next time
- cloud_tablet()->last_sync_time_s = 0;
+ cloud_tablet()->last_sync_rowsets_time_s = 0;
} else if (start_resp.status().code() == cloud::TABLET_NOT_FOUND) {
// tablet not found
cloud_tablet()->clear_cache();
diff --git a/be/src/cloud/cloud_full_compaction.cpp
b/be/src/cloud/cloud_full_compaction.cpp
index 12ea9c87c99..1133330baa4 100644
--- a/be/src/cloud/cloud_full_compaction.cpp
+++ b/be/src/cloud/cloud_full_compaction.cpp
@@ -113,7 +113,7 @@ Status CloudFullCompaction::request_global_lock() {
if (!st.ok()) {
if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
// set last_sync_time to 0 to force sync tablet next time
- cloud_tablet()->last_sync_time_s = 0;
+ cloud_tablet()->last_sync_rowsets_time_s = 0;
} else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
// tablet not found
cloud_tablet()->clear_cache();
diff --git a/be/src/cloud/cloud_index_change_compaction.cpp
b/be/src/cloud/cloud_index_change_compaction.cpp
index 3bb55f46ce8..f9bbd5dea3d 100644
--- a/be/src/cloud/cloud_index_change_compaction.cpp
+++ b/be/src/cloud/cloud_index_change_compaction.cpp
@@ -170,7 +170,7 @@ Status
CloudIndexChangeCompaction::request_global_lock(bool& should_skip_err) {
if (!st.ok()) {
if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
// set last_sync_time to 0 to force sync tablet next time
- cloud_tablet()->last_sync_time_s = 0;
+ cloud_tablet()->last_sync_rowsets_time_s = 0;
should_skip_err = true;
} else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
// tablet not found
@@ -380,7 +380,7 @@ void
CloudIndexChangeCompaction::_update_tablet_for_base_compaction(
// the tablet to be unable to synchronize the rowset meta changes
generated by cumu compaction.
cloud_tablet()->set_base_compaction_cnt(stats.base_compaction_cnt());
if (stats.cumulative_point() >
cloud_tablet()->cumulative_layer_point()) {
- cloud_tablet()->last_sync_time_s = 0;
+ cloud_tablet()->last_sync_rowsets_time_s = 0;
}
if (output_rowset_delete_bitmap) {
_tablet->tablet_meta()->delete_bitmap().merge(*output_rowset_delete_bitmap);
diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp
index 688c18ecb9f..be2124d2c50 100644
--- a/be/src/cloud/cloud_meta_mgr.cpp
+++ b/be/src/cloud/cloud_meta_mgr.cpp
@@ -885,7 +885,7 @@ Status
CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet,
}
int64_t now =
duration_cast<seconds>(system_clock::now().time_since_epoch()).count();
- tablet->last_sync_time_s = now;
+ tablet->last_sync_rowsets_time_s = now;
if (sync_stats) {
sync_stats->get_remote_rowsets_rpc_ns +=
diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp
index da45f5e87db..de3394688af 100644
--- a/be/src/cloud/cloud_tablet.cpp
+++ b/be/src/cloud/cloud_tablet.cpp
@@ -1623,6 +1623,7 @@ Status CloudTablet::sync_meta() {
}
}
+ last_sync_tablet_meta_time_s = ::time(nullptr);
return Status::OK();
}
diff --git a/be/src/cloud/cloud_tablet.h b/be/src/cloud/cloud_tablet.h
index 96fa5e6faa5..6a2bfff05a1 100644
--- a/be/src/cloud/cloud_tablet.h
+++ b/be/src/cloud/cloud_tablet.h
@@ -348,7 +348,14 @@ public:
const auto& rowset_map() const { return _rs_version_map; }
- int64_t last_sync_time_s = 0;
+ // How long since this tablet's ROWSETS were pulled from MS. Only
sync_rowsets() advances
+ // it, and only when it actually issues the RPC.
+ int64_t last_sync_rowsets_time_s = 0;
+ // How long since this tablet's META was pulled from MS, which is what
carries properties
+ // such as the file cache TTL. Only sync_meta() advances it. Tracked
separately on
+ // purpose: a tablet under continuous ingest keeps the rowsets clock
permanently fresh,
+ // so gating meta work on that one starves the meta refresh entirely.
+ int64_t last_sync_tablet_meta_time_s = 0;
int64_t last_load_time_ms = 0;
int64_t last_base_compaction_success_time_ms = 0;
int64_t last_cumu_compaction_success_time_ms = 0;
diff --git a/be/src/cloud/cloud_tablet_mgr.cpp
b/be/src/cloud/cloud_tablet_mgr.cpp
index 82dd46a924d..d65c2fcbeef 100644
--- a/be/src/cloud/cloud_tablet_mgr.cpp
+++ b/be/src/cloud/cloud_tablet_mgr.cpp
@@ -19,7 +19,10 @@
#include <bthread/countdown_event.h>
+#include <algorithm>
#include <chrono>
+#include <set>
+#include <utility>
#include "cloud/cloud_cluster_info.h"
#include "cloud/cloud_meta_mgr.h"
@@ -40,6 +43,8 @@ bvar::Adder<uint64_t> g_base_compaction_not_frozen_tablet_num(
"base_compaction_not_frozen_tablet_num");
bvar::Adder<uint64_t> g_cumu_compaction_not_frozen_tablet_num(
"cumu_compaction_not_frozen_tablet_num");
+bvar::Adder<uint64_t> g_sync_tablets_meta_num("sync_tablets_meta_num");
+bvar::Adder<uint64_t> g_sync_tablets_rowsets_num("sync_tablets_rowsets_num");
namespace {
// port from
@@ -386,53 +391,93 @@ std::vector<std::weak_ptr<CloudTablet>>
CloudTabletMgr::get_weak_tablets() {
void CloudTabletMgr::sync_tablets(const CountDownLatch& stop_latch) {
LOG_INFO("begin to sync tablets");
- int64_t last_sync_time_bound = ::time(nullptr) -
config::tablet_sync_interval_s;
- auto weak_tablets = get_weak_tablets();
+ // A tablet carries two staleness clocks and each one gates a different
RPC:
+ //
+ // last_sync_rowsets_time_s
+ // how long since we pulled this tablet's ROWSETS from MS. Only
sync_rowsets()
+ // advances it, and only when it actually issues the RPC -- a query
whose requested
+ // version we already hold returns early and leaves the clock
untouched.
+ //
+ // last_sync_tablet_meta_time_s
+ // how long since we pulled this tablet's META from MS, which is
what carries
+ // properties such as the file cache TTL. Only sync_meta() advances
it.
+ //
+ // They have to be read separately. A tablet under continuous ingest keeps
the rowsets
+ // clock permanently fresh, so selecting meta work by it -- as this used
to -- means such a
+ // tablet never has its meta refreshed at all, and it keeps serving
whatever TTL it was
+ // built with.
+ const int64_t stale_before = ::time(nullptr) -
config::tablet_sync_interval_s;
+
+ struct Work {
+ std::weak_ptr<CloudTablet> tablet;
+ bool needs_meta = false;
+ bool needs_rowsets = false;
+ };
- // sort by last_sync_time
+ // Ordered by the older of the two clocks, so that if we are told to stop
half way, the
+ // tablets that have been waiting longest have already been served.
static auto cmp = [](const auto& a, const auto& b) { return a.first <
b.first; };
- std::multiset<std::pair<int64_t, std::weak_ptr<CloudTablet>>,
decltype(cmp)>
- sync_time_tablet_set(cmp);
+ std::multiset<std::pair<int64_t, Work>, decltype(cmp)> due(cmp);
- for (auto& weak_tablet : weak_tablets) {
- if (auto tablet = weak_tablet.lock()) {
- int64_t last_sync_time = tablet->last_sync_time_s;
- if (last_sync_time <= last_sync_time_bound) {
- sync_time_tablet_set.emplace(last_sync_time, weak_tablet);
- }
+ for (auto& weak_tablet : get_weak_tablets()) {
+ auto tablet = weak_tablet.lock();
+ if (!tablet) {
+ continue;
+ }
+ const bool needs_rowsets = tablet->last_sync_rowsets_time_s <=
stale_before;
+ Work work {
+ .tablet = weak_tablet,
+ // Pulling rowsets implies pulling the tablet meta: the
rowsets we are about
+ // to take are only as trustworthy as the meta they belong to,
and this is
+ // the relationship the previous single pass had.
+ .needs_meta = needs_rowsets ||
tablet->last_sync_tablet_meta_time_s <= stale_before,
+ .needs_rowsets = needs_rowsets};
+ if (!work.needs_meta && !work.needs_rowsets) {
+ continue;
}
+ due.emplace(
+ std::min(tablet->last_sync_tablet_meta_time_s,
tablet->last_sync_rowsets_time_s),
+ std::move(work));
}
int num_sync = 0;
- for (auto&& [_, weak_tablet] : sync_time_tablet_set) {
+ int num_sync_meta = 0;
+ for (auto&& [_, work] : due) {
if (stop_latch.count() <= 0) {
break;
}
+ auto tablet = work.tablet.lock();
+ if (!tablet) {
+ continue;
+ }
- if (auto tablet = weak_tablet.lock()) {
- if (tablet->last_sync_time_s > last_sync_time_bound) {
- continue;
- }
-
- ++num_sync;
+ if (work.needs_meta) {
+ ++num_sync_meta;
+ g_sync_tablets_meta_num << 1;
auto st = tablet->sync_meta();
if (!st) {
LOG_WARNING("failed to sync tablet meta {}",
tablet->tablet_id()).error(st);
if (st.is<ErrorCode::NOT_FOUND>()) {
+ // the tablet is gone from MS, there is nothing left to
sync
continue;
}
}
+ }
+
+ if (work.needs_rowsets) {
+ ++num_sync;
+ g_sync_tablets_rowsets_num << 1;
SyncOptions options;
options.query_version = -1;
options.merge_schema = true;
- st = tablet->sync_rowsets(options);
+ auto st = tablet->sync_rowsets(options);
if (!st) {
LOG_WARNING("failed to sync tablet rowsets {}",
tablet->tablet_id()).error(st);
}
}
}
- LOG_INFO("finish sync tablets").tag("num_sync", num_sync);
+ LOG_INFO("finish sync tablets").tag("num_sync",
num_sync).tag("num_sync_meta", num_sync_meta);
}
Status CloudTabletMgr::get_topn_tablets_to_compact(
diff --git a/be/src/storage/compaction/compaction.cpp
b/be/src/storage/compaction/compaction.cpp
index d4e2ec24030..015c6ee497a 100644
--- a/be/src/storage/compaction/compaction.cpp
+++ b/be/src/storage/compaction/compaction.cpp
@@ -2112,7 +2112,7 @@ bool
CloudCompactionMixin::should_apply_cumulative_compaction_result(
}
if (response_cumulative_compaction_cnt != local_cumulative_compaction_cnt
+ 1) {
// Only the current task's output is available locally. Sync all
missing outputs instead.
- cloud_tablet()->last_sync_time_s = 0;
+ cloud_tablet()->last_sync_rowsets_time_s = 0;
LOG_INFO("defer applying cumulative compaction result until tablet
sync")
.tag("tablet_id", _tablet->tablet_id())
.tag("job_id", _uuid)
diff --git a/be/test/cloud/cloud_compaction_test.cpp
b/be/test/cloud/cloud_compaction_test.cpp
index 9f0ac135bc0..683e72b3064 100644
--- a/be/test/cloud/cloud_compaction_test.cpp
+++ b/be/test/cloud/cloud_compaction_test.cpp
@@ -456,7 +456,7 @@ TEST_F(CloudCompactionTest,
cumulative_global_lock_failure_keeps_thread_count_ba
std::unique_lock lock(tablet->get_header_lock());
tablet->add_rowsets(std::move(rowsets), false, lock);
}
- tablet->last_sync_time_s = 1;
+ tablet->last_sync_rowsets_time_s = 1;
tablet->_approximate_num_rowsets = 0;
ASSERT_TRUE(ThreadPoolBuilder("CumuCompactionTaskThreadPoolTest")
@@ -579,16 +579,16 @@ public:
TEST_F(CloudCompactionTest, cumulative_result_requires_next_counter) {
auto tablet = std::make_shared<CloudTablet>(_engine, _tablet_meta);
tablet->set_cumulative_compaction_cnt(1);
- tablet->last_sync_time_s = 1;
+ tablet->last_sync_rowsets_time_s = 1;
TestableCloudCompaction compaction(_engine, tablet);
std::unique_lock lock(tablet->get_header_lock());
EXPECT_FALSE(compaction.test_should_apply_cumulative_compaction_result(1));
- EXPECT_EQ(tablet->last_sync_time_s, 1);
+ EXPECT_EQ(tablet->last_sync_rowsets_time_s, 1);
EXPECT_TRUE(compaction.test_should_apply_cumulative_compaction_result(2));
- EXPECT_EQ(tablet->last_sync_time_s, 1);
+ EXPECT_EQ(tablet->last_sync_rowsets_time_s, 1);
EXPECT_FALSE(compaction.test_should_apply_cumulative_compaction_result(3));
- EXPECT_EQ(tablet->last_sync_time_s, 0);
+ EXPECT_EQ(tablet->last_sync_rowsets_time_s, 0);
}
class TestableCloudCumulativeCompaction : public CloudCumulativeCompaction {
@@ -627,7 +627,7 @@ static CloudTabletSPtr
create_cloud_tablet_with_rowsets(CloudStorageEngine& engi
tablet->set_cumulative_layer_point(cumulative_point);
tablet->fetch_add_approximate_num_rowsets(static_cast<int64_t>(num_rowsets) -
tablet->fetch_add_approximate_num_rowsets(0));
- tablet->last_sync_time_s = 1;
+ tablet->last_sync_rowsets_time_s = 1;
return tablet;
}
@@ -694,7 +694,7 @@ TEST_F(CloudCompactionTest,
base_result_with_newer_cumulative_point_forces_sync)
ASSERT_TRUE(compaction.modify_rowsets().ok());
EXPECT_EQ(tablet->cumulative_layer_point(), 6);
- EXPECT_EQ(tablet->last_sync_time_s, expected_sync_time);
+ EXPECT_EQ(tablet->last_sync_rowsets_time_s, expected_sync_time);
};
run_case(10008, 8, 0);
@@ -903,7 +903,7 @@ TEST_F(CloudCompactionTest,
parallel_pick_keeps_mode_after_dynamic_config_change
EXPECT_TRUE(commit_called);
EXPECT_EQ(tablet->cumulative_compaction_cnt(), 0);
EXPECT_EQ(tablet->cumulative_layer_point(), 2);
- EXPECT_EQ(tablet->last_sync_time_s, 0);
+ EXPECT_EQ(tablet->last_sync_rowsets_time_s, 0);
}
TEST_F(CloudCompactionTest,
parallel_pick_advances_continuous_low_prefix_through_delete) {
diff --git a/be/test/cloud/cloud_tablet_mgr_test.cpp
b/be/test/cloud/cloud_tablet_mgr_test.cpp
index c893b72a475..2859a476e3e 100644
--- a/be/test/cloud/cloud_tablet_mgr_test.cpp
+++ b/be/test/cloud/cloud_tablet_mgr_test.cpp
@@ -26,10 +26,15 @@
#include <memory>
#include <mutex>
#include <thread>
+#include <unordered_map>
+#include <vector>
#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablet.h"
+#include "cloud/config.h"
#include "cpp/sync_point.h"
#include "storage/tablet/tablet_meta.h"
+#include "util/countdown_latch.h"
#include "util/uid_util.h"
namespace doris {
@@ -210,4 +215,100 @@ TEST_F(CloudTabletMgrTest,
TestGetTabletIfCachedOnlyReturnsCachedTablet) {
sp->clear_all_call_backs();
}
+// A tablet under continuous ingest keeps last_sync_rowsets_time_s permanently
fresh, because every
+// rowset sync advances it. Selecting meta work by that same clock meant such
a tablet never had
+// sync_meta() called on it at all, so it kept serving the tablet properties
-- the file cache
+// TTL among them -- that it happened to be built with.
+TEST_F(CloudTabletMgrTest,
SyncTabletsRefreshesMetaOfContinuouslyIngestedTablet) {
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ std::mutex mutex;
+ std::unordered_map<int64_t, TabletMetaSharedPtr> metas;
+ std::unordered_map<int64_t, int> meta_syncs;
+ std::unordered_map<int64_t, int> rowset_syncs;
+
+ auto meta_for = [&](int64_t tablet_id) {
+ std::lock_guard<std::mutex> lock(mutex);
+ auto it = metas.find(tablet_id);
+ if (it == metas.end()) {
+ it = metas.emplace(tablet_id, std::make_shared<TabletMeta>(
+ 1, 2, tablet_id, 15674, 4,
5, TTabletSchema(), 6,
+ std::unordered_map<uint32_t,
uint32_t> {{7, 8}},
+ UniqueId(9, 10),
TTabletType::TABLET_TYPE_DISK,
+ TCompressionType::LZ4F))
+ .first;
+ }
+ return it->second;
+ };
+
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta", [&](auto&& args) {
+ auto tablet_id = try_any_cast<int64_t>(args[0]);
+ auto* tablet_meta_ptr = try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ *tablet_meta_ptr = meta_for(tablet_id);
+ {
+ std::lock_guard<std::mutex> lock(mutex);
+ ++meta_syncs[tablet_id];
+ }
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+ sp->set_call_back("CloudMetaMgr::sync_tablet_rowsets", [&](auto&& args) {
+ auto* tablet = try_any_cast<CloudTablet*>(args[0]);
+ {
+ std::lock_guard<std::mutex> lock(mutex);
+ ++rowset_syncs[tablet->tablet_id()];
+ }
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+
+ CloudTabletMgr mgr(_engine);
+ constexpr int64_t kBothStale = 70001;
+ constexpr int64_t kIngesting = 70002;
+ constexpr int64_t kFresh = 70003;
+
+ std::vector<std::shared_ptr<CloudTablet>> tablets;
+ for (int64_t tablet_id : {kBothStale, kIngesting, kFresh}) {
+ auto res = mgr.get_tablet(tablet_id);
+ ASSERT_TRUE(res.has_value()) << res.error();
+ tablets.push_back(res.value());
+ }
+
+ const int64_t now = ::time(nullptr);
+ const int64_t stale = now - config::tablet_sync_interval_s - 10;
+
+ tablets[0]->last_sync_rowsets_time_s = stale;
+ tablets[0]->last_sync_tablet_meta_time_s = stale;
+ // Rowsets pulled a moment ago, meta left behind.
+ tablets[1]->last_sync_rowsets_time_s = now;
+ tablets[1]->last_sync_tablet_meta_time_s = stale;
+ tablets[2]->last_sync_rowsets_time_s = now;
+ tablets[2]->last_sync_tablet_meta_time_s = now;
+
+ {
+ std::lock_guard<std::mutex> lock(mutex);
+ meta_syncs.clear();
+ rowset_syncs.clear();
+ }
+
+ CountDownLatch latch(1);
+ mgr.sync_tablets(latch);
+
+ std::lock_guard<std::mutex> lock(mutex);
+ EXPECT_EQ(1, meta_syncs[kBothStale]);
+ EXPECT_EQ(1, rowset_syncs[kBothStale]);
+
+ // The regression: meta has to be refreshed even though the rowset clock
never goes stale,
+ // and it costs one RPC rather than the two a full sync would.
+ EXPECT_EQ(1, meta_syncs[kIngesting]);
+ EXPECT_EQ(0, rowset_syncs[kIngesting]);
+ EXPECT_GE(tablets[1]->last_sync_tablet_meta_time_s, now);
+
+ EXPECT_EQ(0, meta_syncs[kFresh]);
+ EXPECT_EQ(0, rowset_syncs[kFresh]);
+
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+}
+
} // namespace doris
diff --git a/be/test/storage/compaction/cloud_index_change_compaction_test.cpp
b/be/test/storage/compaction/cloud_index_change_compaction_test.cpp
index 62e5f27d7a4..cf0d8871d2f 100644
--- a/be/test/storage/compaction/cloud_index_change_compaction_test.cpp
+++ b/be/test/storage/compaction/cloud_index_change_compaction_test.cpp
@@ -421,7 +421,7 @@ TEST_F(CloudIndexChangeCompactionTest, ms_ret_status_test) {
{
tablet->set_base_compaction_cnt(0);
tablet->set_cumulative_layer_point(6);
- tablet->last_sync_time_s = 1;
+ tablet->last_sync_rowsets_time_s = 1;
auto index_change_compact =
std::make_shared<CloudIndexChangeCompaction>(
*_engine, tablet, 0, index_list, columns);
index_change_compact->_input_rowsets.push_back(rowset_ptr);
@@ -435,7 +435,7 @@ TEST_F(CloudIndexChangeCompactionTest, ms_ret_status_test) {
index_change_compact->_update_tablet_for_base_compaction(response,
nullptr);
EXPECT_EQ(tablet->cumulative_layer_point(), 6);
- EXPECT_EQ(tablet->last_sync_time_s, 0);
+ EXPECT_EQ(tablet->last_sync_rowsets_time_s, 0);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
