You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/04/01 05:33:41 UTC
(doris) branch branch-4.1 updated: branch-4.1: [pick] Cherry-pick #60520 #60239 #60310 (#61921)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new b387fb96de7 branch-4.1: [pick] Cherry-pick #60520 #60239 #60310 (#61921)
b387fb96de7 is described below
commit b387fb96de7e90f565efa6d3759170f636d3d14a
Author: Jimmy <li...@selectdb.com>
AuthorDate: Wed Apr 1 13:33:34 2026 +0800
branch-4.1: [pick] Cherry-pick #60520 #60239 #60310 (#61921)
## Proposed changes
Cherry-pick the following PRs to branch-4.1:
- [Enhancement](Recycler) Add bvar metrics for operation logs #60520
- [Feat](Snapshot) require root privilege for cluster snapshot commands
#60239
- [feature](cloud) Support compaction read-write separation for cloud
mode #60310
---
be/src/cloud/cloud_cluster_info.cpp | 201 +++++++++
be/src/cloud/cloud_cluster_info.h | 78 +++-
be/src/cloud/cloud_meta_mgr.cpp | 39 +-
be/src/cloud/cloud_meta_mgr.h | 6 +
be/src/cloud/cloud_tablet.h | 20 +
be/src/cloud/cloud_tablet_mgr.cpp | 9 +-
be/src/cloud/config.cpp | 11 +
be/src/cloud/config.h | 11 +
be/src/runtime/exec_env_init.cpp | 8 +
be/test/cloud/cloud_cluster_info_test.cpp | 280 ++++++++++++
cloud/src/common/bvars.cpp | 40 ++
cloud/src/common/bvars.h | 24 +-
cloud/src/meta-service/meta_service_resource.cpp | 13 +
cloud/src/meta-service/meta_service_txn.cpp | 70 +++
cloud/src/recycler/recycler.cpp | 40 ++
cloud/src/recycler/recycler.h | 25 +-
cloud/src/recycler/recycler_operation_log.cpp | 90 +++-
cloud/test/CMakeLists.txt | 1 +
cloud/test/compaction_rw_separation_test.cpp | 181 ++++++++
cloud/test/recycler_operation_log_test.cpp | 478 +++++++++++++++++++++
.../main/java/org/apache/doris/common/Config.java | 8 +
.../transaction/CloudGlobalTransactionMgr.java | 16 +
.../nereids/rules/analysis/UserAuthentication.java | 30 +-
.../AdminCreateClusterSnapshotCommand.java | 18 +-
.../commands/AdminDropClusterSnapshotCommand.java | 18 +-
.../AdminSetAutoClusterSnapshotCommand.java | 18 +-
...dminSetClusterSnapshotFeatureSwitchCommand.java | 18 +-
.../rules/analysis/UserAuthenticationTest.java | 396 +++++++++++++++++
.../AdminCreateClusterSnapshotCommandTest.java | 102 ++++-
.../AdminDropClusterSnapshotCommandTest.java | 99 ++++-
.../AdminSetAutoClusterSnapshotCommandTest.java | 103 ++++-
...SetClusterSnapshotFeatureSwitchCommandTest.java | 99 ++++-
gensrc/proto/cloud.proto | 18 +
.../test_compaction_cluster_takeover.groovy | 264 ++++++++++++
.../test_compaction_rw_separation.groovy | 158 +++++++
.../test_compaction_rw_separation_disabled.groovy | 151 +++++++
.../test_compaction_rw_separation_force.groovy | 161 +++++++
37 files changed, 3243 insertions(+), 59 deletions(-)
diff --git a/be/src/cloud/cloud_cluster_info.cpp b/be/src/cloud/cloud_cluster_info.cpp
new file mode 100644
index 00000000000..cef8040ca3f
--- /dev/null
+++ b/be/src/cloud/cloud_cluster_info.cpp
@@ -0,0 +1,201 @@
+// 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 "cloud/cloud_cluster_info.h"
+
+#include <glog/logging.h>
+
+#include "cloud/cloud_meta_mgr.h"
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablet.h"
+#include "cloud/config.h"
+#include "gen_cpp/cloud.pb.h"
+#include "runtime/exec_env.h"
+#include "util/time.h"
+
+namespace doris {
+
+CloudClusterInfo::~CloudClusterInfo() {
+ stop_bg_worker();
+}
+
+void CloudClusterInfo::start_bg_worker() {
+ bool expected = true;
+ if (!_bg_worker_stopped.compare_exchange_strong(expected, false)) {
+ // Already running
+ return;
+ }
+
+ _bg_worker = std::thread(&CloudClusterInfo::_bg_worker_func, this);
+ LOG(INFO) << "CloudClusterInfo background worker started, "
+ << "refresh_interval=" << config::cluster_status_cache_refresh_interval_sec << "s";
+}
+
+void CloudClusterInfo::stop_bg_worker() {
+ bool expected = false;
+ if (!_bg_worker_stopped.compare_exchange_strong(expected, true)) {
+ // Already stopped
+ return;
+ }
+
+ {
+ std::lock_guard lock(_bg_worker_mutex);
+ _bg_worker_cv.notify_all();
+ }
+
+ if (_bg_worker.joinable()) {
+ _bg_worker.join();
+ }
+
+ LOG(INFO) << "CloudClusterInfo background worker stopped";
+}
+
+void CloudClusterInfo::_bg_worker_func() {
+ LOG(INFO) << "CloudClusterInfo background worker thread running";
+
+ while (!_bg_worker_stopped.load()) {
+ _refresh_cluster_status();
+
+ std::unique_lock lock(_bg_worker_mutex);
+ _bg_worker_cv.wait_for(
+ lock, std::chrono::seconds(config::cluster_status_cache_refresh_interval_sec),
+ [this] { return _bg_worker_stopped.load(); });
+ }
+}
+
+void CloudClusterInfo::_refresh_cluster_status() {
+ if (!config::is_cloud_mode()) {
+ return;
+ }
+ auto* cloud_engine =
+ dynamic_cast<CloudStorageEngine*>(&ExecEnv::GetInstance()->storage_engine());
+ if (!cloud_engine) {
+ return;
+ }
+
+ std::unordered_map<std::string, std::pair<int32_t, int64_t>> cluster_status;
+ std::string resolved_cluster_id;
+ Status st = cloud_engine->meta_mgr().get_cluster_status(
+ &cluster_status, my_cluster_id().empty() ? &resolved_cluster_id : nullptr);
+ if (!st.ok()) {
+ LOG(WARNING) << "Failed to refresh cluster status: " << st;
+ return;
+ }
+
+ // Update cache
+ {
+ std::unique_lock lock(_mutex);
+ _cluster_status_cache.clear();
+ for (const auto& [cluster_id, status_pair] : cluster_status) {
+ _cluster_status_cache[cluster_id] = {status_pair.first, status_pair.second};
+ }
+ }
+
+ VLOG_DEBUG << "Refreshed cluster status cache, " << cluster_status.size() << " clusters";
+
+ // Set our own cluster_id if resolved from the response
+ if (my_cluster_id().empty() && !resolved_cluster_id.empty()) {
+ set_my_cluster_id(resolved_cluster_id);
+ LOG(INFO) << "Resolved my cluster_id: " << resolved_cluster_id;
+ }
+}
+
+bool CloudClusterInfo::should_skip_compaction(CloudTablet* tablet) const {
+ if (!config::enable_compaction_rw_separation) {
+ return false;
+ }
+
+ std::string last_active_cluster = tablet->last_active_cluster_id();
+ std::string my_cluster = my_cluster_id();
+ int64_t tablet_id = tablet->tablet_id();
+
+ // Case 1: No active cluster record, any cluster can compact
+ if (last_active_cluster.empty()) {
+ VLOG_DEBUG << "tablet " << tablet_id << " has no last_active_cluster record, "
+ << "my_cluster=" << my_cluster << ", allow compaction";
+ return false;
+ }
+
+ // Case 2: This is the active cluster, allow compaction
+ if (last_active_cluster == my_cluster) {
+ VLOG_DEBUG << "tablet " << tablet_id << " last_active_cluster=" << last_active_cluster
+ << " equals my_cluster=" << my_cluster << ", allow compaction";
+ return false;
+ }
+
+ // Case 3: Check if the last active cluster is available
+ ClusterStatusCache cache;
+ if (!get_cluster_status(last_active_cluster, &cache)) {
+ // Cluster not found in cache, might be deleted, allow takeover
+ LOG(INFO) << "compaction_rw_separation: tablet " << tablet_id
+ << " last_active_cluster=" << last_active_cluster
+ << " not found in cache (maybe deleted), my_cluster=" << my_cluster
+ << ", allow takeover";
+ return false;
+ }
+
+ // Force compaction if tablet has too many rowsets (>80% of max_tablet_version_num),
+ // even on read clusters, to prevent version count from growing unbounded
+ // when the write cluster can't keep up or has compaction disabled.
+ int64_t num_rowsets = tablet->fetch_add_approximate_num_rowsets(0);
+ auto threshold = static_cast<int64_t>(tablet->max_version_config() *
+ config::compaction_rw_separation_version_threshold_ratio);
+ if (num_rowsets > threshold) {
+ LOG(INFO) << "compaction_rw_separation: force compaction on tablet " << tablet_id
+ << ", num_rowsets=" << num_rowsets << " > threshold=" << threshold << " (80% of "
+ << tablet->max_version_config() << ")"
+ << ", my_cluster=" << my_cluster;
+ return false;
+ }
+
+ auto status = static_cast<cloud::ClusterStatus>(cache.status);
+ int64_t status_mtime = cache.mtime_ms;
+ int64_t now = UnixMillis();
+ int64_t elapsed = now - status_mtime;
+ int64_t timeout = config::compaction_cluster_takeover_timeout_ms;
+
+ // Case 4: Original cluster is NORMAL (still active), cannot takeover
+ if (status == cloud::ClusterStatus::NORMAL) {
+ LOG_EVERY_N(INFO, 100) << "compaction_rw_separation: skip tablet " << tablet_id
+ << ", last_active_cluster=" << last_active_cluster
+ << " is NORMAL (active), my_cluster=" << my_cluster;
+ return true;
+ }
+
+ // Case 5: Original cluster is unavailable (SUSPENDED/MANUAL_SHUTDOWN/deleted)
+ if (elapsed > timeout) {
+ // Takeover successful
+ LOG(INFO) << "compaction_rw_separation: takeover tablet " << tablet_id
+ << ", last_active_cluster=" << last_active_cluster
+ << " status=" << cloud::ClusterStatus_Name(status)
+ << " status_mtime=" << status_mtime << " elapsed=" << elapsed
+ << "ms > timeout=" << timeout << "ms"
+ << ", my_cluster=" << my_cluster;
+ return false;
+ } else {
+ // Timeout not reached yet, waiting
+ LOG_EVERY_N(INFO, 100) << "compaction_rw_separation: skip tablet " << tablet_id
+ << ", last_active_cluster=" << last_active_cluster
+ << " status=" << cloud::ClusterStatus_Name(status)
+ << " status_mtime=" << status_mtime << " elapsed=" << elapsed
+ << "ms <= timeout=" << timeout << "ms"
+ << ", my_cluster=" << my_cluster << ", waiting for takeover";
+ return true;
+ }
+}
+
+} // namespace doris
diff --git a/be/src/cloud/cloud_cluster_info.h b/be/src/cloud/cloud_cluster_info.h
index e62db51934a..a008ab22ea9 100644
--- a/be/src/cloud/cloud_cluster_info.h
+++ b/be/src/cloud/cloud_cluster_info.h
@@ -15,18 +15,92 @@
// specific language governing permissions and limitations
// under the License.
+#pragma once
+
+#include <atomic>
+#include <condition_variable>
+#include <mutex>
+#include <shared_mutex>
+#include <string>
+#include <thread>
+#include <unordered_map>
+
#include "runtime/cluster_info.h"
namespace doris {
+class CloudTablet;
+
+// Cached cluster status information
+struct ClusterStatusCache {
+ int32_t status {0}; // ClusterStatus enum value
+ int64_t mtime_ms {0}; // Timestamp when status was last changed
+};
+
class CloudClusterInfo : public ClusterInfo {
public:
- bool is_in_standby() const { return _is_in_standby; }
+ ~CloudClusterInfo();
+ bool is_in_standby() const { return _is_in_standby; }
void set_is_in_standby(bool flag) { _is_in_standby = flag; }
+ // Get this BE's cluster ID
+ std::string my_cluster_id() const {
+ std::shared_lock lock(_mutex);
+ return _my_cluster_id;
+ }
+ void set_my_cluster_id(const std::string& id) {
+ std::unique_lock lock(_mutex);
+ _my_cluster_id = id;
+ }
+
+ // Get cached cluster status, returns false if not found
+ bool get_cluster_status(const std::string& id, ClusterStatusCache* cache) const {
+ std::shared_lock lock(_mutex);
+ auto it = _cluster_status_cache.find(id);
+ if (it != _cluster_status_cache.end()) {
+ *cache = it->second;
+ return true;
+ }
+ return false;
+ }
+
+ // Update cluster status cache
+ void set_cluster_status(const std::string& id, int32_t status, int64_t mtime_ms) {
+ std::unique_lock lock(_mutex);
+ _cluster_status_cache[id] = {status, mtime_ms};
+ }
+
+ // Clear all cached cluster status
+ void clear_cluster_status_cache() {
+ std::unique_lock lock(_mutex);
+ _cluster_status_cache.clear();
+ }
+
+ // Start background refresh thread
+ void start_bg_worker();
+ // Stop background refresh thread
+ void stop_bg_worker();
+
+ // Check if this cluster should skip compaction for the given tablet
+ // Returns true if should skip (i.e., another cluster should do the compaction)
+ bool should_skip_compaction(CloudTablet* tablet) const;
+
private:
+ void _bg_worker_func();
+ void _refresh_cluster_status();
+
bool _is_in_standby = false;
+
+ mutable std::shared_mutex _mutex;
+ std::string _my_cluster_id;
+ std::unordered_map<std::string, ClusterStatusCache> _cluster_status_cache;
+
+ // Background worker
+ std::thread _bg_worker;
+ std::atomic<bool> _bg_worker_stopped {true};
+ std::mutex _bg_worker_mutex;
+ std::condition_variable _bg_worker_cv;
};
-} // namespace doris
\ No newline at end of file
+} // namespace doris
diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp
index ca6c6ff3e79..5a73f095e8f 100644
--- a/be/src/cloud/cloud_meta_mgr.cpp
+++ b/be/src/cloud/cloud_meta_mgr.cpp
@@ -352,7 +352,7 @@ static std::string debug_info(const Request& req) {
} else if constexpr (is_any_v<Request, GetTabletRequest>) {
return fmt::format(" tablet_id={}", req.tablet_id());
} else if constexpr (is_any_v<Request, GetObjStoreInfoRequest, ListSnapshotRequest,
- GetInstanceRequest>) {
+ GetInstanceRequest, GetClusterStatusRequest>) {
return "";
} else if constexpr (is_any_v<Request, CreateRowsetRequest>) {
return fmt::format(" tablet_id={}", req.rowset_meta().tablet_id());
@@ -789,6 +789,12 @@ Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet,
tablet->set_cumulative_layer_point(stats.cumulative_point());
tablet->reset_approximate_stats(stats.num_rowsets(), stats.num_segments(),
stats.num_rows(), stats.data_size());
+
+ // Sync last active cluster info for compaction read-write separation
+ if (config::enable_compaction_rw_separation && stats.has_last_active_cluster_id()) {
+ tablet->set_last_active_cluster_info(stats.last_active_cluster_id(),
+ stats.last_active_time_ms());
+ }
}
return Status::OK();
}
@@ -2304,5 +2310,36 @@ Status CloudMetaMgr::update_packed_file_info(const std::string& packed_file_path
&cloud::MetaService_Stub::update_packed_file_info);
}
+Status CloudMetaMgr::get_cluster_status(
+ std::unordered_map<std::string, std::pair<int32_t, int64_t>>* result,
+ std::string* my_cluster_id) {
+ GetClusterStatusRequest req;
+ GetClusterStatusResponse resp;
+ req.add_cloud_unique_ids(config::cloud_unique_id);
+
+ Status s = retry_rpc("get cluster status", req, &resp, &MetaService_Stub::get_cluster_status);
+ if (!s.ok()) {
+ return s;
+ }
+
+ result->clear();
+ for (const auto& detail : resp.details()) {
+ for (const auto& cluster : detail.clusters()) {
+ // Store cluster status and mtime (mtime is in seconds from MS, convert to ms).
+ // If mtime is not set, use current time as a conservative default
+ // to avoid immediate takeover due to elapsed being huge.
+ int64_t mtime_ms = cluster.has_mtime() ? cluster.mtime() * 1000 : UnixMillis();
+ (*result)[cluster.cluster_id()] = {static_cast<int32_t>(cluster.cluster_status()),
+ mtime_ms};
+ }
+ }
+
+ if (my_cluster_id && resp.has_requester_cluster_id()) {
+ *my_cluster_id = resp.requester_cluster_id();
+ }
+
+ return Status::OK();
+}
+
#include "common/compile_check_end.h"
} // namespace doris::cloud
diff --git a/be/src/cloud/cloud_meta_mgr.h b/be/src/cloud/cloud_meta_mgr.h
index e469a90592e..2713ce12921 100644
--- a/be/src/cloud/cloud_meta_mgr.h
+++ b/be/src/cloud/cloud_meta_mgr.h
@@ -169,6 +169,12 @@ public:
int64_t& max_reserved_snapshots,
int64_t& snapshot_interval_seconds);
+ // Get all cluster status for the instance
+ // Returns cluster_id -> (status, mtime_ms)
+ // If my_cluster_id is not null, also returns the requesting node's cluster_id
+ Status get_cluster_status(std::unordered_map<std::string, std::pair<int32_t, int64_t>>* result,
+ std::string* my_cluster_id = nullptr);
+
private:
bool sync_tablet_delete_bitmap_by_cache(CloudTablet* tablet, int64_t old_max_version,
std::ranges::range auto&& rs_metas,
diff --git a/be/src/cloud/cloud_tablet.h b/be/src/cloud/cloud_tablet.h
index 984df0427a4..c872a54c0e5 100644
--- a/be/src/cloud/cloud_tablet.h
+++ b/be/src/cloud/cloud_tablet.h
@@ -251,6 +251,21 @@ public:
int64_t alter_version() const { return _alter_version; }
void set_alter_version(int64_t alter_version) { _alter_version = alter_version; }
+ // Last active cluster info for compaction read-write separation
+ std::string last_active_cluster_id() const {
+ std::shared_lock lock(_cluster_info_mutex);
+ return _last_active_cluster_id;
+ }
+ int64_t last_active_time_ms() const {
+ std::shared_lock lock(_cluster_info_mutex);
+ return _last_active_time_ms;
+ }
+ void set_last_active_cluster_info(const std::string& cluster_id, int64_t time_ms) {
+ std::unique_lock lock(_cluster_info_mutex);
+ _last_active_cluster_id = cluster_id;
+ _last_active_time_ms = time_ms;
+ }
+
std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_base_compaction();
inline Version max_version() const {
@@ -487,6 +502,11 @@ private:
mutable std::shared_mutex _warmed_up_rowsets_mutex;
std::unordered_set<RowsetId> _warmed_up_rowsets;
+
+ // Cluster info for compaction read-write separation
+ mutable std::shared_mutex _cluster_info_mutex;
+ std::string _last_active_cluster_id;
+ int64_t _last_active_time_ms {0};
};
using CloudTabletSPtr = std::shared_ptr<CloudTablet>;
diff --git a/be/src/cloud/cloud_tablet_mgr.cpp b/be/src/cloud/cloud_tablet_mgr.cpp
index dc09560b870..6161ef484ea 100644
--- a/be/src/cloud/cloud_tablet_mgr.cpp
+++ b/be/src/cloud/cloud_tablet_mgr.cpp
@@ -435,12 +435,19 @@ Status CloudTabletMgr::get_topn_tablets_to_compact(
auto now = duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();
auto skip = [now, compaction_type](CloudTablet* t) {
auto* cloud_cluster_info = static_cast<CloudClusterInfo*>(ExecEnv::GetInstance()->cluster_info());
+
if (config::enable_standby_passive_compaction && cloud_cluster_info->is_in_standby()) {
if (t->fetch_add_approximate_num_rowsets(0) < config::max_tablet_version_num * config::standby_compaction_version_ratio) {
return true;
}
}
+ // Compaction read-write separation: skip tablets that should be compacted by other clusters.
+ // Placed after standby check so standby invariants (version count threshold) are preserved.
+ if (cloud_cluster_info->should_skip_compaction(t)) {
+ return true;
+ }
+
int32_t max_version_config = t->max_version_config();
if (compaction_type == CompactionType::BASE_COMPACTION) {
bool is_recent_failure = now - t->last_base_compaction_failure_time() < config::min_compaction_failure_interval_ms;
@@ -450,7 +457,7 @@ Status CloudTabletMgr::get_topn_tablets_to_compact(
g_base_compaction_not_frozen_tablet_num << !is_frozen;
return is_recent_failure || is_frozen;
}
-
+
// If tablet has too many rowsets but not be compacted for a long time, compaction should be performed
// regardless of whether there is a load job recently.
bool is_recent_failure = now - t->last_cumu_compaction_failure_time() < config::min_compaction_failure_interval_ms;
diff --git a/be/src/cloud/config.cpp b/be/src/cloud/config.cpp
index 57d817b9dce..f13af24224a 100644
--- a/be/src/cloud/config.cpp
+++ b/be/src/cloud/config.cpp
@@ -148,6 +148,17 @@ DEFINE_mBool(enable_standby_passive_compaction, "true");
DEFINE_mDouble(standby_compaction_version_ratio, "0.8");
+// Compaction read-write separation: only the "last active" cluster (the one that most recently
+// performed load) is allowed to compact a tablet
+DEFINE_mBool(enable_compaction_rw_separation, "false");
+// Timeout in ms for takeover when last active cluster becomes unavailable (default 5 min)
+DEFINE_mInt64(compaction_cluster_takeover_timeout_ms, "300000");
+// Interval in seconds to refresh cluster status cache (default 1 min)
+DEFINE_mInt64(cluster_status_cache_refresh_interval_sec, "60");
+// When version count exceeds this ratio of max_tablet_version_num, force compaction
+// even on read-only clusters (safety valve)
+DEFINE_mDouble(compaction_rw_separation_version_threshold_ratio, "0.8");
+
DEFINE_mBool(enable_cache_read_from_peer, "true");
// Rate limit for warmup download in bytes per second, default 100MB/s
diff --git a/be/src/cloud/config.h b/be/src/cloud/config.h
index 32d82ea6090..bb7d0a7f73e 100644
--- a/be/src/cloud/config.h
+++ b/be/src/cloud/config.h
@@ -191,6 +191,17 @@ DECLARE_mBool(enable_standby_passive_compaction);
DECLARE_mDouble(standby_compaction_version_ratio);
+// Compaction read-write separation: only the "last active" cluster (the one that most recently
+// performed load) is allowed to compact a tablet
+DECLARE_mBool(enable_compaction_rw_separation);
+// Timeout in ms for takeover when last active cluster becomes unavailable (default 5 min)
+DECLARE_mInt64(compaction_cluster_takeover_timeout_ms);
+// Interval in seconds to refresh cluster status cache for compaction read-write separation
+DECLARE_mInt64(cluster_status_cache_refresh_interval_sec);
+// When version count exceeds this ratio of max_tablet_version_num, force compaction
+// even on read-only clusters (safety valve to prevent unbounded version growth)
+DECLARE_mDouble(compaction_rw_separation_version_threshold_ratio);
+
DECLARE_mBool(enable_cache_read_from_peer);
// Rate limit for warmup download in bytes per second, default 100MB/s
diff --git a/be/src/runtime/exec_env_init.cpp b/be/src/runtime/exec_env_init.cpp
index 0c081a9d9d8..7857762af37 100644
--- a/be/src/runtime/exec_env_init.cpp
+++ b/be/src/runtime/exec_env_init.cpp
@@ -429,6 +429,9 @@ Status ExecEnv::_init(const std::vector<StorePath>& store_paths,
if (config::is_cloud_mode()) {
RETURN_IF_ERROR(_packed_file_manager->init());
_packed_file_manager->start_background_manager();
+
+ // Start cluster info background worker for compaction read-write separation
+ static_cast<CloudClusterInfo*>(_cluster_info)->start_bg_worker();
}
_index_policy_mgr = new IndexPolicyMgr();
@@ -837,6 +840,11 @@ void ExecEnv::destroy() {
// _id_manager must be destoried before tablet schema cache
SAFE_DELETE(_id_manager);
+ // Stop cluster info background worker before storage engine is destroyed
+ if (config::is_cloud_mode() && _cluster_info) {
+ static_cast<CloudClusterInfo*>(_cluster_info)->stop_bg_worker();
+ }
+
// StorageEngine must be destoried before _cache_manager destory
SAFE_STOP(_storage_engine);
_storage_engine.reset();
diff --git a/be/test/cloud/cloud_cluster_info_test.cpp b/be/test/cloud/cloud_cluster_info_test.cpp
new file mode 100644
index 00000000000..36e0334a150
--- /dev/null
+++ b/be/test/cloud/cloud_cluster_info_test.cpp
@@ -0,0 +1,280 @@
+// 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 "cloud/cloud_cluster_info.h"
+
+#include <gtest/gtest.h>
+
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablet.h"
+#include "cloud/config.h"
+#include "gen_cpp/cloud.pb.h"
+#include "storage/tablet/tablet_meta.h"
+#include "util/time.h"
+#include "util/uid_util.h"
+
+namespace doris {
+
+class CloudClusterInfoTest : public testing::Test {
+public:
+ CloudClusterInfoTest() : _engine(CloudStorageEngine(EngineOptions {})) {}
+
+ void SetUp() override {
+ _cluster_info = std::make_unique<CloudClusterInfo>();
+
+ _tablet_meta.reset(new TabletMeta(1, 2, 15673, 15674, 4, 5, TTabletSchema(), 6, {{7, 8}},
+ UniqueId(9, 10), TTabletType::TABLET_TYPE_DISK,
+ TCompressionType::LZ4F));
+ _tablet =
+ std::make_shared<CloudTablet>(_engine, std::make_shared<TabletMeta>(*_tablet_meta));
+ }
+
+ void TearDown() override {
+ _cluster_info.reset();
+ _tablet.reset();
+ }
+
+protected:
+ CloudStorageEngine _engine;
+ std::unique_ptr<CloudClusterInfo> _cluster_info;
+ std::shared_ptr<TabletMeta> _tablet_meta;
+ std::shared_ptr<CloudTablet> _tablet;
+};
+
+// Test my_cluster_id get/set
+TEST_F(CloudClusterInfoTest, MyClusterId) {
+ EXPECT_EQ(_cluster_info->my_cluster_id(), "");
+
+ _cluster_info->set_my_cluster_id("cluster_a");
+ EXPECT_EQ(_cluster_info->my_cluster_id(), "cluster_a");
+
+ _cluster_info->set_my_cluster_id("cluster_b");
+ EXPECT_EQ(_cluster_info->my_cluster_id(), "cluster_b");
+}
+
+// Test cluster status cache get/set/clear
+TEST_F(CloudClusterInfoTest, ClusterStatusCache) {
+ ClusterStatusCache cache;
+
+ // Not found initially
+ EXPECT_FALSE(_cluster_info->get_cluster_status("cluster_a", &cache));
+
+ // Set and get
+ _cluster_info->set_cluster_status("cluster_a", cloud::ClusterStatus::NORMAL, 1000);
+ EXPECT_TRUE(_cluster_info->get_cluster_status("cluster_a", &cache));
+ EXPECT_EQ(cache.status, cloud::ClusterStatus::NORMAL);
+ EXPECT_EQ(cache.mtime_ms, 1000);
+
+ // Set another cluster
+ _cluster_info->set_cluster_status("cluster_b", cloud::ClusterStatus::SUSPENDED, 2000);
+ EXPECT_TRUE(_cluster_info->get_cluster_status("cluster_b", &cache));
+ EXPECT_EQ(cache.status, cloud::ClusterStatus::SUSPENDED);
+ EXPECT_EQ(cache.mtime_ms, 2000);
+
+ // Clear
+ _cluster_info->clear_cluster_status_cache();
+ EXPECT_FALSE(_cluster_info->get_cluster_status("cluster_a", &cache));
+ EXPECT_FALSE(_cluster_info->get_cluster_status("cluster_b", &cache));
+}
+
+// Test is_in_standby
+TEST_F(CloudClusterInfoTest, IsInStandby) {
+ EXPECT_FALSE(_cluster_info->is_in_standby());
+ _cluster_info->set_is_in_standby(true);
+ EXPECT_TRUE(_cluster_info->is_in_standby());
+ _cluster_info->set_is_in_standby(false);
+ EXPECT_FALSE(_cluster_info->is_in_standby());
+}
+
+// Case 1: Feature disabled, should never skip
+TEST_F(CloudClusterInfoTest, ShouldSkipCompactionDisabled) {
+ config::enable_compaction_rw_separation = false;
+
+ _cluster_info->set_my_cluster_id("cluster_a");
+ _tablet->set_last_active_cluster_info("cluster_b", UnixMillis());
+
+ EXPECT_FALSE(_cluster_info->should_skip_compaction(_tablet.get()));
+}
+
+// Case 2: No last_active_cluster, allow compaction
+TEST_F(CloudClusterInfoTest, ShouldSkipCompactionNoActiveCluster) {
+ config::enable_compaction_rw_separation = true;
+
+ _cluster_info->set_my_cluster_id("cluster_a");
+ // tablet has no last_active_cluster set (empty)
+
+ EXPECT_FALSE(_cluster_info->should_skip_compaction(_tablet.get()));
+}
+
+// Case 3: This is the active cluster, allow compaction
+TEST_F(CloudClusterInfoTest, ShouldSkipCompactionSameCluster) {
+ config::enable_compaction_rw_separation = true;
+
+ _cluster_info->set_my_cluster_id("cluster_a");
+ _tablet->set_last_active_cluster_info("cluster_a", UnixMillis());
+
+ EXPECT_FALSE(_cluster_info->should_skip_compaction(_tablet.get()));
+}
+
+// Case 4: Active cluster not in cache (deleted), allow takeover
+TEST_F(CloudClusterInfoTest, ShouldSkipCompactionClusterDeleted) {
+ config::enable_compaction_rw_separation = true;
+
+ _cluster_info->set_my_cluster_id("cluster_a");
+ _tablet->set_last_active_cluster_info("cluster_deleted", UnixMillis());
+ // cluster_deleted is not in cache
+
+ EXPECT_FALSE(_cluster_info->should_skip_compaction(_tablet.get()));
+}
+
+// Case 5: Active cluster is NORMAL, skip compaction
+TEST_F(CloudClusterInfoTest, ShouldSkipCompactionActiveClusterNormal) {
+ config::enable_compaction_rw_separation = true;
+
+ _cluster_info->set_my_cluster_id("cluster_a");
+ _tablet->set_last_active_cluster_info("cluster_b", UnixMillis());
+ _cluster_info->set_cluster_status("cluster_b", cloud::ClusterStatus::NORMAL, UnixMillis());
+
+ EXPECT_TRUE(_cluster_info->should_skip_compaction(_tablet.get()));
+}
+
+// Case 6: Active cluster SUSPENDED, timeout not reached, skip
+TEST_F(CloudClusterInfoTest, ShouldSkipCompactionSuspendedNotTimedOut) {
+ config::enable_compaction_rw_separation = true;
+ config::compaction_cluster_takeover_timeout_ms = 60000; // 60s
+
+ _cluster_info->set_my_cluster_id("cluster_a");
+ _tablet->set_last_active_cluster_info("cluster_b", UnixMillis());
+ // Set suspended just now — timeout not reached
+ _cluster_info->set_cluster_status("cluster_b", cloud::ClusterStatus::SUSPENDED, UnixMillis());
+
+ EXPECT_TRUE(_cluster_info->should_skip_compaction(_tablet.get()));
+}
+
+// Case 7: Active cluster SUSPENDED, timeout exceeded, allow takeover
+TEST_F(CloudClusterInfoTest, ShouldSkipCompactionSuspendedTimedOut) {
+ config::enable_compaction_rw_separation = true;
+ config::compaction_cluster_takeover_timeout_ms = 1000; // 1s
+
+ _cluster_info->set_my_cluster_id("cluster_a");
+ _tablet->set_last_active_cluster_info("cluster_b", UnixMillis());
+ // Set suspended long ago
+ _cluster_info->set_cluster_status("cluster_b", cloud::ClusterStatus::SUSPENDED,
+ UnixMillis() - 5000);
+
+ EXPECT_FALSE(_cluster_info->should_skip_compaction(_tablet.get()));
+}
+
+// Case 8: Active cluster MANUAL_SHUTDOWN, timeout exceeded, allow takeover
+TEST_F(CloudClusterInfoTest, ShouldSkipCompactionManualShutdownTimedOut) {
+ config::enable_compaction_rw_separation = true;
+ config::compaction_cluster_takeover_timeout_ms = 1000;
+
+ _cluster_info->set_my_cluster_id("cluster_a");
+ _tablet->set_last_active_cluster_info("cluster_b", UnixMillis());
+ _cluster_info->set_cluster_status("cluster_b", cloud::ClusterStatus::MANUAL_SHUTDOWN,
+ UnixMillis() - 5000);
+
+ EXPECT_FALSE(_cluster_info->should_skip_compaction(_tablet.get()));
+}
+
+// Test CloudTablet last_active_cluster_info accessors
+TEST_F(CloudClusterInfoTest, TabletLastActiveClusterInfo) {
+ EXPECT_EQ(_tablet->last_active_cluster_id(), "");
+ EXPECT_EQ(_tablet->last_active_time_ms(), 0);
+
+ int64_t now = UnixMillis();
+ _tablet->set_last_active_cluster_info("cluster_x", now);
+ EXPECT_EQ(_tablet->last_active_cluster_id(), "cluster_x");
+ EXPECT_EQ(_tablet->last_active_time_ms(), now);
+
+ _tablet->set_last_active_cluster_info("cluster_y", now + 1000);
+ EXPECT_EQ(_tablet->last_active_cluster_id(), "cluster_y");
+ EXPECT_EQ(_tablet->last_active_time_ms(), now + 1000);
+}
+
+// Case 9: Active cluster is NORMAL but version count exceeds 80% threshold, force compaction
+TEST_F(CloudClusterInfoTest, ShouldSkipCompactionForceOnHighVersionCount) {
+ config::enable_compaction_rw_separation = true;
+
+ _cluster_info->set_my_cluster_id("cluster_a");
+ _tablet->set_last_active_cluster_info("cluster_b", UnixMillis());
+ _cluster_info->set_cluster_status("cluster_b", cloud::ClusterStatus::NORMAL, UnixMillis());
+
+ // Default max_tablet_version_num is 2000, 80% threshold = 1600
+ // _approximate_num_rowsets starts at -1, so add (1600 - (-1)) = 1601 to reach 1600
+ int64_t cur = _tablet->fetch_add_approximate_num_rowsets(0);
+ _tablet->fetch_add_approximate_num_rowsets(1600 - cur);
+ EXPECT_TRUE(_cluster_info->should_skip_compaction(_tablet.get()));
+
+ // Now bump to 1601 (above threshold) => should NOT skip (force compaction)
+ _tablet->fetch_add_approximate_num_rowsets(1);
+ EXPECT_FALSE(_cluster_info->should_skip_compaction(_tablet.get()));
+}
+
+// Case 10: Active cluster SUSPENDED, not timed out, but version count exceeds threshold
+TEST_F(CloudClusterInfoTest, ShouldSkipCompactionForceOnHighVersionCountSuspended) {
+ config::enable_compaction_rw_separation = true;
+ config::compaction_cluster_takeover_timeout_ms = 60000; // 60s
+
+ _cluster_info->set_my_cluster_id("cluster_a");
+ _tablet->set_last_active_cluster_info("cluster_b", UnixMillis());
+ _cluster_info->set_cluster_status("cluster_b", cloud::ClusterStatus::SUSPENDED, UnixMillis());
+
+ // Set approximate_num_rowsets above 80% threshold => force compaction
+ int64_t cur = _tablet->fetch_add_approximate_num_rowsets(0);
+ _tablet->fetch_add_approximate_num_rowsets(1601 - cur);
+ EXPECT_FALSE(_cluster_info->should_skip_compaction(_tablet.get()));
+}
+
+// Test start/stop bg worker lifecycle — covers start_bg_worker, stop_bg_worker,
+// _bg_worker_func, and _refresh_cluster_status (early return branch since no CloudStorageEngine)
+TEST_F(CloudClusterInfoTest, BgWorkerStartStop) {
+ // Use a very short refresh interval so the bg thread loops quickly
+ auto old_interval = config::cluster_status_cache_refresh_interval_sec;
+ config::cluster_status_cache_refresh_interval_sec = 1;
+
+ // Start the bg worker — this spawns a thread that calls _bg_worker_func,
+ // which calls _refresh_cluster_status (returns early since no CloudStorageEngine)
+ _cluster_info->start_bg_worker();
+
+ // Let it run at least one iteration
+ std::this_thread::sleep_for(std::chrono::milliseconds(1500));
+
+ // Double start should be safe (already running)
+ _cluster_info->start_bg_worker();
+
+ // Stop
+ _cluster_info->stop_bg_worker();
+
+ // Double stop should be safe
+ _cluster_info->stop_bg_worker();
+
+ config::cluster_status_cache_refresh_interval_sec = old_interval;
+}
+
+// Test destructor stops bg worker
+TEST_F(CloudClusterInfoTest, DestructorStopsBgWorker) {
+ auto info = std::make_unique<CloudClusterInfo>();
+ config::cluster_status_cache_refresh_interval_sec = 1;
+ info->start_bg_worker();
+ std::this_thread::sleep_for(std::chrono::milliseconds(1500));
+ // Destructor should call stop_bg_worker
+ info.reset();
+}
+
+} // namespace doris
diff --git a/cloud/src/common/bvars.cpp b/cloud/src/common/bvars.cpp
index 03aa6ad5049..e2f587e151a 100644
--- a/cloud/src/common/bvars.cpp
+++ b/cloud/src/common/bvars.cpp
@@ -189,6 +189,46 @@ BvarStatusWithTag<int64_t> g_bvar_recycler_batch_delete_rowset_plan_count(
BvarStatusWithTag<int64_t> g_bvar_recycler_batch_delete_failures(
"recycler", "batch_delete_failures");
+// Operation Log Recycler BVars
+mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_total_num(
+ "recycler_oplog_last_round_total_num", {"instance_id"});
+mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_not_recycled_num(
+ "recycler_oplog_last_round_not_recycled_num", {"instance_id"});
+mBvarIntAdder g_bvar_recycler_oplog_recycle_failed_num(
+ "recycler_oplog_recycle_failed_num", {"instance_id"});
+mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_commit_partition_num(
+ "recycler_oplog_last_round_recycled_commit_partition_num", {"instance_id"});
+mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_drop_partition_num(
+ "recycler_oplog_last_round_recycled_drop_partition_num", {"instance_id"});
+mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_commit_index_num(
+ "recycler_oplog_last_round_recycled_commit_index_num", {"instance_id"});
+mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_drop_index_num(
+ "recycler_oplog_last_round_recycled_drop_index_num", {"instance_id"});
+mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_update_tablet_num(
+ "recycler_oplog_last_round_recycled_update_tablet_num", {"instance_id"});
+mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_compaction_num(
+ "recycler_oplog_last_round_recycled_compaction_num", {"instance_id"});
+mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_schema_change_num(
+ "recycler_oplog_last_round_recycled_schema_change_num", {"instance_id"});
+mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_commit_txn_num(
+ "recycler_oplog_last_round_recycled_commit_txn_num", {"instance_id"});
+mBvarIntAdder g_bvar_recycler_oplog_recycled_commit_partition_num(
+ "recycler_oplog_recycled_commit_partition_num", {"instance_id"});
+mBvarIntAdder g_bvar_recycler_oplog_recycled_drop_partition_num(
+ "recycler_oplog_recycled_drop_partition_num", {"instance_id"});
+mBvarIntAdder g_bvar_recycler_oplog_recycled_commit_index_num(
+ "recycler_oplog_recycled_commit_index_num", {"instance_id"});
+mBvarIntAdder g_bvar_recycler_oplog_recycled_drop_index_num(
+ "recycler_oplog_recycled_drop_index_num", {"instance_id"});
+mBvarIntAdder g_bvar_recycler_oplog_recycled_update_tablet_num(
+ "recycler_oplog_recycled_update_tablet_num", {"instance_id"});
+mBvarIntAdder g_bvar_recycler_oplog_recycled_compaction_num(
+ "recycler_oplog_recycled_compaction_num", {"instance_id"});
+mBvarIntAdder g_bvar_recycler_oplog_recycled_schema_change_num(
+ "recycler_oplog_recycled_schema_change_num", {"instance_id"});
+mBvarIntAdder g_bvar_recycler_oplog_recycled_commit_txn_num(
+ "recycler_oplog_recycled_commit_txn_num", {"instance_id"});
+
// txn_kv's bvars
bvar::LatencyRecorder g_bvar_txn_kv_get("txn_kv", "get");
bvar::LatencyRecorder g_bvar_txn_kv_range_get("txn_kv", "range_get");
diff --git a/cloud/src/common/bvars.h b/cloud/src/common/bvars.h
index fc73d8d3149..5497cf6a754 100644
--- a/cloud/src/common/bvars.h
+++ b/cloud/src/common/bvars.h
@@ -676,8 +676,28 @@ extern BvarStatusWithTag<int64_t> g_bvar_recycler_batch_delete_failures;
extern BvarStatusWithTag<int64_t> g_bvar_recycler_packed_file_bytes_object_deleted;
extern BvarStatusWithTag<int64_t> g_bvar_recycler_packed_file_rowset_scanned_num;
-extern BvarStatusWithTag<int64_t> g_bvar_recycler_batch_delete_rowset_plan_count;
-extern BvarStatusWithTag<int64_t> g_bvar_recycler_batch_delete_failures;
+// Operation Log Recycler BVars
+// Note: generic metrics (last_round_to_recycle_num/bytes, last_round_recycled_num/bytes, etc.)
+// are reported by RecyclerMetricsContext with operation_type = "recycle_operation_logs".
+extern mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_total_num;
+extern mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_not_recycled_num;
+extern mBvarIntAdder g_bvar_recycler_oplog_recycle_failed_num;
+extern mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_commit_partition_num;
+extern mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_drop_partition_num;
+extern mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_commit_index_num;
+extern mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_drop_index_num;
+extern mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_update_tablet_num;
+extern mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_compaction_num;
+extern mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_schema_change_num;
+extern mBvarStatus<int64_t> g_bvar_recycler_oplog_last_round_recycled_commit_txn_num;
+extern mBvarIntAdder g_bvar_recycler_oplog_recycled_commit_partition_num;
+extern mBvarIntAdder g_bvar_recycler_oplog_recycled_drop_partition_num;
+extern mBvarIntAdder g_bvar_recycler_oplog_recycled_commit_index_num;
+extern mBvarIntAdder g_bvar_recycler_oplog_recycled_drop_index_num;
+extern mBvarIntAdder g_bvar_recycler_oplog_recycled_update_tablet_num;
+extern mBvarIntAdder g_bvar_recycler_oplog_recycled_compaction_num;
+extern mBvarIntAdder g_bvar_recycler_oplog_recycled_schema_change_num;
+extern mBvarIntAdder g_bvar_recycler_oplog_recycled_commit_txn_num;
// txn_kv's bvars
extern bvar::LatencyRecorder g_bvar_txn_kv_get;
diff --git a/cloud/src/meta-service/meta_service_resource.cpp b/cloud/src/meta-service/meta_service_resource.cpp
index 57127e642d0..dad23b45b2c 100644
--- a/cloud/src/meta-service/meta_service_resource.cpp
+++ b/cloud/src/meta-service/meta_service_resource.cpp
@@ -4718,6 +4718,9 @@ void MetaServiceImpl::get_cluster_status(google::protobuf::RpcController* contro
} else {
pb.set_cluster_status(ClusterStatus::NORMAL);
}
+ if (cluster.has_mtime()) {
+ pb.set_mtime(cluster.mtime());
+ }
detail.add_clusters()->CopyFrom(pb);
}
if (detail.clusters().size() == 0) {
@@ -4728,6 +4731,16 @@ void MetaServiceImpl::get_cluster_status(google::protobuf::RpcController* contro
std::for_each(instance_ids.begin(), instance_ids.end(), get_clusters_info);
+ // Resolve the requesting node's cluster_id from cloud_unique_id
+ if (!request->cloud_unique_ids().empty()) {
+ const auto& cloud_unique_id = request->cloud_unique_ids(0);
+ std::vector<NodeInfo> nodes;
+ std::string node_err = resource_mgr_->get_node(cloud_unique_id, &nodes);
+ if (node_err.empty() && !nodes.empty()) {
+ response->set_requester_cluster_id(nodes[0].cluster_id);
+ }
+ }
+
msg = proto_to_json(*response);
}
diff --git a/cloud/src/meta-service/meta_service_txn.cpp b/cloud/src/meta-service/meta_service_txn.cpp
index 799299d11c3..88fd0b81053 100644
--- a/cloud/src/meta-service/meta_service_txn.cpp
+++ b/cloud/src/meta-service/meta_service_txn.cpp
@@ -40,6 +40,7 @@
#include "meta-store/txn_kv.h"
#include "meta-store/txn_kv_error.h"
#include "meta-store/versioned_value.h"
+#include "resource-manager/resource_manager.h"
using namespace std::chrono;
@@ -1182,6 +1183,48 @@ void scan_tmp_rowset(
return;
}
+// Update the last active cluster info for a tablet
+void update_tablet_last_active_cluster(const StatsTabletKeyInfo& info,
+ const std::string& cluster_id,
+ std::unique_ptr<Transaction>& txn, MetaServiceCode& code,
+ std::string& msg) {
+ if (cluster_id.empty()) {
+ return;
+ }
+
+ std::string key;
+ stats_tablet_key(info, &key);
+ std::string val;
+ TxnErrorCode err = txn->get(key, &val);
+ if (err != TxnErrorCode::TXN_OK) {
+ // If tablet stats not found, skip (will be created later)
+ if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
+ return;
+ }
+ code = cast_as<ErrCategory::READ>(err);
+ msg = fmt::format("failed to get tablet stats for cluster update, err={} tablet_id={}", err,
+ std::get<4>(info));
+ return;
+ }
+
+ TabletStatsPB stats_pb;
+ if (!stats_pb.ParseFromString(val)) {
+ code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+ msg = fmt::format("malformed tablet stats value for cluster update, key={}", hex(key));
+ return;
+ }
+
+ stats_pb.set_last_active_cluster_id(cluster_id);
+ stats_pb.set_last_active_time_ms(::time(nullptr) * 1000);
+ // Clear the mtime when updating cluster to allow dynamic filling on next get_rowset
+ stats_pb.clear_last_active_cluster_status_mtime_ms();
+
+ stats_pb.SerializeToString(&val);
+ txn->put(key, val);
+ LOG(INFO) << "update last_active_cluster, key=" << hex(key) << " cluster_id=" << cluster_id
+ << " tablet_id=" << std::get<4>(info);
+}
+
void update_tablet_stats(const StatsTabletKeyInfo& info, const TabletStats& stats,
std::unique_ptr<Transaction>& txn, MetaServiceCode& code,
std::string& msg) {
@@ -1877,6 +1920,14 @@ void MetaServiceImpl::commit_txn_immediately(
<< " versioned tablet stats, txn_id=" << txn_id;
}
+ // Get cluster_id for updating last active cluster.
+ // Use load_cluster_id stored in TxnInfoPB by prepare_rowset (called by BE),
+ // because commit_txn is called by FE whose cloud_unique_id resolves to SQL server cluster.
+ std::string requester_cluster_id;
+ if (txn_info.has_load_cluster_id()) {
+ requester_cluster_id = txn_info.load_cluster_id();
+ }
+
// Update stats of affected tablet
for (auto& [tablet_id, stats] : tablet_stats) {
DCHECK(tablet_ids.count(tablet_id));
@@ -1886,6 +1937,12 @@ void MetaServiceImpl::commit_txn_immediately(
update_tablet_stats(info, stats, txn, code, msg);
if (code != MetaServiceCode::OK) return;
+ // Update last active cluster if load has data
+ if (!requester_cluster_id.empty() && stats.num_segs > 0) {
+ update_tablet_last_active_cluster(info, requester_cluster_id, txn, code, msg);
+ if (code != MetaServiceCode::OK) return;
+ }
+
if (is_versioned_write) {
TabletStatsPB stats_pb = existing_versioned_stats[tablet_id];
merge_tablet_stats(stats_pb, stats);
@@ -3051,6 +3108,13 @@ void MetaServiceImpl::commit_txn_with_sub_txn(const CommitTxnRequest* request,
<< " versioned tablet stats, txn_id=" << txn_id;
}
+ // Get cluster_id for updating last active cluster.
+ // Use load_cluster_id stored in TxnInfoPB by prepare_rowset (called by BE).
+ std::string requester_cluster_id_ev;
+ if (txn_info.has_load_cluster_id()) {
+ requester_cluster_id_ev = txn_info.load_cluster_id();
+ }
+
// Update stats of affected tablet
for (auto& [tablet_id, stats] : tablet_stats) {
DCHECK(tablet_ids.count(tablet_id));
@@ -3060,6 +3124,12 @@ void MetaServiceImpl::commit_txn_with_sub_txn(const CommitTxnRequest* request,
update_tablet_stats(info, stats, txn, code, msg);
if (code != MetaServiceCode::OK) return;
+ // Update last active cluster if load has data
+ if (!requester_cluster_id_ev.empty() && stats.num_segs > 0) {
+ update_tablet_last_active_cluster(info, requester_cluster_id_ev, txn, code, msg);
+ if (code != MetaServiceCode::OK) return;
+ }
+
if (is_versioned_write) {
TabletStatsPB stats_pb = existing_versioned_stats[tablet_id];
merge_tablet_stats(stats_pb, stats);
diff --git a/cloud/src/recycler/recycler.cpp b/cloud/src/recycler/recycler.cpp
index acd572e71c7..47958da4f5b 100644
--- a/cloud/src/recycler/recycler.cpp
+++ b/cloud/src/recycler/recycler.cpp
@@ -7298,6 +7298,46 @@ int InstanceRecycler::scan_and_statistics_restore_jobs() {
return ret;
}
+void InstanceRecycler::scan_and_statistics_operation_logs() {
+ if (!should_recycle_versioned_keys()) {
+ return;
+ }
+
+ RecyclerMetricsContext metrics_context(instance_id_, "recycle_operation_logs");
+
+ OperationLogRecycleChecker recycle_checker(instance_id_, txn_kv_.get(), instance_info_);
+ if (recycle_checker.init() != 0) {
+ return;
+ }
+
+ std::string log_key_prefix = versioned::log_key(instance_id_);
+ std::string begin_key = encode_versioned_key(log_key_prefix, Versionstamp::min());
+ std::string end_key = encode_versioned_key(log_key_prefix, Versionstamp::max());
+
+ std::unique_ptr<BlobIterator> iter = blob_get_range(txn_kv_, begin_key, end_key);
+ for (; iter->valid(); iter->next()) {
+ OperationLogPB operation_log;
+ if (!iter->parse_value(&operation_log)) {
+ continue;
+ }
+
+ std::string_view key = iter->key();
+ Versionstamp log_versionstamp;
+ if (!decode_versioned_key(&key, &log_versionstamp)) {
+ continue;
+ }
+
+ OperationLogReferenceInfo ref_info;
+ if (recycle_checker.can_recycle(log_versionstamp, operation_log.min_timestamp(),
+ &ref_info)) {
+ metrics_context.total_need_recycle_num++;
+ metrics_context.total_need_recycle_data_size += operation_log.ByteSizeLong();
+ }
+ }
+
+ metrics_context.report(true);
+}
+
int InstanceRecycler::classify_rowset_task_by_ref_count(
RowsetDeleteTask& task, std::vector<RowsetDeleteTask>& batch_delete_tasks) {
constexpr int MAX_RETRY = 10;
diff --git a/cloud/src/recycler/recycler.h b/cloud/src/recycler/recycler.h
index cf5cd4ea034..1802d7aed40 100644
--- a/cloud/src/recycler/recycler.h
+++ b/cloud/src/recycler/recycler.h
@@ -247,6 +247,8 @@ public:
: RecyclerMetricsContext("global_recycler", "recycle_segment") {}
};
+struct OplogRecycleStats;
+
class InstanceRecycler {
public:
struct PackedFileRecycleStats {
@@ -401,6 +403,8 @@ public:
int scan_and_statistics_restore_jobs();
+ void scan_and_statistics_operation_logs();
+
/**
* Decode the key of a packed-file metadata record into the persisted object path.
*
@@ -489,7 +493,8 @@ private:
//
// Both `operation_log` and `raw_keys` will be removed in the same transaction, to ensure atomicity.
int recycle_operation_log(Versionstamp log_version, const std::vector<std::string>& raw_keys,
- OperationLogPB operation_log);
+ OperationLogPB operation_log,
+ OplogRecycleStats* oplog_stats = nullptr);
// Recycle rowset meta and data, return 0 for success otherwise error
//
@@ -613,6 +618,24 @@ struct OperationLogReferenceInfo {
Versionstamp referenced_snapshot_timestamp;
};
+struct OplogRecycleStats {
+ // Total oplog count scanned per round
+ std::atomic<int64_t> total_num {0};
+ // Oplogs not recycled this round (per round, written to mBvarStatus)
+ std::atomic<int64_t> not_recycled_num {0};
+ // Recycle failures (per round, accumulated to mBvarIntAdder at end)
+ std::atomic<int64_t> failed_num {0};
+ // Per-oplog-type recycled counts (incremented after successful commit)
+ std::atomic<int64_t> recycled_commit_partition {0};
+ std::atomic<int64_t> recycled_drop_partition {0};
+ std::atomic<int64_t> recycled_commit_index {0};
+ std::atomic<int64_t> recycled_drop_index {0};
+ std::atomic<int64_t> recycled_update_tablet {0};
+ std::atomic<int64_t> recycled_compaction {0};
+ std::atomic<int64_t> recycled_schema_change {0};
+ std::atomic<int64_t> recycled_commit_txn {0};
+};
+
// Helper class to check if operation logs can be recycled based on snapshots and versionstamps
class OperationLogRecycleChecker {
public:
diff --git a/cloud/src/recycler/recycler_operation_log.cpp b/cloud/src/recycler/recycler_operation_log.cpp
index 6e4129b23ae..f6743a5dabb 100644
--- a/cloud/src/recycler/recycler_operation_log.cpp
+++ b/cloud/src/recycler/recycler_operation_log.cpp
@@ -31,6 +31,8 @@
#include <utility>
#include <vector>
+#include "common/bvars.h"
+#include "common/config.h"
#include "common/defer.h"
#include "common/encryption_util.h"
#include "common/logging.h"
@@ -651,6 +653,50 @@ static TxnErrorCode get_txn_info(TxnKv* txn_kv, std::string_view instance_id, in
return TxnErrorCode::TXN_OK;
}
+static void report_oplog_recycle_stats(const std::string& instance_id,
+ const OplogRecycleStats& stats) {
+ g_bvar_recycler_oplog_last_round_total_num.put({instance_id}, stats.total_num.load());
+ g_bvar_recycler_oplog_last_round_not_recycled_num.put({instance_id},
+ stats.not_recycled_num.load());
+ if (stats.failed_num.load() > 0) {
+ g_bvar_recycler_oplog_recycle_failed_num.put({instance_id}, stats.failed_num.load());
+ }
+ // Per-type last round counts (mBvarStatus, overwritten each round)
+ g_bvar_recycler_oplog_last_round_recycled_commit_partition_num.put(
+ {instance_id}, stats.recycled_commit_partition.load());
+ g_bvar_recycler_oplog_last_round_recycled_drop_partition_num.put(
+ {instance_id}, stats.recycled_drop_partition.load());
+ g_bvar_recycler_oplog_last_round_recycled_commit_index_num.put(
+ {instance_id}, stats.recycled_commit_index.load());
+ g_bvar_recycler_oplog_last_round_recycled_drop_index_num.put({instance_id},
+ stats.recycled_drop_index.load());
+ g_bvar_recycler_oplog_last_round_recycled_update_tablet_num.put(
+ {instance_id}, stats.recycled_update_tablet.load());
+ g_bvar_recycler_oplog_last_round_recycled_compaction_num.put({instance_id},
+ stats.recycled_compaction.load());
+ g_bvar_recycler_oplog_last_round_recycled_schema_change_num.put(
+ {instance_id}, stats.recycled_schema_change.load());
+ g_bvar_recycler_oplog_last_round_recycled_commit_txn_num.put({instance_id},
+ stats.recycled_commit_txn.load());
+ // Per-type cumulative counts (mBvarIntAdder, accumulated across rounds)
+ g_bvar_recycler_oplog_recycled_commit_partition_num.put({instance_id},
+ stats.recycled_commit_partition.load());
+ g_bvar_recycler_oplog_recycled_drop_partition_num.put({instance_id},
+ stats.recycled_drop_partition.load());
+ g_bvar_recycler_oplog_recycled_commit_index_num.put({instance_id},
+ stats.recycled_commit_index.load());
+ g_bvar_recycler_oplog_recycled_drop_index_num.put({instance_id},
+ stats.recycled_drop_index.load());
+ g_bvar_recycler_oplog_recycled_update_tablet_num.put({instance_id},
+ stats.recycled_update_tablet.load());
+ g_bvar_recycler_oplog_recycled_compaction_num.put({instance_id},
+ stats.recycled_compaction.load());
+ g_bvar_recycler_oplog_recycled_schema_change_num.put({instance_id},
+ stats.recycled_schema_change.load());
+ g_bvar_recycler_oplog_recycled_commit_txn_num.put({instance_id},
+ stats.recycled_commit_txn.load());
+}
+
int InstanceRecycler::recycle_operation_logs() {
if (!should_recycle_versioned_keys()) {
VLOG_DEBUG << "instance " << instance_id_
@@ -665,6 +711,18 @@ int InstanceRecycler::recycle_operation_logs() {
AnnotateTag tag("instance_id", instance_id_);
LOG_WARNING("begin to recycle operation logs");
+ const std::string task_name = "recycle_operation_logs";
+ RecyclerMetricsContext metrics_context(instance_id_, task_name);
+ OplogRecycleStats oplog_stats;
+
+ // scan_and_statistics_operation_logs() is expensive (scans lots of KVs),
+ // so it's controlled by enable_recycler_stats_metrics.
+ // The other stats (counting what was actually recycled) are lightweight
+ // and always collected.
+ if (config::enable_recycler_stats_metrics) {
+ scan_and_statistics_operation_logs();
+ }
+
StopWatch stop_watch;
size_t total_operation_logs = 0;
size_t recycled_operation_logs = 0;
@@ -673,6 +731,9 @@ int InstanceRecycler::recycle_operation_logs() {
size_t recycled_operation_log_data_size = 0;
DORIS_CLOUD_DEFER {
+ metrics_context.finish_report();
+ report_oplog_recycle_stats(instance_id_, oplog_stats);
+
int64_t cost = stop_watch.elapsed_us() / 1000'000;
LOG_WARNING("recycle operation logs, cost={}s", cost)
.tag("total_operation_logs", total_operation_logs)
@@ -706,16 +767,24 @@ int InstanceRecycler::recycle_operation_logs() {
OperationLogReferenceInfo reference_info;
if (recycle_checker.can_recycle(log_versionstamp, operation_log.min_timestamp(),
&reference_info)) {
+ metrics_context.total_need_recycle_num++;
+ metrics_context.total_need_recycle_data_size += value_size;
+
AnnotateTag tag("log_key", hex(key));
- int res = recycle_operation_log(log_versionstamp, raw_keys, std::move(operation_log));
+ int res = recycle_operation_log(log_versionstamp, raw_keys, std::move(operation_log),
+ &oplog_stats);
if (res != 0) {
LOG_WARNING("failed to recycle operation log").tag("error_code", res);
+ oplog_stats.failed_num.fetch_add(1, std::memory_order_relaxed);
return res;
}
recycled_operation_logs++;
recycled_operation_log_data_size += value_size;
+ metrics_context.total_recycled_num++;
+ metrics_context.total_recycled_data_size += value_size;
} else {
+ oplog_stats.not_recycled_num.fetch_add(1, std::memory_order_relaxed);
int res = calculator.calculate_operation_log_data_size(key, operation_log,
reference_info);
if (res != 0) {
@@ -727,6 +796,8 @@ int InstanceRecycler::recycle_operation_logs() {
total_operation_logs++;
operation_log_data_size += value_size;
max_operation_log_data_size = std::max(max_operation_log_data_size, value_size);
+ oplog_stats.total_num.fetch_add(1, std::memory_order_relaxed);
+ metrics_context.report();
return 0;
};
@@ -801,8 +872,11 @@ int InstanceRecycler::recycle_operation_logs() {
int InstanceRecycler::recycle_operation_log(Versionstamp log_version,
const std::vector<std::string>& raw_keys,
- OperationLogPB operation_log) {
+ OperationLogPB operation_log,
+ OplogRecycleStats* oplog_stats) {
int recycle_log_count = 0;
+ // Track which oplog type was recycled (only one per log entry)
+ std::atomic<int64_t>* recycled_counter = nullptr;
OperationLogRecycler log_recycler(instance_id_, txn_kv_.get(), log_version,
operation_log.min_timestamp(), raw_keys);
RETURN_ON_FAILURE(log_recycler.begin());
@@ -818,6 +892,9 @@ int InstanceRecycler::recycle_operation_log(Versionstamp log_version,
return res; \
} \
recycle_log_count++; \
+ if (oplog_stats) { \
+ recycled_counter = &oplog_stats->recycled_##log_type; \
+ } \
} \
} while (0)
@@ -859,6 +936,9 @@ int InstanceRecycler::recycle_operation_log(Versionstamp log_version,
}
recycle_log_count++;
+ if (oplog_stats) {
+ recycled_counter = &oplog_stats->recycled_commit_txn;
+ }
}
if (recycle_log_count > 1) {
@@ -869,7 +949,11 @@ int InstanceRecycler::recycle_operation_log(Versionstamp log_version,
return -1; // This is an unexpected condition, should not happen
}
- return log_recycler.commit();
+ int ret = log_recycler.commit();
+ if (ret == 0 && recycled_counter) {
+ recycled_counter->fetch_add(1, std::memory_order_relaxed);
+ }
+ return ret;
}
} // namespace doris::cloud
diff --git a/cloud/test/CMakeLists.txt b/cloud/test/CMakeLists.txt
index 9edd059321f..f57c9d46282 100644
--- a/cloud/test/CMakeLists.txt
+++ b/cloud/test/CMakeLists.txt
@@ -46,6 +46,7 @@ add_executable(meta_service_test
meta_service_tablet_stats_test.cpp
meta_service_versioned_read_test.cpp
schema_kv_test.cpp
+ compaction_rw_separation_test.cpp
)
add_executable(meta_server_test meta_server_test.cpp)
diff --git a/cloud/test/compaction_rw_separation_test.cpp b/cloud/test/compaction_rw_separation_test.cpp
new file mode 100644
index 00000000000..adecbba327a
--- /dev/null
+++ b/cloud/test/compaction_rw_separation_test.cpp
@@ -0,0 +1,181 @@
+// 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 <brpc/controller.h>
+#include <gen_cpp/cloud.pb.h>
+#include <gtest/gtest.h>
+
+#include "common/config.h"
+#include "meta-service/meta_service.h"
+#include "meta-store/keys.h"
+#include "meta-store/mem_txn_kv.h"
+#include "mock_resource_manager.h"
+#include "rate-limiter/rate_limiter.h"
+
+namespace doris::cloud {
+
+// Test: commit_txn always updates last_active_cluster_id
+// MS always updates this field, BE controls whether to use it for compaction skipping
+TEST(CompactionRWSeparationTest, CommitTxnUpdatesLastActiveCluster) {
+ auto txn_kv = std::make_shared<MemTxnKv>();
+ ASSERT_EQ(txn_kv->init(), 0);
+
+ // Clear data
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ txn->remove("\x00", "\xfe");
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ auto rs = std::make_shared<MockResourceManager>(txn_kv);
+ auto rl = std::make_shared<RateLimiter>();
+ auto snapshot = std::make_shared<SnapshotManager>(txn_kv);
+ MetaServiceImpl meta_service(txn_kv, rs, rl, snapshot);
+
+ int64_t db_id = 1;
+ int64_t table_id = 100;
+ int64_t index_id = 100;
+ int64_t partition_id = 100;
+ int64_t tablet_id = 10001;
+
+ // Create tablet
+ {
+ brpc::Controller cntl;
+ CreateTabletsRequest req;
+ CreateTabletsResponse res;
+ req.set_cloud_unique_id("test_cloud_unique_id");
+ req.set_db_id(db_id);
+
+ auto tablet = req.add_tablet_metas();
+ tablet->set_table_id(table_id);
+ tablet->set_index_id(index_id);
+ tablet->set_partition_id(partition_id);
+ tablet->set_tablet_id(tablet_id);
+ auto schema = tablet->mutable_schema();
+ schema->set_schema_version(0);
+ auto first_rowset = tablet->add_rs_metas();
+ first_rowset->set_rowset_id(0);
+ first_rowset->set_rowset_id_v2("rowset_0");
+ first_rowset->set_start_version(0);
+ first_rowset->set_end_version(1);
+ first_rowset->mutable_tablet_schema()->CopyFrom(*schema);
+
+ meta_service.create_tablets(&cntl, &req, &res, nullptr);
+ ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().msg();
+ }
+
+ // Begin txn
+ int64_t txn_id = -1;
+ {
+ brpc::Controller cntl;
+ BeginTxnRequest req;
+ BeginTxnResponse res;
+ req.set_cloud_unique_id("test_cloud_unique_id");
+ auto txn_info = req.mutable_txn_info();
+ txn_info->set_db_id(db_id);
+ txn_info->set_label("test_label_1");
+ txn_info->add_table_ids(table_id);
+ txn_info->set_timeout_ms(36000);
+ // FE sets load_cluster_id during begin_txn for compaction RW separation
+ txn_info->set_load_cluster_id(mock_cluster_id);
+
+ meta_service.begin_txn(&cntl, &req, &res, nullptr);
+ ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().msg();
+ txn_id = res.txn_id();
+ }
+
+ // Prepare rowset
+ {
+ brpc::Controller cntl;
+ CreateRowsetRequest req;
+ CreateRowsetResponse res;
+ req.set_cloud_unique_id("test_cloud_unique_id");
+
+ auto rowset = req.mutable_rowset_meta();
+ rowset->set_rowset_id(0);
+ rowset->set_rowset_id_v2("rowset_txn_1");
+ rowset->set_tablet_id(tablet_id);
+ rowset->set_partition_id(partition_id);
+ rowset->set_txn_id(txn_id);
+ rowset->set_num_segments(1);
+ rowset->set_num_rows(100);
+ rowset->set_data_disk_size(1024);
+ rowset->mutable_tablet_schema()->set_schema_version(0);
+ rowset->set_txn_expiration(::time(nullptr) + 3600);
+
+ meta_service.prepare_rowset(&cntl, &req, &res, nullptr);
+ ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().msg();
+ }
+
+ // Commit rowset
+ {
+ brpc::Controller cntl;
+ CreateRowsetRequest req;
+ CreateRowsetResponse res;
+ req.set_cloud_unique_id("test_cloud_unique_id");
+
+ auto rowset = req.mutable_rowset_meta();
+ rowset->set_rowset_id(0);
+ rowset->set_rowset_id_v2("rowset_txn_1");
+ rowset->set_tablet_id(tablet_id);
+ rowset->set_partition_id(partition_id);
+ rowset->set_txn_id(txn_id);
+ rowset->set_num_segments(1);
+ rowset->set_num_rows(100);
+ rowset->set_data_disk_size(1024);
+ rowset->mutable_tablet_schema()->set_schema_version(0);
+ rowset->set_txn_expiration(::time(nullptr) + 3600);
+
+ meta_service.commit_rowset(&cntl, &req, &res, nullptr);
+ ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().msg();
+ }
+
+ // Commit txn
+ {
+ brpc::Controller cntl;
+ CommitTxnRequest req;
+ CommitTxnResponse res;
+ req.set_cloud_unique_id("test_cloud_unique_id");
+ req.set_db_id(db_id);
+ req.set_txn_id(txn_id);
+
+ meta_service.commit_txn(&cntl, &req, &res, nullptr);
+ ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().msg();
+ }
+
+ // Get rowset and check last_active_cluster_id
+ {
+ brpc::Controller cntl;
+ GetRowsetRequest req;
+ GetRowsetResponse res;
+ req.set_cloud_unique_id("test_cloud_unique_id");
+ req.mutable_idx()->set_tablet_id(tablet_id);
+ req.set_start_version(0);
+ req.set_end_version(-1);
+ req.set_base_compaction_cnt(0);
+ req.set_cumulative_compaction_cnt(0);
+ req.set_cumulative_point(2);
+
+ meta_service.get_rowset(&cntl, &req, &res, nullptr);
+ ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().msg();
+
+ // Check that last_active_cluster_id is always set (MS always updates it)
+ EXPECT_TRUE(res.stats().has_last_active_cluster_id());
+ EXPECT_EQ(res.stats().last_active_cluster_id(), mock_cluster_id);
+ }
+}
+
+} // namespace doris::cloud
diff --git a/cloud/test/recycler_operation_log_test.cpp b/cloud/test/recycler_operation_log_test.cpp
index 8eb1cb6f4b2..1d3c3e65065 100644
--- a/cloud/test/recycler_operation_log_test.cpp
+++ b/cloud/test/recycler_operation_log_test.cpp
@@ -16,15 +16,18 @@
// under the License.
#include <butil/strings/string_split.h>
+#include <bvar/variable.h>
#include <fmt/core.h>
#include <gen_cpp/cloud.pb.h>
#include <gen_cpp/olap_file.pb.h>
#include <gtest/gtest.h>
+#include <climits>
#include <cstdint>
#include <memory>
#include <string>
+#include "common/bvars.h"
#include "common/config.h"
#include "common/util.h"
#include "cpp/sync_point.h"
@@ -2807,4 +2810,479 @@ TEST(RecycleOperationLogTest, RecycleCompactionLogKeepKeyBoundsWhenDisabled) {
}
}
+// ============================================================================
+// Tests for OplogRecycleStats and recycle_operation_logs metrics integration
+// ============================================================================
+
+TEST(RecycleOperationLogTest, OplogRecycleStatsAtomicOperations) {
+ OplogRecycleStats stats;
+
+ // Test initial values
+ EXPECT_EQ(stats.total_num.load(), 0);
+ EXPECT_EQ(stats.not_recycled_num.load(), 0);
+ EXPECT_EQ(stats.failed_num.load(), 0);
+ EXPECT_EQ(stats.recycled_commit_partition.load(), 0);
+ EXPECT_EQ(stats.recycled_drop_partition.load(), 0);
+ EXPECT_EQ(stats.recycled_commit_index.load(), 0);
+ EXPECT_EQ(stats.recycled_drop_index.load(), 0);
+ EXPECT_EQ(stats.recycled_update_tablet.load(), 0);
+ EXPECT_EQ(stats.recycled_compaction.load(), 0);
+ EXPECT_EQ(stats.recycled_schema_change.load(), 0);
+ EXPECT_EQ(stats.recycled_commit_txn.load(), 0);
+
+ // Test fetch_add
+ stats.not_recycled_num.fetch_add(3, std::memory_order_relaxed);
+ EXPECT_EQ(stats.not_recycled_num.load(), 3);
+
+ stats.failed_num.fetch_add(1, std::memory_order_relaxed);
+ EXPECT_EQ(stats.failed_num.load(), 1);
+
+ stats.recycled_drop_partition.fetch_add(5, std::memory_order_relaxed);
+ stats.recycled_drop_index.fetch_add(2, std::memory_order_relaxed);
+ stats.recycled_compaction.fetch_add(10, std::memory_order_relaxed);
+ stats.recycled_commit_txn.fetch_add(7, std::memory_order_relaxed);
+ EXPECT_EQ(stats.recycled_drop_partition.load(), 5);
+ EXPECT_EQ(stats.recycled_drop_index.load(), 2);
+ EXPECT_EQ(stats.recycled_compaction.load(), 10);
+ EXPECT_EQ(stats.recycled_commit_txn.load(), 7);
+
+ // Test total_num
+ stats.total_num.fetch_add(10, std::memory_order_relaxed);
+ EXPECT_EQ(stats.total_num.load(), 10);
+}
+
+TEST(RecycleOperationLogTest, OplogRecycleStatsPerTypeCounters) {
+ OplogRecycleStats stats;
+
+ // Simulate multiple oplog recycling events
+ stats.recycled_drop_partition.fetch_add(1, std::memory_order_relaxed);
+ stats.recycled_drop_partition.fetch_add(1, std::memory_order_relaxed);
+ stats.recycled_drop_index.fetch_add(1, std::memory_order_relaxed);
+ stats.recycled_compaction.fetch_add(1, std::memory_order_relaxed);
+ stats.recycled_compaction.fetch_add(1, std::memory_order_relaxed);
+ stats.recycled_compaction.fetch_add(1, std::memory_order_relaxed);
+ stats.recycled_commit_txn.fetch_add(1, std::memory_order_relaxed);
+
+ EXPECT_EQ(stats.recycled_commit_partition.load(), 0);
+ EXPECT_EQ(stats.recycled_drop_partition.load(), 2);
+ EXPECT_EQ(stats.recycled_commit_index.load(), 0);
+ EXPECT_EQ(stats.recycled_drop_index.load(), 1);
+ EXPECT_EQ(stats.recycled_update_tablet.load(), 0);
+ EXPECT_EQ(stats.recycled_compaction.load(), 3);
+ EXPECT_EQ(stats.recycled_schema_change.load(), 0);
+ EXPECT_EQ(stats.recycled_commit_txn.load(), 1);
+}
+
+// Test recycle_operation_logs with stats enabled: drop partition + drop index logs
+// Verifies per-type recycled counts and bvar reporting
+TEST(RecycleOperationLogTest, RecycleOperationLogsWithStatsEnabled) {
+ auto old_flag = config::enable_recycler_stats_metrics;
+ config::enable_recycler_stats_metrics = true;
+ DORIS_CLOUD_DEFER {
+ config::enable_recycler_stats_metrics = old_flag;
+ };
+
+ auto txn_kv = std::make_shared<MemTxnKv>();
+ txn_kv->update_commit_version(1000);
+ ASSERT_EQ(txn_kv->init(), 0);
+
+ InstanceInfoPB instance;
+ instance.set_instance_id(instance_id);
+ instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
+ auto* obj_info = instance.add_obj_info();
+ obj_info->set_id("recycle_oplog_stats");
+ obj_info->set_ak(config::test_s3_ak);
+ obj_info->set_sk(config::test_s3_sk);
+ obj_info->set_endpoint(config::test_s3_endpoint);
+ obj_info->set_region(config::test_s3_region);
+ obj_info->set_bucket(config::test_s3_bucket);
+ obj_info->set_prefix("recycle_oplog_stats");
+ update_instance_info(txn_kv.get(), instance);
+
+ InstanceRecycler recycler(txn_kv, instance, thread_group,
+ std::make_shared<TxnLazyCommitter>(txn_kv));
+ ASSERT_EQ(recycler.init(), 0);
+
+ uint64_t db_id = 1;
+ uint64_t table_id = 100;
+ uint64_t index_id1 = 201;
+ uint64_t index_id2 = 202;
+ uint64_t partition_id1 = 301;
+ uint64_t partition_id2 = 302;
+ int64_t expiration = ::time(nullptr) + 3600;
+
+ // Create table version so the recycler can remove it
+ {
+ std::string ver_key = versioned::table_version_key({instance_id, table_id});
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ versioned_put(txn.get(), ver_key, "");
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ // Put a drop partition log with 2 partitions
+ {
+ std::string log_key = versioned::log_key(instance_id);
+ OperationLogPB operation_log;
+ operation_log.set_min_timestamp(100);
+ auto* drop_partition = operation_log.mutable_drop_partition();
+ drop_partition->set_db_id(db_id);
+ drop_partition->set_table_id(table_id);
+ drop_partition->add_index_ids(index_id1);
+ drop_partition->add_partition_ids(partition_id1);
+ drop_partition->add_partition_ids(partition_id2);
+ drop_partition->set_expired_at_s(expiration);
+ drop_partition->set_update_table_version(true);
+
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ versioned::blob_put(txn.get(), log_key, operation_log);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ // Put a drop index log with 2 indexes
+ {
+ std::string log_key = versioned::log_key(instance_id);
+ OperationLogPB operation_log;
+ operation_log.set_min_timestamp(101);
+ auto* drop_index = operation_log.mutable_drop_index();
+ drop_index->set_db_id(db_id);
+ drop_index->set_table_id(table_id);
+ drop_index->add_index_ids(index_id1);
+ drop_index->add_index_ids(index_id2);
+
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ versioned::blob_put(txn.get(), log_key, operation_log);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ // Save cumulative baselines before recycling (mBvarIntAdder accumulates across tests)
+ int64_t baseline_drop_partition =
+ g_bvar_recycler_oplog_recycled_drop_partition_num.get({instance_id});
+ int64_t baseline_drop_index = g_bvar_recycler_oplog_recycled_drop_index_num.get({instance_id});
+
+ // Recycle the operation logs - stats should be collected
+ ASSERT_EQ(recycler.recycle_operation_logs(), 0);
+
+ // Verify recycle partition records were created
+ {
+ std::string recycle_key1 = recycle_partition_key({instance_id, partition_id1});
+ std::string recycle_key2 = recycle_partition_key({instance_id, partition_id2});
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ std::string value;
+ ASSERT_EQ(txn->get(recycle_key1, &value), TxnErrorCode::TXN_OK);
+ ASSERT_EQ(txn->get(recycle_key2, &value), TxnErrorCode::TXN_OK);
+ }
+
+ // Verify recycle index records were created
+ {
+ std::string recycle_key1 = recycle_index_key({instance_id, index_id1});
+ std::string recycle_key2 = recycle_index_key({instance_id, index_id2});
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ std::string value;
+ ASSERT_EQ(txn->get(recycle_key1, &value), TxnErrorCode::TXN_OK);
+ ASSERT_EQ(txn->get(recycle_key2, &value), TxnErrorCode::TXN_OK);
+ }
+
+ // Verify total oplog count: 2 logs (1 drop_partition + 1 drop_index)
+ {
+ int64_t total = g_bvar_recycler_oplog_last_round_total_num.get({instance_id});
+ EXPECT_EQ(total, 2) << "Should have scanned 2 operation logs in total";
+ }
+
+ // Verify per-type last round recycled counts: 1 drop_partition log and 1 drop_index log
+ {
+ int last_round_drop_partition =
+ g_bvar_recycler_oplog_last_round_recycled_drop_partition_num.get({instance_id});
+ EXPECT_EQ(last_round_drop_partition, 1) << "Should have recycled 1 drop_partition log";
+ }
+ {
+ int last_round_drop_index =
+ g_bvar_recycler_oplog_last_round_recycled_drop_index_num.get({instance_id});
+ EXPECT_EQ(last_round_drop_index, 1) << "Should have recycled 1 drop_index log";
+ }
+
+ // Verify per-type cumulative recycled counts (check delta from baseline)
+ {
+ int64_t recycled_drop_partition =
+ g_bvar_recycler_oplog_recycled_drop_partition_num.get({instance_id});
+ EXPECT_EQ(recycled_drop_partition - baseline_drop_partition, 1)
+ << "Cumulative drop_partition delta should be 1";
+ }
+ {
+ int64_t recycled_drop_index =
+ g_bvar_recycler_oplog_recycled_drop_index_num.get({instance_id});
+ EXPECT_EQ(recycled_drop_index - baseline_drop_index, 1)
+ << "Cumulative drop_index delta should be 1";
+ }
+}
+
+// Test recycle_operation_logs with snapshot protection: some logs skipped
+TEST(RecycleOperationLogTest, RecycleOperationLogsSkippedBySnapshot) {
+ auto old_flag = config::enable_recycler_stats_metrics;
+ config::enable_recycler_stats_metrics = true;
+ DORIS_CLOUD_DEFER {
+ config::enable_recycler_stats_metrics = old_flag;
+ };
+
+ auto txn_kv = std::make_shared<MemTxnKv>();
+ txn_kv->update_commit_version(1000);
+ ASSERT_EQ(txn_kv->init(), 0);
+
+ std::string test_instance_id = "test_oplog_not_recycled";
+
+ InstanceInfoPB instance;
+ instance.set_instance_id(test_instance_id);
+ instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
+ auto* obj_info = instance.add_obj_info();
+ obj_info->set_id("recycle_snapshot_skip");
+ obj_info->set_ak(config::test_s3_ak);
+ obj_info->set_sk(config::test_s3_sk);
+ obj_info->set_endpoint(config::test_s3_endpoint);
+ obj_info->set_region(config::test_s3_region);
+ obj_info->set_bucket(config::test_s3_bucket);
+ obj_info->set_prefix("recycle_snapshot_skip");
+
+ // Store instance info
+ {
+ std::string key = instance_key({test_instance_id});
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ txn->put(key, instance.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ auto insert_empty_value = [&]() {
+ std::unique_ptr<Transaction> txn;
+ EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ txn->put("dummy_snapshot_test", "");
+ EXPECT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ };
+
+ // Create table version
+ uint64_t table_id = 500;
+ {
+ std::string ver_key = versioned::table_version_key({test_instance_id, table_id});
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ versioned_put(txn.get(), ver_key, "");
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ // Put an old operation log (will be recyclable)
+ {
+ std::string log_key = versioned::log_key(test_instance_id);
+ OperationLogPB operation_log;
+ operation_log.set_min_timestamp(100);
+
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ versioned::blob_put(txn.get(), log_key, operation_log);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ insert_empty_value();
+
+ // Write a snapshot (this creates a protection boundary)
+ {
+ SnapshotPB snapshot;
+ std::string snapshot_key = versioned::snapshot_full_key(test_instance_id);
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ versioned_put(txn.get(), snapshot_key, snapshot.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ insert_empty_value();
+
+ // Put a new operation log after the snapshot (min_timestamp before snapshot)
+ // This log should be skipped by snapshot protection
+ {
+ std::string log_key = versioned::log_key(test_instance_id);
+ OperationLogPB operation_log;
+ // min_timestamp is set to a value before the snapshot, so the snapshot
+ // falls within [min_timestamp, log_version) and the log can't be recycled
+ operation_log.set_min_timestamp(100);
+
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ versioned::blob_put(txn.get(), log_key, operation_log);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ insert_empty_value();
+
+ InstanceRecycler recycler(txn_kv, instance, thread_group,
+ std::make_shared<TxnLazyCommitter>(txn_kv));
+ ASSERT_EQ(recycler.init(), 0);
+
+ // Recycle - some logs should be recycled, some not recycled due to snapshot
+ ASSERT_EQ(recycler.recycle_operation_logs(), 0);
+
+ // Verify not_recycled bvar is set
+ {
+ int64_t not_recycled =
+ g_bvar_recycler_oplog_last_round_not_recycled_num.get({test_instance_id});
+ EXPECT_GE(not_recycled, 1) << "At least one log should not be recycled";
+ }
+}
+
+// Test recycle_operation_logs with stats disabled (default)
+TEST(RecycleOperationLogTest, RecycleOperationLogsStatsDisabled) {
+ auto old_flag = config::enable_recycler_stats_metrics;
+ config::enable_recycler_stats_metrics = false;
+ DORIS_CLOUD_DEFER {
+ config::enable_recycler_stats_metrics = old_flag;
+ };
+
+ auto txn_kv = std::make_shared<MemTxnKv>();
+ txn_kv->update_commit_version(1000);
+ ASSERT_EQ(txn_kv->init(), 0);
+
+ InstanceInfoPB instance;
+ instance.set_instance_id(instance_id);
+ instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
+ auto* obj_info = instance.add_obj_info();
+ obj_info->set_id("recycle_stats_disabled");
+ obj_info->set_ak(config::test_s3_ak);
+ obj_info->set_sk(config::test_s3_sk);
+ obj_info->set_endpoint(config::test_s3_endpoint);
+ obj_info->set_region(config::test_s3_region);
+ obj_info->set_bucket(config::test_s3_bucket);
+ obj_info->set_prefix("recycle_stats_disabled");
+ update_instance_info(txn_kv.get(), instance);
+
+ InstanceRecycler recycler(txn_kv, instance, thread_group,
+ std::make_shared<TxnLazyCommitter>(txn_kv));
+ ASSERT_EQ(recycler.init(), 0);
+
+ // Put an empty operation log
+ {
+ std::string log_key = versioned::log_key(instance_id);
+ Versionstamp versionstamp(123, 0);
+ OperationLogPB operation_log;
+ operation_log.set_min_timestamp(versionstamp.version());
+
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ versioned::blob_put(txn.get(), log_key, versionstamp, operation_log);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ // Recycle should succeed without stats collection
+ ASSERT_EQ(recycler.recycle_operation_logs(), 0);
+
+ // Verify all logs were recycled
+ remove_instance_info(txn_kv.get());
+ ASSERT_TRUE(is_empty_range(txn_kv.get())) << dump_range(txn_kv.get());
+}
+
+// Test recycle_operation_logs compaction log with stats: recycled_compaction tracking
+TEST(RecycleOperationLogTest, RecycleCompactionLogWithStats) {
+ auto old_flag = config::enable_recycler_stats_metrics;
+ config::enable_recycler_stats_metrics = true;
+ DORIS_CLOUD_DEFER {
+ config::enable_recycler_stats_metrics = old_flag;
+ };
+
+ auto txn_kv = std::make_shared<MemTxnKv>();
+ txn_kv->update_commit_version(1000);
+ ASSERT_EQ(txn_kv->init(), 0);
+
+ std::string test_instance_id = "test_compaction_log_stats";
+
+ InstanceInfoPB instance;
+ instance.set_instance_id(test_instance_id);
+ instance.set_multi_version_status(MultiVersionStatus::MULTI_VERSION_ENABLED);
+ auto* obj_info = instance.add_obj_info();
+ obj_info->set_id("compaction_stats");
+ obj_info->set_ak(config::test_s3_ak);
+ obj_info->set_sk(config::test_s3_sk);
+ obj_info->set_endpoint(config::test_s3_endpoint);
+ obj_info->set_region(config::test_s3_region);
+ obj_info->set_bucket(config::test_s3_bucket);
+ obj_info->set_prefix("compaction_stats");
+
+ {
+ std::string key = instance_key({test_instance_id});
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ txn->put(key, instance.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ InstanceRecycler recycler(txn_kv, instance, thread_group,
+ std::make_shared<TxnLazyCommitter>(txn_kv));
+ ASSERT_EQ(recycler.init(), 0);
+
+ int64_t table_id = 600;
+ int64_t tablet_id = 603;
+
+ // Create table version
+ {
+ std::string ver_key = versioned::table_version_key({test_instance_id, table_id});
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ versioned_put(txn.get(), ver_key, "");
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ // Put a compaction log with 3 recycle rowsets
+ {
+ std::string log_key = versioned::log_key(test_instance_id);
+ OperationLogPB operation_log;
+ operation_log.set_min_timestamp(100);
+ auto* compaction = operation_log.mutable_compaction();
+ compaction->set_tablet_id(tablet_id);
+
+ for (int i = 0; i < 3; i++) {
+ auto* recycle_rowset = compaction->add_recycle_rowsets();
+ auto* rowset_meta = recycle_rowset->mutable_rowset_meta();
+ rowset_meta->set_rowset_id(0);
+ rowset_meta->set_tablet_id(tablet_id);
+ rowset_meta->set_rowset_id_v2(fmt::format("compaction_rowset_{}", i));
+ rowset_meta->set_start_version(i * 10);
+ rowset_meta->set_end_version(i * 10 + 9);
+ }
+
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ versioned::blob_put(txn.get(), log_key, operation_log);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ // Recycle the operation logs
+ ASSERT_EQ(recycler.recycle_operation_logs(), 0);
+
+ // Verify recycle rowset records were created (3 rowsets from compaction log)
+ {
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
+ for (int i = 0; i < 3; i++) {
+ std::string rowset_id = fmt::format("compaction_rowset_{}", i);
+ std::string recycle_key = recycle_rowset_key({test_instance_id, tablet_id, rowset_id});
+ std::string value;
+ TxnErrorCode err = txn->get(recycle_key, &value);
+ EXPECT_EQ(err, TxnErrorCode::TXN_OK)
+ << "Recycle rowset key not found for rowset " << rowset_id;
+ }
+ }
+
+ // Verify last_round recycled_compaction bvar count (1 compaction log recycled)
+ {
+ int last_round_compaction =
+ g_bvar_recycler_oplog_last_round_recycled_compaction_num.get({test_instance_id});
+ EXPECT_EQ(last_round_compaction, 1) << "Last round should have recycled 1 compaction log";
+ }
+ // Verify cumulative recycled_compaction bvar count
+ {
+ int recycled_compaction =
+ g_bvar_recycler_oplog_recycled_compaction_num.get({test_instance_id});
+ EXPECT_EQ(recycled_compaction, 1) << "Cumulative compaction should be 1";
+ }
+}
+
} // namespace doris::cloud
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 53b73dd2b04..52fea2e8c9b 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
@@ -3787,6 +3787,14 @@ public class Config extends ConfigBase {
@ConfField(mutable = true)
public static long cloud_auto_snapshot_min_interval_seconds = 3600;
+ @ConfField(mutable = true, description = {
+ "cluster snapshot 相关操作的最低权限要求。可选值：'root'（仅 root 用户可执行）或 'admin'（ADMIN 权限用户可执行）。默认值为 'root'。",
+ "The minimum privilege required for cluster snapshot operations. "
+ + "Valid values: 'root' (only root user can execute)"
+ + " or 'admin' (users with ADMIN privilege can execute). "
+ + "Default is 'root'."})
+ public static String cluster_snapshot_min_privilege = "root";
+
@ConfField(mutable = true)
public static long multi_part_upload_part_size_in_bytes = 256 * 1024 * 1024L; // 256MB
@ConfField(mutable = true)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
index bf7e82072c3..90c95f4c2b4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
@@ -306,6 +306,22 @@ public class CloudGlobalTransactionMgr implements GlobalTransactionMgrIface {
txnInfoBuilder.setTimeoutMs(timeoutSecond * 1000);
txnInfoBuilder.setPrecommitTimeoutMs(Config.stream_load_default_precommit_timeout_second * 1000);
+ // Set load_cluster_id for compaction read-write separation.
+ // commit_txn uses this to update last_active_cluster_id on tablet stats.
+ if (Config.isCloudMode()) {
+ try {
+ ConnectContext ctx = ConnectContext.get();
+ if (ctx != null) {
+ String clusterId = ctx.getComputeGroup().getId();
+ if (clusterId != null && !clusterId.isEmpty()) {
+ txnInfoBuilder.setLoadClusterId(clusterId);
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to get compute group id for load_cluster_id, label: {}", label, e);
+ }
+ }
+
final BeginTxnRequest beginTxnRequest = BeginTxnRequest.newBuilder()
.setTxnInfo(txnInfoBuilder.build())
.setCloudUniqueId(Config.cloud_unique_id)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/UserAuthentication.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/UserAuthentication.java
index 0a10f7d9f48..b695254ca83 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/UserAuthentication.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/UserAuthentication.java
@@ -17,8 +17,13 @@
package org.apache.doris.nereids.rules.analysis;
+import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.InfoSchemaDb;
import org.apache.doris.catalog.TableIf;
+import org.apache.doris.cluster.ClusterNamespace;
+import org.apache.doris.common.Config;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.common.UserException;
@@ -59,7 +64,30 @@ public class UserAuthentication {
if (db == null) {
return;
}
- String dbName = db.getFullName();
+ String dbName = ClusterNamespace.getNameFromFullName(db.getFullName());
+
+ // Special handling: cluster snapshot related tables in information_schema
+ // require privilege based on configuration
+ if (dbName.equalsIgnoreCase(InfoSchemaDb.DATABASE_NAME)
+ && (tableName.equalsIgnoreCase("cluster_snapshots")
+ || tableName.equalsIgnoreCase("cluster_snapshot_properties"))) {
+ if ("admin".equalsIgnoreCase(Config.cluster_snapshot_min_privilege)) {
+ // When configured as admin, check ADMIN privilege
+ if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(connectContext, PrivPredicate.ADMIN)) {
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+ PrivPredicate.ADMIN.getPrivs().toString());
+ }
+ } else {
+ // Default or configured as root, check if user is root
+ UserIdentity currentUser = connectContext.getCurrentUserIdentity();
+ if (currentUser == null || !currentUser.isRootUser()) {
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+ "root privilege");
+ }
+ }
+ return; // privilege check passed, allow access
+ }
+
CatalogIf catalog = db.getCatalog();
if (catalog == null) {
return;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminCreateClusterSnapshotCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminCreateClusterSnapshotCommand.java
index ab5c9460bf8..b62a960bede 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminCreateClusterSnapshotCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminCreateClusterSnapshotCommand.java
@@ -18,6 +18,7 @@
package org.apache.doris.nereids.trees.plans.commands;
import org.apache.doris.analysis.StmtType;
+import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
import org.apache.doris.cloud.catalog.CloudEnv;
import org.apache.doris.cloud.snapshot.CloudSnapshotHandler;
@@ -73,9 +74,20 @@ public class AdminCreateClusterSnapshotCommand extends Command implements Forwar
if (!Config.isCloudMode()) {
throw new AnalysisException("The sql is illegal in disk mode ");
}
- if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)) {
- ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
- PrivPredicate.ADMIN.getPrivs().toString());
+ // Check privilege based on configuration
+ if ("admin".equalsIgnoreCase(Config.cluster_snapshot_min_privilege)) {
+ // When configured as admin, check ADMIN privilege
+ if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)) {
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+ PrivPredicate.ADMIN.getPrivs().toString());
+ }
+ } else {
+ // Default or configured as root, check if user is root
+ UserIdentity currentUser = ctx.getCurrentUserIdentity();
+ if (currentUser == null || !currentUser.isRootUser()) {
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+ "root privilege");
+ }
}
for (Map.Entry<String, String> entry : properties.entrySet()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminDropClusterSnapshotCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminDropClusterSnapshotCommand.java
index b131f8a8cc8..a545113321e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminDropClusterSnapshotCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminDropClusterSnapshotCommand.java
@@ -18,6 +18,7 @@
package org.apache.doris.nereids.trees.plans.commands;
import org.apache.doris.analysis.StmtType;
+import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
import org.apache.doris.cloud.proto.Cloud;
import org.apache.doris.cloud.rpc.MetaServiceProxy;
@@ -70,9 +71,20 @@ public class AdminDropClusterSnapshotCommand extends Command implements ForwardW
if (!Config.isCloudMode()) {
throw new AnalysisException("The sql is illegal in disk mode ");
}
- if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)) {
- ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
- PrivPredicate.ADMIN.getPrivs().toString());
+ // Check privilege based on configuration
+ if ("admin".equalsIgnoreCase(Config.cluster_snapshot_min_privilege)) {
+ // When configured as admin, check ADMIN privilege
+ if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)) {
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+ PrivPredicate.ADMIN.getPrivs().toString());
+ }
+ } else {
+ // Default or configured as root, check if user is root
+ UserIdentity currentUser = ctx.getCurrentUserIdentity();
+ if (currentUser == null || !currentUser.isRootUser()) {
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+ "root privilege");
+ }
}
if (key == null || !key.equalsIgnoreCase(SNAPSHOT_ID)) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetAutoClusterSnapshotCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetAutoClusterSnapshotCommand.java
index b450ada674f..2a873c32790 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetAutoClusterSnapshotCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetAutoClusterSnapshotCommand.java
@@ -18,6 +18,7 @@
package org.apache.doris.nereids.trees.plans.commands;
import org.apache.doris.analysis.StmtType;
+import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
import org.apache.doris.cloud.catalog.CloudEnv;
import org.apache.doris.cloud.proto.Cloud;
@@ -92,9 +93,20 @@ public class AdminSetAutoClusterSnapshotCommand extends Command implements Forwa
if (!Config.isCloudMode()) {
throw new AnalysisException("The sql is illegal in disk mode ");
}
- if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)) {
- ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
- PrivPredicate.ADMIN.getPrivs().toString());
+ // Check privilege based on configuration
+ if ("admin".equalsIgnoreCase(Config.cluster_snapshot_min_privilege)) {
+ // When configured as admin, check ADMIN privilege
+ if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)) {
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+ PrivPredicate.ADMIN.getPrivs().toString());
+ }
+ } else {
+ // Default or configured as root, check if user is root
+ UserIdentity currentUser = ctx.getCurrentUserIdentity();
+ if (currentUser == null || !currentUser.isRootUser()) {
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+ "root privilege");
+ }
}
if (properties.isEmpty()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetClusterSnapshotFeatureSwitchCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetClusterSnapshotFeatureSwitchCommand.java
index 7b873ae65dd..9b2d8612315 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetClusterSnapshotFeatureSwitchCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetClusterSnapshotFeatureSwitchCommand.java
@@ -18,6 +18,7 @@
package org.apache.doris.nereids.trees.plans.commands;
import org.apache.doris.analysis.StmtType;
+import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
import org.apache.doris.cloud.catalog.CloudEnv;
import org.apache.doris.cloud.proto.Cloud;
@@ -73,9 +74,20 @@ public class AdminSetClusterSnapshotFeatureSwitchCommand extends Command impleme
if (!Config.isCloudMode()) {
throw new AnalysisException("The sql is illegal in disk mode ");
}
- if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)) {
- ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
- PrivPredicate.ADMIN.getPrivs().toString());
+ // Check privilege based on configuration
+ if ("admin".equalsIgnoreCase(Config.cluster_snapshot_min_privilege)) {
+ // When configured as admin, check ADMIN privilege
+ if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)) {
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+ PrivPredicate.ADMIN.getPrivs().toString());
+ }
+ } else {
+ // Default or configured as root, check if user is root
+ UserIdentity currentUser = ctx.getCurrentUserIdentity();
+ if (currentUser == null || !currentUser.isRootUser()) {
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+ "root privilege");
+ }
}
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/UserAuthenticationTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/UserAuthenticationTest.java
new file mode 100644
index 00000000000..b026dad95d4
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/UserAuthenticationTest.java
@@ -0,0 +1,396 @@
+// 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.
+
+package org.apache.doris.nereids.rules.analysis;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.InfoSchemaDb;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
+
+import mockit.Expectations;
+import mockit.Mocked;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test for UserAuthentication privilege checks.
+ */
+public class UserAuthenticationTest {
+ @Mocked
+ private Env env;
+ @Mocked
+ private ConnectContext connectContext;
+ @Mocked
+ private AccessControllerManager accessControllerManager;
+ @Mocked
+ private SessionVariable sessionVariable;
+ @Mocked
+ private TableIf table;
+ @Mocked
+ private DatabaseIf db;
+ @Mocked
+ private CatalogIf catalog;
+
+ private String originalMinPrivilege;
+
+ @BeforeEach
+ public void setUp() {
+ originalMinPrivilege = Config.cluster_snapshot_min_privilege;
+ }
+
+ @AfterEach
+ public void tearDown() {
+ Config.cluster_snapshot_min_privilege = originalMinPrivilege;
+ }
+
+ /**
+ * Test that a table named "cluster_snapshots" in a normal database (not information_schema)
+ * does NOT trigger the special privilege check for cluster snapshot tables.
+ * This verifies that only information_schema.cluster_snapshots requires special privileges.
+ */
+ @Test
+ public void testSameNameTableInNormalDbNotTriggerSpecialCheck() throws Exception {
+ // Set to root mode - if special check is triggered, non-root user would be denied
+ Config.cluster_snapshot_min_privilege = "root";
+
+ UserIdentity normalUser = new UserIdentity("normal_user", "%");
+ normalUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ // Table setup - same name as cluster snapshot table but in a normal database
+ table.getName();
+ minTimes = 0;
+ result = "cluster_snapshots"; // Same name as the special table
+
+ table.getDatabase();
+ minTimes = 0;
+ result = db;
+
+ // Database is NOT information_schema, it's a normal user database
+ db.getFullName();
+ minTimes = 0;
+ result = "mydb"; // Normal database, not information_schema
+
+ db.getCatalog();
+ minTimes = 0;
+ result = catalog;
+
+ catalog.getName();
+ minTimes = 0;
+ result = "internal";
+
+ // ConnectContext setup
+ connectContext.getSessionVariable();
+ minTimes = 0;
+ result = sessionVariable;
+
+ sessionVariable.isPlayNereidsDump();
+ minTimes = 0;
+ result = false;
+
+ connectContext.getEnv();
+ minTimes = 0;
+ result = env;
+
+ env.getAccessManager();
+ minTimes = 0;
+ result = accessControllerManager;
+
+ // Normal SELECT privilege check should be called (not special root check)
+ // and should return true to allow access
+ accessControllerManager.checkTblPriv(connectContext, "internal", "mydb",
+ "cluster_snapshots", PrivPredicate.SELECT);
+ minTimes = 1; // This MUST be called, proving we're using normal privilege check
+ result = true;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = normalUser;
+ }
+ };
+
+ // Should NOT throw exception - normal user can access mydb.cluster_snapshots
+ // because it's not the special information_schema table
+ Assertions.assertDoesNotThrow(() ->
+ UserAuthentication.checkPermission(table, connectContext, null));
+ }
+
+ /**
+ * Test that a table named "cluster_snapshot_properties" in a normal database
+ * does NOT trigger the special privilege check.
+ */
+ @Test
+ public void testSameNamePropertiesTableInNormalDbNotTriggerSpecialCheck() throws Exception {
+ Config.cluster_snapshot_min_privilege = "root";
+
+ UserIdentity normalUser = new UserIdentity("normal_user", "%");
+ normalUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ table.getName();
+ minTimes = 0;
+ result = "cluster_snapshot_properties";
+
+ table.getDatabase();
+ minTimes = 0;
+ result = db;
+
+ db.getFullName();
+ minTimes = 0;
+ result = "user_database";
+
+ db.getCatalog();
+ minTimes = 0;
+ result = catalog;
+
+ catalog.getName();
+ minTimes = 0;
+ result = "internal";
+
+ connectContext.getSessionVariable();
+ minTimes = 0;
+ result = sessionVariable;
+
+ sessionVariable.isPlayNereidsDump();
+ minTimes = 0;
+ result = false;
+
+ connectContext.getEnv();
+ minTimes = 0;
+ result = env;
+
+ env.getAccessManager();
+ minTimes = 0;
+ result = accessControllerManager;
+
+ accessControllerManager.checkTblPriv(connectContext, "internal", "user_database",
+ "cluster_snapshot_properties", PrivPredicate.SELECT);
+ minTimes = 1;
+ result = true;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = normalUser;
+ }
+ };
+
+ Assertions.assertDoesNotThrow(() ->
+ UserAuthentication.checkPermission(table, connectContext, null));
+ }
+
+ /**
+ * Test that information_schema.cluster_snapshots DOES trigger special privilege check
+ * and denies non-root user in root mode.
+ */
+ @Test
+ public void testInfoSchemaClusterSnapshotsRequiresRootPrivilege() {
+ Config.cluster_snapshot_min_privilege = "root";
+
+ UserIdentity normalUser = new UserIdentity("normal_user", "%");
+ normalUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ table.getName();
+ minTimes = 0;
+ result = "cluster_snapshots";
+
+ table.getDatabase();
+ minTimes = 0;
+ result = db;
+
+ // This IS information_schema
+ db.getFullName();
+ minTimes = 0;
+ result = InfoSchemaDb.DATABASE_NAME;
+
+ connectContext.getSessionVariable();
+ minTimes = 0;
+ result = sessionVariable;
+
+ sessionVariable.isPlayNereidsDump();
+ minTimes = 0;
+ result = false;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = normalUser;
+ }
+ };
+
+ // Should throw AnalysisException because non-root user cannot access
+ // information_schema.cluster_snapshots in root mode
+ Assertions.assertThrows(AnalysisException.class, () ->
+ UserAuthentication.checkPermission(table, connectContext, null));
+ }
+
+ /**
+ * Test that information_schema.cluster_snapshots allows root user.
+ */
+ @Test
+ public void testInfoSchemaClusterSnapshotsAllowsRootUser() {
+ Config.cluster_snapshot_min_privilege = "root";
+
+ new Expectations() {
+ {
+ table.getName();
+ minTimes = 0;
+ result = "cluster_snapshots";
+
+ table.getDatabase();
+ minTimes = 0;
+ result = db;
+
+ db.getFullName();
+ minTimes = 0;
+ result = InfoSchemaDb.DATABASE_NAME;
+
+ connectContext.getSessionVariable();
+ minTimes = 0;
+ result = sessionVariable;
+
+ sessionVariable.isPlayNereidsDump();
+ minTimes = 0;
+ result = false;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = UserIdentity.ROOT;
+ }
+ };
+
+ // Root user should be able to access
+ Assertions.assertDoesNotThrow(() ->
+ UserAuthentication.checkPermission(table, connectContext, null));
+ }
+
+ /**
+ * Test that cluster:information_schema.cluster_snapshots (with cluster prefix)
+ * DOES trigger special privilege check.
+ * This verifies ClusterNamespace.getNameFromFullName correctly strips the cluster prefix.
+ */
+ @Test
+ public void testInfoSchemaWithClusterPrefixTriggersSpecialCheck() {
+ Config.cluster_snapshot_min_privilege = "root";
+
+ UserIdentity normalUser = new UserIdentity("normal_user", "%");
+ normalUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ table.getName();
+ minTimes = 0;
+ result = "cluster_snapshots";
+
+ table.getDatabase();
+ minTimes = 0;
+ result = db;
+
+ // Database name with cluster prefix - ClusterNamespace.getNameFromFullName
+ // should strip "default_cluster:" and return "information_schema"
+ db.getFullName();
+ minTimes = 0;
+ result = "default_cluster:information_schema";
+
+ connectContext.getSessionVariable();
+ minTimes = 0;
+ result = sessionVariable;
+
+ sessionVariable.isPlayNereidsDump();
+ minTimes = 0;
+ result = false;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = normalUser;
+ }
+ };
+
+ // Should throw AnalysisException because after stripping cluster prefix,
+ // the db name is "information_schema", which triggers special privilege check
+ Assertions.assertThrows(AnalysisException.class, () ->
+ UserAuthentication.checkPermission(table, connectContext, null));
+ }
+
+ /**
+ * Test that information_schema.cluster_snapshots allows admin user in admin mode.
+ */
+ @Test
+ public void testInfoSchemaClusterSnapshotsAllowsAdminInAdminMode() {
+ Config.cluster_snapshot_min_privilege = "admin";
+
+ UserIdentity adminUser = new UserIdentity("admin", "%");
+ adminUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ table.getName();
+ minTimes = 0;
+ result = "cluster_snapshots";
+
+ table.getDatabase();
+ minTimes = 0;
+ result = db;
+
+ db.getFullName();
+ minTimes = 0;
+ result = InfoSchemaDb.DATABASE_NAME;
+
+ connectContext.getSessionVariable();
+ minTimes = 0;
+ result = sessionVariable;
+
+ sessionVariable.isPlayNereidsDump();
+ minTimes = 0;
+ result = false;
+
+ Env.getCurrentEnv();
+ minTimes = 0;
+ result = env;
+
+ env.getAccessManager();
+ minTimes = 0;
+ result = accessControllerManager;
+
+ accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
+ minTimes = 0;
+ result = true;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = adminUser;
+ }
+ };
+
+ // Admin user should be able to access in admin mode
+ Assertions.assertDoesNotThrow(() ->
+ UserAuthentication.checkPermission(table, connectContext, null));
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminCreateClusterSnapshotCommandTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminCreateClusterSnapshotCommandTest.java
index 31f63a5bf38..c1a3f2e523e 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminCreateClusterSnapshotCommandTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminCreateClusterSnapshotCommandTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.nereids.trees.plans.commands;
+import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
@@ -24,9 +25,12 @@ import org.apache.doris.mysql.privilege.AccessControllerManager;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.qe.ConnectContext;
+import com.google.common.collect.ImmutableMap;
import mockit.Expectations;
import mockit.Mocked;
+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.util.HashMap;
@@ -36,9 +40,21 @@ public class AdminCreateClusterSnapshotCommandTest {
@Mocked
private Env env;
@Mocked
- private AccessControllerManager accessControllerManager;
- @Mocked
private ConnectContext connectContext;
+ @Mocked
+ private AccessControllerManager accessControllerManager;
+
+ private String originalMinPrivilege;
+
+ @BeforeEach
+ public void setUp() {
+ originalMinPrivilege = Config.cluster_snapshot_min_privilege;
+ }
+
+ @AfterEach
+ public void tearDown() {
+ Config.cluster_snapshot_min_privilege = originalMinPrivilege;
+ }
private void runBefore() throws Exception {
new Expectations() {
@@ -47,10 +63,6 @@ public class AdminCreateClusterSnapshotCommandTest {
minTimes = 0;
result = env;
- env.getAccessManager();
- minTimes = 0;
- result = accessControllerManager;
-
ConnectContext.get();
minTimes = 0;
result = connectContext;
@@ -59,9 +71,10 @@ public class AdminCreateClusterSnapshotCommandTest {
minTimes = 0;
result = true;
- accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
+ // Mock root user for privilege check
+ connectContext.getCurrentUserIdentity();
minTimes = 0;
- result = true;
+ result = UserIdentity.ROOT;
}
};
}
@@ -97,7 +110,74 @@ public class AdminCreateClusterSnapshotCommandTest {
}
@Test
- public void testValidateNoPriviledge() {
+ public void testValidateNoPrivilegeRootMode() {
+ // Test root mode (default): admin user should be denied
+ Config.cluster_snapshot_min_privilege = "root";
+
+ UserIdentity nonRootUser = new UserIdentity("admin", "%");
+ nonRootUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ Env.getCurrentEnv();
+ minTimes = 0;
+ result = env;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = nonRootUser;
+ }
+ };
+ Config.deploy_mode = "cloud";
+
+ Map<String, String> properties = new HashMap<>();
+ AdminCreateClusterSnapshotCommand command = new AdminCreateClusterSnapshotCommand(properties);
+ Assertions.assertThrows(AnalysisException.class, () -> command.validate(connectContext),
+ "Access denied; you need (at least one of) the (root privilege) privilege(s) for this operation");
+ }
+
+ @Test
+ public void testValidateAdminModeWithAdminUser() {
+ // Test admin mode: admin user with ADMIN privilege should be allowed
+ Config.cluster_snapshot_min_privilege = "admin";
+
+ UserIdentity adminUser = new UserIdentity("admin", "%");
+ adminUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ Env.getCurrentEnv();
+ minTimes = 0;
+ result = env;
+
+ env.getAccessManager();
+ minTimes = 0;
+ result = accessControllerManager;
+
+ accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
+ minTimes = 0;
+ result = true;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = adminUser;
+ }
+ };
+ Config.deploy_mode = "cloud";
+
+ Map<String, String> properties = ImmutableMap.of("ttl", "3600", "label", "test");
+ AdminCreateClusterSnapshotCommand command = new AdminCreateClusterSnapshotCommand(properties);
+ Assertions.assertDoesNotThrow(() -> command.validate(connectContext));
+ }
+
+ @Test
+ public void testValidateAdminModeWithNormalUser() {
+ // Test admin mode: normal user without ADMIN privilege should be denied
+ Config.cluster_snapshot_min_privilege = "admin";
+
+ UserIdentity normalUser = new UserIdentity("normal_user", "%");
+ normalUser.setIsAnalyzed();
+
new Expectations() {
{
Env.getCurrentEnv();
@@ -111,6 +191,10 @@ public class AdminCreateClusterSnapshotCommandTest {
accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
minTimes = 0;
result = false;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = normalUser;
}
};
Config.deploy_mode = "cloud";
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminDropClusterSnapshotCommandTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminDropClusterSnapshotCommandTest.java
index f9a289e7e90..49a13623b61 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminDropClusterSnapshotCommandTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminDropClusterSnapshotCommandTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.nereids.trees.plans.commands;
+import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
@@ -26,16 +27,30 @@ import org.apache.doris.qe.ConnectContext;
import mockit.Expectations;
import mockit.Mocked;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
public class AdminDropClusterSnapshotCommandTest {
@Mocked
private Env env;
@Mocked
- private AccessControllerManager accessControllerManager;
- @Mocked
private ConnectContext connectContext;
+ @Mocked
+ private AccessControllerManager accessControllerManager;
+
+ private String originalMinPrivilege;
+
+ @BeforeEach
+ public void setUp() {
+ originalMinPrivilege = Config.cluster_snapshot_min_privilege;
+ }
+
+ @AfterEach
+ public void tearDown() {
+ Config.cluster_snapshot_min_privilege = originalMinPrivilege;
+ }
private void runBefore() throws Exception {
new Expectations() {
@@ -44,10 +59,6 @@ public class AdminDropClusterSnapshotCommandTest {
minTimes = 0;
result = env;
- env.getAccessManager();
- minTimes = 0;
- result = accessControllerManager;
-
ConnectContext.get();
minTimes = 0;
result = connectContext;
@@ -56,9 +67,10 @@ public class AdminDropClusterSnapshotCommandTest {
minTimes = 0;
result = true;
- accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
+ // Mock root user for privilege check
+ connectContext.getCurrentUserIdentity();
minTimes = 0;
- result = true;
+ result = UserIdentity.ROOT;
}
};
}
@@ -94,7 +106,72 @@ public class AdminDropClusterSnapshotCommandTest {
}
@Test
- public void testValidateNoPriviledge() {
+ public void testValidateNoPrivilegeRootMode() {
+ // Test root mode (default): admin user should be denied
+ Config.cluster_snapshot_min_privilege = "root";
+
+ UserIdentity nonRootUser = new UserIdentity("admin", "%");
+ nonRootUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ Env.getCurrentEnv();
+ minTimes = 0;
+ result = env;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = nonRootUser;
+ }
+ };
+ Config.deploy_mode = "cloud";
+
+ AdminDropClusterSnapshotCommand command = new AdminDropClusterSnapshotCommand("snapshot_id", "323741");
+ Assertions.assertThrows(AnalysisException.class, () -> command.validate(connectContext),
+ "Access denied; you need (at least one of) the (root privilege) privilege(s) for this operation");
+ }
+
+ @Test
+ public void testValidateAdminModeWithAdminUser() {
+ // Test admin mode: admin user with ADMIN privilege should be allowed
+ Config.cluster_snapshot_min_privilege = "admin";
+
+ UserIdentity adminUser = new UserIdentity("admin", "%");
+ adminUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ Env.getCurrentEnv();
+ minTimes = 0;
+ result = env;
+
+ env.getAccessManager();
+ minTimes = 0;
+ result = accessControllerManager;
+
+ accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
+ minTimes = 0;
+ result = true;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = adminUser;
+ }
+ };
+ Config.deploy_mode = "cloud";
+
+ AdminDropClusterSnapshotCommand command = new AdminDropClusterSnapshotCommand("snapshot_id", "323741");
+ Assertions.assertDoesNotThrow(() -> command.validate(connectContext));
+ }
+
+ @Test
+ public void testValidateAdminModeWithNormalUser() {
+ // Test admin mode: normal user without ADMIN privilege should be denied
+ Config.cluster_snapshot_min_privilege = "admin";
+
+ UserIdentity normalUser = new UserIdentity("normal_user", "%");
+ normalUser.setIsAnalyzed();
+
new Expectations() {
{
Env.getCurrentEnv();
@@ -108,6 +185,10 @@ public class AdminDropClusterSnapshotCommandTest {
accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
minTimes = 0;
result = false;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = normalUser;
}
};
Config.deploy_mode = "cloud";
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminSetAutoClusterSnapshotCommandTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminSetAutoClusterSnapshotCommandTest.java
index 6779d42fce7..65c8cd8f8f6 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminSetAutoClusterSnapshotCommandTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminSetAutoClusterSnapshotCommandTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.nereids.trees.plans.commands;
+import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
@@ -26,7 +27,9 @@ import org.apache.doris.qe.ConnectContext;
import mockit.Expectations;
import mockit.Mocked;
+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.util.HashMap;
@@ -36,9 +39,21 @@ public class AdminSetAutoClusterSnapshotCommandTest {
@Mocked
private Env env;
@Mocked
- private AccessControllerManager accessControllerManager;
- @Mocked
private ConnectContext connectContext;
+ @Mocked
+ private AccessControllerManager accessControllerManager;
+
+ private String originalMinPrivilege;
+
+ @BeforeEach
+ public void setUp() {
+ originalMinPrivilege = Config.cluster_snapshot_min_privilege;
+ }
+
+ @AfterEach
+ public void tearDown() {
+ Config.cluster_snapshot_min_privilege = originalMinPrivilege;
+ }
private void runBefore() throws Exception {
new Expectations() {
@@ -47,10 +62,6 @@ public class AdminSetAutoClusterSnapshotCommandTest {
minTimes = 0;
result = env;
- env.getAccessManager();
- minTimes = 0;
- result = accessControllerManager;
-
ConnectContext.get();
minTimes = 0;
result = connectContext;
@@ -59,9 +70,10 @@ public class AdminSetAutoClusterSnapshotCommandTest {
minTimes = 0;
result = true;
- accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
+ // Mock root user for privilege check
+ connectContext.getCurrentUserIdentity();
minTimes = 0;
- result = true;
+ result = UserIdentity.ROOT;
}
};
}
@@ -117,7 +129,76 @@ public class AdminSetAutoClusterSnapshotCommandTest {
}
@Test
- public void testValidateNoPriviledge() {
+ public void testValidateNoPrivilegeRootMode() {
+ // Test root mode (default): admin user should be denied
+ Config.cluster_snapshot_min_privilege = "root";
+
+ UserIdentity nonRootUser = new UserIdentity("admin", "%");
+ nonRootUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ Env.getCurrentEnv();
+ minTimes = 0;
+ result = env;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = nonRootUser;
+ }
+ };
+ Config.deploy_mode = "cloud";
+
+ Map<String, String> properties = new HashMap<>();
+ AdminSetAutoClusterSnapshotCommand command = new AdminSetAutoClusterSnapshotCommand(properties);
+ Assertions.assertThrows(AnalysisException.class, () -> command.validate(connectContext),
+ "Access denied; you need (at least one of) the (root privilege) privilege(s) for this operation");
+ }
+
+ @Test
+ public void testValidateAdminModeWithAdminUser() {
+ // Test admin mode: admin user with ADMIN privilege should be allowed
+ Config.cluster_snapshot_min_privilege = "admin";
+
+ UserIdentity adminUser = new UserIdentity("admin", "%");
+ adminUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ Env.getCurrentEnv();
+ minTimes = 0;
+ result = env;
+
+ env.getAccessManager();
+ minTimes = 0;
+ result = accessControllerManager;
+
+ accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
+ minTimes = 0;
+ result = true;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = adminUser;
+ }
+ };
+ Config.deploy_mode = "cloud";
+
+ Map<String, String> properties = new HashMap<>();
+ properties.put("max_reserved_snapshots", "10");
+ properties.put("snapshot_interval_seconds", "3600");
+ AdminSetAutoClusterSnapshotCommand command = new AdminSetAutoClusterSnapshotCommand(properties);
+ Assertions.assertDoesNotThrow(() -> command.validate(connectContext));
+ }
+
+ @Test
+ public void testValidateAdminModeWithNormalUser() {
+ // Test admin mode: normal user without ADMIN privilege should be denied
+ Config.cluster_snapshot_min_privilege = "admin";
+
+ UserIdentity normalUser = new UserIdentity("normal_user", "%");
+ normalUser.setIsAnalyzed();
+
new Expectations() {
{
Env.getCurrentEnv();
@@ -131,6 +212,10 @@ public class AdminSetAutoClusterSnapshotCommandTest {
accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
minTimes = 0;
result = false;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = normalUser;
}
};
Config.deploy_mode = "cloud";
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminSetClusterSnapshotFeatureSwitchCommandTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminSetClusterSnapshotFeatureSwitchCommandTest.java
index 63121ebc01b..a36a3493369 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminSetClusterSnapshotFeatureSwitchCommandTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminSetClusterSnapshotFeatureSwitchCommandTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.nereids.trees.plans.commands;
+import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
@@ -26,16 +27,30 @@ import org.apache.doris.qe.ConnectContext;
import mockit.Expectations;
import mockit.Mocked;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
public class AdminSetClusterSnapshotFeatureSwitchCommandTest {
@Mocked
private Env env;
@Mocked
- private AccessControllerManager accessControllerManager;
- @Mocked
private ConnectContext connectContext;
+ @Mocked
+ private AccessControllerManager accessControllerManager;
+
+ private String originalMinPrivilege;
+
+ @BeforeEach
+ public void setUp() {
+ originalMinPrivilege = Config.cluster_snapshot_min_privilege;
+ }
+
+ @AfterEach
+ public void tearDown() {
+ Config.cluster_snapshot_min_privilege = originalMinPrivilege;
+ }
private void runBefore() throws Exception {
new Expectations() {
@@ -44,10 +59,6 @@ public class AdminSetClusterSnapshotFeatureSwitchCommandTest {
minTimes = 0;
result = env;
- env.getAccessManager();
- minTimes = 0;
- result = accessControllerManager;
-
ConnectContext.get();
minTimes = 0;
result = connectContext;
@@ -56,9 +67,10 @@ public class AdminSetClusterSnapshotFeatureSwitchCommandTest {
minTimes = 0;
result = true;
- accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
+ // Mock root user for privilege check
+ connectContext.getCurrentUserIdentity();
minTimes = 0;
- result = true;
+ result = UserIdentity.ROOT;
}
};
}
@@ -81,7 +93,72 @@ public class AdminSetClusterSnapshotFeatureSwitchCommandTest {
}
@Test
- public void testValidateNoPriviledge() {
+ public void testValidateNoPrivilegeRootMode() {
+ // Test root mode (default): admin user should be denied
+ Config.cluster_snapshot_min_privilege = "root";
+
+ UserIdentity nonRootUser = new UserIdentity("admin", "%");
+ nonRootUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ Env.getCurrentEnv();
+ minTimes = 0;
+ result = env;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = nonRootUser;
+ }
+ };
+ Config.deploy_mode = "cloud";
+
+ AdminSetClusterSnapshotFeatureSwitchCommand command = new AdminSetClusterSnapshotFeatureSwitchCommand(true);
+ Assertions.assertThrows(AnalysisException.class, () -> command.validate(connectContext),
+ "Access denied; you need (at least one of) the (root privilege) privilege(s) for this operation");
+ }
+
+ @Test
+ public void testValidateAdminModeWithAdminUser() {
+ // Test admin mode: admin user with ADMIN privilege should be allowed
+ Config.cluster_snapshot_min_privilege = "admin";
+
+ UserIdentity adminUser = new UserIdentity("admin", "%");
+ adminUser.setIsAnalyzed();
+
+ new Expectations() {
+ {
+ Env.getCurrentEnv();
+ minTimes = 0;
+ result = env;
+
+ env.getAccessManager();
+ minTimes = 0;
+ result = accessControllerManager;
+
+ accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
+ minTimes = 0;
+ result = true;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = adminUser;
+ }
+ };
+ Config.deploy_mode = "cloud";
+
+ AdminSetClusterSnapshotFeatureSwitchCommand command = new AdminSetClusterSnapshotFeatureSwitchCommand(true);
+ Assertions.assertDoesNotThrow(() -> command.validate(connectContext));
+ }
+
+ @Test
+ public void testValidateAdminModeWithNormalUser() {
+ // Test admin mode: normal user without ADMIN privilege should be denied
+ Config.cluster_snapshot_min_privilege = "admin";
+
+ UserIdentity normalUser = new UserIdentity("normal_user", "%");
+ normalUser.setIsAnalyzed();
+
new Expectations() {
{
Env.getCurrentEnv();
@@ -95,6 +172,10 @@ public class AdminSetClusterSnapshotFeatureSwitchCommandTest {
accessControllerManager.checkGlobalPriv(connectContext, PrivPredicate.ADMIN);
minTimes = 0;
result = false;
+
+ connectContext.getCurrentUserIdentity();
+ minTimes = 0;
+ result = normalUser;
}
};
Config.deploy_mode = "cloud";
diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto
index 5f454bbfd42..6e2c46672cd 100644
--- a/gensrc/proto/cloud.proto
+++ b/gensrc/proto/cloud.proto
@@ -517,6 +517,9 @@ message TxnInfoPB {
// This is used for lazy commit to reduce contention during transaction commit.
// The delete bitmap locks will be extended and removed during lazy commit phase.
optional bool defer_deleting_pending_delete_bitmaps = 21;
+ // The cluster_id of the compute cluster that performed the load
+ // Set by FE during begin_txn, used by commit_txn for compaction read-write separation
+ optional string load_cluster_id = 22;
}
// For check txn conflict and txn timeout
@@ -897,6 +900,17 @@ message TabletStatsPB {
optional int64 last_full_compaction_time_ms = 13;
optional int64 index_size = 14;
optional int64 segment_size = 15;
+ // The cluster ID that most recently performed load or compaction on this tablet
+ optional string last_active_cluster_id = 16;
+ // Timestamp (ms) when the last_active_cluster was updated
+ optional int64 last_active_time_ms = 17;
+ // Current status of the last_active_cluster (filled by MS at get_rowset time)
+ // NORMAL means available, other states (SUSPENDED/MANUAL_SHUTDOWN) or UNKNOWN means unavailable
+ optional ClusterStatus last_active_cluster_status = 18;
+ // Timestamp for timeout judgment
+ // - Normal case: dynamically filled with ClusterPB.mtime (not persisted)
+ // - Deleted case: persisted when first detected as deleted, reused thereafter
+ optional int64 last_active_cluster_status_mtime_ms = 19;
}
message ObjectFilePB {
@@ -1359,6 +1373,8 @@ message GetRowsetResponse {
optional SchemaCloudDictionary schema_dict = 4;
// The current max version of the partition
optional int64 partition_max_version = 5;
+ // The cluster_id of the requester
+ optional string requester_cluster_id = 6;
}
message GetSchemaDictRequest {
@@ -1547,6 +1563,8 @@ message GetClusterStatusResponse {
}
optional MetaServiceResponseStatus status = 1;
repeated GetClusterStatusResponseDetail details = 2;
+ // The cluster_id of the requesting node (resolved from cloud_unique_id)
+ optional string requester_cluster_id = 3;
}
message GetClusterResponse {
diff --git a/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_cluster_takeover.groovy b/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_cluster_takeover.groovy
new file mode 100644
index 00000000000..6eaf27b9f49
--- /dev/null
+++ b/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_cluster_takeover.groovy
@@ -0,0 +1,264 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import groovy.json.JsonSlurper
+import groovy.json.JsonOutput
+
+/**
+ * Test compaction cluster takeover when the original write cluster becomes unavailable.
+ *
+ * Scenario:
+ * 1. Write cluster performs load, auto compaction is disabled at BE level initially
+ * 2. Write cluster becomes SUSPENDED via MS API
+ * 3. After timeout, enable auto compaction on read cluster BE via HTTP config update
+ * 4. Read cluster should take over compaction via auto compaction
+ *
+ * This test uses auto compaction (not manual trigger) because manual trigger bypasses
+ * the get_topn_tablets_to_compact logic where should_skip_compaction is checked.
+ * Auto compaction is controlled via BE config (not table property) because ALTER TABLE
+ * disable_auto_compaction has a known bug and may not take effect immediately.
+ */
+suite('test_compaction_cluster_takeover', 'docker') {
+ def options = new ClusterOptions()
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'cloud_tablet_rebalancer_interval_second=1',
+ ]
+ options.beConfigs += [
+ 'enable_compaction_rw_separation=true',
+ 'compaction_cluster_takeover_timeout_ms=10000', // 10 seconds for faster testing
+ 'cluster_status_cache_refresh_interval_sec=3', // 3 seconds for faster testing
+ 'file_cache_enter_disk_resource_limit_mode_percent=99',
+ 'disable_auto_compaction=true', // Disable auto compaction initially, enable via HTTP later
+ ]
+ options.cloudMode = true
+
+ def getTabletStatus = { ip, port, tablet_id ->
+ def url = "http://${ip}:${port}/api/compaction/show?tablet_id=${tablet_id}"
+ def response = new URL(url).text
+ return new JsonSlurper().parseText(response)
+ }
+
+ def getBeIpAndPort = { cluster_name ->
+ def backends = sql """SHOW BACKENDS"""
+ def cluster_bes = backends.findAll { it[19].contains("""\"compute_group_name\" : \"${cluster_name}\"""") }
+
+ if (cluster_bes.isEmpty()) {
+ throw new RuntimeException("No BE found for cluster: ${cluster_name}")
+ }
+
+ def firstBe = cluster_bes[0]
+ return [ip: firstBe[1], http_port: firstBe[4], brpc_port: firstBe[5], backend_id: firstBe[0]]
+ }
+
+ def getCloudClusterId = { cluster_name ->
+ def backends = sql """SHOW BACKENDS"""
+ def cluster_be = backends.find { it[19].contains("""\"compute_group_name\" : \"${cluster_name}\"""") }
+ if (cluster_be == null) {
+ throw new RuntimeException("No BE found for cluster: ${cluster_name}")
+ }
+ def tag = cluster_be[19]
+ def jsonSlurper = new JsonSlurper()
+ def jsonObject = jsonSlurper.parseText(tag)
+ return [cluster_id: jsonObject.compute_group_id, unique_id: jsonObject.cloud_unique_id]
+ }
+
+ // Helper function to set cluster status via MS HTTP API
+ def setClusterStatus = { String unique_id, String cluster_id, String status, def ms ->
+ def jsonOutput = new JsonOutput()
+ def reqBody = [
+ cloud_unique_id: unique_id,
+ cluster: [
+ cluster_id: cluster_id,
+ cluster_status: status
+ ]
+ ]
+ def js = jsonOutput.toJson(reqBody)
+ logger.info("set_cluster_status request: ${js}")
+
+ httpTest {
+ endpoint ms.host + ':' + ms.httpPort
+ uri "/MetaService/http/set_cluster_status?token=greedisgood9999"
+ body js
+ check { respCode, body ->
+ logger.info("set_cluster_status response: ${body} ${respCode}")
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"), "Failed to set cluster status: ${body}")
+ }
+ }
+ }
+
+ docker(options) {
+ def writeCluster = "primary_cluster"
+ def readCluster = "secondary_cluster"
+
+ // Add two clusters
+ cluster.addBackend(1, writeCluster)
+ cluster.addBackend(1, readCluster)
+
+ // Get MS instance
+ def ms = cluster.getAllMetaservices().get(0)
+ logger.info("Meta service: ${ms.host}:${ms.httpPort}")
+
+ def writeBe = getBeIpAndPort(writeCluster)
+ def readBe = getBeIpAndPort(readCluster)
+ def writeClusterInfo = getCloudClusterId(writeCluster)
+
+ logger.info("Primary BE: ${writeBe.ip}:${writeBe.http_port}")
+ logger.info("Secondary BE: ${readBe.ip}:${readBe.http_port}")
+ logger.info("Primary cluster ID: ${writeClusterInfo.cluster_id}")
+
+ // Use write cluster to create table and load data
+ // Auto compaction is disabled at BE level, so no compaction will run initially
+ sql """use @${writeCluster}"""
+
+ def tableName = "test_compaction_takeover"
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE ${tableName} (
+ k1 INT NOT NULL,
+ v1 STRING NOT NULL
+ ) UNIQUE KEY(`k1`)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ );
+ """
+
+ // Insert data from primary cluster (this sets last_active_cluster_id)
+ for (int i = 0; i < 10; i++) {
+ sql """INSERT INTO ${tableName} VALUES (${i}, 'value_${i}')"""
+ }
+
+ // Get tablet info
+ def tablets = sql_return_maparray """ SHOW TABLETS FROM ${tableName} """
+ assertEquals(1, tablets.size())
+ def tablet = tablets[0]
+ def tablet_id = tablet.TabletId
+ logger.info("Tablet ID: ${tablet_id}")
+
+ sleep(3000)
+
+ // Sync rowsets to read cluster
+ sql """use @${readCluster}"""
+ def result = sql """SELECT COUNT(*) FROM ${tableName}"""
+ assertEquals(10, result[0][0])
+
+ // Now suspend the primary cluster using MS HTTP API
+ logger.info("Suspending primary cluster to simulate unavailability...")
+ setClusterStatus(writeClusterInfo.unique_id, writeClusterInfo.cluster_id, "SUSPENDED", ms)
+
+ // Wait for cluster status cache to refresh + takeover timeout
+ // cluster_status_cache_refresh_interval_sec=3, compaction_cluster_takeover_timeout_ms=10000
+ logger.info("Waiting for cluster status cache refresh and takeover timeout...")
+ sleep(15000)
+
+ // Enable auto compaction on the read cluster BE via HTTP config update
+ // (Using BE config instead of ALTER TABLE because ALTER TABLE disable_auto_compaction
+ // has a known bug and may not take effect immediately)
+ logger.info("Enabling auto compaction on read cluster BE...")
+ def (code, out, err) = curl("POST",
+ String.format("http://%s:%s/api/update_config?disable_auto_compaction=false",
+ readBe.ip, readBe.http_port))
+ logger.info("Update config response: code=${code}, out=${out}")
+
+ // Wait for auto compaction to run on the read cluster
+ logger.info("Waiting for auto compaction on read cluster after takeover...")
+ sleep(30000)
+
+ // Check that read cluster has compacted
+ def readStatus = getTabletStatus(readBe.ip, readBe.http_port, tablet_id)
+ logger.info("Read cluster tablet status: ${readStatus}")
+ def readLastCumuTime = readStatus["last cumulative success time"]
+ logger.info("Read cluster last cumulative success time: ${readLastCumuTime}")
+
+ assertTrue(readLastCumuTime != "1970-01-01 08:00:00.000",
+ "Read cluster should have taken over compaction after timeout, but last cumulative success time is: ${readLastCumuTime}")
+
+ logger.info("Secondary cluster successfully took over compaction!")
+
+ // ==========================================
+ // Phase 2: Primary resumes, loads data, and takes back compaction control
+ // ==========================================
+ logger.info("=== Phase 2: Testing primary cluster recovery ===")
+
+ // Resume the primary cluster
+ logger.info("Resuming primary cluster...")
+ setClusterStatus(writeClusterInfo.unique_id, writeClusterInfo.cluster_id, "NORMAL", ms)
+
+ // Record secondary's compaction timestamp before primary recovery
+ def readStatusBefore = getTabletStatus(readBe.ip, readBe.http_port, tablet_id)
+ def readCumuTimeBefore = readStatusBefore["last cumulative success time"]
+ logger.info("Secondary last cumulative success time before primary recovery: ${readCumuTimeBefore}")
+
+ // Insert more data from primary cluster — this updates last_active_cluster_id back to primary
+ sql """use @${writeCluster}"""
+ for (int i = 10; i < 20; i++) {
+ sql """INSERT INTO ${tableName} VALUES (${i}, 'value_${i}')"""
+ }
+ logger.info("Inserted more data from primary cluster")
+
+ // Sync to secondary so both clusters have the new rowsets
+ sql """use @${readCluster}"""
+ def result2 = sql """SELECT COUNT(*) FROM ${tableName}"""
+ assertEquals(20, result2[0][0])
+
+ // Wait for cluster status cache to refresh so both clusters see updated last_active_cluster
+ sleep(5000)
+
+ // Enable auto compaction on both clusters
+ logger.info("Enabling auto compaction on both cluster BEs...")
+ def (code3, out3, err3) = curl("POST",
+ String.format("http://%s:%s/api/update_config?disable_auto_compaction=false",
+ writeBe.ip, writeBe.http_port))
+ logger.info("Enable primary auto compaction response: code=${code3}, out=${out3}")
+ def (code4, out4, err4) = curl("POST",
+ String.format("http://%s:%s/api/update_config?disable_auto_compaction=false",
+ readBe.ip, readBe.http_port))
+ logger.info("Enable secondary auto compaction response: code=${code4}, out=${out4}")
+
+ // Wait for auto compaction to run
+ logger.info("Waiting for auto compaction after primary recovery...")
+ sleep(30000)
+
+ // Check that primary cluster has compacted after recovery
+ def writeStatus = getTabletStatus(writeBe.ip, writeBe.http_port, tablet_id)
+ logger.info("Primary cluster tablet status after recovery: ${writeStatus}")
+ def writeLastCumuTime = writeStatus["last cumulative success time"]
+ logger.info("Primary cluster last cumulative success time: ${writeLastCumuTime}")
+
+ assertTrue(writeLastCumuTime != "1970-01-01 08:00:00.000",
+ "Primary cluster should have resumed compaction after recovery, but last cumulative success time is: ${writeLastCumuTime}")
+
+ logger.info("Primary cluster successfully took back compaction control!")
+
+ // Check that secondary cluster did NOT do new compaction after primary took back control
+ def readStatusAfter = getTabletStatus(readBe.ip, readBe.http_port, tablet_id)
+ def readCumuTimeAfter = readStatusAfter["last cumulative success time"]
+ logger.info("Secondary last cumulative success time after primary recovery: ${readCumuTimeAfter}")
+
+ assertEquals(readCumuTimeBefore, readCumuTimeAfter,
+ "Secondary cluster should have stopped compaction after primary recovered, " +
+ "but cumulative time changed from ${readCumuTimeBefore} to ${readCumuTimeAfter}")
+
+ // Clean up
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+
+ logger.info("All takeover tests passed!")
+ }
+}
diff --git a/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_rw_separation.groovy b/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_rw_separation.groovy
new file mode 100644
index 00000000000..f4755396daf
--- /dev/null
+++ b/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_rw_separation.groovy
@@ -0,0 +1,158 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import groovy.json.JsonSlurper
+
+/**
+ * Test compaction read-write separation feature.
+ *
+ * When enable_compaction_rw_separation is enabled:
+ * - Only the cluster that last performed load (write cluster) should execute compaction
+ * - Read-only clusters should skip compaction to avoid file cache invalidation
+ *
+ * This test uses auto compaction (not manual trigger) because manual trigger bypasses
+ * the get_topn_tablets_to_compact logic where should_skip_compaction is checked.
+ */
+suite('test_compaction_rw_separation', 'docker') {
+ def options = new ClusterOptions()
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'cloud_tablet_rebalancer_interval_second=1',
+ ]
+ options.beConfigs += [
+ 'enable_compaction_rw_separation=true',
+ 'compaction_cluster_takeover_timeout_ms=600000', // 10 min, ensure no takeover in test
+ 'cluster_status_cache_refresh_interval_sec=5',
+ 'file_cache_enter_disk_resource_limit_mode_percent=99',
+ 'disable_auto_compaction=true', // Disable initially, enable via HTTP after data load
+ ]
+ options.cloudMode = true
+
+ def getTabletStatus = { ip, port, tablet_id ->
+ def url = "http://${ip}:${port}/api/compaction/show?tablet_id=${tablet_id}"
+ def response = new URL(url).text
+ return new JsonSlurper().parseText(response)
+ }
+
+ def getBeIpAndPort = { cluster_name ->
+ def backends = sql """SHOW BACKENDS"""
+ def cluster_bes = backends.findAll { it[19].contains("""\"compute_group_name\" : \"${cluster_name}\"""") }
+
+ if (cluster_bes.isEmpty()) {
+ throw new RuntimeException("No BE found for cluster: ${cluster_name}")
+ }
+
+ def firstBe = cluster_bes[0]
+ return [ip: firstBe[1], http_port: firstBe[4], brpc_port: firstBe[5], backend_id: firstBe[0]]
+ }
+
+ docker(options) {
+ def writeCluster = "write_cluster"
+ def readCluster = "read_cluster"
+
+ // Add two clusters
+ cluster.addBackend(1, writeCluster)
+ cluster.addBackend(1, readCluster)
+
+ logger.info("Created write cluster: ${writeCluster}")
+ logger.info("Created read cluster: ${readCluster}")
+
+ def writeBe = getBeIpAndPort(writeCluster)
+ def readBe = getBeIpAndPort(readCluster)
+
+ logger.info("Write BE: ${writeBe.ip}:${writeBe.http_port}")
+ logger.info("Read BE: ${readBe.ip}:${readBe.http_port}")
+
+ // Use write cluster to create table (auto compaction enabled)
+ sql """use @${writeCluster}"""
+
+ def tableName = "test_compaction_rw_sep"
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE ${tableName} (
+ k1 INT NOT NULL,
+ v1 INT NOT NULL
+ ) UNIQUE KEY(`k1`)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ );
+ """
+
+ // Insert multiple rowsets from write cluster to trigger auto compaction
+ for (int i = 0; i < 10; i++) {
+ sql """INSERT INTO ${tableName} VALUES (${i}, ${i * 10})"""
+ }
+
+ // Get tablet info
+ def tablets = sql_return_maparray """ SHOW TABLETS FROM ${tableName} """
+ assertEquals(1, tablets.size())
+ def tablet = tablets[0]
+ def tablet_id = tablet.TabletId
+ logger.info("Tablet ID: ${tablet_id}")
+
+ // Sync rowsets to read cluster
+ sql """use @${readCluster}"""
+ def result = sql """SELECT COUNT(*) FROM ${tableName}"""
+ assertEquals(10, result[0][0])
+
+ // Enable auto compaction on both BEs via HTTP config update
+ // (Using BE config instead of table property because ALTER TABLE disable_auto_compaction
+ // has a known bug and may not take effect immediately)
+ logger.info("Enabling auto compaction on both BEs...")
+ def (code1, out1, err1) = curl("POST",
+ String.format("http://%s:%s/api/update_config?disable_auto_compaction=false",
+ writeBe.ip, writeBe.http_port))
+ logger.info("Update write BE config response: code=${code1}, out=${out1}")
+ def (code2, out2, err2) = curl("POST",
+ String.format("http://%s:%s/api/update_config?disable_auto_compaction=false",
+ readBe.ip, readBe.http_port))
+ logger.info("Update read BE config response: code=${code2}, out=${out2}")
+
+ // Wait for auto compaction to run on both clusters
+ // Write cluster should compact, read cluster should skip
+ logger.info("Waiting for auto compaction to run...")
+ sleep(30000)
+
+ // Check compaction status on write cluster
+ def writeStatus = getTabletStatus(writeBe.ip, writeBe.http_port, tablet_id)
+ logger.info("Write cluster tablet status: ${writeStatus}")
+ def writeLastCumuTime = writeStatus["last cumulative success time"]
+ logger.info("Write cluster last cumulative success time: ${writeLastCumuTime}")
+
+ // Check compaction status on read cluster
+ def readStatus = getTabletStatus(readBe.ip, readBe.http_port, tablet_id)
+ logger.info("Read cluster tablet status: ${readStatus}")
+ def readLastCumuTime = readStatus["last cumulative success time"]
+ logger.info("Read cluster last cumulative success time: ${readLastCumuTime}")
+
+ // Write cluster should have compacted (last success time is not the initial value)
+ assertTrue(writeLastCumuTime != "1970-01-01 08:00:00.000",
+ "Write cluster should have executed compaction, but last cumulative success time is: ${writeLastCumuTime}")
+
+ // Read cluster should NOT have compacted (last success time should still be initial)
+ assertTrue(readLastCumuTime == "1970-01-01 08:00:00.000",
+ "Read cluster should NOT have executed compaction, but last cumulative success time is: ${readLastCumuTime}")
+
+ logger.info("Test passed: Write cluster compacted, read cluster skipped!")
+
+ // Clean up
+ sql """use @${writeCluster}"""
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ }
+}
diff --git a/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_rw_separation_disabled.groovy b/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_rw_separation_disabled.groovy
new file mode 100644
index 00000000000..55ff31ffaec
--- /dev/null
+++ b/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_rw_separation_disabled.groovy
@@ -0,0 +1,151 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import groovy.json.JsonSlurper
+
+/**
+ * Test behavior when compaction read-write separation is DISABLED.
+ *
+ * When enable_compaction_rw_separation is false (default):
+ * - All clusters should be able to execute compaction regardless of which cluster last performed load
+ *
+ * This test uses auto compaction to verify that both clusters can compact.
+ */
+suite('test_compaction_rw_separation_disabled', 'docker') {
+ def options = new ClusterOptions()
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'cloud_tablet_rebalancer_interval_second=1',
+ ]
+ options.beConfigs += [
+ 'enable_compaction_rw_separation=false', // Feature disabled
+ 'file_cache_enter_disk_resource_limit_mode_percent=99',
+ 'disable_auto_compaction=true', // Disable initially, enable via HTTP after data load
+ ]
+ options.cloudMode = true
+
+ def getTabletStatus = { ip, port, tablet_id ->
+ def url = "http://${ip}:${port}/api/compaction/show?tablet_id=${tablet_id}"
+ def response = new URL(url).text
+ return new JsonSlurper().parseText(response)
+ }
+
+ def getBeIpAndPort = { cluster_name ->
+ def backends = sql """SHOW BACKENDS"""
+ def cluster_bes = backends.findAll { it[19].contains("""\"compute_group_name\" : \"${cluster_name}\"""") }
+
+ if (cluster_bes.isEmpty()) {
+ throw new RuntimeException("No BE found for cluster: ${cluster_name}")
+ }
+
+ def firstBe = cluster_bes[0]
+ return [ip: firstBe[1], http_port: firstBe[4], brpc_port: firstBe[5], backend_id: firstBe[0]]
+ }
+
+ docker(options) {
+ def clusterA = "cluster_a"
+ def clusterB = "cluster_b"
+
+ // Add two clusters
+ cluster.addBackend(1, clusterA)
+ cluster.addBackend(1, clusterB)
+
+ logger.info("Created cluster A: ${clusterA}")
+ logger.info("Created cluster B: ${clusterB}")
+
+ def beA = getBeIpAndPort(clusterA)
+ def beB = getBeIpAndPort(clusterB)
+
+ logger.info("Cluster A BE: ${beA.ip}:${beA.http_port}")
+ logger.info("Cluster B BE: ${beB.ip}:${beB.http_port}")
+
+ // Use cluster A to create table and load data (auto compaction enabled)
+ sql """use @${clusterA}"""
+
+ def tableName = "test_compaction_disabled"
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE ${tableName} (
+ k1 INT NOT NULL,
+ v1 INT NOT NULL
+ ) UNIQUE KEY(`k1`)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ );
+ """
+
+ // Insert data from cluster A
+ for (int i = 0; i < 10; i++) {
+ sql """INSERT INTO ${tableName} VALUES (${i}, ${i * 100})"""
+ }
+
+ // Get tablet info
+ def tablets = sql_return_maparray """ SHOW TABLETS FROM ${tableName} """
+ assertEquals(1, tablets.size())
+ def tablet = tablets[0]
+ def tablet_id = tablet.TabletId
+ logger.info("Tablet ID: ${tablet_id}")
+
+ // Sync rowsets to cluster B
+ sql """use @${clusterB}"""
+ def result = sql """SELECT COUNT(*) FROM ${tableName}"""
+ assertEquals(10, result[0][0])
+
+ // Enable auto compaction on both BEs via HTTP config update
+ // (Using BE config instead of table property because ALTER TABLE disable_auto_compaction
+ // has a known bug and may not take effect immediately)
+ logger.info("Enabling auto compaction on both BEs...")
+ def (code1, out1, err1) = curl("POST",
+ String.format("http://%s:%s/api/update_config?disable_auto_compaction=false",
+ beA.ip, beA.http_port))
+ logger.info("Update cluster A BE config response: code=${code1}, out=${out1}")
+ def (code2, out2, err2) = curl("POST",
+ String.format("http://%s:%s/api/update_config?disable_auto_compaction=false",
+ beB.ip, beB.http_port))
+ logger.info("Update cluster B BE config response: code=${code2}, out=${out2}")
+
+ // Wait for auto compaction to run on both clusters
+ logger.info("Waiting for auto compaction to run on both clusters...")
+ sleep(30000)
+
+ // Check compaction status on both clusters
+ def statusA = getTabletStatus(beA.ip, beA.http_port, tablet_id)
+ logger.info("Cluster A tablet status: ${statusA}")
+ def lastCumuTimeA = statusA["last cumulative success time"]
+ logger.info("Cluster A last cumulative success time: ${lastCumuTimeA}")
+
+ def statusB = getTabletStatus(beB.ip, beB.http_port, tablet_id)
+ logger.info("Cluster B tablet status: ${statusB}")
+ def lastCumuTimeB = statusB["last cumulative success time"]
+ logger.info("Cluster B last cumulative success time: ${lastCumuTimeB}")
+
+ // When feature is disabled, both clusters should be able to compact
+ // At least one of them should have compacted within 30 seconds
+ // (Both may not compact if they contend, but at least cluster A which has the tablet should)
+ assertTrue(lastCumuTimeA != "1970-01-01 08:00:00.000" || lastCumuTimeB != "1970-01-01 08:00:00.000",
+ "At least one cluster should have compacted when feature is disabled. " +
+ "Cluster A time: ${lastCumuTimeA}, Cluster B time: ${lastCumuTimeB}")
+
+ logger.info("Feature disabled test passed - compaction ran successfully!")
+
+ // Clean up
+ sql """use @${clusterA}"""
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ }
+}
diff --git a/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_rw_separation_force.groovy b/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_rw_separation_force.groovy
new file mode 100644
index 00000000000..eea6addf93e
--- /dev/null
+++ b/regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_rw_separation_force.groovy
@@ -0,0 +1,161 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import groovy.json.JsonSlurper
+
+/**
+ * Test force compaction on read cluster when version count exceeds 80% of max_tablet_version_num.
+ *
+ * When the write cluster cannot keep up with compaction (disabled, overloaded, etc.),
+ * version count accumulates. If it exceeds 80% of max_tablet_version_num, read clusters
+ * should do compaction anyway despite RW separation being enabled.
+ *
+ * Setup:
+ * - max_tablet_version_num=20 on both BEs (so threshold = 16)
+ * - Disable auto compaction initially
+ * - Load 18 rowsets from write cluster (exceeds threshold)
+ * - Sync to read cluster
+ * - Keep auto compaction disabled on write cluster
+ * - Enable auto compaction only on read cluster
+ * - Verify read cluster performs compaction despite RW separation
+ */
+suite('test_compaction_rw_separation_force', 'docker') {
+ def options = new ClusterOptions()
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'cloud_tablet_rebalancer_interval_second=1',
+ ]
+ options.beConfigs += [
+ 'enable_compaction_rw_separation=true',
+ 'compaction_cluster_takeover_timeout_ms=600000', // 10 min, ensure no takeover in test
+ 'cluster_status_cache_refresh_interval_sec=5',
+ 'file_cache_enter_disk_resource_limit_mode_percent=99',
+ 'disable_auto_compaction=true', // Disable initially
+ 'max_tablet_version_num=20', // Low limit so 80% threshold = 16
+ ]
+ options.cloudMode = true
+
+ def getTabletStatus = { ip, port, tablet_id ->
+ def url = "http://${ip}:${port}/api/compaction/show?tablet_id=${tablet_id}"
+ def response = new URL(url).text
+ return new JsonSlurper().parseText(response)
+ }
+
+ def getBeIpAndPort = { cluster_name ->
+ def backends = sql """SHOW BACKENDS"""
+ def cluster_bes = backends.findAll { it[19].contains("""\"compute_group_name\" : \"${cluster_name}\"""") }
+
+ if (cluster_bes.isEmpty()) {
+ throw new RuntimeException("No BE found for cluster: ${cluster_name}")
+ }
+
+ def firstBe = cluster_bes[0]
+ return [ip: firstBe[1], http_port: firstBe[4], brpc_port: firstBe[5], backend_id: firstBe[0]]
+ }
+
+ docker(options) {
+ def writeCluster = "write_cluster"
+ def readCluster = "read_cluster"
+
+ // Add two clusters
+ cluster.addBackend(1, writeCluster)
+ cluster.addBackend(1, readCluster)
+
+ logger.info("Created write cluster: ${writeCluster}")
+ logger.info("Created read cluster: ${readCluster}")
+
+ def writeBe = getBeIpAndPort(writeCluster)
+ def readBe = getBeIpAndPort(readCluster)
+
+ logger.info("Write BE: ${writeBe.ip}:${writeBe.http_port}")
+ logger.info("Read BE: ${readBe.ip}:${readBe.http_port}")
+
+ // Use write cluster to create table
+ sql """use @${writeCluster}"""
+
+ def tableName = "test_compaction_rw_sep_force"
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE ${tableName} (
+ k1 INT NOT NULL,
+ v1 INT NOT NULL
+ ) UNIQUE KEY(`k1`)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ );
+ """
+
+ // Insert 18 rowsets from write cluster to exceed 80% of max_tablet_version_num=20 (threshold=16)
+ for (int i = 0; i < 18; i++) {
+ sql """INSERT INTO ${tableName} VALUES (${i}, ${i * 10})"""
+ }
+
+ // Get tablet info
+ def tablets = sql_return_maparray """ SHOW TABLETS FROM ${tableName} """
+ assertEquals(1, tablets.size())
+ def tablet = tablets[0]
+ def tablet_id = tablet.TabletId
+ logger.info("Tablet ID: ${tablet_id}")
+
+ // Sync rowsets to read cluster
+ sql """use @${readCluster}"""
+ def result = sql """SELECT COUNT(*) FROM ${tableName}"""
+ assertEquals(18, result[0][0])
+
+ // Keep auto compaction DISABLED on write cluster (simulating "can't compact")
+ // Enable auto compaction ONLY on read cluster
+ logger.info("Enabling auto compaction only on read cluster...")
+ def (code2, out2, err2) = curl("POST",
+ String.format("http://%s:%s/api/update_config?disable_auto_compaction=false",
+ readBe.ip, readBe.http_port))
+ logger.info("Update read BE config response: code=${code2}, out=${out2}")
+
+ // Wait for auto compaction to run on read cluster
+ // Since version count (18) > 80% of max_tablet_version_num (20) = 16,
+ // read cluster should force compaction despite RW separation
+ logger.info("Waiting for read cluster to force compaction...")
+ sleep(30000)
+
+ // Check compaction status on write cluster (should NOT have compacted, auto compaction disabled)
+ def writeStatus = getTabletStatus(writeBe.ip, writeBe.http_port, tablet_id)
+ logger.info("Write cluster tablet status: ${writeStatus}")
+ def writeLastCumuTime = writeStatus["last cumulative success time"]
+ logger.info("Write cluster last cumulative success time: ${writeLastCumuTime}")
+
+ // Check compaction status on read cluster (should have compacted due to force)
+ def readStatus = getTabletStatus(readBe.ip, readBe.http_port, tablet_id)
+ logger.info("Read cluster tablet status: ${readStatus}")
+ def readLastCumuTime = readStatus["last cumulative success time"]
+ logger.info("Read cluster last cumulative success time: ${readLastCumuTime}")
+
+ // Write cluster should NOT have compacted (auto compaction is disabled)
+ assertTrue(writeLastCumuTime == "1970-01-01 08:00:00.000",
+ "Write cluster should NOT have executed compaction (disabled), but last cumulative success time is: ${writeLastCumuTime}")
+
+ // Read cluster SHOULD have compacted (forced by high version count)
+ assertTrue(readLastCumuTime != "1970-01-01 08:00:00.000",
+ "Read cluster should have force-compacted due to high version count, but last cumulative success time is: ${readLastCumuTime}")
+
+ logger.info("Test passed: Read cluster force-compacted despite RW separation!")
+
+ // Clean up
+ sql """use @${writeCluster}"""
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org