This is an automated email from the ASF dual-hosted git repository.
liaoxin01 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 e7b7f1d1359 [opt](cloud) Add warm up job count metric on BE (#64734)
e7b7f1d1359 is described below
commit e7b7f1d1359a46622dbdc98a493681674698e73e
Author: Xin Liao <[email protected]>
AuthorDate: Mon Jul 27 20:52:48 2026 +0800
[opt](cloud) Add warm up job count metric on BE (#64734)
## Proposed changes
Add a `file_cache_warm_up_job_num` bvar metric that tracks the number of
warm up jobs currently held in a BE's memory. This gives operators
per-BE visibility into how many warm up jobs each backend is currently
holding.
### What changed
In `be/src/cloud/cloud_warm_up_manager.cpp`:
- New `bvar::Adder<int64_t>
g_file_cache_warm_up_job_num("file_cache_warm_up_job_num")`.
- **+1** when FE dispatches a new job to this BE:
- `check_and_set_job_id` — regular (cluster/table) warm up `SET_JOB`, on
`_cur_job_id` transition `0 -> job_id`.
- `check_and_set_batch_id` — defensive, same `0 -> job_id` transition
(e.g. if a `SET_BATCH` were to arrive first).
- `set_event` — event-driven `SET_JOB`, when a new `job_id` is inserted
into `_tablet_replica_cache`.
- **-1** when the job is cleared:
- `clear_job` — regular `CLEAR_JOB`, only when a live job actually
existed.
- `set_event` (clear) — event-driven `CLEAR_JOB`, only when
`_tablet_replica_cache.erase()` actually removed an entry.
---
be/src/cloud/cloud_warm_up_manager.cpp | 22 ++++++++++++++++++++--
1 file changed, 20 insertions(+), 2 deletions(-)
diff --git a/be/src/cloud/cloud_warm_up_manager.cpp
b/be/src/cloud/cloud_warm_up_manager.cpp
index e6fad053f07..dc7ddc01b59 100644
--- a/be/src/cloud/cloud_warm_up_manager.cpp
+++ b/be/src/cloud/cloud_warm_up_manager.cpp
@@ -109,6 +109,10 @@ bvar::Status<int64_t>
g_file_cache_warm_up_rowset_last_call_unix_ts(
"file_cache_warm_up_rowset_last_call_unix_ts", 0);
bvar::Adder<uint64_t> file_cache_warm_up_failed_task_num("file_cache_warm_up",
"failed_task_num");
bvar::Adder<int64_t>
g_balance_tablet_be_mapping_size("balance_tablet_be_mapping_size");
+// Number of warm up jobs currently held in this BE's memory.
+// Incremented when FE dispatches a new job to this BE (SET_JOB / SET_BATCH /
event SET_JOB),
+// decremented when the job is cleared (CLEAR_JOB / event CLEAR_JOB).
+bvar::Adder<int64_t>
g_file_cache_warm_up_job_num("file_cache_warm_up_job_num");
bvar::LatencyRecorder g_file_cache_warm_up_rowset_wait_for_compaction_latency(
"file_cache_warm_up_rowset_wait_for_compaction_latency");
@@ -440,6 +444,7 @@ Status CloudWarmUpManager::check_and_set_job_id(int64_t
job_id) {
std::lock_guard lock(_mtx);
if (_cur_job_id == 0) {
_cur_job_id = job_id;
+ g_file_cache_warm_up_job_num << 1;
}
Status st = Status::OK();
if (_cur_job_id != job_id) {
@@ -458,6 +463,7 @@ Status CloudWarmUpManager::check_and_set_batch_id(int64_t
job_id, int64_t batch_
}
if (_cur_job_id == 0) {
_cur_job_id = job_id;
+ g_file_cache_warm_up_job_num << 1;
}
if (_cur_batch_id == batch_id) {
*retry = true;
@@ -503,6 +509,9 @@ Status CloudWarmUpManager::clear_job(int64_t job_id) {
std::lock_guard lock(_mtx);
Status st = Status::OK();
if (job_id == _cur_job_id) {
+ if (_cur_job_id != 0) {
+ g_file_cache_warm_up_job_num << -1;
+ }
_cur_job_id = 0;
_cur_batch_id = -1;
_pending_job_metas.clear();
@@ -525,11 +534,14 @@ Status CloudWarmUpManager::set_event(int64_t job_id,
TWarmUpEventType::type even
Status st = Status::OK();
if (event == TWarmUpEventType::type::LOAD) {
if (clear) {
- _tablet_replica_cache.erase(job_id);
+ if (_tablet_replica_cache.erase(job_id) > 0) {
+ g_file_cache_warm_up_job_num << -1;
+ }
_event_driven_filters.erase(job_id);
LOG(INFO) << "Clear event driven sync, job_id=" << job_id << ",
event=" << event;
} else if (!_tablet_replica_cache.contains(job_id)) {
static_cast<void>(_tablet_replica_cache[job_id]);
+ g_file_cache_warm_up_job_num << 1;
if (table_ids != nullptr) {
// table-level filter: set to the given table_id set (may be
empty,
// meaning all matched tables were deleted — warm up nothing)
@@ -663,7 +675,13 @@ std::vector<JobReplicaInfo>
CloudWarmUpManager::get_replica_info(int64_t tablet_
}
for (auto job_id : cancelled_jobs) {
LOG(INFO) << "get_replica_info: erasing cancelled job, job_id=" <<
job_id;
- _tablet_replica_cache.erase(job_id);
+ // Lazy cleanup path: FE reported the warm up job as CANCELLED, so the
job is
+ // no longer held in memory. Keep the job-count metric and the event
filter
+ // consistent with the explicit CLEAR_JOB path in set_event().
+ if (_tablet_replica_cache.erase(job_id) > 0) {
+ g_file_cache_warm_up_job_num << -1;
+ }
+ _event_driven_filters.erase(job_id);
}
VLOG_DEBUG << "get_replica_info: return " << replicas.size()
<< " replicas, tablet id=" << tablet_id;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
