Merge branch 'master' into iceberg_stats_orders

Author: morningman

.licenserc.yaml

@@ -47,6 +47,7 @@ header:
     - "fe/fe-core/src/main/java/software/amazon/awssdk/core/client/builder/SdkDefaultClientBuilder.java"
     - "fe/fe-core/src/main/antlr4/org/apache/doris/nereids/JavaLexer.g4"
     - "fe/fe-core/src/main/antlr4/org/apache/doris/nereids/JavaParser.g4"
+    - "fe/.idea/vcs.xml"
     - "be/dict/ik/*"
     - "be/dict/pinyin/*"
     - "be/src/common/signal_handler.h"

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

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
+} // namespace doris

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());
@@ -775,6 +775,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();
     }
@@ -2260,5 +2266,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

be/src/cloud/cloud_meta_mgr.h

@@ -169,6 +169,12 @@ class CloudMetaMgr {
                                    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,