This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new d5c3d056d25 [improvement](tablet) Source active tablet stats from BE
reports (#67621) (#70111)
d5c3d056d25 is described below
commit d5c3d056d25cffb590122d2ead2d4845ecdd17f9
Author: deardeng <[email protected]>
AuthorDate: Thu Nov 5 09:21:33 2026 +0800
[improvement](tablet) Source active tablet stats from BE reports (#67621)
(#70111)
pick from https://github.com/apache/doris/pull/67621
Problem Summary:
FE tracks tablet activity with per-tablet sliding windows, consuming
substantial heap and retaining stale entries on followers and observers.
Use BE query and load counters to report bounded sets of active tablets.
Replace FE sliding windows and asynchronous recording with per-BE report
snapshots, and rank query and load activity separately for cloud
balancing. Also fix inflated tablet counts when cloud tablet reports
retry.
Reduce FE memory usage for tablet activity tracking. Rename
`WindowAccessCount` to `AccessCount`, reflecting activity over a BE
report interval. Tablets outside the reported top-N show zero activity.
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/agent/task_worker_pool.cpp | 57 +-
be/src/cloud/cloud_tablet_mgr.cpp | 14 +-
be/src/cloud/cloud_tablet_mgr.h | 4 +-
be/src/common/config.cpp | 2 +
be/src/common/config.h | 14 +
be/src/exec/scan/olap_scanner.cpp | 5 +
be/src/load/delta_writer/delta_writer.cpp | 27 +-
be/src/service/point_query_executor.cpp | 6 +
be/src/storage/active_tablet_stats.cpp | 109 ++++
be/src/storage/active_tablet_stats.h | 87 +++
be/src/storage/tablet/base_tablet.cpp | 13 +
be/src/storage/tablet/base_tablet.h | 26 +
be/src/storage/tablet/tablet_manager.cpp | 10 +-
be/src/storage/tablet/tablet_manager.h | 4 +-
be/src/storage/tablet/tablet_meta.h | 4 +-
be/test/storage/active_tablet_stats_test.cpp | 316 +++++++++
.../main/java/org/apache/doris/common/Config.java | 13 +-
.../main/java/org/apache/doris/catalog/Tablet.java | 2 -
.../catalog/TabletSlidingWindowAccessStats.java | 706 +++++++++------------
.../apache/doris/cloud/catalog/CloudTablet.java | 3 -
.../doris/cloud/system/CloudSystemInfoService.java | 2 +
.../org/apache/doris/master/ReportHandler.java | 11 +
.../org/apache/doris/system/SystemInfoService.java | 3 +
.../TabletSlidingWindowAccessStatsTest.java | 532 +++++++++++-----
gensrc/thrift/MasterService.thrift | 23 +
25 files changed, 1423 insertions(+), 570 deletions(-)
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 013f34112f7..c9e7ab75c7e 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -72,6 +72,7 @@
#include "runtime/snapshot_loader.h"
#include "runtime/user_function_cache.h"
#include "service/backend_options.h"
+#include "storage/active_tablet_stats.h"
#include "storage/compaction/cumulative_compaction_time_series_policy.h"
#include "storage/data_dir.h"
#include "storage/olap_common.h"
@@ -119,6 +120,29 @@ void increase_report_version() {
s_report_version.fetch_add(1, std::memory_order_relaxed);
}
+enum class ActiveDim { QUERY, LOAD };
+
+// The two top-N lists differ only in which pair of thrift fields carries the
dimension;
+// tablet_id and delta_window_ms are common, so they are written once here.
+template <ActiveDim Dim>
+std::vector<TActiveTabletStat> to_thrift(const
std::vector<ActiveTabletCandidate>& candidates) {
+ std::vector<TActiveTabletStat> result;
+ result.reserve(candidates.size());
+ for (const auto& candidate : candidates) {
+ auto& stat = result.emplace_back();
+ stat.__set_tablet_id(candidate.tablet_id);
+ if constexpr (Dim == ActiveDim::QUERY) {
+ stat.__set_scan_count_delta(candidate.delta);
+ stat.__set_last_query_time_ms(candidate.last_time_ms);
+ } else {
+ stat.__set_load_count_delta(candidate.delta);
+ stat.__set_last_load_time_ms(candidate.last_time_ms);
+ }
+ stat.__set_delta_window_ms(candidate.window_ms);
+ }
+ return result;
+}
+
// FIXME(plat1ko): Paired register and remove task info
bool register_task_info(const TTaskType::type task_type, int64_t signature) {
if (task_type == TTaskType::type::PUSH_STORAGE_POLICY ||
@@ -521,6 +545,7 @@ bvar::Adder<uint64_t> report_disk_total("report",
"disk_total");
bvar::Adder<uint64_t> report_disk_failed("report", "disk_failed");
bvar::Adder<uint64_t> report_tablet_total("report", "tablet_total");
bvar::Adder<uint64_t> report_tablet_failed("report", "tablet_failed");
+bvar::Adder<uint64_t> report_active_tablet_truncated("report",
"active_tablet_truncated");
bvar::Adder<uint64_t> report_index_policy_total("report",
"index_policy_total");
bvar::Adder<uint64_t> report_index_policy_failed("report",
"index_policy_failed");
@@ -1171,10 +1196,12 @@ void report_tablet_callback(StorageEngine& engine,
const ClusterInfo* cluster_in
increase_report_version();
uint64_t report_version;
+ ActiveTabletCollector active;
for (int i = 0; i < 5; i++) {
request.tablets.clear();
+ active.clear();
report_version = s_report_version;
-
engine.tablet_manager()->build_all_report_tablets_info(&request.tablets);
+
engine.tablet_manager()->build_all_report_tablets_info(&request.tablets,
&active);
if (report_version == s_report_version) {
break;
}
@@ -1222,7 +1249,17 @@ void report_tablet_callback(StorageEngine& engine, const
ClusterInfo* cluster_in
}
request.__isset.resource = true;
+ active.take_top_n();
+
request.__set_top_query_tablets(to_thrift<ActiveDim::QUERY>(active.query_candidates()));
+
request.__set_top_load_tablets(to_thrift<ActiveDim::LOAD>(active.load_candidates()));
+ if (active.truncated()) {
+ report_active_tablet_truncated << 1;
+ }
+
bool succ = handle_report(request, cluster_info, "tablet");
+ if (succ) {
+ active.commit();
+ }
report_tablet_total << 1;
if (!succ) [[unlikely]] {
report_tablet_failed << 1;
@@ -1244,10 +1281,16 @@ void report_tablet_callback(CloudStorageEngine& engine,
const ClusterInfo* clust
increase_report_version();
uint64_t report_version;
uint64_t total_num_tablets = 0;
+ ActiveTabletCollector active;
for (int i = 0; i < 5; i++) {
request.tablets.clear();
+ // build_all_report_tablets_info() only increments tablet_num, so a
retry would
+ // otherwise report a multiple of the real tablet count to FE.
+ total_num_tablets = 0;
+ active.clear();
report_version = s_report_version;
- engine.tablet_mgr().build_all_report_tablets_info(&request.tablets,
&total_num_tablets);
+ engine.tablet_mgr().build_all_report_tablets_info(&request.tablets,
&total_num_tablets,
+ &active);
if (report_version == s_report_version) {
break;
}
@@ -1262,7 +1305,17 @@ void report_tablet_callback(CloudStorageEngine& engine,
const ClusterInfo* clust
request.__set_report_version(report_version);
request.__set_num_tablets(total_num_tablets);
+ active.take_top_n();
+
request.__set_top_query_tablets(to_thrift<ActiveDim::QUERY>(active.query_candidates()));
+
request.__set_top_load_tablets(to_thrift<ActiveDim::LOAD>(active.load_candidates()));
+ if (active.truncated()) {
+ report_active_tablet_truncated << 1;
+ }
+
bool succ = handle_report(request, cluster_info, "tablet");
+ if (succ) {
+ active.commit();
+ }
report_tablet_total << 1;
if (!succ) [[unlikely]] {
report_tablet_failed << 1;
diff --git a/be/src/cloud/cloud_tablet_mgr.cpp
b/be/src/cloud/cloud_tablet_mgr.cpp
index 3e5c7d0b764..a4f209ace17 100644
--- a/be/src/cloud/cloud_tablet_mgr.cpp
+++ b/be/src/cloud/cloud_tablet_mgr.cpp
@@ -32,6 +32,7 @@
#include "common/status.h"
#include "cpp/sync_point.h"
#include "runtime/memory/cache_policy.h"
+#include "storage/active_tablet_stats.h"
#include "storage/compaction/cumulative_compaction_time_series_policy.h"
#include "util/debug_points.h"
#include "util/lru_cache.h"
@@ -591,8 +592,12 @@ Status CloudTabletMgr::get_topn_tablets_to_compact(
}
void CloudTabletMgr::build_all_report_tablets_info(std::map<TTabletId,
TTablet>* tablets_info,
- uint64_t* tablet_num) {
+ uint64_t* tablet_num,
+ ActiveTabletCollector*
active) {
DCHECK(tablets_info != nullptr);
+ if (active) {
+ active->start();
+ }
VLOG_NOTICE << "begin to build all report cloud tablets info";
HistogramStat tablet_version_num_hist;
@@ -601,6 +606,13 @@ void
CloudTabletMgr::build_all_report_tablets_info(std::map<TTabletId, TTablet>*
auto tablet = tablet_wk.lock();
if (!tablet) return;
(*tablet_num)++;
+ if (active) {
+ // Known limitation: the counters live on the cached CloudTablet,
so a tablet
+ // evicted from the cache before its first report takes its
activity with it and
+ // FE sees it as cold. Not fixed - keeping a sample independent of
the metadata
+ // cache is a lifetime redesign, and this is a scheduling hint,
not accounting.
+ active->collect(tablet);
+ }
TTabletInfo tablet_info;
tablet->build_tablet_report_info(&tablet_info);
using namespace std::chrono;
diff --git a/be/src/cloud/cloud_tablet_mgr.h b/be/src/cloud/cloud_tablet_mgr.h
index 5513468809b..bf380a9e5fe 100644
--- a/be/src/cloud/cloud_tablet_mgr.h
+++ b/be/src/cloud/cloud_tablet_mgr.h
@@ -32,6 +32,7 @@ namespace doris {
class CloudTablet;
class CloudStorageEngine;
+class ActiveTabletCollector;
class LRUCachePolicy;
class CountDownLatch;
struct SyncRowsetStats;
@@ -96,7 +97,8 @@ public:
* @param tablet_num tablets in be tabletMgr, total num
*/
void build_all_report_tablets_info(std::map<TTabletId, TTablet>*
tablets_info,
- uint64_t* tablet_num);
+ uint64_t* tablet_num,
+ ActiveTabletCollector* active =
nullptr);
void get_tablet_info(int64_t num_tablets, std::vector<TabletInfo>*
tablets_info);
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 12cefd73884..106e578ebc2 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -274,6 +274,8 @@ DEFINE_mInt32(report_task_interval_seconds, "10");
DEFINE_mInt32(report_disk_state_interval_seconds, "30");
// the interval time(seconds) for agent report olap table to FE
DEFINE_mInt32(report_tablet_interval_seconds, "60");
+DEFINE_mInt32(report_active_tablet_max_num, "5000");
+DEFINE_mInt32(report_active_tablet_window_second, "3600");
// the max download speed(KB/s)
DEFINE_mInt32(max_download_speed_kbps, "50000");
// download low speed limit(KB/s)
diff --git a/be/src/common/config.h b/be/src/common/config.h
index e61abf25db5..7910e627cd8 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -329,6 +329,20 @@ DECLARE_mInt32(report_task_interval_seconds);
DECLARE_mInt32(report_disk_state_interval_seconds);
// the interval time(seconds) for agent report olap table to FE
DECLARE_mInt32(report_tablet_interval_seconds);
+// Per-list cap: a report carries at most 2 * report_active_tablet_max_num
entries
+// (one top-N list for queries, one for loads).
+//
+// Sizing against fe cloud_active_partition_scheduling_topn: FE reserves only
topn/2 for the
+// query bucket and backfills whatever the load bucket leaves free from the
query list, so
+// one dimension can claim the whole budget. The worst case -- the cluster's
entire hot set
+// on a single BE -- therefore wants this >= topn. The default is deliberately
half of the
+// FE default instead: FE merges across backends, so it takes a one-BE hot set
for the cap
+// to bind at all, and paying 2x the report size on every BE in every cluster
to cover that
+// is not worth it. Raise it if you raise the FE value, or if a BE reports
+// report.active_tablet_truncated far above zero.
+DECLARE_mInt32(report_active_tablet_max_num);
+// Only tablets queried / loaded within this window are reported as active.
+DECLARE_mInt32(report_active_tablet_window_second);
// the max download speed(KB/s)
DECLARE_mInt32(max_download_speed_kbps);
// download low speed limit(KB/s)
diff --git a/be/src/exec/scan/olap_scanner.cpp
b/be/src/exec/scan/olap_scanner.cpp
index 8e12cfad762..fd6eb28c780 100644
--- a/be/src/exec/scan/olap_scanner.cpp
+++ b/be/src/exec/scan/olap_scanner.cpp
@@ -65,6 +65,7 @@
#include "storage/storage_engine.h"
#include "storage/tablet/tablet_schema.h"
#include "util/json/path_in_data.h"
+#include "util/time.h"
namespace doris {
#include "common/compile_check_avoid_begin.h"
@@ -994,6 +995,10 @@ void OlapScanner::_collect_profile_before_close() {
auto& tablet = _tablet_reader_params.tablet;
tablet->query_scan_bytes->increment(local_state->_read_uncompressed_counter->value());
tablet->query_scan_rows->increment(local_state->_scan_rows->value());
+ // Timestamp first: the report walk reads the counter and the timestamp
without a
+ // lock, so the reverse order lets it see a fresh delta paired with a
stale timestamp
+ // and drop the tablet as inactive.
+ tablet->last_query_scan_time_ms.store(UnixMillis(),
std::memory_order_relaxed);
tablet->query_scan_count->increment(1);
COUNTER_UPDATE(local_state->_ann_range_search_filter_counter,
diff --git a/be/src/load/delta_writer/delta_writer.cpp
b/be/src/load/delta_writer/delta_writer.cpp
index 89b174411c1..482d087263d 100644
--- a/be/src/load/delta_writer/delta_writer.cpp
+++ b/be/src/load/delta_writer/delta_writer.cpp
@@ -23,6 +23,7 @@
#include <gen_cpp/internal_service.pb.h>
#include <gen_cpp/olap_file.pb.h>
+#include <chrono>
#include <filesystem>
#include <ostream>
#include <string>
@@ -103,10 +104,30 @@ BaseDeltaWriter::~BaseDeltaWriter() {
return;
}
- if (_rowset_builder->tablet() != nullptr) {
+ const BaseTabletSPtr& flushed_tablet = _rowset_builder->tablet();
+ if (flushed_tablet != nullptr) {
const FlushStatistic& stat = _memtable_writer->get_flush_token_stats();
-
_rowset_builder->tablet()->flush_bytes->increment(stat.flush_size_bytes);
-
_rowset_builder->tablet()->flush_finish_count->increment(stat.flush_finish_count);
+ flushed_tablet->flush_bytes->increment(stat.flush_size_bytes);
+ // Stamped here and nowhere else, so the load timestamp and the
counter it
+ // accompanies always move together: a non-zero flush delta always
comes with a
+ // fresh timestamp, and the active-window filter can only ever retire
a stale
+ // delta, never invent one. Timestamp first, for the same reason as
the query
+ // path -- the report walk reads both without a lock.
+ //
+ // Known limitation: DeltaWriterV2 (enable_memtable_on_sink_node,
local mode only)
+ // commits through LoadStreamWriter and never reaches this destructor,
so those loads
+ // are missing from SHOW TABLET / PROC. Not fixed: the only scheduling
consumer is the
+ // cloud rebalancer, and V2 is excluded from cloud.
+ //
+ // Only when this writer actually flushed something. close()
initializes an
+ // empty-rowset writer for every tablet of a touched partition, and
those flush
+ // nothing. Stamping unconditionally would let such a writer refresh
the timestamp
+ // of a delta that is still uncommitted because reports have been
failing, and the
+ // next successful report would then resurrect that whole stale delta
as fresh heat.
+ if (stat.flush_finish_count > 0) {
+ flushed_tablet->last_load_flush_time_ms.store(UnixMillis(),
std::memory_order_relaxed);
+
flushed_tablet->flush_finish_count->increment(stat.flush_finish_count);
+ }
}
}
diff --git a/be/src/service/point_query_executor.cpp
b/be/src/service/point_query_executor.cpp
index 1b4e6fcb04c..018afe2f389 100644
--- a/be/src/service/point_query_executor.cpp
+++ b/be/src/service/point_query_executor.cpp
@@ -61,6 +61,7 @@
#include "util/lru_cache.h"
#include "util/simd/bits.h"
#include "util/thrift_util.h"
+#include "util/time.h"
namespace doris {
@@ -367,6 +368,11 @@ Status PointQueryExecutor::lookup_up() {
RETURN_IF_ERROR(_lookup_row_key());
RETURN_IF_ERROR(_lookup_row_data());
RETURN_IF_ERROR(_output_data());
+ if (_tablet != nullptr) {
+ // Timestamp before the counter -- see olap_scanner.cpp.
+ _tablet->last_query_scan_time_ms.store(UnixMillis(),
std::memory_order_relaxed);
+ _tablet->query_scan_count->increment(1);
+ }
return Status::OK();
}
diff --git a/be/src/storage/active_tablet_stats.cpp
b/be/src/storage/active_tablet_stats.cpp
new file mode 100644
index 00000000000..e634ec5c634
--- /dev/null
+++ b/be/src/storage/active_tablet_stats.cpp
@@ -0,0 +1,109 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "storage/active_tablet_stats.h"
+
+#include <algorithm>
+#include <cstddef>
+
+#include "common/config.h"
+#include "storage/tablet/base_tablet.h"
+#include "util/time.h"
+
+namespace doris {
+
+void ActiveTabletCollector::start() {
+ start(UnixMillis(), MonotonicMillis());
+}
+
+void ActiveTabletCollector::start(int64_t now_ms, int64_t now_mono_ms) {
+ _now_ms = now_ms;
+ _now_mono_ms = now_mono_ms;
+}
+
+void ActiveTabletCollector::collect(const std::shared_ptr<BaseTablet>& tablet)
{
+ if (!tablet) {
+ return;
+ }
+ const int64_t scan_cur = tablet->query_scan_count->value();
+ const int64_t flush_cur = tablet->flush_finish_count->value();
+ // Snapshot only -- the baselines on the tablet stay untouched until
commit().
+ _pending.push_back({.tablet = tablet, .scan_count = scan_cur, .flush_count
= flush_cur});
+
+ const int64_t prev_mono_ms =
tablet->last_reported_mono_ms.load(std::memory_order_relaxed);
+ if (prev_mono_ms == 0) {
+ // Unreachable for normally constructed tablets: BaseTablet
initializes the baseline time.
+ return;
+ }
+ const int64_t scan_delta =
+ scan_cur -
tablet->last_reported_scan_count.load(std::memory_order_relaxed);
+ const int64_t load_delta =
+ flush_cur -
tablet->last_reported_flush_count.load(std::memory_order_relaxed);
+ const int64_t win_ms = std::max<int64_t>(1, _now_mono_ms - prev_mono_ms);
+ const int64_t active_window_ms =
+ static_cast<int64_t>(config::report_active_tablet_window_second) *
1000;
+
+ const int64_t last_q =
tablet->last_query_scan_time_ms.load(std::memory_order_relaxed);
+ const int64_t last_l =
tablet->last_load_flush_time_ms.load(std::memory_order_relaxed);
+ if (scan_delta > 0 && _now_ms - last_q <= active_window_ms) {
+ _query_cands.push_back({tablet->tablet_id(), scan_delta, win_ms,
last_q});
+ }
+ if (load_delta > 0 && _now_ms - last_l <= active_window_ms) {
+ _load_cands.push_back({tablet->tablet_id(), load_delta, win_ms,
last_l});
+ }
+ // A tablet hot on both dimensions appears once in each list; FE merges by
tablet_id.
+}
+
+void ActiveTabletCollector::take_top_n() {
+ const auto truncate_to_cap = [this](std::vector<ActiveTabletCandidate>&
candidates) {
+ // Clamp before the cast: a negative value would wrap to SIZE_MAX and
silently
+ // uncap the list. <= 0 therefore means "report nothing" (feature
off), never
+ // "report everything" -- an uncapped list can push the report past
FE's
+ // thrift_max_message_size, and that failure is permanent: the message
never
+ // gets smaller, so tablet reporting for this BE stops succeeding for
good.
+ const auto cap = static_cast<std::size_t>(
+ std::max(0,
static_cast<int>(config::report_active_tablet_max_num)));
+ if (candidates.size() > cap) {
+ _truncated = true;
+ std::nth_element(
+ candidates.begin(), candidates.begin() + cap,
candidates.end(),
+ [](const auto& lhs, const auto& rhs) { return lhs.rate() >
rhs.rate(); });
+ candidates.resize(cap);
+ }
+ };
+ truncate_to_cap(_query_cands);
+ truncate_to_cap(_load_cands);
+}
+
+void ActiveTabletCollector::commit() {
+ for (auto& pending : _pending) {
+ if (auto tablet = pending.tablet.lock()) {
+ tablet->last_reported_scan_count.store(pending.scan_count,
std::memory_order_relaxed);
+ tablet->last_reported_flush_count.store(pending.flush_count,
std::memory_order_relaxed);
+ tablet->last_reported_mono_ms.store(_now_mono_ms,
std::memory_order_relaxed);
+ }
+ }
+}
+
+void ActiveTabletCollector::clear() {
+ _query_cands.clear();
+ _load_cands.clear();
+ _pending.clear();
+ _truncated = false;
+}
+
+} // namespace doris
diff --git a/be/src/storage/active_tablet_stats.h
b/be/src/storage/active_tablet_stats.h
new file mode 100644
index 00000000000..fc0d7e0571a
--- /dev/null
+++ b/be/src/storage/active_tablet_stats.h
@@ -0,0 +1,87 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+namespace doris {
+class BaseTablet;
+
+// One entry of a per-dimension top-N list. Kept thrift-free on purpose so
this header
+// does not drag gen_cpp into the storage layer; conversion lives in
task_worker_pool.cpp.
+struct ActiveTabletCandidate {
+ int64_t tablet_id = 0;
+ int64_t delta = 0;
+ int64_t window_ms = 1;
+ int64_t last_time_ms = 0;
+
+ // Rate, not raw delta. Raw deltas are not comparable: a dropped report
round commits
+ // no baseline, so the next delta covers several intervals, and backends
report on
+ // independent phases.
+ double rate() const {
+ return static_cast<double>(delta) * 1000.0 /
static_cast<double>(window_ms);
+ }
+};
+
+// Collected inside the EXISTING report-tablet walk (no extra traversal).
Collection is
+// strictly read-only on the tablet baselines; commit() is called only after
+// handle_report() returned true, so a retried or dropped report loses nothing.
+class ActiveTabletCollector {
+public:
+ // Called at the top of each build_all_report_tablets_info() pass.
+ void start();
+ // Clock-injecting overload, so tests get exact delta windows. now_ms is
wall clock
+ // (compared against the tablets' activity timestamps, and shipped to FE);
now_mono_ms
+ // is monotonic and feeds the delta window only.
+ void start(int64_t now_ms, int64_t now_mono_ms);
+ // Called once per tablet during the walk.
+ void collect(const std::shared_ptr<BaseTablet>& tablet);
+ // Per-dimension nth_element by rate. No cross-dimension merging or
weighting here:
+ // query and load counts differ by one to two orders of magnitude, so
ranking them
+ // against each other drops load-heavy tablets as a class. FE owns that
trade-off,
+ // where the split is a mutable config instead of a be.conf restart.
+ void take_top_n();
+ // Advance the baselines. ONLY after handle_report() succeeded.
+ void commit();
+ // Reset everything; called on each retry of the report loop.
+ void clear();
+
+ const std::vector<ActiveTabletCandidate>& query_candidates() const {
return _query_cands; }
+ const std::vector<ActiveTabletCandidate>& load_candidates() const { return
_load_cands; }
+ bool truncated() const { return _truncated; }
+
+private:
+ struct Pending {
+ std::weak_ptr<BaseTablet> tablet;
+ int64_t scan_count = 0;
+ int64_t flush_count = 0;
+ };
+ std::vector<ActiveTabletCandidate> _query_cands;
+ std::vector<ActiveTabletCandidate> _load_cands;
+ std::vector<Pending> _pending;
+ // Wall clock, compared against the tablets' last query / load timestamps
and shipped
+ // to FE. _now_mono_ms only ever feeds the delta window -- see
+ // BaseTablet::last_reported_mono_ms.
+ int64_t _now_ms = 0;
+ int64_t _now_mono_ms = 0;
+ bool _truncated = false;
+};
+
+} // namespace doris
diff --git a/be/src/storage/tablet/base_tablet.cpp
b/be/src/storage/tablet/base_tablet.cpp
index fd54b902374..1ccf300c3a2 100644
--- a/be/src/storage/tablet/base_tablet.cpp
+++ b/be/src/storage/tablet/base_tablet.cpp
@@ -60,6 +60,7 @@
#include "util/debug_points.h"
#include "util/jsonb/serialize.h"
#include "util/string_util.h"
+#include "util/time.h"
namespace doris {
#include "common/compile_check_begin.h"
@@ -229,6 +230,18 @@ BaseTablet::BaseTablet(TabletMetaSharedPtr tablet_meta) :
_tablet_meta(std::move
INT_COUNTER_METRIC_REGISTER(_metric_entity, flush_bytes);
INT_COUNTER_METRIC_REGISTER(_metric_entity, flush_finish_count);
+ // Seed the baselines from whatever the counters already hold -- they are
NOT
+ // necessarily zero. The metric entity is keyed by tablet id, so
register_metric() above
+ // hands back the EXISTING counters whenever another BaseTablet for this
id is still
+ // alive: local migration constructs the replacement before dropping the
original, and in
+ // cloud a background shared_ptr can outlive cache eviction while a
replacement is loaded.
+ // Assuming zero there would make this object's first report bill the old
object's entire
+ // history to a single window. Leaving the time at zero instead would make
collect() skip
+ // the first round while commit() still advanced the baseline, discarding
that activity.
+ last_reported_scan_count.store(query_scan_count->value(),
std::memory_order_relaxed);
+ last_reported_flush_count.store(flush_finish_count->value(),
std::memory_order_relaxed);
+ last_reported_mono_ms.store(MonotonicMillis(), std::memory_order_relaxed);
+
// construct _timestamped_versioned_tracker from rs and stale rs meta
_timestamped_version_tracker.construct_versioned_tracker(_tablet_meta->all_rs_metas(),
_tablet_meta->all_stale_rs_metas());
diff --git a/be/src/storage/tablet/base_tablet.h
b/be/src/storage/tablet/base_tablet.h
index 12d51a0c656..9cdecf9ed6e 100644
--- a/be/src/storage/tablet/base_tablet.h
+++ b/be/src/storage/tablet/base_tablet.h
@@ -407,6 +407,32 @@ public:
IntCounter* query_scan_count = nullptr;
IntCounter* flush_bytes = nullptr;
IntCounter* flush_finish_count = nullptr;
+ // Last time this tablet was scanned by a query / flushed by a load (ms
since epoch),
+ // named after the counter each one accompanies (query_scan_count /
flush_finish_count).
+ // Set only on the query scan path and the memtable flush path
respectively, so they are
+ // NOT polluted by compaction / snapshot / agent tasks the way
CloudTablet::
+ // last_access_time_ms is. Do NOT shorten to last_load_time_ms:
CloudTablet already has
+ // a field by that name tracking load start for the compaction freeze
check.
+ std::atomic<int64_t> last_query_scan_time_ms {0};
+ std::atomic<int64_t> last_load_flush_time_ms {0};
+ // Snapshots taken at the last SUCCESSFULLY DELIVERED report. Committed
only after
+ // handle_report() returns true, never with exchange() during collection
-- see
+ // ActiveTabletCollector::commit(). Seeded in the constructor from the
counters' current
+ // values, which are shared with any other live BaseTablet for the same
tablet id.
+ std::atomic<int64_t> last_reported_scan_count {0};
+ std::atomic<int64_t> last_reported_flush_count {0};
+ // Monotonic clock of that commit -- NOT wall clock. Its only use is the
+ // `now - last_reported_mono_ms` subtraction that yields the delta window;
a backward
+ // NTP step on the wall clock would clamp that window to 1 ms and inflate
every rate
+ // by orders of magnitude. The two last_*_time_ms fields above must stay
wall clock:
+ // they are shipped to FE, which compares them against its own clock.
+ //
+ // Shared by every tablet in a round, because commit() is all-or-nothing:
a dropped
+ // report (5 retries exhausted, or handle_report() returning false)
advances no
+ // baseline at all, so the next round's delta spans several intervals.
Backends also
+ // report on independent phases. FE compares entries across backends and
across
+ // rounds, so it must normalise delta by this window rather than compare
raw counts.
+ std::atomic<int64_t> last_reported_mono_ms {0};
std::atomic<int64_t> published_count = 0;
std::atomic<int64_t> read_block_count = 0;
std::atomic<int64_t> write_count = 0;
diff --git a/be/src/storage/tablet/tablet_manager.cpp
b/be/src/storage/tablet/tablet_manager.cpp
index 7cdaebe2c2f..138899bf068 100644
--- a/be/src/storage/tablet/tablet_manager.cpp
+++ b/be/src/storage/tablet/tablet_manager.cpp
@@ -44,6 +44,7 @@
#include "io/fs/local_file_system.h"
#include "runtime/exec_env.h"
#include "service/backend_options.h"
+#include "storage/active_tablet_stats.h"
#include "storage/compaction/cumulative_compaction_policy.h"
#include "storage/compaction/cumulative_compaction_time_series_policy.h"
#include "storage/data_dir.h"
@@ -1075,8 +1076,12 @@ Status TabletManager::report_tablet_info(TTabletInfo*
tablet_info) {
return res;
}
-void TabletManager::build_all_report_tablets_info(std::map<TTabletId,
TTablet>* tablets_info) {
+void TabletManager::build_all_report_tablets_info(std::map<TTabletId,
TTablet>* tablets_info,
+ ActiveTabletCollector*
active) {
DCHECK(tablets_info != nullptr);
+ if (active) {
+ active->start();
+ }
VLOG_NOTICE << "begin to build all report tablets info";
// build the expired txn map first, outside the tablet map lock
@@ -1087,6 +1092,9 @@ void
TabletManager::build_all_report_tablets_info(std::map<TTabletId, TTablet>*
HistogramStat tablet_version_num_hist;
auto local_cache = std::make_shared<std::vector<TTabletStat>>();
auto handler = [&](const TabletSharedPtr& tablet) {
+ if (active) {
+ active->collect(tablet);
+ }
auto& t_tablet = (*tablets_info)[tablet->tablet_id()];
TTabletInfo& tablet_info = t_tablet.tablet_infos.emplace_back();
tablet->build_tablet_report_info(&tablet_info, true, true);
diff --git a/be/src/storage/tablet/tablet_manager.h
b/be/src/storage/tablet/tablet_manager.h
index 3cae490c877..edf671bebf0 100644
--- a/be/src/storage/tablet/tablet_manager.h
+++ b/be/src/storage/tablet/tablet_manager.h
@@ -45,6 +45,7 @@ namespace doris {
class DataDir;
class CumulativeCompactionPolicy;
+class ActiveTabletCollector;
class MemTracker;
class TCreateTabletReq;
class TTablet;
@@ -140,7 +141,8 @@ public:
// Status::Error<INVALID_ARGUMENT>(), if tables is null
Status report_tablet_info(TTabletInfo* tablet_info);
- void build_all_report_tablets_info(std::map<TTabletId, TTablet>*
tablets_info);
+ void build_all_report_tablets_info(std::map<TTabletId, TTablet>*
tablets_info,
+ ActiveTabletCollector* active =
nullptr);
Status start_trash_sweep();
diff --git a/be/src/storage/tablet/tablet_meta.h
b/be/src/storage/tablet/tablet_meta.h
index 972768bd405..d8b4a264b22 100644
--- a/be/src/storage/tablet/tablet_meta.h
+++ b/be/src/storage/tablet/tablet_meta.h
@@ -126,7 +126,9 @@ public:
// UT
#ifdef BE_TEST
- TabletMeta(TabletSchemaSPtr tablet_schema) : _schema(tablet_schema) {}
+ TabletMeta(TabletSchemaSPtr tablet_schema, int64_t tablet_id = 0) :
_schema(tablet_schema) {
+ _tablet_id = tablet_id;
+ }
#endif
// Function create_from_file is used to be compatible with previous
tablet_meta.
diff --git a/be/test/storage/active_tablet_stats_test.cpp
b/be/test/storage/active_tablet_stats_test.cpp
new file mode 100644
index 00000000000..f8c07054515
--- /dev/null
+++ b/be/test/storage/active_tablet_stats_test.cpp
@@ -0,0 +1,316 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "storage/active_tablet_stats.h"
+
+#include <gtest/gtest.h>
+
+#include <memory>
+
+#include "common/config.h"
+#include "storage/rowset/rowset_writer.h"
+#include "storage/tablet/base_tablet.h"
+#include "storage/tablet/tablet_meta.h"
+#include "storage/tablet/tablet_schema.h"
+#include "util/time.h"
+
+namespace doris {
+
+namespace {
+
+constexpr int64_t kNowMs = 1'700'000'000'000;
+constexpr int64_t kReportIntervalMs = 60'000;
+
+// Minimal concrete BaseTablet, same shape as the FakeTablet stubs in the
scanner and
+// file-cache tests. The collector only reads the counters and the atomics
added for
+// active-tablet stats, so none of these overrides is ever reached.
+class StatsTestTablet final : public BaseTablet {
+public:
+ explicit StatsTestTablet(int64_t tablet_id) :
BaseTablet(create_meta(tablet_id)) {}
+
+ std::string tablet_path() const override { return ""; }
+
+ bool exceed_version_limit(int32_t /*limit*/) override { return false; }
+
+ Result<std::unique_ptr<RowsetWriter>>
create_rowset_writer(RowsetWriterContext& /*context*/,
+ bool
/*vertical*/) override {
+ return ResultError(Status::NotSupported("stats test tablet"));
+ }
+
+ Result<std::unique_ptr<RowsetWriter>> create_transient_rowset_writer(
+ const Rowset& /*rowset*/, std::shared_ptr<PartialUpdateInfo>
/*partial_update_info*/,
+ int64_t /*txn_expiration*/ = 0) override {
+ return ResultError(Status::NotSupported("stats test tablet"));
+ }
+
+ Status capture_rs_readers(const Version& /*spec_version*/,
+ std::vector<RowSetSplits>* /*rs_splits*/,
+ const CaptureRowsetOps& /*opts*/) override {
+ return Status::NotSupported("stats test tablet");
+ }
+
+ Status save_delete_bitmap(const TabletTxnInfo* /*txn_info*/, int64_t
/*txn_id*/,
+ DeleteBitmapPtr /*delete_bitmap*/, RowsetWriter*
/*rowset_writer*/,
+ const RowsetIdUnorderedSet& /*cur_rowset_ids*/,
+ int64_t /*lock_id*/ = -1,
+ int64_t /*next_visible_version*/ = -1) override {
+ return Status::NotSupported("stats test tablet");
+ }
+
+ CalcDeleteBitmapExecutor* calc_delete_bitmap_executor() override { return
nullptr; }
+
+ void clear_cache() override {}
+
+ Versions calc_missed_versions(int64_t /*spec_version*/,
+ Versions /*existing_versions*/) const
override {
+ return {};
+ }
+
+ size_t tablet_footprint() override { return 0; }
+
+private:
+ static TabletMetaSharedPtr create_meta(int64_t tablet_id) {
+ return std::make_shared<TabletMeta>(std::make_shared<TabletSchema>(),
tablet_id);
+ }
+};
+
+// A tablet with a committed baseline at a fixed time for deterministic report
windows.
+// The activity timestamps are wall clock and the reported baseline is
monotonic, but the
+// tests drive both from the same number: only differences are ever asserted.
+std::shared_ptr<StatsTestTablet> make_baselined_tablet(int64_t tablet_id,
int64_t baseline_ms) {
+ auto tablet = std::make_shared<StatsTestTablet>(tablet_id);
+ tablet->last_reported_mono_ms.store(baseline_ms);
+ tablet->last_query_scan_time_ms.store(baseline_ms);
+ tablet->last_load_flush_time_ms.store(baseline_ms);
+ return tablet;
+}
+
+// Drives one report round at fixed clocks instead of UnixMillis() /
MonotonicMillis(), so
+// the delta windows the assertions check are exact. The two clocks are driven
from the
+// same number because only differences are ever asserted.
+void collect_at(ActiveTabletCollector& collector, const
std::shared_ptr<BaseTablet>& tablet,
+ int64_t now_ms) {
+ collector.start(now_ms, now_ms);
+ collector.collect(tablet);
+}
+
+} // namespace
+
+// The reason BE ranks by delta/window and not by raw delta: tablet 1 has 5x
the raw
+// scan delta of tablet 2 but accumulated it over 10x the window, so it is the
colder
+// one. Also pins that the two dimensions are truncated independently -- a
load-only
+// tablet must not be crowded out by query traffic.
+TEST(ActiveTabletStatsTest, TakesIndependentTopListsByRate) {
+ const auto old_cap = config::report_active_tablet_max_num;
+ config::report_active_tablet_max_num = 1;
+
+ // Same shape as the report walk: one start(), then one collect() per
tablet.
+ auto slow_query = make_baselined_tablet(1, kNowMs - 10 *
kReportIntervalMs);
+ slow_query->query_scan_count->increment(100);
+ auto fast_query = make_baselined_tablet(2, kNowMs - kReportIntervalMs);
+ fast_query->query_scan_count->increment(20);
+ auto fast_load = make_baselined_tablet(3, kNowMs - kReportIntervalMs);
+ fast_load->flush_finish_count->increment(30);
+ auto slow_load = make_baselined_tablet(4, kNowMs - 10 * kReportIntervalMs);
+ slow_load->flush_finish_count->increment(200);
+
+ ActiveTabletCollector collector;
+ collector.start(kNowMs, kNowMs);
+ for (const auto& tablet : {slow_query, fast_query, fast_load, slow_load}) {
+ collector.collect(tablet);
+ }
+ collector.take_top_n();
+ config::report_active_tablet_max_num = old_cap;
+
+ ASSERT_EQ(collector.query_candidates().size(), 1);
+ EXPECT_EQ(collector.query_candidates().front().tablet_id, 2);
+ ASSERT_EQ(collector.load_candidates().size(), 1);
+ EXPECT_EQ(collector.load_candidates().front().tablet_id, 3);
+ EXPECT_TRUE(collector.truncated());
+}
+
+// Construction establishes the zero-counter baseline, so the first report
includes all
+// activity since construction and normalizes it by the actual elapsed window.
+TEST(ActiveTabletStatsTest, FirstRoundReportsActivitySinceConstruction) {
+ const int64_t before_ms = MonotonicMillis();
+ auto tablet = std::make_shared<StatsTestTablet>(101);
+ const int64_t baseline_ms = tablet->last_reported_mono_ms.load();
+ EXPECT_GE(baseline_ms, before_ms);
+ EXPECT_LE(baseline_ms, MonotonicMillis());
+ ASSERT_GT(baseline_ms, 0);
+ EXPECT_EQ(tablet->last_reported_scan_count.load(), 0);
+ EXPECT_EQ(tablet->last_reported_flush_count.load(), 0);
+
+ const int64_t report_ms = baseline_ms + kReportIntervalMs;
+ tablet->query_scan_count->increment(42);
+ tablet->flush_finish_count->increment(7);
+ tablet->last_query_scan_time_ms.store(report_ms);
+ tablet->last_load_flush_time_ms.store(report_ms);
+
+ ActiveTabletCollector collector;
+ collect_at(collector, tablet, report_ms);
+
+ ASSERT_EQ(collector.query_candidates().size(), 1);
+ EXPECT_EQ(collector.query_candidates().front().delta, 42);
+ EXPECT_EQ(collector.query_candidates().front().window_ms,
kReportIntervalMs);
+ ASSERT_EQ(collector.load_candidates().size(), 1);
+ EXPECT_EQ(collector.load_candidates().front().delta, 7);
+ EXPECT_EQ(collector.load_candidates().front().window_ms,
kReportIntervalMs);
+ EXPECT_EQ(tablet->last_reported_mono_ms.load(), baseline_ms);
+
+ collector.commit();
+ EXPECT_EQ(tablet->last_reported_scan_count.load(), 42);
+ EXPECT_EQ(tablet->last_reported_flush_count.load(), 7);
+ EXPECT_EQ(tablet->last_reported_mono_ms.load(), report_ms);
+
+ collector.clear();
+ collect_at(collector, tablet, report_ms + kReportIntervalMs);
+ EXPECT_TRUE(collector.query_candidates().empty());
+ EXPECT_TRUE(collector.load_candidates().empty());
+}
+
+// The metric entity is keyed by tablet id, so a replacement object for a
tablet whose original
+// is still alive shares its counters. Seeding the baselines from the counters
is what stops the
+// replacement's first report from billing the whole of the original's history
to one window.
+TEST(ActiveTabletStatsTest, ReplacementTabletDoesNotInheritTheOldCountAsDelta)
{
+ auto original = make_baselined_tablet(104, kNowMs - kReportIntervalMs);
+ original->query_scan_count->increment(1'000'000);
+
+ // Built while the original is still alive, the way local migration does
it.
+ auto replacement = std::make_shared<StatsTestTablet>(104);
+ EXPECT_EQ(replacement->query_scan_count->value(), 1'000'000) << "counters
are shared by id";
+ EXPECT_EQ(replacement->last_reported_scan_count.load(), 1'000'000);
+
+ replacement->query_scan_count->increment(1);
+ replacement->last_query_scan_time_ms.store(kNowMs);
+ replacement->last_reported_mono_ms.store(kNowMs - kReportIntervalMs);
+
+ ActiveTabletCollector collector;
+ collect_at(collector, replacement, kNowMs);
+ ASSERT_EQ(collector.query_candidates().size(), 1);
+ EXPECT_EQ(collector.query_candidates().front().delta, 1);
+}
+
+// Collection must be read-only on the baselines. The report path retries the
walk up
+// to 5 times on a report-version conflict, so a destructive read (exchange)
would make
+// every retry after the first see a zero delta.
+TEST(ActiveTabletStatsTest, CollectDoesNotAdvanceBaseline) {
+ auto tablet = make_baselined_tablet(102, kNowMs - kReportIntervalMs);
+ tablet->query_scan_count->increment(10);
+
+ ActiveTabletCollector collector;
+ for (int retry = 0; retry < 3; ++retry) {
+ collector.clear();
+ collect_at(collector, tablet, kNowMs);
+
+ ASSERT_EQ(collector.query_candidates().size(), 1) << "retry " << retry;
+ EXPECT_EQ(collector.query_candidates().front().delta, 10);
+ EXPECT_EQ(collector.query_candidates().front().window_ms,
kReportIntervalMs);
+ EXPECT_EQ(tablet->last_reported_scan_count.load(), 0);
+ EXPECT_EQ(tablet->last_reported_mono_ms.load(), kNowMs -
kReportIntervalMs);
+ }
+}
+
+// Once the report is delivered, commit() advances the baseline so the next
round
+// reports only what happened since.
+TEST(ActiveTabletStatsTest, CommitAdvancesBaselineSoNextDeltaIsIncremental) {
+ auto tablet = make_baselined_tablet(103, kNowMs - kReportIntervalMs);
+ tablet->query_scan_count->increment(10);
+ tablet->flush_finish_count->increment(4);
+
+ ActiveTabletCollector collector;
+ collect_at(collector, tablet, kNowMs);
+ collector.commit();
+ EXPECT_EQ(tablet->last_reported_scan_count.load(), 10);
+ EXPECT_EQ(tablet->last_reported_flush_count.load(), 4);
+ EXPECT_EQ(tablet->last_reported_mono_ms.load(), kNowMs);
+
+ const int64_t next_ms = kNowMs + kReportIntervalMs;
+ tablet->query_scan_count->increment(3);
+ tablet->flush_finish_count->increment(1);
+ tablet->last_query_scan_time_ms.store(next_ms);
+ tablet->last_load_flush_time_ms.store(next_ms);
+
+ collector.clear();
+ collect_at(collector, tablet, next_ms);
+
+ ASSERT_EQ(collector.query_candidates().size(), 1);
+ EXPECT_EQ(collector.query_candidates().front().delta, 3);
+ EXPECT_EQ(collector.query_candidates().front().window_ms,
kReportIntervalMs);
+ ASSERT_EQ(collector.load_candidates().size(), 1);
+ EXPECT_EQ(collector.load_candidates().front().delta, 1);
+}
+
+// When a report is dropped (5 failed retries, or handle_report() returning
false) the
+// baseline is not committed, so the next round's delta must span both
intervals and
+// window_ms must widen with it -- that is what keeps the rate comparable.
+TEST(ActiveTabletStatsTest, DroppedReportCarriesDeltaIntoAWiderWindow) {
+ auto tablet = make_baselined_tablet(104, kNowMs - kReportIntervalMs);
+ tablet->query_scan_count->increment(10);
+
+ ActiveTabletCollector collector;
+ collect_at(collector, tablet, kNowMs);
+ // report failed: no commit()
+
+ const int64_t next_ms = kNowMs + kReportIntervalMs;
+ tablet->query_scan_count->increment(10);
+ tablet->last_query_scan_time_ms.store(next_ms);
+
+ collector.clear();
+ collect_at(collector, tablet, next_ms);
+
+ ASSERT_EQ(collector.query_candidates().size(), 1);
+ const auto& candidate = collector.query_candidates().front();
+ EXPECT_EQ(candidate.delta, 20);
+ EXPECT_EQ(candidate.window_ms, 2 * kReportIntervalMs);
+ // Same rate as a single round of 10 -- the point of normalising by the
window.
+ EXPECT_DOUBLE_EQ(candidate.rate(), 10.0 * 1000.0 / kReportIntervalMs);
+}
+
+// A tablet whose delta is stale -- reporting kept failing, so no baseline was
committed
+// for a long stretch -- must age out once it stops being touched. This is the
case
+// report_active_tablet_window_second exists for: without it, a tablet that
went cold
+// hours ago would be reported as active the moment reporting recovers.
+TEST(ActiveTabletStatsTest, StaleDeltaOutsideActiveWindowIsDropped) {
+ const int64_t window_ms = int64_t
{config::report_active_tablet_window_second} * 1000;
+ auto tablet = make_baselined_tablet(105, kNowMs - window_ms -
kReportIntervalMs);
+ tablet->query_scan_count->increment(10);
+ // Last actually queried just before the active window opened.
+ tablet->last_query_scan_time_ms.store(kNowMs - window_ms - 1);
+
+ ActiveTabletCollector collector;
+ collect_at(collector, tablet, kNowMs);
+
+ EXPECT_TRUE(collector.query_candidates().empty());
+ // Still pending, so the baseline moves on and the stale delta is retired.
+ collector.commit();
+ EXPECT_EQ(tablet->last_reported_scan_count.load(), 10);
+}
+
+// A tablet dropped between collection and delivery must not crash commit().
+TEST(ActiveTabletStatsTest, CommitSkipsTabletsDroppedAfterCollection) {
+ auto tablet = make_baselined_tablet(106, kNowMs - kReportIntervalMs);
+ tablet->query_scan_count->increment(10);
+
+ ActiveTabletCollector collector;
+ collect_at(collector, tablet, kNowMs);
+
+ tablet.reset();
+ collector.commit();
+}
+
+} // namespace doris
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 4bfa681adc8..85b5f8ea659 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -3667,8 +3667,8 @@ public class Config extends ConfigBase {
+ "is the active tablet priority scheduling strategy enabled?
(Default: Enabled)"})
public static boolean enable_cloud_active_tablet_priority_scheduling =
false;
- @ConfField(masterOnly = true, description = {"是否启用活跃tablet滑动窗口访问统计功能，默认打开",
- "Whether to enable active tablet sliding window access statistics
feature, default true"})
+ @ConfField(mutable = true, masterOnly = true, description =
{"是否启用活跃tablet滑动窗口访问统计功能，默认关闭",
+ "Whether to enable active tablet sliding window access statistics
feature, default false"})
public static boolean enable_active_tablet_sliding_window_access_stats =
false;
@ConfField(mutable = true, masterOnly = true, description =
{"活跃tablet滑动窗口访问统计的时间窗口大小（秒），默认3600秒（1小时）",
@@ -3678,11 +3678,16 @@ public class Config extends ConfigBase {
@ConfField(mutable = true, masterOnly = true, description = {
"活跃 tablet 优先调度开启时：partition 级调度将优先处理 TopN 的活跃 partition，"
- + "再处理其余活跃 partition、非活跃 partition，最后处理 internal db。默认
10000，<=0 表示不做 TopN 分段。",
+ + "再处理其余活跃 partition、非活跃 partition，最后处理 internal db。默认
10000，<=0 表示不做 TopN 分段。"
+ + "每个 BE 每个维度最多上报 report_active_tablet_max_num 个
tablet，热点集中在单个 BE 时，"
+ + "该 BE 配置需不小于此值才能填满预算；调大前请关注 BE bvar
report.active_tablet_truncated。",
"When active tablet priority scheduling is enabled:
partition-level scheduling processes TopN active "
+ "partitions first, then other active partitions,"
+ "then inactive partitions, and internal db at last. "
- + "Default 10000. <=0 disables TopN segmentation."})
+ + "Default 10000. <=0 disables TopN segmentation. "
+ + "A BE reports at most report_active_tablet_max_num
tablets per dimension, so a BE holding "
+ + "a concentrated hot set cannot fill this budget on its
own unless that be config is >= this "
+ + "value; watch the BE bvar report.active_tablet_truncated
before raising this one."})
public static int cloud_active_partition_scheduling_topn = 10000;
@ConfField(mutable = true, masterOnly = false)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
index 50c4ebf5599..d3850e95eea 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
@@ -206,7 +206,6 @@ public abstract class Tablet {
// return map of (BE id -> path hash) of normal replicas
// for load plan.
public Multimap<Long, Long> getNormalReplicaBackendPathMap() throws
UserException {
- TabletSlidingWindowAccessStats.recordTablet(getId());
return getNormalReplicaBackendPathMapImpl(null, (rep, be) ->
rep.getBackendId());
}
@@ -228,7 +227,6 @@ public abstract class Tablet {
List<Replica> mayMissingVersionReplica =
Lists.newArrayListWithCapacity(replicaNum);
List<Replica> notCatchupReplica =
Lists.newArrayListWithCapacity(replicaNum);
List<Replica> userDropReplica =
Lists.newArrayListWithCapacity(replicaNum);
- TabletSlidingWindowAccessStats.recordTablet(getId());
for (Replica replica : replicas) {
if (replica.isBad()) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/TabletSlidingWindowAccessStats.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/TabletSlidingWindowAccessStats.java
index 90da80dd1af..30c611222b6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/TabletSlidingWindowAccessStats.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/TabletSlidingWindowAccessStats.java
@@ -18,294 +18,236 @@
package org.apache.doris.catalog;
import org.apache.doris.common.Config;
-import org.apache.doris.common.util.MasterDaemon;
+import org.apache.doris.thrift.TActiveTabletStat;
-import com.google.common.hash.HashFunction;
-import com.google.common.hash.Hashing;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
+import com.google.common.collect.Maps;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.PriorityQueue;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicLongArray;
/**
- * Sliding window access statistics utility class.
- * Supports tracking access statistics for different types of IDs (tablet,
replica, backend, etc.)
+ * Active tablet access statistics reported by backends.
*/
public class TabletSlidingWindowAccessStats {
- private static final Logger LOG =
LogManager.getLogger(TabletSlidingWindowAccessStats.class);
-
private static volatile TabletSlidingWindowAccessStats instance;
- private static final HashFunction SHARD_HASH = Hashing.murmur3_128();
-
- // Sort active IDs by accessCount desc, then lastAccessTime desc
- private static final Comparator<AccessStatsResult> TOPN_ACTIVE_COMPARATOR =
- Comparator.comparingLong((AccessStatsResult r) ->
r.accessCount).reversed()
- .thenComparing(Comparator.comparingLong((AccessStatsResult
r) -> r.lastAccessTime).reversed());
-
- // Time window in milliseconds (default: 1 hour)
- private final long timeWindowMs;
+ // Hottest first, most recently touched breaking a tie. Reversing the
whole chain is the
+ // same as reversing each key, and reads as the one sentence above.
+ private static final Comparator<AccessStatsResult> QUERY_RATE_COMPARATOR =
+ Comparator.comparingDouble((AccessStatsResult r) -> r.scanRate)
+ .thenComparingLong(r -> r.lastAccessTime)
+ .reversed();
+ private static final Comparator<AccessStatsResult> LOAD_RATE_COMPARATOR =
+ Comparator.comparingDouble((AccessStatsResult r) -> r.loadRate)
+ .thenComparingLong(r -> r.lastAccessTime)
+ .reversed();
+
+ // beId -> (tabletId -> stats). A report updates the tablets it carries
and ages out the
+ // rest by active_tablet_sliding_window_time_window_second. Reads also
filter expired entries
+ // and reclaim expired snapshots when reports stop; backend removal calls
removeBackend().
+ private final ConcurrentHashMap<Long, Map<Long, AccessStatsResult>>
beToStats = new ConcurrentHashMap<>();
+ private final AtomicLong totalAccessCount = new AtomicLong(0);
- // Bucket size in milliseconds (default: 1 minute)
- // The time window is divided into multiple buckets, each bucket stores
access count for a time period.
- // For example: if timeWindowMs=1hour and bucketSizeMs=1minute, there will
be 60 buckets.
- // Smaller bucket size = more accurate statistics but more memory usage.
- private final long bucketSizeMs;
+ // Merging every backend snapshot is O(total RETAINED tablets), which is
not the same as
+ // what was last reported -- retainWithinWindow() states the per-backend
bound and it is not
+ // the BE per-list cap. MetricRepo scrapes two of the aggregate getters
+ // below on every /metrics request, so they share a short-lived snapshot
the same way the
+ // pre-BE-report implementation cached its aggregates. getTopNActive()
deliberately does NOT
+ // use this cache: it runs once per
cloud_active_tablet_ids_refresh_interval_second and feeds
+ // scheduling decisions, so it always merges fresh.
+ private static final long MERGED_CACHE_TTL_MS = 10_000L;
+ private volatile Map<Long, AccessStatsResult> mergedCache =
Collections.emptyMap();
+ private final AtomicLong mergedCacheTimeMs = new AtomicLong(0);
- // Number of buckets in the sliding window
- private final int numBuckets;
+ TabletSlidingWindowAccessStats() {
+ }
- // Cleanup interval in milliseconds (default: 5 minutes)
- private final long cleanupIntervalMs;
+ public void updateFromReport(long beId, List<TActiveTabletStat> topQuery,
List<TActiveTabletStat> topLoad) {
+ if (!Config.enable_active_tablet_sliding_window_access_stats) {
+ beToStats.remove(beId);
+ return;
+ }
- // Shard size to reduce lock contention
- private static final int SHARD_SIZE = 1024;
+ // A tablet hot on both dimensions arrives once in each list, each
entry carrying only
+ // its own dimension, so the two lists are merged by tablet id before
anything is built.
+ Map<Long, Accumulator> accumulated = Maps.newHashMapWithExpectedSize(
+ topQuery.size() + topLoad.size());
+ for (TActiveTabletStat stat : topQuery) {
+ Accumulator acc = accumulated.computeIfAbsent(stat.getTabletId(),
key -> new Accumulator());
+ acc.scanDelta = stat.getScanCountDelta();
+ acc.lastQueryMs = stat.getLastQueryTimeMs();
+ acc.scanWindowMs = Math.max(1L, stat.getDeltaWindowMs());
+ }
+ for (TActiveTabletStat stat : topLoad) {
+ Accumulator acc = accumulated.computeIfAbsent(stat.getTabletId(),
key -> new Accumulator());
+ acc.loadDelta = stat.getLoadCountDelta();
+ acc.lastLoadMs = stat.getLastLoadTimeMs();
+ acc.loadWindowMs = Math.max(1L, stat.getDeltaWindowMs());
+ }
+
+ Map<Long, AccessStatsResult> backendStats =
Maps.newHashMapWithExpectedSize(accumulated.size());
+ long reportedAccesses = 0;
+ for (Map.Entry<Long, Accumulator> entry : accumulated.entrySet()) {
+ long tabletId = entry.getKey();
+ Accumulator acc = entry.getValue();
+ reportedAccesses += acc.scanDelta + acc.loadDelta;
+ backendStats.put(tabletId, new AccessStatsResult(tabletId,
acc.scanDelta, acc.loadDelta,
+ Math.max(acc.lastQueryMs, acc.lastLoadMs), acc.scanRate(),
acc.loadRate()));
+ }
+ totalAccessCount.addAndGet(reportedAccesses);
+ beToStats.put(beId, retainWithinWindow(beToStats.get(beId),
backendStats));
+ }
/**
- * Sliding window counter for a single replica/tablet
+ * A report only carries the tablets that saw traffic since the previous
one, so taking it
+ * as the whole truth would shrink "active" to one report interval - a
tablet queried hard
+ * four minutes ago would read as cold and become a migration candidate.
Entries the backend
+ * did not repeat are therefore kept until they fall outside
+ * active_tablet_sliding_window_time_window_second, which is the window
this feature has
+ * advertised since it lived in FE memory.
+ *
+ * <p>Retention is bounded by cloud_active_partition_scheduling_topn (K
below): the scheduler
+ * never consumes more actives than that in total, and a backend whose hot
set churns would
+ * otherwise accumulate a whole window's worth of distinct tablets. The
coldest are dropped
+ * first, by rate rather than by age - everything that survives the cutoff
above is inside
+ * the window already, and dropping by age alone would evict a tablet
hammered early in the
+ * window in favour of one touched once at the end.
+ *
+ * <p>Size of a backend's snapshot, which is NOT simply K. The capped path
below keeps up to
+ * K per dimension, so up to 2K ids. The early return keeps the incoming
report whole, which
+ * the backend caps at report_active_tablet_max_num (B) per dimension, so
up to 2B ids - and
+ * that path is taken on a backend's first report and whenever K <= 0.
For fixed positive
+ * K and B the bound is therefore max(2K, 2B), and lowering either config
does not shrink a
+ * snapshot that already exists until its next update or expiry.
+ *
+ * <p>Two cheaper capping rules were both wrong. One ordering over both
dimensions
+ * compares unlike units - scans outnumber flushes by one to two orders of
magnitude, so a
+ * churning query set evicts every load tablet before the reserved load
quota downstream
+ * ever sees it. Spending getTopNActive()'s unique-id budget here is wrong
for a subtler
+ * reason: it allocates the final quota per backend, before cross-backend
overlap is known.
+ * A load candidate dropped here as the backend's own runner-up is exactly
the one the
+ * global load share needs once another backend turns the leader into a
query winner, whose
+ * load entry is then a duplicate; a weak query backfills the freed slot
instead. Neither
+ * failure trips the BE truncation flag, because both backend lists fit
their own cap.
+ *
+ * <p>lastAccessTime comes from the backend's clock while the cutoff comes
from FE's, but a
+ * window measured in hours absorbs the skew between them.
*/
- private static class SlidingWindowCounter {
- // Each bucket stores count for a time period
- private final AtomicLongArray buckets;
- // Timestamp of each bucket (to detect expired buckets)
- private final AtomicLongArray bucketTimestamps;
- // Last access time (for TopN sorting)
- private volatile long lastAccessTime = 0;
- // Total count in current window (cached for performance)
- private volatile long cachedTotalCount = 0;
- private volatile long cachedCountTime = 0;
-
- SlidingWindowCounter(int numBuckets) {
- this.buckets = new AtomicLongArray(numBuckets);
- this.bucketTimestamps = new AtomicLongArray(numBuckets);
- }
-
- /**
- * Get current bucket index based on current time
- */
- private int getBucketIndex(long currentTimeMs, long bucketSizeMs, int
numBuckets) {
- return (int) ((currentTimeMs / bucketSizeMs) % numBuckets);
- }
-
- /**
- * Add an access count
- */
- void add(long currentTimeMs, long bucketSizeMs, int numBuckets) {
- int bucketIndex = getBucketIndex(currentTimeMs, bucketSizeMs,
numBuckets);
- long bucketStartTime = (currentTimeMs / bucketSizeMs) *
bucketSizeMs;
-
- // Check if this bucket is expired (belongs to a different time
window)
- long bucketTimestamp = bucketTimestamps.get(bucketIndex);
- if (bucketTimestamp != bucketStartTime) {
- // Reset expired bucket
- buckets.set(bucketIndex, 0);
- bucketTimestamps.set(bucketIndex, bucketStartTime);
+ private static Map<Long, AccessStatsResult> retainWithinWindow(
+ Map<Long, AccessStatsResult> previous, Map<Long,
AccessStatsResult> reported) {
+ int retainLimit = Config.cloud_active_partition_scheduling_topn;
+ if (previous == null || previous.isEmpty() || retainLimit <= 0) {
+ // retainLimit <= 0 disables TopN segmentation, so getTopNActive()
returns nothing
+ // and anything retained here would only cost memory.
+ return reported;
+ }
+
+ long windowMs = Math.max(1L,
Config.active_tablet_sliding_window_time_window_second) * 1000L;
+ long oldestKept = System.currentTimeMillis() - windowMs;
+ Map<Long, AccessStatsResult> merged = Maps.newHashMapWithExpectedSize(
+ previous.size() + reported.size());
+ for (AccessStatsResult stale : previous.values()) {
+ if (stale.lastAccessTime >= oldestKept) {
+ merged.put(stale.id, stale);
}
-
- // Increment count
- buckets.addAndGet(bucketIndex, 1);
- lastAccessTime = currentTimeMs;
- cachedTotalCount = -1; // Invalidate cache
}
-
- /**
- * Get total count within the time window
- */
- long getCount(long currentTimeMs, long timeWindowMs, long
bucketSizeMs, int numBuckets) {
- // Use cached value if still valid (within 1 second)
- if (cachedTotalCount >= 0 && (currentTimeMs - cachedCountTime) <
1000) {
- return cachedTotalCount;
+ // Whatever the backend just reported is fresher than anything
retained for it.
+ //
+ // Known limitation: a report carrying only one dimension replaces the
whole record, so
+ // a query-only report zeroes a load rate that is still inside the
window, and a
+ // load-only report does the same to a query rate. Not fixed -
expiring the dimensions
+ // independently needs a separate timestamp per dimension, which is
more machinery than
+ // a scheduling hint is worth. Losing a dimension drops the tablet out
of that bucket's
+ // ranking, and once the cap below binds it can leave this map
entirely, so the tablet
+ // can end up treated as cold.
+ merged.putAll(reported);
+ if (merged.size() <= retainLimit) {
+ return merged;
+ }
+
+ List<AccessStatsResult> byQuery = new ArrayList<>();
+ List<AccessStatsResult> byLoad = new ArrayList<>();
+ for (AccessStatsResult result : merged.values()) {
+ if (result.scanRate > 0) {
+ byQuery.add(result);
}
-
- long windowStart = currentTimeMs - timeWindowMs;
- long count = 0;
-
- for (int i = 0; i < numBuckets; i++) {
- long bucketTimestamp = bucketTimestamps.get(i);
- if (bucketTimestamp >= windowStart && bucketTimestamp > 0) {
- count += buckets.get(i);
- }
+ if (result.loadRate > 0) {
+ byLoad.add(result);
}
-
- cachedTotalCount = count;
- cachedCountTime = currentTimeMs;
- return count;
}
-
- long getLastAccessTime() {
- return lastAccessTime;
+ byQuery.sort(QUERY_RATE_COMPARATOR);
+ byLoad.sort(LOAD_RATE_COMPARATOR);
+ Map<Long, AccessStatsResult> capped = Maps.newHashMapWithExpectedSize(
+ Math.min(merged.size(), 2 * retainLimit));
+ for (int i = 0; i < Math.min(retainLimit, byQuery.size()); i++) {
+ capped.put(byQuery.get(i).id, byQuery.get(i));
}
-
- /**
- * Clean up expired buckets
- */
- void cleanup(long currentTimeMs, long timeWindowMs) {
- long windowStart = currentTimeMs - timeWindowMs;
- for (int i = 0; i < buckets.length(); i++) {
- long bucketTimestamp = bucketTimestamps.get(i);
- if (bucketTimestamp > 0 && bucketTimestamp < windowStart) {
- buckets.set(i, 0);
- bucketTimestamps.set(i, 0);
- }
- }
- cachedTotalCount = -1; // Invalidate cache
- }
-
- /**
- * Check if this counter has any recent activity
- */
- boolean hasRecentActivity(long currentTimeMs, long timeWindowMs) {
- return lastAccessTime >= (currentTimeMs - timeWindowMs);
+ for (int i = 0; i < Math.min(retainLimit, byLoad.size()); i++) {
+ capped.put(byLoad.get(i).id, byLoad.get(i));
}
+ return capped;
}
/**
- * Shard structure to reduce lock contention
+ * One tablet's half-built stats while the query and load lists are being
merged.
+ * Each dimension keeps its own window: the two entries for one tablet
come from the
+ * same report, but a backend that skipped a round carries a wider window
on the
+ * dimension that was reported then.
*/
- private static class AccessStatsShard {
- // ID counters: id -> SlidingWindowCounter
- private final ConcurrentHashMap<Long, SlidingWindowCounter> idCounters
= new ConcurrentHashMap<>();
- }
+ private static class Accumulator {
+ private long scanDelta;
+ private long loadDelta;
+ private long lastQueryMs;
+ private long lastLoadMs;
+ // Never zero, so no division guard is needed below.
+ private long scanWindowMs = 1L;
+ private long loadWindowMs = 1L;
- // Sharded access stats to reduce lock contention
- private final AccessStatsShard[] shards = new AccessStatsShard[SHARD_SIZE];
-
- // Access counter for monitoring
- private final AtomicLong totalAccessCount = new AtomicLong(0);
-
- // Aggregated stats cache for metrics/observability
- // - recentAccessCountInWindow: sum of accessCount of all active IDs in
current time window
- // - activeIdsInWindow: number of IDs that have recent activity in current
time window
- // These are computed on-demand with a TTL to avoid expensive full scans
on every metric scrape.
- private static final long AGGREGATE_REFRESH_INTERVAL_MS = 10_000L;
- private final AtomicLong recentAccessCountInWindow = new AtomicLong(0);
- private final AtomicLong activeIdsInWindow = new AtomicLong(0);
- private final AtomicLong lastAggregateRefreshTimeMs = new AtomicLong(0);
-
- // Cleanup daemon
- private AccessStatsCleanupDaemon cleanupDaemon;
-
- // Thread pool for async recordAccess execution
- private ThreadPoolExecutor asyncExecutor;
-
- // Default bucket size: 1 minute (60 buckets for 1 hour window)
- private static final long DEFAULT_BUCKET_SIZE_SECOND = 60L;
-
- // Default cleanup interval: 5 minutes
- private static final long DEFAULT_CLEANUP_INTERVAL_SECOND = 300L;
-
- TabletSlidingWindowAccessStats() {
- long configuredWindowSecond =
Config.active_tablet_sliding_window_time_window_second;
- long effectiveWindowSecond = Math.max(1L, configuredWindowSecond);
- if (configuredWindowSecond < DEFAULT_BUCKET_SIZE_SECOND) {
- LOG.warn("active_tablet_sliding_window_time_window_second={} is
less than default bucket size {}, "
- + "using one bucket to avoid zero-bucket window",
- configuredWindowSecond, DEFAULT_BUCKET_SIZE_SECOND);
- }
- this.timeWindowMs = effectiveWindowSecond * 1000L;
- this.bucketSizeMs = DEFAULT_BUCKET_SIZE_SECOND * 1000L;
- this.numBuckets = (int) Math.max(1L, effectiveWindowSecond /
DEFAULT_BUCKET_SIZE_SECOND);
- this.cleanupIntervalMs = DEFAULT_CLEANUP_INTERVAL_SECOND * 1000L;
-
- // Initialize shards
- for (int i = 0; i < SHARD_SIZE; i++) {
- shards[i] = new AccessStatsShard();
+ // Accesses per minute. Backends must be compared by rate, not by raw
delta: a
+ // skipped report makes the next delta cover several periods, and
backends report
+ // on independent phases.
+ private double scanRate() {
+ return scanDelta * 60_000.0 / scanWindowMs;
}
- // Start cleanup daemon and async executor if enabled
- if (Config.enable_active_tablet_sliding_window_access_stats) {
- this.cleanupDaemon = new AccessStatsCleanupDaemon();
- this.cleanupDaemon.start();
- // Initialize async executor for recordAccess
- // Use a small thread pool with bounded queue to avoid blocking
- // If queue is full, discard the task (statistics can tolerate
some loss)
- this.asyncExecutor = new ThreadPoolExecutor(
- 2, // core pool size
- 4, // maximum pool size
- 60L, TimeUnit.SECONDS,
- new LinkedBlockingQueue<>(1000), // queue capacity
- r -> {
- Thread t = new Thread(r,
"sliding-window-access-stats-async-tablet-record");
- t.setDaemon(true);
- return t;
- },
- new ThreadPoolExecutor.DiscardPolicy() // discard when
queue is full
- );
-
- LOG.info("SlidingWindowAccessStats initialized for type tablet:
timeWindow={}ms, bucketSize={}ms, "
- + "numBuckets={}, shardSize={}, cleanupInterval={}ms",
- timeWindowMs, bucketSizeMs, numBuckets, SHARD_SIZE,
cleanupIntervalMs);
+ private double loadRate() {
+ return loadDelta * 60_000.0 / loadWindowMs;
}
}
- /**
- * Get shard index for a given ID
- */
- private int getShardIndex(long id) {
- int hash = SHARD_HASH.hashLong(id).asInt();
- return Math.floorMod(hash, SHARD_SIZE);
- }
-
- private void refreshAggregatesIfNeeded(long currentTimeMs) {
- long last = lastAggregateRefreshTimeMs.get();
- if (currentTimeMs - last < AGGREGATE_REFRESH_INTERVAL_MS) {
- return;
- }
- if (!lastAggregateRefreshTimeMs.compareAndSet(last, currentTimeMs)) {
- return;
- }
-
- int activeIds = 0;
- long totalAccess = 0;
- for (AccessStatsShard shard : shards) {
- for (SlidingWindowCounter counter : shard.idCounters.values()) {
- if (counter.hasRecentActivity(currentTimeMs, timeWindowMs)) {
- activeIds++;
- totalAccess += counter.getCount(currentTimeMs,
timeWindowMs, bucketSizeMs, numBuckets);
- }
- }
- }
- activeIdsInWindow.set(activeIds);
- recentAccessCountInWindow.set(totalAccess);
+ public void removeBackend(long beId) {
+ beToStats.remove(beId);
}
/**
- * Get total access count within current time window across all IDs
(cached).
+ * Get total access count in the latest backend snapshots.
*/
public long getRecentAccessCountInWindow() {
if (!Config.enable_active_tablet_sliding_window_access_stats) {
return 0L;
}
- long now = System.currentTimeMillis();
- refreshAggregatesIfNeeded(now);
- return recentAccessCountInWindow.get();
+ return cachedMergedStats().values().stream().mapToLong(result ->
result.accessCount).sum();
}
/**
- * Get number of active IDs within current time window (cached).
+ * Get the number of distinct tablets in the latest backend snapshots.
*/
public long getActiveIdsInWindow() {
if (!Config.enable_active_tablet_sliding_window_access_stats) {
return 0L;
}
- long now = System.currentTimeMillis();
- refreshAggregatesIfNeeded(now);
- return activeIdsInWindow.get();
+ return cachedMergedStats().size();
}
/**
- * Get total access count since FE start (monotonic increasing while
enabled).
+ * Get total access count reported since FE start.
*/
public long getTotalAccessCount() {
if (!Config.enable_active_tablet_sliding_window_access_stats) {
@@ -315,80 +257,52 @@ public class TabletSlidingWindowAccessStats {
}
/**
- * Record an access asynchronously
- * This method is non-blocking and should be used in high-frequency call
paths
- * to avoid blocking the caller thread.
- */
- public void recordAccessAsync(long id) {
- if (asyncExecutor == null) {
- return;
- }
-
- try {
- asyncExecutor.execute(() -> {
- try {
- recordAccess(id);
- } catch (Exception e) {
- // Log but don't propagate exception to avoid affecting
caller
- LOG.warn("Failed to record access asynchronously for
tablet id={}", id, e);
- }
- });
- } catch (Exception e) {
- // If executor is shutdown or queue is full, silently ignore
- // Statistics can tolerate some loss
- LOG.warn("Failed to submit async recordAccess task for tablet
id={}", id, e);
- }
- }
-
- /**
- * Record an access
- */
- public void recordAccess(long id) {
- long currentTime = System.currentTimeMillis();
- int shardIndex = getShardIndex(id);
- AccessStatsShard shard = shards[shardIndex];
-
- SlidingWindowCounter counter = shard.idCounters.computeIfAbsent(id,
- k -> new SlidingWindowCounter(numBuckets));
- counter.add(currentTime, bucketSizeMs, numBuckets);
- totalAccessCount.incrementAndGet();
- }
-
- /**
- * Get access count for an ID within the time window
+ * Get access information for a tablet, merged across the backends holding
its replicas the
+ * same way getTopNActive() does. Returning a single backend's record
instead would report
+ * that backend's LastAccessTime even when another replica was touched
more recently, and
+ * would show one replica's share of a split query load as the whole
tablet's.
*/
public AccessStatsResult getAccessInfo(long id) {
if (!Config.enable_active_tablet_sliding_window_access_stats) {
return null;
}
- int shardIndex = getShardIndex(id);
- AccessStatsShard shard = shards[shardIndex];
- SlidingWindowCounter counter = shard.idCounters.get(id);
-
- if (counter == null) {
- return null;
+ long oldestKept = System.currentTimeMillis()
+ - Math.max(1L,
Config.active_tablet_sliding_window_time_window_second) * 1000L;
+ AccessStatsResult result = null;
+ for (Map<Long, AccessStatsResult> backendStats : beToStats.values()) {
+ AccessStatsResult candidate = backendStats.get(id);
+ if (candidate != null && candidate.lastAccessTime >= oldestKept) {
+ result = (result == null) ? candidate : mergeReplicas(result,
candidate);
+ }
}
-
- long currentTime = System.currentTimeMillis();
- return new AccessStatsResult(
- id,
- counter.getCount(currentTime, timeWindowMs, bucketSizeMs,
numBuckets),
- counter.getLastAccessTime());
+ return result;
}
/**
- * Result for top N query
+ * Result for top N query.
*/
public static class AccessStatsResult {
public final long id;
+ // Kept apart because the two dimensions merge differently across
replicas -- see
+ // mergeReplicas(). accessCount is their sum, the single number the
SHOW / PROC
+ // views display.
+ public final long scanCount;
+ public final long loadCount;
public final long accessCount;
public final long lastAccessTime;
+ public final double scanRate;
+ public final double loadRate;
- public AccessStatsResult(long id, long accessCount, long
lastAccessTime) {
+ public AccessStatsResult(long id, long scanCount, long loadCount, long
lastAccessTime,
+ double scanRate, double loadRate) {
this.id = id;
- this.accessCount = accessCount;
+ this.scanCount = scanCount;
+ this.loadCount = loadCount;
+ this.accessCount = scanCount + loadCount;
this.lastAccessTime = lastAccessTime;
+ this.scanRate = scanRate;
+ this.loadRate = loadRate;
}
@Override
@@ -397,143 +311,151 @@ public class TabletSlidingWindowAccessStats {
+ "id=" + id
+ ", accessCount=" + accessCount
+ ", lastAccessTime=" + lastAccessTime
+ + ", scanRate=" + scanRate
+ + ", loadRate=" + loadRate
+ '}';
}
}
/**
- * Get top N most active IDs
- * Uses a min-heap (PriorityQueue) to maintain TopN efficiently without
sorting all results.
+ * Get top N active tablets, reserving half of the capacity for each
access type.
*/
public List<AccessStatsResult> getTopNActive(int topN) {
- if (!Config.enable_active_tablet_sliding_window_access_stats) {
- return Collections.emptyList();
- }
-
- if (topN <= 0) {
+ if (!Config.enable_active_tablet_sliding_window_access_stats || topN
<= 0) {
return Collections.emptyList();
}
-
- long currentTime = System.currentTimeMillis();
- // Use a min-heap with reversed comparator to maintain TopN
- // The heap keeps the smallest element at the top, so we can
efficiently replace it
- // when we find a larger element
- PriorityQueue<AccessStatsResult> minHeap = new PriorityQueue<>(
- topN + 1, // Initial capacity: topN + 1 to avoid resizing
- Collections.reverseOrder(TOPN_ACTIVE_COMPARATOR) // Reversed:
min-heap for TopN
- );
-
- // Collect from all shards and maintain TopN using min-heap
- for (AccessStatsShard shard : shards) {
- for (Map.Entry<Long, SlidingWindowCounter> entry :
shard.idCounters.entrySet()) {
- long id = entry.getKey();
- SlidingWindowCounter counter = entry.getValue();
-
- // Skip if no recent activity
- if (!counter.hasRecentActivity(currentTime, timeWindowMs)) {
- continue;
- }
-
- long accessCount = counter.getCount(currentTime, timeWindowMs,
bucketSizeMs, numBuckets);
- if (accessCount > 0) {
- AccessStatsResult result = new AccessStatsResult(id,
accessCount, counter.getLastAccessTime());
-
- if (minHeap.size() < topN) {
- // Heap not full, directly add
- minHeap.offer(result);
- } else {
- // Heap is full, compare with the smallest element
(heap top)
- // If current element is larger, replace the heap top
- if (TOPN_ACTIVE_COMPARATOR.compare(result,
minHeap.peek()) > 0) {
- minHeap.poll();
- minHeap.offer(result);
- }
- }
- }
- }
- }
-
- // Convert heap to list and sort in descending order (TopN)
- List<AccessStatsResult> results = new ArrayList<>(minHeap);
- results.sort(TOPN_ACTIVE_COMPARATOR);
- return results;
+ return pickAcrossDimensions(mergeBackendStats().values(), topN);
}
/**
- * Clean up expired access records
+ * Spend a budget of {@code limit} tablets across the two dimensions, each
ranked on its own
+ * rate. Half is reserved for queries, counted in tablets actually added
so a tablet hot on
+ * both dimensions costs one slot and not two; loads then fill the rest,
and queries backfill
+ * whatever is still free - which happens when loads ran out, or when the
two lists
+ * overlapped. Neither dimension is ever ranked against the other: scans
and flushes differ
+ * by one to two orders of magnitude, so a single ordering drops
load-heavy tablets as a class.
*/
- private void cleanupExpiredRecords() {
- if (!Config.enable_active_tablet_sliding_window_access_stats) {
- return;
- }
-
- long currentTime = System.currentTimeMillis();
- int cleaned = 0;
-
- // Clean each shard
- for (AccessStatsShard shard : shards) {
- // Clean ID counters
- for (Map.Entry<Long, SlidingWindowCounter> entry :
shard.idCounters.entrySet()) {
- SlidingWindowCounter counter = entry.getValue();
- counter.cleanup(currentTime, timeWindowMs);
-
- if (!counter.hasRecentActivity(currentTime, timeWindowMs)) {
- shard.idCounters.remove(entry.getKey());
- cleaned++;
- }
+ private static List<AccessStatsResult> pickAcrossDimensions(
+ Collection<AccessStatsResult> candidates, int limit) {
+ List<AccessStatsResult> queryStats = new ArrayList<>();
+ List<AccessStatsResult> loadStats = new ArrayList<>();
+ for (AccessStatsResult result : candidates) {
+ if (result.scanRate > 0) {
+ queryStats.add(result);
+ }
+ if (result.loadRate > 0) {
+ loadStats.add(result);
}
}
+ queryStats.sort(QUERY_RATE_COMPARATOR);
+ loadStats.sort(LOAD_RATE_COMPARATOR);
- if (LOG.isDebugEnabled() && cleaned > 0) {
- LOG.debug("Cleaned up {} expired access records for type tablet",
cleaned);
+ Map<Long, AccessStatsResult> selected = new LinkedHashMap<>();
+ int queryIdx = 0;
+ int loadIdx = 0;
+ int queryReserve = limit / 2;
+ while (queryIdx < queryStats.size() && selected.size() < queryReserve)
{
+ AccessStatsResult result = queryStats.get(queryIdx++);
+ selected.putIfAbsent(result.id, result);
+ }
+ while (loadIdx < loadStats.size() && selected.size() < limit) {
+ AccessStatsResult result = loadStats.get(loadIdx++);
+ selected.putIfAbsent(result.id, result);
}
+ while (queryIdx < queryStats.size() && selected.size() < limit) {
+ AccessStatsResult result = queryStats.get(queryIdx++);
+ selected.putIfAbsent(result.id, result);
+ }
+ return new ArrayList<>(selected.values());
}
/**
- * Get statistics summary
+ * Get statistics summary.
*/
public String getStatsSummary() {
if (!Config.enable_active_tablet_sliding_window_access_stats) {
return String.format("Active tablet sliding window access stats is
disabled");
}
- long currentTime = System.currentTimeMillis();
- refreshAggregatesIfNeeded(currentTime);
- int activeIds = (int) getActiveIdsInWindow();
- long totalAccess = getRecentAccessCountInWindow();
-
+ Map<Long, AccessStatsResult> mergedStats = cachedMergedStats();
+ long totalAccess = mergedStats.values().stream().mapToLong(result ->
result.accessCount).sum();
return String.format(
- "SlidingWindowAccessStats{type=tablet, timeWindow=%ds,
bucketSize=%ds, numBuckets=%d, "
- + "shardSize=%d, activeIds=%d, "
- + "totalAccess=%d, totalAccessCount=%d}",
- timeWindowMs / 1000, bucketSizeMs / 1000, numBuckets, SHARD_SIZE,
- activeIds, totalAccess, totalAccessCount.get());
+ "SlidingWindowAccessStats{type=tablet, beCount=%d,
activeIds=%d, "
+ + "totalAccess=%d, totalAccessCount=%d}",
+ beToStats.size(), mergedStats.size(), totalAccess,
totalAccessCount.get());
}
- /**
- * Cleanup daemon for expired records
- */
- private class AccessStatsCleanupDaemon extends MasterDaemon {
- public AccessStatsCleanupDaemon() {
- super("sliding-window-access-stats-cleanup-tablet" +
cleanupIntervalMs);
+ private Map<Long, AccessStatsResult> cachedMergedStats() {
+ long now = System.currentTimeMillis();
+ long last = mergedCacheTimeMs.get();
+ if (now - last < MERGED_CACHE_TTL_MS) {
+ return mergedCache;
}
+ if (!mergedCacheTimeMs.compareAndSet(last, now)) {
+ // Another thread is already refreshing; the previous snapshot is
good enough here.
+ return mergedCache;
+ }
+ Map<Long, AccessStatsResult> merged = mergeBackendStats();
+ mergedCache = merged;
+ return merged;
+ }
- @Override
- protected void runAfterCatalogReady() {
- if (!Env.getCurrentEnv().isMaster()) {
- return;
- }
-
- try {
- cleanupExpiredRecords();
- if (LOG.isDebugEnabled()) {
- LOG.debug("tablet stat = {}, top 10 active = {}",
- getStatsSummary(), getTopNActive(10));
+ /**
+ * Flatten beId -> tabletId -> stats into one view keyed by tablet. A
tablet with several
+ * replicas is reported once per backend holding one, so the collisions
are resolved by
+ * mergeReplicas().
+ */
+ private Map<Long, AccessStatsResult> mergeBackendStats() {
+ int upperBound = 0;
+ for (Map<Long, AccessStatsResult> backendStats : beToStats.values()) {
+ upperBound += backendStats.size();
+ }
+ Map<Long, AccessStatsResult> mergedStats =
Maps.newHashMapWithExpectedSize(upperBound);
+ long oldestKept = System.currentTimeMillis()
+ - Math.max(1L,
Config.active_tablet_sliding_window_time_window_second) * 1000L;
+ for (Map.Entry<Long, Map<Long, AccessStatsResult>> backend :
beToStats.entrySet()) {
+ Map<Long, AccessStatsResult> backendStats = backend.getValue();
+ boolean hasActiveStats = false;
+ for (AccessStatsResult result : backendStats.values()) {
+ if (result.lastAccessTime >= oldestKept) {
+ hasActiveStats = true;
+ mergedStats.merge(result.id, result,
TabletSlidingWindowAccessStats::mergeReplicas);
}
- } catch (Exception e) {
- LOG.warn("Failed to cleanup expired access records for type
tablet", e);
+ }
+ if (!hasActiveStats) {
+ // A concurrent report may have replaced this snapshot; only
remove the one read.
+ beToStats.remove(backend.getKey(), backendStats);
}
}
+ return mergedStats;
+ }
+
+ /**
+ * Query traffic always SUMS across backends: the planner assigns each
scan range to exactly
+ * one replica, so a tablet's query traffic is split between the backends
holding it and only
+ * the sum is the tablet's real rate - taking the max would make a
three-replica tablet read
+ * three times colder than a single-replica one carrying the same load.
+ *
+ * <p>Load traffic depends on the deployment. Shared-nothing replicates
every write, so all
+ * three backends report the same flushes and summing would multiply them
by the replication
+ * factor - max is the tablet's real rate there. Cloud does not replicate
on the write path:
+ * a CloudTablet holds a single CloudReplica, and
CloudReplica#getBackendIdImpl resolves it
+ * to one backend per compute group (a random one of cloud_replica_num when
+ * enable_cloud_multi_replica is on), so two backends reporting the same
tablet did
+ * independent work and max would silently discard half of it.
+ *
+ * <p>lastAccessTime is the most recent touch either way.
+ */
+ private static AccessStatsResult mergeReplicas(AccessStatsResult left,
AccessStatsResult right) {
+ boolean loadIsSplit = Config.isCloudMode();
+ return new AccessStatsResult(left.id,
+ left.scanCount + right.scanCount,
+ loadIsSplit ? left.loadCount + right.loadCount
+ : Math.max(left.loadCount, right.loadCount),
+ Math.max(left.lastAccessTime, right.lastAccessTime),
+ left.scanRate + right.scanRate,
+ loadIsSplit ? left.loadRate + right.loadRate
+ : Math.max(left.loadRate, right.loadRate));
}
public static TabletSlidingWindowAccessStats getInstance() {
@@ -546,10 +468,4 @@ public class TabletSlidingWindowAccessStats {
}
return instance;
}
-
- // async record tablet instance access
- public static void recordTablet(long id) {
- TabletSlidingWindowAccessStats sas = getInstance();
- sas.recordAccessAsync(id);
- }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java
index 541570b079e..7c4d4f39a58 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java
@@ -21,7 +21,6 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.Replica.ReplicaState;
import org.apache.doris.catalog.Tablet;
-import org.apache.doris.catalog.TabletSlidingWindowAccessStats;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.InternalErrorCode;
import org.apache.doris.common.UserException;
@@ -85,14 +84,12 @@ public class CloudTablet extends Tablet implements
GsonPostProcessable {
// Cluster id is supplied by the caller (resolved lazily once per request),
// bypassing the per-replica
ConnectContext/priv/status/autoStart/existence pipeline.
public Multimap<Long, Long>
getNormalReplicaBackendPathMapByClusterId(String clusterId) throws
UserException {
- TabletSlidingWindowAccessStats.recordTablet(getId());
Multimap<Long, Long> pathMap =
super.getNormalReplicaBackendPathMapImpl(null,
(rep, be) -> ((CloudReplica)
rep).getBackendIdWithClusterId(clusterId));
return backendPathMapReprocess(pathMap);
}
public Multimap<Long, Long> getNormalReplicaBackendPathMap(String
beEndpoint) throws UserException {
- TabletSlidingWindowAccessStats.recordTablet(getId());
Multimap<Long, Long> pathMap =
super.getNormalReplicaBackendPathMapImpl(beEndpoint,
(rep, be) -> ((CloudReplica) rep).getBackendId(be));
return backendPathMapReprocess(pathMap);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
index f53c4c0833a..1be04971b7e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
@@ -24,6 +24,7 @@ import org.apache.doris.catalog.ColocateGroupSchema;
import org.apache.doris.catalog.ColocateTableIndex.GroupId;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.ReplicaAllocation;
+import org.apache.doris.catalog.TabletSlidingWindowAccessStats;
import org.apache.doris.cloud.catalog.CloudColocatePlacement;
import org.apache.doris.cloud.catalog.CloudComputeGroupMeta;
import org.apache.doris.cloud.catalog.CloudEnv;
@@ -577,6 +578,7 @@ public class CloudSystemInfoService extends
SystemInfoService {
be.setAlive(false);
be.setLastMissingHeartbeatTime(System.currentTimeMillis());
Env.getCurrentEnv().getEditLog().logDropBackend(be);
+
TabletSlidingWindowAccessStats.getInstance().removeBackend(be.getId());
LOG.info("dropped cloud backend={}, and
lastMissingHeartbeatTime={}", be, be.getLastMissingHeartbeatTime());
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java
index 4ddbced69a9..8457eca1493 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java
@@ -41,6 +41,7 @@ import org.apache.doris.catalog.Tablet;
import org.apache.doris.catalog.Tablet.TabletStatus;
import org.apache.doris.catalog.TabletInvertedIndex;
import org.apache.doris.catalog.TabletMeta;
+import org.apache.doris.catalog.TabletSlidingWindowAccessStats;
import org.apache.doris.common.Config;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.Pair;
@@ -110,6 +111,7 @@ import org.apache.logging.log4j.Logger;
import org.apache.thrift.TException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
@@ -165,6 +167,15 @@ public class ReportHandler extends Daemon {
}
long beId = backend.getId();
+ // A backend too old to send these fields contributes no activity at
all, for as long as
+ // it stays un-upgraded. Not fixed: a capability handshake costs far
more than the
+ // precision is worth here. Upgrade BEs before FE if it matters.
+ if (request.isSetTopQueryTablets() || request.isSetTopLoadTablets()) {
+ TabletSlidingWindowAccessStats.getInstance().updateFromReport(
+ beId,
+ request.isSetTopQueryTablets() ?
request.getTopQueryTablets() : Collections.emptyList(),
+ request.isSetTopLoadTablets() ?
request.getTopLoadTablets() : Collections.emptyList());
+ }
Map<TTaskType, Set<Long>> tasks = null;
Map<String, TDisk> disks = null;
Map<Long, TTablet> tablets = null;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
index ae2f365cd16..ec51655095c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
@@ -23,6 +23,7 @@ import org.apache.doris.catalog.ColocateTableIndex.GroupId;
import org.apache.doris.catalog.DiskInfo;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.ReplicaAllocation;
+import org.apache.doris.catalog.TabletSlidingWindowAccessStats;
import org.apache.doris.cloud.qe.ComputeGroupException;
import org.apache.doris.cluster.ClusterGuard;
import org.apache.doris.cluster.ClusterGuardException;
@@ -306,6 +307,7 @@ public class SystemInfoService {
copiedReportVersions.remove(droppedBackend.getId());
ImmutableMap<Long, AtomicLong> newIdToReportVersion =
ImmutableMap.copyOf(copiedReportVersions);
idToReportVersionRef = newIdToReportVersion;
+
TabletSlidingWindowAccessStats.getInstance().removeBackend(droppedBackend.getId());
// log
Env.getCurrentEnv().getEditLog().logDropBackend(droppedBackend);
@@ -855,6 +857,7 @@ public class SystemInfoService {
copiedReportVersions.remove(backend.getId());
ImmutableMap<Long, AtomicLong> newIdToReportVersion =
ImmutableMap.copyOf(copiedReportVersions);
idToReportVersionRef = newIdToReportVersion;
+
TabletSlidingWindowAccessStats.getInstance().removeBackend(backend.getId());
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/TabletSlidingWindowAccessStatsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/TabletSlidingWindowAccessStatsTest.java
index d29e7bda0bf..61b9ac5fbd4 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/catalog/TabletSlidingWindowAccessStatsTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/catalog/TabletSlidingWindowAccessStatsTest.java
@@ -18,207 +18,427 @@
package org.apache.doris.catalog;
import org.apache.doris.common.Config;
+import org.apache.doris.thrift.TActiveTabletStat;
+import com.google.common.collect.Sets;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import java.lang.reflect.Constructor;
-import java.lang.reflect.Field;
-import java.lang.reflect.Method;
import java.util.ArrayList;
-import java.util.Comparator;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
-
-/**
- * Unit tests for TabletSlidingWindowAccessStatsTest sliding window logic.
- *
- * <p>We use reflection to test the private static inner class
SlidingWindowCounter,
- * focusing on time-window counting, bucket expiration, cleanup and cache
invalidation.
- */
public class TabletSlidingWindowAccessStatsTest {
+ private boolean originalEnabled;
+ private TabletSlidingWindowAccessStats stats;
+
+ @BeforeEach
+ public void setUp() {
+ originalEnabled =
Config.enable_active_tablet_sliding_window_access_stats;
+ Config.enable_active_tablet_sliding_window_access_stats = true;
+ stats = new TabletSlidingWindowAccessStats();
+ }
- private static class CounterHarness {
- private final Object counter;
- private final Method add;
- private final Method getCount;
- private final Method cleanup;
- private final Method hasRecentActivity;
- private final Method getLastAccessTime;
-
- CounterHarness(int numBuckets) throws Exception {
- Class<?> clazz = Class.forName(
-
"org.apache.doris.catalog.TabletSlidingWindowAccessStats$SlidingWindowCounter");
- Constructor<?> ctor = clazz.getDeclaredConstructor(int.class);
- ctor.setAccessible(true);
- this.counter = ctor.newInstance(numBuckets);
-
- this.add = clazz.getDeclaredMethod("add", long.class, long.class,
int.class);
- this.add.setAccessible(true);
- this.getCount = clazz.getDeclaredMethod("getCount", long.class,
long.class, long.class, int.class);
- this.getCount.setAccessible(true);
- this.cleanup = clazz.getDeclaredMethod("cleanup", long.class,
long.class);
- this.cleanup.setAccessible(true);
- this.hasRecentActivity =
clazz.getDeclaredMethod("hasRecentActivity", long.class, long.class);
- this.hasRecentActivity.setAccessible(true);
- this.getLastAccessTime =
clazz.getDeclaredMethod("getLastAccessTime");
- this.getLastAccessTime.setAccessible(true);
- }
-
- void add(long currentTimeMs, long bucketSizeMs, int numBuckets) throws
Exception {
- add.invoke(counter, currentTimeMs, bucketSizeMs, numBuckets);
- }
-
- long getCount(long currentTimeMs, long timeWindowMs, long
bucketSizeMs, int numBuckets) throws Exception {
- Object v = getCount.invoke(counter, currentTimeMs, timeWindowMs,
bucketSizeMs, numBuckets);
- return (long) v;
- }
+ @AfterEach
+ public void tearDown() {
+ Config.enable_active_tablet_sliding_window_access_stats =
originalEnabled;
+ }
- void cleanup(long currentTimeMs, long timeWindowMs) throws Exception {
- cleanup.invoke(counter, currentTimeMs, timeWindowMs);
- }
+ @Test
+ public void testUpdateFromReportProvidesAccessInfo() {
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L,
+ Collections.singletonList(queryStat(10L, 12L, now - 100L,
60_000L)),
+ Collections.singletonList(loadStat(10L, 3L, now - 200L,
60_000L)));
+
+ TabletSlidingWindowAccessStats.AccessStatsResult result =
stats.getAccessInfo(10L);
+ Assertions.assertNotNull(result);
+ Assertions.assertEquals(15L, result.accessCount);
+ Assertions.assertEquals(now - 100L, result.lastAccessTime);
+ Assertions.assertEquals(12.0, result.scanRate);
+ Assertions.assertEquals(3.0, result.loadRate);
+ }
- boolean hasRecentActivity(long currentTimeMs, long timeWindowMs)
throws Exception {
- Object v = hasRecentActivity.invoke(counter, currentTimeMs,
timeWindowMs);
- return (boolean) v;
- }
+ // A report carries only the tablets that saw traffic since the previous
one. Dropping the
+ // rest would shrink "active" to a single report interval, which in a
cluster reporting
+ // every second means a tablet is active for one second after it stops
being queried.
+ @Test
+ public void testTabletsMissingFromTheNextReportStayActive() {
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L, Collections.singletonList(queryStat(10L,
12L, now, 60_000L)),
+ Collections.emptyList());
+ // Next report: tablet 10 saw no traffic, tablet 20 did.
+ stats.updateFromReport(1L, Collections.singletonList(queryStat(20L,
7L, now, 60_000L)),
+ Collections.emptyList());
+
+ Assertions.assertEquals(12L, stats.getAccessInfo(10L).accessCount);
+ Assertions.assertEquals(7L, stats.getAccessInfo(20L).accessCount);
+ Assertions.assertEquals(2L, stats.getActiveIdsInWindow());
+ Assertions.assertEquals(19L, stats.getRecentAccessCountInWindow());
+ }
- long getLastAccessTime() throws Exception {
- Object v = getLastAccessTime.invoke(counter);
- return (long) v;
- }
+ // A repeated tablet takes the fresh values, it is not accumulated across
reports.
+ @Test
+ public void testRepeatedTabletTakesTheLatestReport() {
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L, Collections.singletonList(queryStat(10L,
12L, now - 1_000L, 60_000L)),
+ Collections.emptyList());
+ stats.updateFromReport(1L, Collections.singletonList(queryStat(10L,
7L, now, 60_000L)),
+ Collections.emptyList());
+
+ Assertions.assertEquals(7L, stats.getAccessInfo(10L).accessCount);
+ Assertions.assertEquals(now, stats.getAccessInfo(10L).lastAccessTime);
+ Assertions.assertEquals(1L, stats.getActiveIdsInWindow());
}
+ // Equal rates fall through to the tie-break, and the tie-break runs in
the same direction
+ // as the rate: most recently touched wins.
@Test
- public void testAddAndGetCountWithinWindow() throws Exception {
- int numBuckets = 5;
- long bucketSizeMs = 1000L;
- long timeWindowMs = 5000L;
- long base = 10_000L; // avoid 0 (bucket timestamp 0 is treated as
"unset" in getCount)
+ public void testEqualRatesPreferTheMostRecentlyTouched() {
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L, Arrays.asList(
+ queryStat(10L, 10L, now - 1_000L, 60_000L),
+ queryStat(20L, 10L, now, 60_000L)), Collections.emptyList());
+
+ List<TabletSlidingWindowAccessStats.AccessStatsResult> top =
stats.getTopNActive(1);
+ Assertions.assertEquals(1, top.size());
+ Assertions.assertEquals(20L, top.get(0).id);
+ }
- CounterHarness c = new CounterHarness(numBuckets);
- c.add(base, bucketSizeMs, numBuckets);
- c.add(base + 1000L, bucketSizeMs, numBuckets);
- c.add(base + 2000L, bucketSizeMs, numBuckets);
+ // The half-and-half budget is a floor, not a cap: whichever dimension has
fewer candidates
+ // hands its unused share to the other, and when both are short everything
is returned.
+ @Test
+ public void testTopNQuotaSpillsToTheOtherDimension() {
+ long now = System.currentTimeMillis();
+
+ // Only queries: the load half is handed over, so all 8 query tablets
fit in topN=8.
+ TabletSlidingWindowAccessStats queryOnly = new
TabletSlidingWindowAccessStats();
+ queryOnly.updateFromReport(1L, queryStats(8, now),
Collections.emptyList());
+ Assertions.assertEquals(8, queryOnly.getTopNActive(8).size());
+
+ // Only loads: mirror image.
+ TabletSlidingWindowAccessStats loadOnly = new
TabletSlidingWindowAccessStats();
+ loadOnly.updateFromReport(1L, Collections.emptyList(), loadStats(8,
now));
+ Assertions.assertEquals(8, loadOnly.getTopNActive(8).size());
+
+ // Both below topN/2: nothing to spill, every candidate is returned
and topN is not met.
+ TabletSlidingWindowAccessStats bothShort = new
TabletSlidingWindowAccessStats();
+ bothShort.updateFromReport(1L, queryStats(2, now), loadStats(3, now));
+ Assertions.assertEquals(5, bothShort.getTopNActive(20).size());
+ }
- Assertions.assertEquals(base + 2000L, c.getLastAccessTime());
- Assertions.assertTrue(c.hasRecentActivity(base + 2000L, timeWindowMs));
+ @Test
+ public void testTopNFillsBudgetWhenQueryAndLoadOverlap() {
+ long now = System.currentTimeMillis();
+ List<TActiveTabletStat> queries = new ArrayList<>();
+ List<TActiveTabletStat> loads = new ArrayList<>();
+ for (long id = 1; id <= 4; id++) {
+ queries.add(queryStat(id, 10L - id, now, 60_000L));
+ loads.add(loadStat(id, 10L - id, now, 60_000L));
+ }
+ stats.updateFromReport(1L, queries, loads);
- long cnt = c.getCount(base + 2000L, timeWindowMs, bucketSizeMs,
numBuckets);
- Assertions.assertEquals(3L, cnt);
+ List<TabletSlidingWindowAccessStats.AccessStatsResult> top =
stats.getTopNActive(4);
+ Assertions.assertEquals(4, top.size());
+ Assertions.assertEquals(Sets.newHashSet(1L, 2L, 3L, 4L),
+ top.stream().map(r -> r.id).collect(Collectors.toSet()));
+ }
- // Cached result (same timestamp) should be consistent
- long cnt2 = c.getCount(base + 2000L, timeWindowMs, bucketSizeMs,
numBuckets);
- Assertions.assertEquals(3L, cnt2);
+ @Test
+ public void testTopNQuotaBoundaries() {
+ long now = System.currentTimeMillis();
+ // topN, query candidates, load candidates, selected queries, selected
loads.
+ int[][] cases = {{10, 100, 100, 5, 5}, {10, 2, 100, 2, 8},
+ {10, 100, 2, 8, 2}, {10, 2, 3, 2, 3}, {1, 2, 2, 0, 1}};
+ for (int[] testCase : cases) {
+ TabletSlidingWindowAccessStats candidateStats = new
TabletSlidingWindowAccessStats();
+ candidateStats.updateFromReport(1L, queryStats(testCase[1], now),
loadStats(testCase[2], now));
+ List<TabletSlidingWindowAccessStats.AccessStatsResult> top =
candidateStats.getTopNActive(testCase[0]);
+ Assertions.assertEquals(testCase[3] + testCase[4], top.size());
+ Assertions.assertEquals(testCase[3], top.stream().filter(r ->
r.scanRate > 0).count());
+ Assertions.assertEquals(testCase[4], top.stream().filter(r ->
r.loadRate > 0).count());
+ }
+ }
- // add() should invalidate cachedTotalCount
- c.add(base + 2500L, bucketSizeMs, numBuckets);
- long cnt3 = c.getCount(base + 2500L, timeWindowMs, bucketSizeMs,
numBuckets);
- Assertions.assertEquals(4L, cnt3);
+ @Test
+ public void testReadsExpireStatsWithoutAnotherReportAndReclaimBackend() {
+ long originalWindow =
Config.active_tablet_sliding_window_time_window_second;
+ try {
+ Config.active_tablet_sliding_window_time_window_second = 600;
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L,
Collections.singletonList(queryStat(10L, 1_000L, now - 120_000L, 60_000L)),
+ Collections.emptyList());
+ stats.updateFromReport(2L, Arrays.asList(queryStat(10L, 7L, now,
60_000L),
+ queryStat(20L, 900L, now - 120_000L, 60_000L)),
Collections.emptyList());
+ Assertions.assertEquals(1_007L,
stats.getAccessInfo(10L).accessCount);
+ Assertions.assertNotNull(stats.getAccessInfo(20L));
+
+ // Move the window past old accesses without delivering another
report or sleeping.
+ Config.active_tablet_sliding_window_time_window_second = 60;
+ Assertions.assertEquals(7L, stats.getAccessInfo(10L).accessCount);
+ Assertions.assertNull(stats.getAccessInfo(20L));
+ List<TabletSlidingWindowAccessStats.AccessStatsResult> top =
stats.getTopNActive(4);
+ Assertions.assertEquals(1, top.size());
+ Assertions.assertEquals(10L, top.get(0).id);
+ Assertions.assertEquals(7.0, top.get(0).scanRate);
+ Assertions.assertEquals(1L, stats.getActiveIdsInWindow());
+ Assertions.assertEquals(7L, stats.getRecentAccessCountInWindow());
+
Assertions.assertTrue(stats.getStatsSummary().contains("beCount=1,"));
+ } finally {
+ Config.active_tablet_sliding_window_time_window_second =
originalWindow;
+ }
}
+ // Retention ends at active_tablet_sliding_window_time_window_second.
@Test
- public void testWindowExpiresOldBuckets() throws Exception {
- int numBuckets = 5;
- long bucketSizeMs = 1000L;
- long timeWindowMs = 5000L;
- long base = 20_000L;
+ public void testEntriesAgeOutOfTheWindow() {
+ long windowMs = Config.active_tablet_sliding_window_time_window_second
* 1000L;
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L,
+ Collections.singletonList(queryStat(10L, 12L, now - windowMs -
60_000L, 60_000L)),
+ Collections.emptyList());
+ stats.updateFromReport(1L, Collections.singletonList(queryStat(20L,
7L, now, 60_000L)),
+ Collections.emptyList());
+
+ Assertions.assertNull(stats.getAccessInfo(10L));
+ Assertions.assertEquals(1L, stats.getActiveIdsInWindow());
+ }
- CounterHarness c = new CounterHarness(numBuckets);
- c.add(base, bucketSizeMs, numBuckets);
- c.add(base + 1000L, bucketSizeMs, numBuckets);
- c.add(base + 2000L, bucketSizeMs, numBuckets);
+ // Retention is bounded, so a backend with a churning hot set cannot grow
FE memory for a
+ // whole window. The coldest entries go first: tablet 1 is the oldest of
the three but by
+ // far the hottest, and everything here is inside the window already, so
age alone would
+ // evict exactly the tablet the scheduler most needs to know about.
+ @Test
+ public void testRetentionIsCappedByTopnKeepingTheHottest() {
+ int originalTopn = Config.cloud_active_partition_scheduling_topn;
+ Config.cloud_active_partition_scheduling_topn = 2;
+ try {
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L, Arrays.asList(
+ queryStat(1L, 100L, now - 3_000L, 60_000L),
+ queryStat(2L, 1L, now - 2_000L, 60_000L)),
Collections.emptyList());
+ stats.updateFromReport(1L,
+ Collections.singletonList(queryStat(3L, 1L, now,
60_000L)), Collections.emptyList());
+
+ Assertions.assertEquals(2L, stats.getActiveIdsInWindow());
+ Assertions.assertNotNull(stats.getAccessInfo(1L));
+ // Same rate as tablet 3, so the tie-break on recency drops it.
+ Assertions.assertNull(stats.getAccessInfo(2L));
+ Assertions.assertNotNull(stats.getAccessInfo(3L));
+ } finally {
+ Config.cloud_active_partition_scheduling_topn = originalTopn;
+ }
+ }
- // At base + 6000, window start is base + 1000 => count should be 2
(1000 and 2000 buckets)
- long cnt1 = c.getCount(base + 6000L, timeWindowMs, bucketSizeMs,
numBuckets);
- Assertions.assertEquals(2L, cnt1);
+ @Test
+ public void testTopNReservesCapacityForLoadTablets() {
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L, Arrays.asList(
+ queryStat(1L, 1_000L, now - 1L, 60_000L),
+ queryStat(2L, 900L, now - 2L, 60_000L),
+ queryStat(3L, 800L, now - 3L, 60_000L),
+ queryStat(4L, 700L, now - 4L, 60_000L)), Arrays.asList(
+ loadStat(101L, 10L, now - 101L, 60_000L),
+ loadStat(102L, 9L, now - 102L, 60_000L),
+ loadStat(103L, 8L, now - 103L, 60_000L),
+ loadStat(104L, 7L, now - 104L, 60_000L)));
+
+ Set<Long> ids = stats.getTopNActive(4).stream().map(r ->
r.id).collect(Collectors.toSet());
+ Assertions.assertEquals(Sets.newHashSet(1L, 2L, 101L, 102L), ids);
+ }
- // At base + 7000, window start is base + 2000 => count should be 1
(2000 bucket)
- long cnt2 = c.getCount(base + 7000L, timeWindowMs, bucketSizeMs,
numBuckets);
- Assertions.assertEquals(1L, cnt2);
+ @Test
+ public void testTopNSortsByRateInsteadOfRawDeltaAcrossBackends() {
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L, Collections.singletonList(queryStat(1L,
100L, now - 1L, 120_000L)),
+ Collections.emptyList());
+ stats.updateFromReport(2L, Collections.singletonList(queryStat(2L,
75L, now - 2L, 60_000L)),
+ Collections.emptyList());
+
+ List<TabletSlidingWindowAccessStats.AccessStatsResult> results =
stats.getTopNActive(2);
+ Assertions.assertEquals(2L, results.get(0).id);
+ Assertions.assertEquals(1L, results.get(1).id);
+ }
- // recent activity should still be true (lastAccessTime=base+2000
within 5000ms of base+7000? exactly 5000ms)
- Assertions.assertTrue(c.hasRecentActivity(base + 7000L, timeWindowMs));
+ // Two backends each holding a replica of tablet 10. A scan range runs on
exactly one
+ // replica, so the tablet's query traffic is 10 + 20; a load writes both
replicas, so its
+ // load traffic is max(2, 1) and not their sum.
+ @Test
+ public void testReplicaQueryStatsSumWhileLoadStatsTakeTheMax() {
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L,
+ Collections.singletonList(queryStat(10L, 10L, now - 100L,
60_000L)),
+ Collections.singletonList(loadStat(10L, 2L, now - 300L,
60_000L)));
+ stats.updateFromReport(2L,
+ Collections.singletonList(queryStat(10L, 20L, now - 200L,
60_000L)),
+ Collections.singletonList(loadStat(10L, 1L, now - 400L,
60_000L)));
+
+ TabletSlidingWindowAccessStats.AccessStatsResult result =
stats.getTopNActive(2).get(0);
+ Assertions.assertEquals(30L, result.scanCount);
+ Assertions.assertEquals(2L, result.loadCount);
+ Assertions.assertEquals(32L, result.accessCount);
+ Assertions.assertEquals(now - 100L, result.lastAccessTime);
+ Assertions.assertEquals(30.0, result.scanRate);
+ Assertions.assertEquals(2.0, result.loadRate);
+ Assertions.assertEquals(32L, stats.getRecentAccessCountInWindow());
+ Assertions.assertEquals(1L, stats.getActiveIdsInWindow());
+
+ // SHOW TABLET reads the same merged view, not whichever backend
reported most.
+ TabletSlidingWindowAccessStats.AccessStatsResult shown =
stats.getAccessInfo(10L);
+ Assertions.assertEquals(32L, shown.accessCount);
+ Assertions.assertEquals(now - 100L, shown.lastAccessTime);
}
+ // The retention cap must not rank the two dimensions against each other:
scans outnumber
+ // flushes by orders of magnitude, so one ordering would evict every load
tablet before the
+ // reserved load quota downstream ever saw it - with both backend lists
well under their own
+ // cap, so nothing would report the loss.
@Test
- public void testCleanupRemovesExpiredBuckets() throws Exception {
- int numBuckets = 5;
- long bucketSizeMs = 1000L;
- long timeWindowMs = 5000L;
- long base = 30_000L;
+ public void testRetentionCapKeepsBothDimensions() {
+ int originalTopn = Config.cloud_active_partition_scheduling_topn;
+ Config.cloud_active_partition_scheduling_topn = 4;
+ try {
+ long now = System.currentTimeMillis();
+ // Round 1: two hot queries and two cold-but-real loads. Nothing
is capped yet.
+ stats.updateFromReport(1L, Arrays.asList(
+ queryStat(1L, 100L, now - 1_000L, 60_000L),
+ queryStat(2L, 100L, now - 1_000L, 60_000L)), Arrays.asList(
+ loadStat(101L, 1L, now - 1_000L, 60_000L),
+ loadStat(102L, 1L, now - 1_000L, 60_000L)));
+ // Round 2: the query set churned, the same two tablets are still
loading. The
+ // query rates tie, so recency decides which pair keeps the query
half.
+ stats.updateFromReport(1L, Arrays.asList(
+ queryStat(3L, 100L, now, 60_000L),
+ queryStat(4L, 100L, now, 60_000L)), Arrays.asList(
+ loadStat(101L, 1L, now, 60_000L),
+ loadStat(102L, 1L, now, 60_000L)));
+
+ // Each dimension keeps its own budget of four, so all six survive
retention. A
+ // single max(scanRate, loadRate) ordering would have dropped both
loads.
+ Assertions.assertNotNull(stats.getAccessInfo(101L));
+ Assertions.assertNotNull(stats.getAccessInfo(102L));
+ Assertions.assertEquals(6L, stats.getActiveIdsInWindow());
+ Set<Long> top = stats.getTopNActive(4).stream().map(r ->
r.id).collect(Collectors.toSet());
+ Assertions.assertEquals(Sets.newHashSet(3L, 4L, 101L, 102L), top);
+ } finally {
+ Config.cloud_active_partition_scheduling_topn = originalTopn;
+ }
+ }
- CounterHarness c = new CounterHarness(numBuckets);
- c.add(base, bucketSizeMs, numBuckets);
- c.add(base + 1000L, bucketSizeMs, numBuckets);
- c.add(base + 2000L, bucketSizeMs, numBuckets);
+ // Retention must not spend getTopNActive()'s unique-id budget: that
allocates the final
+ // quota per backend, before cross-backend overlap is known. Here backend
1's own load
+ // runner-up is exactly what the global load share needs, because backend
2 turns the leader
+ // into a query winner whose load entry is then a duplicate.
+ @Test
+ public void testRetentionKeepsTheLoadRunnerUpNeededAfterBackendsMerge() {
+ int originalTopn = Config.cloud_active_partition_scheduling_topn;
+ Config.cloud_active_partition_scheduling_topn = 2;
+ try {
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L, Collections.singletonList(queryStat(1L,
1L, now, 60_000L)),
+ Collections.singletonList(loadStat(10L, 100L, now,
60_000L)));
+ // Still loading 10, and 11 starts loading. No queries this round.
+ stats.updateFromReport(1L, Collections.emptyList(), Arrays.asList(
+ loadStat(10L, 100L, now, 60_000L),
+ loadStat(11L, 90L, now, 60_000L)));
+ // Another compute group queries 10 hard, making it the global
query winner.
+ stats.updateFromReport(2L,
Collections.singletonList(queryStat(10L, 200L, now, 60_000L)),
+ Collections.emptyList());
+
+ // 10 takes the query slot; its own load entry is a duplicate, so
the load slot must
+ // go to 11. Dropping 11 during backend 1's retention would
backfill with query 1.
+ Set<Long> top = stats.getTopNActive(2).stream().map(r ->
r.id).collect(Collectors.toSet());
+ Assertions.assertEquals(Sets.newHashSet(10L, 11L), top);
+ } finally {
+ Config.cloud_active_partition_scheduling_topn = originalTopn;
+ }
+ }
- // cleanup at base + 7000: windowStart=base+2000, so base/base+1000
buckets should be cleared
- c.cleanup(base + 7000L, timeWindowMs);
- long cnt = c.getCount(base + 7000L, timeWindowMs, bucketSizeMs,
numBuckets);
- Assertions.assertEquals(1L, cnt);
+ // Cloud routes a load to one backend per compute group rather than
replicating it, so two
+ // backends reporting the same tablet did independent work and their
flushes must be summed.
+ @Test
+ public void testCloudModeSumsLoadAcrossBackends() {
+ String originalUniqueId = Config.cloud_unique_id;
+ Config.cloud_unique_id = "test_cloud_unique_id";
+ try {
+ Assertions.assertTrue(Config.isCloudMode());
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L, Collections.emptyList(),
+ Collections.singletonList(loadStat(10L, 20L, now - 100L,
60_000L)));
+ stats.updateFromReport(2L, Collections.emptyList(),
+ Collections.singletonList(loadStat(10L, 20L, now - 200L,
60_000L)));
+
+ TabletSlidingWindowAccessStats.AccessStatsResult result =
stats.getTopNActive(2).get(0);
+ Assertions.assertEquals(40L, result.loadCount);
+ Assertions.assertEquals(40.0, result.loadRate);
+ } finally {
+ Config.cloud_unique_id = originalUniqueId;
+ }
}
@Test
- public void testBucketWrapAroundResetsExpiredBucket() throws Exception {
- int numBuckets = 5;
- long bucketSizeMs = 1000L;
- long timeWindowMs = 5000L;
- long base = 40_000L;
+ public void testRemoveBackendRemovesItsSnapshot() {
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L, Collections.singletonList(queryStat(10L,
1L, now - 100L, 60_000L)),
+ Collections.emptyList());
- CounterHarness c = new CounterHarness(numBuckets);
- // bucket index wraps every numBuckets * bucketSizeMs (5s here)
- c.add(base, bucketSizeMs, numBuckets); // idx = 0
- c.add(base + 5000L, bucketSizeMs, numBuckets); // idx = 0 again,
should reset old bucket
+ stats.removeBackend(1L);
- // With current implementation, the old "base" bucket is overwritten,
so only 1 remains.
- long cnt = c.getCount(base + 5000L, timeWindowMs, bucketSizeMs,
numBuckets);
- Assertions.assertEquals(1L, cnt);
+ Assertions.assertNull(stats.getAccessInfo(10L));
+ Assertions.assertTrue(stats.getTopNActive(10).isEmpty());
}
@Test
- public void testTimeWindowLessThanBucketSizeStillUsesValidBucketCount()
throws Exception {
- long originWindowSecond =
Config.active_tablet_sliding_window_time_window_second;
- boolean originEnabled =
Config.enable_active_tablet_sliding_window_access_stats;
- try {
- Config.active_tablet_sliding_window_time_window_second = 59L;
- Config.enable_active_tablet_sliding_window_access_stats = false;
-
- TabletSlidingWindowAccessStats stats = new
TabletSlidingWindowAccessStats();
- Field numBucketsField =
TabletSlidingWindowAccessStats.class.getDeclaredField("numBuckets");
- numBucketsField.setAccessible(true);
- int numBuckets = (int) numBucketsField.get(stats);
- Assertions.assertEquals(1, numBuckets);
+ public void testDisabledStatsReturnEmptyValues() {
+ long now = System.currentTimeMillis();
+ stats.updateFromReport(1L, Collections.singletonList(queryStat(10L,
1L, now - 100L, 60_000L)),
+ Collections.emptyList());
+ Config.enable_active_tablet_sliding_window_access_stats = false;
+
+ Assertions.assertNull(stats.getAccessInfo(10L));
+ Assertions.assertTrue(stats.getTopNActive(10).isEmpty());
+ Assertions.assertEquals(0L, stats.getRecentAccessCountInWindow());
+ Assertions.assertEquals(0L, stats.getActiveIdsInWindow());
+ Assertions.assertEquals(0L, stats.getTotalAccessCount());
+ Assertions.assertEquals("Active tablet sliding window access stats is
disabled", stats.getStatsSummary());
+ }
- // No ArithmeticException should be thrown by modulo numBuckets
path.
- stats.recordAccess(123L);
- } finally {
- Config.active_tablet_sliding_window_time_window_second =
originWindowSecond;
- Config.enable_active_tablet_sliding_window_access_stats =
originEnabled;
+ private static List<TActiveTabletStat> queryStats(int count, long
lastAccessTime) {
+ List<TActiveTabletStat> stats = new ArrayList<>(count);
+ for (int i = 0; i < count; i++) {
+ stats.add(queryStat(1_000L + i, 10L + i, lastAccessTime, 60_000L));
}
+ return stats;
}
- @Test
- public void testTopNComparatorOrdersByAccessCountThenLastAccessTimeDesc()
throws Exception {
- Field f =
TabletSlidingWindowAccessStats.class.getDeclaredField("TOPN_ACTIVE_COMPARATOR");
- f.setAccessible(true);
- @SuppressWarnings("unchecked")
- Comparator<TabletSlidingWindowAccessStats.AccessStatsResult> cmp =
- (Comparator<TabletSlidingWindowAccessStats.AccessStatsResult>)
f.get(null);
-
- List<TabletSlidingWindowAccessStats.AccessStatsResult> results = new
ArrayList<>();
- results.add(new TabletSlidingWindowAccessStats.AccessStatsResult(1L,
1L, 200L));
- results.add(new TabletSlidingWindowAccessStats.AccessStatsResult(2L,
10L, 100L));
- results.add(new TabletSlidingWindowAccessStats.AccessStatsResult(3L,
10L, 300L));
- results.add(new TabletSlidingWindowAccessStats.AccessStatsResult(4L,
2L, 400L));
+ private static List<TActiveTabletStat> loadStats(int count, long
lastAccessTime) {
+ List<TActiveTabletStat> stats = new ArrayList<>(count);
+ for (int i = 0; i < count; i++) {
+ stats.add(loadStat(2_000L + i, 10L + i, lastAccessTime, 60_000L));
+ }
+ return stats;
+ }
- results.sort(cmp);
+ private static TActiveTabletStat queryStat(long tabletId, long delta, long
lastAccessTime, long windowMs) {
+ TActiveTabletStat stat = new TActiveTabletStat();
+ stat.setTabletId(tabletId);
+ stat.setScanCountDelta(delta);
+ stat.setLastQueryTimeMs(lastAccessTime);
+ stat.setDeltaWindowMs(windowMs);
+ return stat;
+ }
- // accessCount desc => 10 first, then 2, then 1
- // tie-breaker lastAccessTime desc => id=3 (300) before id=2 (100)
- Assertions.assertEquals(3L, results.get(0).id);
- Assertions.assertEquals(2L, results.get(1).id);
- Assertions.assertEquals(4L, results.get(2).id);
- Assertions.assertEquals(1L, results.get(3).id);
+ private static TActiveTabletStat loadStat(long tabletId, long delta, long
lastAccessTime, long windowMs) {
+ TActiveTabletStat stat = new TActiveTabletStat();
+ stat.setTabletId(tabletId);
+ stat.setLoadCountDelta(delta);
+ stat.setLastLoadTimeMs(lastAccessTime);
+ stat.setDeltaWindowMs(windowMs);
+ return stat;
}
}
diff --git a/gensrc/thrift/MasterService.thrift
b/gensrc/thrift/MasterService.thrift
index 2caf3c76965..ac1a6a23456 100644
--- a/gensrc/thrift/MasterService.thrift
+++ b/gensrc/thrift/MasterService.thrift
@@ -56,6 +56,23 @@ struct TTabletInfo {
1000: optional bool is_persistent
}
+struct TActiveTabletStat {
+ 1: required Types.TTabletId tablet_id
+ // Query scan count since the previous SUCCESSFULLY REPORTED baseline
(delta, not cumulative)
+ 2: optional i64 scan_count_delta
+ // Memtable flush count since that baseline (delta) -- the load-side signal
+ 3: optional i64 load_count_delta
+ // Last query scan time, ms since epoch
+ 4: optional i64 last_query_time_ms
+ // Last load (memtable write) time, ms since epoch
+ 5: optional i64 last_load_time_ms
+ // Wall-clock span this delta covers, in ms. NOT always the report
interval: a dropped
+ // report round commits no baseline, so the next delta spans several
intervals, and
+ // backends report on independent phases.
+ // Consumers MUST rank by delta/delta_window_ms, never by raw delta.
+ 6: optional i64 delta_window_ms
+}
+
struct TFinishTaskRequest {
1: required Types.TBackend backend
2: required Types.TTaskType task_type
@@ -123,6 +140,12 @@ struct TReportRequest {
15: optional list<AgentService.TIndexPolicy> index_policy
// Running query/loading tasks
16: optional i64 running_tasks
+ // Top-N tablets by query scan count on this BE since the previous report.
+ // Entries set only tablet_id / scan_count_delta / last_query_time_ms /
delta_window_ms.
+ 18: optional list<TActiveTabletStat> top_query_tablets
+ // Top-N tablets by load (memtable flush) count since the previous report.
+ // Entries set only tablet_id / load_count_delta / last_load_time_ms /
delta_window_ms.
+ 19: optional list<TActiveTabletStat> top_load_tablets
}
struct TMasterResult {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
