Skip to content

[feat](warm up) introduce immediate warmup on read cluster #54611

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zhannngchen merged 4 commits into from
Aug 18, 2025
Merged

[feat](warm up) introduce immediate warmup on read cluster #54611

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8187217
[feat](warm up) introduce immediate warmup on read cluster
Aug 12, 2025
3790fff
fix typo
Aug 12, 2025
2120175
remove useless code
Aug 12, 2025
79d7fee
Merge branch 'master' into warmup-immediately
dataroaring Aug 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions be/src/cloud/cloud_internal_service.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet_mgr.h"
#include "cloud/cloud_warm_up_manager.h"
#include "cloud/config.h"
#include "io/cache/block_file_cache.h"
#include "io/cache/block_file_cache_downloader.h"
Expand Down Expand Up @@ -219,6 +220,12 @@ void CloudInternalServiceImpl::warm_up_rowset(google::protobuf::RpcController* c
expiration_time = 0;
}

if (!tablet->add_rowset_warmup_state(rs_meta, WarmUpState::TRIGGERED_BY_JOB)) {
LOG(INFO) << "found duplicate warmup task for rowset " << rs_meta.rowset_id()
<< ", skip it";
continue;
}

for (int64_t segment_id = 0; segment_id < rs_meta.num_segments(); segment_id++) {
auto download_done = [&, tablet_id = rs_meta.tablet_id(),
rowset_id = rs_meta.rowset_id().to_string(),
Expand Down Expand Up @@ -252,6 +259,11 @@ void CloudInternalServiceImpl::warm_up_rowset(google::protobuf::RpcController* c
LOG(WARNING) << "download segment failed, tablet_id: " << tablet_id
<< " rowset_id: " << rowset_id << ", error: " << st;
}
if (tablet->complete_rowset_segment_warmup(rs_meta.rowset_id(), st) ==
WarmUpState::DONE) {
VLOG_DEBUG << "warmup rowset " << rs_meta.version() << "(" << rowset_id
<< ") completed";
}
if (wait) {
wait->signal();
}
Expand Down
3 changes: 2 additions & 1 deletion be/src/cloud/cloud_meta_mgr.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -737,7 +737,8 @@ Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet,
bool version_overlap =
tablet->max_version_unlocked() >= rowsets.front()->start_version();
tablet->add_rowsets(std::move(rowsets), version_overlap, wlock,
options.warmup_delta_data);
options.warmup_delta_data ||
config::enable_warmup_immediately_on_new_rowset);
}
tablet->last_base_compaction_success_time_ms = stats.last_base_compaction_time_ms();
tablet->last_cumu_compaction_success_time_ms = stats.last_cumu_compaction_time_ms();
Expand Down
72 changes: 71 additions & 1 deletion be/src/cloud/cloud_tablet.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,17 @@ bvar::Adder<uint64_t> g_file_cache_recycle_cached_data_segment_size(
bvar::Adder<uint64_t> g_file_cache_recycle_cached_data_index_num(
"file_cache_recycle_cached_data_index_num");

bvar::Adder<uint64_t> g_file_cache_warm_up_segment_complete_num(
"file_cache_warm_up_segment_complete_num");
bvar::Adder<uint64_t> g_file_cache_warm_up_segment_failed_num(
"file_cache_warm_up_segment_failed_num");
bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_complete_num(
"file_cache_warm_up_rowset_complete_num");
bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_triggered_by_job_num(
"file_cache_warm_up_rowset_triggered_by_job_num");
bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_triggered_by_sync_rowset_num(
"file_cache_warm_up_rowset_triggered_by_sync_rowset_num");

CloudTablet::CloudTablet(CloudStorageEngine& engine, TabletMetaSharedPtr tablet_meta)
: BaseTablet(std::move(tablet_meta)), _engine(engine) {}

Expand Down Expand Up @@ -243,6 +254,7 @@ void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_
for (auto& rs : rowsets) {
if (version_overlap || warmup_delta_data) {
#ifndef BE_TEST
bool warm_up_state_updated = false;
// Warmup rowset data in background
for (int seg_id = 0; seg_id < rs->num_segments(); ++seg_id) {
const auto& rowset_meta = rs->rowset_meta();
Expand Down Expand Up @@ -271,7 +283,19 @@ void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_
g_file_cache_cloud_tablet_submitted_segment_size
<< rs->rowset_meta()->segment_file_size(seg_id);
}
if (!warm_up_state_updated) {
VLOG_DEBUG << "warm up rowset " << rs->version() << "(" << rs->rowset_id()
<< ") triggerd by sync rowset";
if (!add_rowset_warmup_state_unlocked(
*(rs->rowset_meta()), WarmUpState::TRIGGERED_BY_SYNC_ROWSET)) {
LOG(INFO) << "found duplicate warmup task for rowset "
<< rs->rowset_id() << ", skip it";
break;
}
warm_up_state_updated = true;
}
// clang-format off
auto self = std::dynamic_pointer_cast<CloudTablet>(shared_from_this());
_engine.file_cache_block_downloader().submit_download_task(io::DownloadFileMeta {
.path = storage_resource.value()->remote_segment_path(*rowset_meta, seg_id),
.file_size = rs->rowset_meta()->segment_file_size(seg_id),
Expand All @@ -281,7 +305,8 @@ void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_
.expiration_time = expiration_time,
.is_dryrun = config::enable_reader_dryrun_when_download_file_cache,
},
.download_done {[](Status st) {
.download_done {[=](Status st) {
self->complete_rowset_segment_warmup(rowset_meta->rowset_id(), st);
if (!st) {
LOG_WARNING("add rowset warm up error ").error(st);
}
Expand Down Expand Up @@ -441,6 +466,7 @@ void CloudTablet::delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete,
_timestamped_version_tracker.add_stale_path_version(rs_metas);
for (auto&& rs : to_delete) {
_rs_version_map.erase(rs->version());
_rowset_warm_up_states.erase(rs->rowset_id());
}

_tablet_meta->modify_rs_metas({}, rs_metas, false);
Expand Down Expand Up @@ -1307,5 +1333,49 @@ Status CloudTablet::check_delete_bitmap_cache(int64_t txn_id,
return Status::OK();
}

WarmUpState CloudTablet::get_rowset_warmup_state(RowsetId rowset_id) {
std::shared_lock rlock(_meta_lock);
if (_rowset_warm_up_states.find(rowset_id) == _rowset_warm_up_states.end()) {
return WarmUpState::NONE;
}
return _rowset_warm_up_states[rowset_id].first;
}

bool CloudTablet::add_rowset_warmup_state(const RowsetMeta& rowset, WarmUpState state) {
std::lock_guard wlock(_meta_lock);
return add_rowset_warmup_state_unlocked(rowset, state);
}

bool CloudTablet::add_rowset_warmup_state_unlocked(const RowsetMeta& rowset, WarmUpState state) {
if (_rowset_warm_up_states.find(rowset.rowset_id()) != _rowset_warm_up_states.end()) {
return false;
}
if (state == WarmUpState::TRIGGERED_BY_JOB) {
g_file_cache_warm_up_rowset_triggered_by_job_num << 1;
} else if (state == WarmUpState::TRIGGERED_BY_SYNC_ROWSET) {
g_file_cache_warm_up_rowset_triggered_by_sync_rowset_num << 1;
}
_rowset_warm_up_states[rowset.rowset_id()] = std::make_pair(state, rowset.num_segments());
return true;
}

WarmUpState CloudTablet::complete_rowset_segment_warmup(RowsetId rowset_id, Status status) {
std::lock_guard wlock(_meta_lock);
if (_rowset_warm_up_states.find(rowset_id) == _rowset_warm_up_states.end()) {
return WarmUpState::NONE;
}
VLOG_DEBUG << "complete rowset segment warmup for rowset " << rowset_id << ", " << status;
g_file_cache_warm_up_segment_complete_num << 1;
if (!status.ok()) {
g_file_cache_warm_up_segment_failed_num << 1;
}
_rowset_warm_up_states[rowset_id].second--;
if (_rowset_warm_up_states[rowset_id].second <= 0) {
g_file_cache_warm_up_rowset_complete_num << 1;
_rowset_warm_up_states[rowset_id].first = WarmUpState::DONE;
}
return _rowset_warm_up_states[rowset_id].first;
}

#include "common/compile_check_end.h"
} // namespace doris
11 changes: 11 additions & 0 deletions be/src/cloud/cloud_tablet.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
namespace doris {

class CloudStorageEngine;
enum class WarmUpState : int;

struct SyncRowsetStats {
int64_t get_remote_rowsets_num {0};
Expand Down Expand Up @@ -289,12 +290,19 @@ class CloudTablet final : public BaseTablet {
static std::vector<RecycledRowsets> recycle_cached_data(
const std::vector<RowsetSharedPtr>& rowsets);

// Add warmup state management
WarmUpState get_rowset_warmup_state(RowsetId rowset_id);
bool add_rowset_warmup_state(const RowsetMeta& rowset, WarmUpState state);
WarmUpState complete_rowset_segment_warmup(RowsetId rowset_id, Status status);

private:
// FIXME(plat1ko): No need to record base size if rowsets are ordered by version
void update_base_size(const Rowset& rs);

Status sync_if_not_running(SyncRowsetStats* stats = nullptr);

bool add_rowset_warmup_state_unlocked(const RowsetMeta& rowset, WarmUpState state);

CloudStorageEngine& _engine;

// this mutex MUST ONLY be used when sync meta
Expand Down Expand Up @@ -350,6 +358,9 @@ class CloudTablet final : public BaseTablet {
std::mutex _gc_mutex;
std::unordered_map<RowsetId, RowsetSharedPtr> _unused_rowsets;
std::vector<std::pair<std::vector<RowsetId>, DeleteBitmapKeyRanges>> _unused_delete_bitmap;

// for warm up states management
std::unordered_map<RowsetId, std::pair<WarmUpState, int32_t>> _rowset_warm_up_states;
};

using CloudTabletSPtr = std::shared_ptr<CloudTablet>;
Expand Down
19 changes: 16 additions & 3 deletions be/src/cloud/cloud_warm_up_manager.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,7 @@ void CloudWarmUpManager::submit_download_tasks(io::Path path, int64_t file_size,
io::FileSystemSPtr file_system,
int64_t expiration_time,
std::shared_ptr<bthread::CountdownEvent> wait,
bool is_index) {
bool is_index, std::function<void(Status)> done_cb) {
if (file_size < 0) {
auto st = file_system->file_size(path, &file_size);
if (!st.ok()) [[unlikely]] {
Expand Down Expand Up @@ -145,7 +145,8 @@ void CloudWarmUpManager::submit_download_tasks(io::Path path, int64_t file_size,
.is_dryrun = config::enable_reader_dryrun_when_download_file_cache,
},
.download_done =
[=](Status st) {
[&](Status st) {
if (done_cb) done_cb(st);
if (!st) {
LOG_WARNING("Warm up error ").error(st);
} else if (is_index) {
Expand Down Expand Up @@ -225,12 +226,24 @@ void CloudWarmUpManager::handle_jobs() {
if (expiration_time <= UnixSeconds()) {
expiration_time = 0;
}
if (!tablet->add_rowset_warmup_state(*rs, WarmUpState::TRIGGERED_BY_JOB)) {
LOG(INFO) << "found duplicate warmup task for rowset " << rs->rowset_id()
<< ", skip it";
continue;
}

// 1st. download segment files
submit_download_tasks(
storage_resource.value()->remote_segment_path(*rs, seg_id),
rs->segment_file_size(seg_id), storage_resource.value()->fs,
expiration_time, wait);
expiration_time, wait, false, [tablet, rs, seg_id](Status st) {
VLOG_DEBUG << "warmup rowset " << rs->version() << " segment "
<< seg_id << " completed";
if (tablet->complete_rowset_segment_warmup(rs->rowset_id(), st) ==
WarmUpState::DONE) {
VLOG_DEBUG << "warmup rowset " << rs->version() << " completed";
}
});

// 2nd. download inverted index files
int64_t file_size = -1;
Expand Down
11 changes: 9 additions & 2 deletions be/src/cloud/cloud_warm_up_manager.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,13 @@ enum class DownloadType {
S3,
};

enum class WarmUpState : int {
NONE,
TRIGGERED_BY_SYNC_ROWSET,
TRIGGERED_BY_JOB,
DONE,
};

struct JobMeta {
JobMeta() = default;
JobMeta(const TJobMeta& meta);
Expand Down Expand Up @@ -91,8 +98,8 @@ class CloudWarmUpManager {

void submit_download_tasks(io::Path path, int64_t file_size, io::FileSystemSPtr file_system,
int64_t expiration_time,
std::shared_ptr<bthread::CountdownEvent> wait,
bool is_index = false);
std::shared_ptr<bthread::CountdownEvent> wait, bool is_index = false,
std::function<void(Status)> done_cb = nullptr);
std::mutex _mtx;
std::condition_variable _cond;
int64_t _cur_job_id {0};
Expand Down
2 changes: 2 additions & 0 deletions be/src/cloud/config.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,5 +101,7 @@ DEFINE_mInt64(warm_up_rowset_sync_wait_min_timeout_ms, "10000");

DEFINE_mInt64(warm_up_rowset_sync_wait_max_timeout_ms, "120000");

DEFINE_mBool(enable_warmup_immediately_on_new_rowset, "false");

#include "common/compile_check_end.h"
} // namespace doris::config
2 changes: 2 additions & 0 deletions be/src/cloud/config.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,5 +142,7 @@ DECLARE_mInt64(warm_up_rowset_sync_wait_min_timeout_ms);

DECLARE_mInt64(warm_up_rowset_sync_wait_max_timeout_ms);

DECLARE_mBool(enable_warmup_immediately_on_new_rowset);

#include "common/compile_check_end.h"
} // namespace doris::config
2 changes: 1 addition & 1 deletion be/src/olap/base_tablet.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ struct TabletWithVersion {
enum class CompactionStage { NOT_SCHEDULED, PENDING, EXECUTING };

// Base class for all tablet classes
class BaseTablet {
class BaseTablet : public std::enable_shared_from_this<BaseTablet> {
public:
explicit BaseTablet(TabletMetaSharedPtr tablet_meta);
virtual ~BaseTablet();
Expand Down
Loading
Loading