// 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_meta_mgr.h"

#include <brpc/channel.h>

#include <brpc/controller.h>

#include <brpc/errno.pb.h>

#include <bthread/bthread.h>

#include <bthread/condition_variable.h>

#include <bthread/mutex.h>

#include <gen_cpp/FrontendService.h>

#include <gen_cpp/HeartbeatService_types.h>

#include <gen_cpp/PlanNodes_types.h>

#include <gen_cpp/Types_types.h>

#include <gen_cpp/cloud.pb.h>

#include <gen_cpp/olap_file.pb.h>

#include <glog/logging.h>

#include <algorithm>

#include <atomic>

#include <chrono>

#include <cstdint>

#include <memory>

#include <mutex>

#include <random>

#include <shared_mutex>

#include <string>

#include <type_traits>

#include <vector>

#include "cloud/cloud_ms_backpressure_handler.h"

#include "cloud/cloud_ms_rpc_rate_limiters.h"

#include "cloud/cloud_storage_engine.h"

#include "cloud/cloud_tablet.h"

#include "cloud/cloud_warm_up_manager.h"

#include "cloud/config.h"

#include "cloud/delete_bitmap_file_reader.h"

#include "cloud/delete_bitmap_file_writer.h"

#include "cloud/pb_convert.h"

#include "common/config.h"

#include "common/logging.h"

#include "common/status.h"

#include "cpp/sync_point.h"

#include "io/fs/obj_storage_client.h"

#include "load/stream_load/stream_load_context.h"

#include "runtime/exec_env.h"

#include "storage/olap_common.h"

#include "storage/rowset/rowset.h"

#include "storage/rowset/rowset_factory.h"

#include "storage/rowset/rowset_fwd.h"

#include "storage/storage_engine.h"

#include "storage/tablet/tablet_meta.h"

#include "util/client_cache.h"

#include "util/client_connection_provider.h"

#include "util/network_util.h"

#include "util/s3_util.h"

#include "util/thrift_rpc_helper.h"

namespace doris::cloud {

using namespace ErrorCode;

void* run_bthread_work(void* arg) {

auto* f = reinterpret_cast<std::function<void()>*>(arg);

(*f)();

delete f;

return nullptr;

}

Status bthread_fork_join(const std::vector<std::function<Status()>>& tasks, int concurrency) {

if (tasks.empty()) {

return Status::OK();

}

bthread::Mutex lock;

bthread::ConditionVariable cond;

Status status; // Guard by lock

int count = 0; // Guard by lock

for (const auto& task : tasks) {

{

std::unique_lock lk(lock);

// Wait until there are available slots

while (status.ok() && count >= concurrency) {

cond.wait(lk);

}

if (!status.ok()) {

break;

}

// Increase running task count

++count;

}

// dispatch task into bthreads

auto* fn = new std::function<void()>([&, &task = task] {

auto st = task();

{

std::lock_guard lk(lock);

--count;

if (!st.ok()) {

std::swap(st, status);

}

cond.notify_one();

}

});

bthread_t bthread_id;

if (bthread_start_background(&bthread_id, nullptr, run_bthread_work, fn) != 0) {

run_bthread_work(fn);

}

}

// Wait until all running tasks have done

{

std::unique_lock lk(lock);

while (count > 0) {

cond.wait(lk);

}

}

return status;

}

Status bthread_fork_join(std::vector<std::function<Status()>>&& tasks, int concurrency,

std::future<Status>* fut) {

// std::function will cause `copy`, we need to use heap memory to avoid copy ctor called

auto prom = std::make_shared<std::promise<Status>>();

*fut = prom->get_future();

std::function<void()>* fn = new std::function<void()>(

[tasks = std::move(tasks), concurrency, p = std::move(prom)]() mutable {

p->set_value(bthread_fork_join(tasks, concurrency));

});

bthread_t bthread_id;

if (bthread_start_background(&bthread_id, nullptr, run_bthread_work, fn) != 0) {

delete fn;

return Status::InternalError<false>("failed to create bthread");

}

return Status::OK();

}

namespace {

constexpr int kBrpcRetryTimes = 3;

bvar::LatencyRecorder _get_rowset_latency("doris_cloud_meta_mgr_get_rowset");

bvar::LatencyRecorder g_cloud_commit_txn_resp_redirect_latency("cloud_table_stats_report_latency");

bvar::Adder<uint64_t> g_cloud_meta_mgr_rpc_timeout_count("cloud_meta_mgr_rpc_timeout_count");

bvar::Window<bvar::Adder<uint64_t>> g_cloud_ms_rpc_timeout_count_window(

"cloud_meta_mgr_rpc_timeout_qps", &g_cloud_meta_mgr_rpc_timeout_count, 30);

bvar::Adder<uint64_t> g_cloud_meta_mgr_ms_too_busy_reason_total_count(

"cloud_meta_mgr_ms_too_busy_reason", "total");

bvar::Adder<uint64_t> g_cloud_meta_mgr_ms_too_busy_reason_fdb_cluster_count(

"cloud_meta_mgr_ms_too_busy_reason", "fdb_cluster");

bvar::Adder<uint64_t> g_cloud_meta_mgr_ms_too_busy_reason_fdb_client_thread_count(

"cloud_meta_mgr_ms_too_busy_reason", "fdb_client_thread");

bvar::Adder<uint64_t> g_cloud_meta_mgr_ms_too_busy_reason_ms_resource_count(

"cloud_meta_mgr_ms_too_busy_reason", "ms_resource");

bvar::Adder<uint64_t> g_cloud_meta_mgr_ms_too_busy_reason_test_injection_count(

"cloud_meta_mgr_ms_too_busy_reason", "test_injection");

bvar::Adder<uint64_t> g_cloud_meta_mgr_ms_too_busy_reason_no_stress_condition_matched_count(

"cloud_meta_mgr_ms_too_busy_reason", "no_stress_condition_matched");

bvar::LatencyRecorder g_cloud_be_mow_get_dbm_lock_backoff_sleep_time(

"cloud_be_mow_get_dbm_lock_backoff_sleep_time");

bvar::Adder<uint64_t> g_cloud_version_hole_filled_count("cloud_version_hole_filled_count");

class MetaServiceProxy {

public:

static Status get_proxy(MetaServiceProxy** proxy) {

// The 'stub' is a useless parameter, added only to reuse the `get_pooled_client` function.

std::shared_ptr<MetaService_Stub> stub;

return get_pooled_client(&stub, proxy);

}

void set_unhealthy() {

std::unique_lock lock(_mutex);

maybe_unhealthy = true;

}

bool need_reconn(long now) {

return maybe_unhealthy && ((now - last_reconn_time_ms.front()) >

config::meta_service_rpc_reconnect_interval_ms);

}

Status get(std::shared_ptr<MetaService_Stub>* stub) {

using namespace std::chrono;

auto now = duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();

{

std::shared_lock lock(_mutex);

if (_deadline_ms >= now && !is_idle_timeout(now) && !need_reconn(now)) {

_last_access_at_ms.store(now, std::memory_order_relaxed);

*stub = _stub;

return Status::OK();

}

}

auto channel = std::make_unique<brpc::Channel>();

Status s = init_channel(channel.get());

if (!s.ok()) [[unlikely]] {

return s;

}

*stub = std::make_shared<MetaService_Stub>(channel.release(),

google::protobuf::Service::STUB_OWNS_CHANNEL);

long deadline = now;

// connection age only works without list endpoint.

if (config::meta_service_connection_age_base_seconds > 0) {

std::default_random_engine rng(static_cast<uint32_t>(now));

std::uniform_int_distribution<> uni(

config::meta_service_connection_age_base_seconds,

config::meta_service_connection_age_base_seconds * 2);

deadline = now + duration_cast<milliseconds>(seconds(uni(rng))).count();

}

// Last one WIN

std::unique_lock lock(_mutex);

_last_access_at_ms.store(now, std::memory_order_relaxed);

_deadline_ms = deadline;

_stub = *stub;

last_reconn_time_ms.push(now);

last_reconn_time_ms.pop();

maybe_unhealthy = false;

return Status::OK();

}

private:

static bool is_meta_service_endpoint_list() {

return config::meta_service_endpoint.find(',') != std::string::npos;

}

/**

* This function initializes a pool of `MetaServiceProxy` objects and selects one using

* round-robin. It returns a client stub via the selected proxy.

*

* @param stub A pointer to a shared pointer of `MetaService_Stub` to be retrieved.

* @param proxy (Optional) A pointer to store the selected `MetaServiceProxy`.

*

* @return Status Returns `Status::OK()` on success or an error status on failure.

*/

static Status get_pooled_client(std::shared_ptr<MetaService_Stub>* stub,

MetaServiceProxy** proxy) {

static std::once_flag proxies_flag;

static size_t num_proxies = 1;

static std::atomic<size_t> index(0);

static std::unique_ptr<MetaServiceProxy[]> proxies;

if (config::meta_service_endpoint.empty()) {

return Status::InvalidArgument(

"Meta service endpoint is empty. Please configure manually or wait for "

"heartbeat to obtain.");

}

std::call_once(

proxies_flag, +[]() {

if (config::meta_service_connection_pooled) {

num_proxies = config::meta_service_connection_pool_size;

}

proxies = std::make_unique<MetaServiceProxy[]>(num_proxies);

});

for (size_t i = 0; i + 1 < num_proxies; ++i) {

size_t next_index = index.fetch_add(1, std::memory_order_relaxed) % num_proxies;

Status s = proxies[next_index].get(stub);

if (proxy != nullptr) {

*proxy = &(proxies[next_index]);

}

if (s.ok()) return Status::OK();

}

size_t next_index = index.fetch_add(1, std::memory_order_relaxed) % num_proxies;

if (proxy != nullptr) {

*proxy = &(proxies[next_index]);

}

return proxies[next_index].get(stub);

}

static Status init_channel(brpc::Channel* channel) {

static std::atomic<size_t> index = 1;

const char* load_balancer_name = nullptr;

std::string endpoint;

if (is_meta_service_endpoint_list()) {

endpoint = fmt::format("list://{}", config::meta_service_endpoint);

load_balancer_name = "random";

} else {

std::string ip;

uint16_t port;

Status s = get_meta_service_ip_and_port(&ip, &port);

if (!s.ok()) {

LOG(WARNING) << "fail to get meta service ip and port: " << s;

return s;

}

endpoint = get_host_port(ip, port);

}

brpc::ChannelOptions options;

RETURN_IF_ERROR(doris::client::configure_brpc_channel_options(&options));

options.connection_group =

fmt::format("ms_{}", index.fetch_add(1, std::memory_order_relaxed));

if (channel->Init(endpoint.c_str(), load_balancer_name, &options) != 0) {

return Status::InvalidArgument("failed to init brpc channel, endpoint: {}", endpoint);

}

return Status::OK();

}

static Status get_meta_service_ip_and_port(std::string* ip, uint16_t* port) {

std::string parsed_host;

if (!parse_endpoint(config::meta_service_endpoint, &parsed_host, port)) {

return Status::InvalidArgument("invalid meta service endpoint: {}",

config::meta_service_endpoint);

}

if (is_valid_ip(parsed_host)) {

*ip = std::move(parsed_host);

return Status::OK();

}

return hostname_to_ip(parsed_host, *ip);

}

bool is_idle_timeout(long now) {

auto idle_timeout_ms = config::meta_service_idle_connection_timeout_ms;

// idle timeout only works without list endpoint.

return !is_meta_service_endpoint_list() && idle_timeout_ms > 0 &&

_last_access_at_ms.load(std::memory_order_relaxed) + idle_timeout_ms < now;

}

std::shared_mutex _mutex;

std::atomic<long> _last_access_at_ms {0};

long _deadline_ms {0};

std::shared_ptr<MetaService_Stub> _stub;

std::queue<long> last_reconn_time_ms {std::deque<long> {0, 0, 0}};

bool maybe_unhealthy = false;

};

template <typename T, typename... Ts>

struct is_any : std::disjunction<std::is_same<T, Ts>...> {};

template <typename T, typename... Ts>

constexpr bool is_any_v = is_any<T, Ts...>::value;

template <typename Request>

static std::string debug_info(const Request& req) {

if constexpr (is_any_v<Request, CommitTxnRequest, AbortTxnRequest, PrecommitTxnRequest>) {

return fmt::format(" txn_id={}", req.txn_id());

} else if constexpr (is_any_v<Request, StartTabletJobRequest, FinishTabletJobRequest>) {

return fmt::format(" tablet_id={}", req.job().idx().tablet_id());

} else if constexpr (is_any_v<Request, UpdateDeleteBitmapRequest>) {

return fmt::format(" tablet_id={}, lock_id={}", req.tablet_id(), req.lock_id());

} else if constexpr (is_any_v<Request, GetDeleteBitmapUpdateLockRequest>) {

return fmt::format(" table_id={}, lock_id={}", req.table_id(), req.lock_id());

} 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, GetClusterStatusRequest>) {

return "";

} else if constexpr (is_any_v<Request, CreateRowsetRequest>) {

return fmt::format(" tablet_id={}", req.rowset_meta().tablet_id());

} else if constexpr (is_any_v<Request, RemoveDeleteBitmapRequest>) {

return fmt::format(" tablet_id={}", req.tablet_id());

} else if constexpr (is_any_v<Request, RemoveDeleteBitmapUpdateLockRequest>) {

return fmt::format(" table_id={}, tablet_id={}, lock_id={}", req.table_id(),

req.tablet_id(), req.lock_id());

} else if constexpr (is_any_v<Request, GetDeleteBitmapRequest>) {

return fmt::format(" tablet_id={}", req.tablet_id());

} else if constexpr (is_any_v<Request, GetSchemaDictRequest>) {

return fmt::format(" index_id={}", req.index_id());

} else if constexpr (is_any_v<Request, RestoreJobRequest>) {

return fmt::format(" tablet_id={}", req.tablet_id());

} else if constexpr (is_any_v<Request, UpdatePackedFileInfoRequest>) {

return fmt::format(" packed_file_path={}", req.packed_file_path());

} else {

static_assert(!sizeof(Request));

}

}

inline std::default_random_engine make_random_engine() {

return std::default_random_engine(

static_cast<uint32_t>(std::chrono::steady_clock::now().time_since_epoch().count()));

}

// Convert MetaServiceRPC to LoadRelatedRpc

// Returns LoadRelatedRpc::COUNT if the RPC is not a load-related RPC

LoadRelatedRpc to_load_related_rpc(MetaServiceRPC rpc) {

switch (rpc) {

case MetaServiceRPC::PREPARE_ROWSET:

return LoadRelatedRpc::PREPARE_ROWSET;

case MetaServiceRPC::COMMIT_ROWSET:

return LoadRelatedRpc::COMMIT_ROWSET;

case MetaServiceRPC::UPDATE_TMP_ROWSET:

return LoadRelatedRpc::UPDATE_TMP_ROWSET;

case MetaServiceRPC::UPDATE_PACKED_FILE_INFO:

return LoadRelatedRpc::UPDATE_PACKED_FILE_INFO;

case MetaServiceRPC::UPDATE_DELETE_BITMAP:

return LoadRelatedRpc::UPDATE_DELETE_BITMAP;

default:

return LoadRelatedRpc::COUNT; // Not a load-related RPC

}

}

template <typename Request, typename Response>

using MetaServiceMethod = void (MetaService_Stub::*)(::google::protobuf::RpcController*,

const Request*, Response*,

::google::protobuf::Closure*);

// Rate limiting context for retry_rpc

struct RpcRateLimitCtx {

HostLevelMSRpcRateLimiters* host_limiters {nullptr};

MSBackpressureHandler* backpressure_handler {nullptr};

int64_t table_id {-1}; // For table-level backpressure, passed from caller

};

// Apply rate limiting before RPC (both host-level and table-level)

void apply_rate_limit(MetaServiceRPC rpc, const RpcRateLimitCtx& ctx) {

// Table-level rate limit (for load-related RPCs only)

if (ctx.backpressure_handler && ctx.table_id > 0) {

LoadRelatedRpc load_rpc = to_load_related_rpc(rpc);

if (load_rpc != LoadRelatedRpc::COUNT) {

auto wait_until = ctx.backpressure_handler->before_rpc(load_rpc, ctx.table_id);

auto now = std::chrono::steady_clock::now();

if (wait_until > now) {

auto wait_us =

std::chrono::duration_cast<std::chrono::microseconds>(wait_until - now)

.count();

if (wait_us > 0) {

if (auto* recorder = get_throttle_wait_recorder(load_rpc);

recorder != nullptr) {

*recorder << wait_us;

}

bthread_usleep(wait_us);

}

}

}

}

// Host-level rate limit

if (ctx.host_limiters) {

ctx.host_limiters->limit(rpc);

}

}

// Record RPC QPS statistics after RPC (for table-level tracking)

void record_rpc_qps(MetaServiceRPC rpc, const RpcRateLimitCtx& ctx) {

if (ctx.backpressure_handler && ctx.table_id > 0) {

LoadRelatedRpc load_rpc = to_load_related_rpc(rpc);

if (load_rpc != LoadRelatedRpc::COUNT) {

ctx.backpressure_handler->after_rpc(load_rpc, ctx.table_id);

}

}

}

template <typename Request, typename Response>

Status retry_rpc(MetaServiceRPC rpc, const Request& req, Response* res,

MetaServiceMethod<Request, Response> method,

const RpcRateLimitCtx& rate_limit_ctx = {}) {

static_assert(std::is_base_of_v<::google::protobuf::Message, Request>);

static_assert(std::is_base_of_v<::google::protobuf::Message, Response>);

std::string_view op_name = meta_service_rpc_display_name(rpc);

// Applies only to the current file, and all req are non-const, but passed as const types.

const_cast<Request&>(req).set_request_ip(BackendOptions::get_be_endpoint());

int retry_times = 0;

uint32_t duration_ms = 0;

std::string error_msg;

std::default_random_engine rng = make_random_engine();

std::uniform_int_distribution<uint32_t> u(20, 200);

std::uniform_int_distribution<uint32_t> u2(500, 1000);

MetaServiceProxy* proxy;

RETURN_IF_ERROR(MetaServiceProxy::get_proxy(&proxy));

while (true) {

std::shared_ptr<MetaService_Stub> stub;

RETURN_IF_ERROR(proxy->get(&stub));

// Apply rate limiting (both host-level and table-level)

apply_rate_limit(rpc, rate_limit_ctx);

TEST_SYNC_POINT_CALLBACK("retry_rpc::after_rate_limit", &rpc);

brpc::Controller cntl;

if (rpc == MetaServiceRPC::GET_DELETE_BITMAP ||

rpc == MetaServiceRPC::UPDATE_DELETE_BITMAP) {

cntl.set_timeout_ms(3 * config::meta_service_brpc_timeout_ms);

} else {

cntl.set_timeout_ms(config::meta_service_brpc_timeout_ms);

}

cntl.set_max_retry(kBrpcRetryTimes);

res->Clear();

int error_code = 0;

(stub.get()->*method)(&cntl, &req, res, nullptr);

// Record QPS statistics for all RPCs sent to MS (success or failure)

record_rpc_qps(rpc, rate_limit_ctx);

if (cntl.Failed()) [[unlikely]] {

error_msg = cntl.ErrorText();

error_code = cntl.ErrorCode();

proxy->set_unhealthy();

} else if (res->status().code() == MetaServiceCode::OK) {

return Status::OK();

} else if (res->status().code() == MetaServiceCode::INVALID_ARGUMENT) {

return Status::Error<ErrorCode::INVALID_ARGUMENT, false>("failed to {}: {}", op_name,

res->status().msg());

} else if (res->status().code() == MetaServiceCode::MS_TOO_BUSY) {

// MS_BUSY should also be retried

if (rate_limit_ctx.backpressure_handler) {

rate_limit_ctx.backpressure_handler->on_ms_busy();

}

} else if (res->status().code() != MetaServiceCode::KV_TXN_CONFLICT) {

return Status::Error<ErrorCode::INTERNAL_ERROR, false>("failed to {}: {}", op_name,

res->status().msg());

} else {

error_msg = res->status().msg();

}

if (error_code == brpc::ERPCTIMEDOUT) {

g_cloud_meta_mgr_rpc_timeout_count << 1;

}

++retry_times;

if (retry_times > config::meta_service_rpc_retry_times ||

(retry_times > config::meta_service_rpc_timeout_retry_times &&

error_code == brpc::ERPCTIMEDOUT) ||

(retry_times > config::meta_service_conflict_error_retry_times &&

res->status().code() == MetaServiceCode::KV_TXN_CONFLICT)) {

break;

}

duration_ms = retry_times <= 100 ? u(rng) : u2(rng);

LOG(WARNING) << "failed to " << op_name << debug_info(req) << " retry_times=" << retry_times

<< " sleep=" << duration_ms << "ms : " << cntl.ErrorText();

bthread_usleep(duration_ms * 1000);

}

return Status::RpcError("failed to {}: rpc timeout, last msg={}", op_name, error_msg);

}

} // namespace

Status CloudMetaMgr::get_tablet_meta(int64_t tablet_id, TabletMetaSharedPtr* tablet_meta) {

VLOG_DEBUG << "send GetTabletRequest, tablet_id: " << tablet_id;

TEST_SYNC_POINT_RETURN_WITH_VALUE("CloudMetaMgr::get_tablet_meta", Status::OK(), tablet_id,

tablet_meta);

GetTabletRequest req;

GetTabletResponse resp;

req.set_cloud_unique_id(config::cloud_unique_id);

req.set_tablet_id(tablet_id);

Status st =

retry_rpc(MetaServiceRPC::GET_TABLET_META, req, &resp, &MetaService_Stub::get_tablet,

{

.host_limiters = host_level_ms_rpc_rate_limiters_,

.backpressure_handler = ms_backpressure_handler_,

});

if (!st.ok()) {

if (resp.status().code() == MetaServiceCode::TABLET_NOT_FOUND) {

return Status::NotFound("failed to get tablet meta: {}", resp.status().msg());

}

return st;

}

*tablet_meta = std::make_shared<TabletMeta>();

(*tablet_meta)

->init_from_pb(cloud_tablet_meta_to_doris(std::move(*resp.mutable_tablet_meta())));

VLOG_DEBUG << "get tablet meta, tablet_id: " << (*tablet_meta)->tablet_id();

return Status::OK();

}

Status CloudMetaMgr::sync_tablet_rowsets(CloudTablet* tablet, const SyncOptions& options,

SyncRowsetStats* sync_stats) {

std::unique_lock lock {tablet->get_sync_meta_lock()};

return sync_tablet_rowsets_unlocked(tablet, lock, options, sync_stats);

}

Status CloudMetaMgr::_log_mow_delete_bitmap(CloudTablet* tablet, GetRowsetResponse& resp,

DeleteBitmap& delete_bitmap, int64_t old_max_version,

bool full_sync, int32_t read_version) {

if (config::enable_mow_verbose_log && !resp.rowset_meta().empty() &&

delete_bitmap.cardinality() > 0) {

int64_t tablet_id = tablet->tablet_id();

std::vector<std::string> new_rowset_msgs;

std::vector<std::string> old_rowset_msgs;

std::unordered_set<RowsetId> new_rowset_ids;

int64_t new_max_version = resp.rowset_meta().rbegin()->end_version();

for (const auto& rs : resp.rowset_meta()) {

RowsetId rowset_id;

rowset_id.init(rs.rowset_id_v2());

new_rowset_ids.insert(rowset_id);

DeleteBitmap rowset_dbm(tablet_id);

delete_bitmap.subset({rowset_id, 0, 0},

{rowset_id, std::numeric_limits<DeleteBitmap::SegmentId>::max(),

std::numeric_limits<DeleteBitmap::Version>::max()},

&rowset_dbm);

size_t cardinality = rowset_dbm.cardinality();

size_t count = rowset_dbm.get_delete_bitmap_count();

if (cardinality > 0) {

new_rowset_msgs.push_back(fmt::format("({}[{}-{}],{},{})", rs.rowset_id_v2(),

rs.start_version(), rs.end_version(), count,

cardinality));

}

}

if (old_max_version > 0) {

std::vector<RowsetSharedPtr> old_rowsets;

RowsetIdUnorderedSet old_rowset_ids;

{

std::lock_guard rlock(tablet->get_header_lock());

RETURN_IF_ERROR(tablet->get_all_rs_id_unlocked(old_max_version, &old_rowset_ids));

old_rowsets = tablet->get_rowset_by_ids(&old_rowset_ids);

}

for (const auto& rs : old_rowsets) {

if (!new_rowset_ids.contains(rs->rowset_id())) {

DeleteBitmap rowset_dbm(tablet_id);

delete_bitmap.subset(

{rs->rowset_id(), 0, 0},

{rs->rowset_id(), std::numeric_limits<DeleteBitmap::SegmentId>::max(),

std::numeric_limits<DeleteBitmap::Version>::max()},

&rowset_dbm);

size_t cardinality = rowset_dbm.cardinality();

size_t count = rowset_dbm.get_delete_bitmap_count();

if (cardinality > 0) {

old_rowset_msgs.push_back(

fmt::format("({}{},{},{})", rs->rowset_id().to_string(),

rs->version().to_string(), count, cardinality));

}

}

}

}

std::string tablet_info = fmt::format(

"tablet_id={} table_id={} index_id={} partition_id={}", tablet->tablet_id(),

tablet->table_id(), tablet->index_id(), tablet->partition_id());

LOG_INFO("[verbose] sync tablet delete bitmap " + tablet_info)

.tag("full_sync", full_sync)

.tag("read_version", read_version)

.tag("old_max_version", old_max_version)

.tag("new_max_version", new_max_version)

.tag("cumu_compaction_cnt", resp.stats().cumulative_compaction_cnt())

.tag("base_compaction_cnt", resp.stats().base_compaction_cnt())

.tag("cumu_point", resp.stats().cumulative_point())

.tag("rowset_num", resp.rowset_meta().size())

.tag("delete_bitmap_cardinality", delete_bitmap.cardinality())

.tag("old_rowsets(rowset,count,cardinality)",

fmt::format("[{}]", fmt::join(old_rowset_msgs, ", ")))

.tag("new_rowsets(rowset,count,cardinality)",

fmt::format("[{}]", fmt::join(new_rowset_msgs, ", ")));

}

return Status::OK();

}

Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet,

std::unique_lock<bthread::Mutex>& lock,

const SyncOptions& options,

SyncRowsetStats* sync_stats) {

using namespace std::chrono;

TEST_SYNC_POINT_RETURN_WITH_VALUE("CloudMetaMgr::sync_tablet_rowsets", Status::OK(), tablet);

DBUG_EXECUTE_IF("CloudMetaMgr::sync_tablet_rowsets.before.inject_error", {

auto target_tablet_id = dp->param<int64_t>("tablet_id", -1);

auto target_table_id = dp->param<int64_t>("table_id", -1);

if (target_tablet_id == tablet->tablet_id() || target_table_id == tablet->table_id()) {

return Status::InternalError(

"[sync_tablet_rowsets_unlocked] injected error for testing");

}

});

MetaServiceProxy* proxy;

RETURN_IF_ERROR(MetaServiceProxy::get_proxy(&proxy));

std::string tablet_info =

fmt::format("tablet_id={} table_id={} index_id={} partition_id={}", tablet->tablet_id(),

tablet->table_id(), tablet->index_id(), tablet->partition_id());

int tried = 0;

while (true) {

std::shared_ptr<MetaService_Stub> stub;

RETURN_IF_ERROR(proxy->get(&stub));

brpc::Controller cntl;

cntl.set_timeout_ms(config::meta_service_brpc_timeout_ms);

GetRowsetRequest req;

GetRowsetResponse resp;

int64_t tablet_id = tablet->tablet_id();

int64_t table_id = tablet->table_id();

int64_t index_id = tablet->index_id();

req.set_cloud_unique_id(config::cloud_unique_id);

auto* idx = req.mutable_idx();

idx->set_tablet_id(tablet_id);

idx->set_table_id(table_id);

idx->set_index_id(index_id);

idx->set_partition_id(tablet->partition_id());

{

auto lock_start = std::chrono::steady_clock::now();

std::shared_lock rlock(tablet->get_header_lock());

if (sync_stats) {

sync_stats->meta_lock_wait_ns +=

std::chrono::duration_cast<std::chrono::nanoseconds>(

std::chrono::steady_clock::now() - lock_start)

.count();

}

if (options.full_sync) {

req.set_start_version(0);

} else {

req.set_start_version(tablet->max_version_unlocked() + 1);

}

req.set_base_compaction_cnt(tablet->base_compaction_cnt());

req.set_cumulative_compaction_cnt(tablet->cumulative_compaction_cnt());

req.set_full_compaction_cnt(tablet->full_compaction_cnt());

req.set_cumulative_point(tablet->cumulative_layer_point());

}

req.set_end_version(-1);

VLOG_DEBUG << "send GetRowsetRequest: " << req.ShortDebugString();

// Host-level rate limiting for get_rowset

if (host_level_ms_rpc_rate_limiters_) {

host_level_ms_rpc_rate_limiters_->limit(MetaServiceRPC::GET_ROWSET);

}

auto start = std::chrono::steady_clock::now();

stub->get_rowset(&cntl, &req, &resp, nullptr);

auto end = std::chrono::steady_clock::now();

int64_t latency = cntl.latency_us();

_get_rowset_latency << latency;

int retry_times = config::meta_service_rpc_retry_times;

if (cntl.Failed()) {

proxy->set_unhealthy();

if (tried++ < retry_times) {

auto rng = make_random_engine();

std::uniform_int_distribution<uint32_t> u(20, 200);

std::uniform_int_distribution<uint32_t> u1(500, 1000);

uint32_t duration_ms = tried >= 100 ? u(rng) : u1(rng);

bthread_usleep(duration_ms * 1000);

LOG_INFO("failed to get rowset meta, " + tablet_info)

.tag("reason", cntl.ErrorText())

.tag("tried", tried)

.tag("sleep", duration_ms);

continue;

}

return Status::RpcError("failed to get rowset meta: {}", cntl.ErrorText());

}

if (resp.status().code() == MetaServiceCode::TABLET_NOT_FOUND) {

LOG(WARNING) << "failed to get rowset meta, err=" << resp.status().msg() << " "

<< tablet_info;

return Status::NotFound("failed to get rowset meta: {}, {}", resp.status().msg(),

tablet_info);

}

if (resp.status().code() == MetaServiceCode::MS_TOO_BUSY) {

// MS_BUSY should also be retried

if (ms_backpressure_handler_) {

ms_backpressure_handler_->on_ms_busy();

}

if (tried++ < retry_times) {

auto rng = make_random_engine();

std::uniform_int_distribution<uint32_t> u(20, 200);

std::uniform_int_distribution<uint32_t> u1(500, 1000);

uint32_t duration_ms = tried >= 100 ? u(rng) : u1(rng);

bthread_usleep(duration_ms * 1000);

LOG_INFO("meta service is too busy when getting rowset meta, " + tablet_info)

.tag("reason", resp.status().msg())

.tag("tried", tried)

.tag("sleep", duration_ms);

continue;

}

return Status::RpcError("failed to get rowset meta: {}", resp.status().msg());

}

if (resp.status().code() != MetaServiceCode::OK) {

LOG(WARNING) << " failed to get rowset meta, err=" << resp.status().msg() << " "

<< tablet_info;

return Status::InternalError("failed to get rowset meta: {}, {}", resp.status().msg(),

tablet_info);

}

if (latency > 100 * 1000) { // 100ms

LOG(INFO) << "finish get_rowset rpc. rowset_meta.size()=" << resp.rowset_meta().size()

<< ", latency=" << latency << "us"

<< " " << tablet_info;

} else {

LOG_EVERY_N(INFO, 100)

<< "finish get_rowset rpc. rowset_meta.size()=" << resp.rowset_meta().size()

<< ", latency=" << latency << "us"

<< " " << tablet_info;

}

int64_t now = duration_cast<seconds>(system_clock::now().time_since_epoch()).count();

tablet->last_sync_time_s = now;

if (sync_stats) {

sync_stats->get_remote_rowsets_rpc_ns +=

std::chrono::duration_cast<std::chrono::nanoseconds>(end - start).count();

sync_stats->get_remote_rowsets_num += resp.rowset_meta().size();

}

// If is mow, the tablet has no delete bitmap in base rowsets.

// So dont need to sync it.

if (options.sync_delete_bitmap && tablet->enable_unique_key_merge_on_write() &&

tablet->tablet_state() == TABLET_RUNNING) {

DBUG_EXECUTE_IF("CloudMetaMgr::sync_tablet_rowsets.sync_tablet_delete_bitmap.block",

DBUG_BLOCK);

DeleteBitmap delete_bitmap(tablet_id);

int64_t old_max_version = req.start_version() - 1;

auto read_version = config::delete_bitmap_store_read_version;

auto st = sync_tablet_delete_bitmap(tablet, old_max_version, resp.rowset_meta(),

resp.stats(), req.idx(), &delete_bitmap,

options.full_sync, sync_stats, read_version, false);

if (st.is<ErrorCode::ROWSETS_EXPIRED>() && tried++ < retry_times) {

LOG_INFO("rowset meta is expired, need to retry, " + tablet_info)

.tag("tried", tried)

.error(st);

continue;

}

if (!st.ok()) {

LOG_WARNING("failed to get delete bitmap, " + tablet_info).error(st);

return st;

}

tablet->tablet_meta()->delete_bitmap().merge(delete_bitmap);

RETURN_IF_ERROR(_log_mow_delete_bitmap(tablet, resp, delete_bitmap, old_max_version,

options.full_sync, read_version));

RETURN_IF_ERROR(

_check_delete_bitmap_v2_correctness(tablet, req, resp, old_max_version));

}

DBUG_EXECUTE_IF("CloudMetaMgr::sync_tablet_rowsets.before.modify_tablet_meta", {

auto target_tablet_id = dp->param<int64_t>("tablet_id", -1);

if (target_tablet_id == tablet->tablet_id()) {

DBUG_BLOCK

}

});

{

const auto& stats = resp.stats();

auto lock_start = std::chrono::steady_clock::now();

std::unique_lock wlock(tablet->get_header_lock());

if (sync_stats) {

sync_stats->meta_lock_wait_ns +=

std::chrono::duration_cast<std::chrono::nanoseconds>(

std::chrono::steady_clock::now() - lock_start)

.count();

}

// ATTN: we are facing following data race

//

// resp_base_compaction_cnt=0|base_compaction_cnt=0|resp_cumulative_compaction_cnt=0|cumulative_compaction_cnt=1|resp_max_version=11|max_version=8

//

// BE-compaction-thread meta-service BE-query-thread

// | | |

// local | commit cumu-compaction | |

// cc_cnt=0 | ---------------------------> | sync rowset (long rpc, local cc_cnt=0 ) | local

// | | <----------------------------------------- | cc_cnt=0

// | | -. |

// local | done cc_cnt=1 | \ |

// cc_cnt=1 | <--------------------------- | \ |

// | | \ returned with resp cc_cnt=0 (snapshot) |

// | | '------------------------------------> | local

// | | | cc_cnt=1

// | | |

// | | | CHECK FAIL

// | | | need retry

// To get rid of just retry syncing tablet

if (stats.base_compaction_cnt() < tablet->base_compaction_cnt() ||

stats.cumulative_compaction_cnt() < tablet->cumulative_compaction_cnt())

[[unlikely]] {

// stale request, ignore

LOG_WARNING("stale get rowset meta request " + tablet_info)

.tag("resp_base_compaction_cnt", stats.base_compaction_cnt())

.tag("base_compaction_cnt", tablet->base_compaction_cnt())

.tag("resp_cumulative_compaction_cnt", stats.cumulative_compaction_cnt())

.tag("cumulative_compaction_cnt", tablet->cumulative_compaction_cnt())

.tag("tried", tried);

if (tried++ < 10) continue;

return Status::OK();

}

std::vector<RowsetSharedPtr> rowsets;

rowsets.reserve(resp.rowset_meta().size());

for (const auto& cloud_rs_meta_pb : resp.rowset_meta()) {

VLOG_DEBUG << "get rowset meta, tablet_id=" << cloud_rs_meta_pb.tablet_id()

<< ", version=[" << cloud_rs_meta_pb.start_version() << '-'

<< cloud_rs_meta_pb.end_version() << ']';

auto existed_rowset = tablet->get_rowset_by_version(

{cloud_rs_meta_pb.start_version(), cloud_rs_meta_pb.end_version()});

if (existed_rowset &&

existed_rowset->rowset_id().to_string() == cloud_rs_meta_pb.rowset_id_v2()) {

continue; // Same rowset, skip it

}

RowsetMetaPB meta_pb = cloud_rowset_meta_to_doris(cloud_rs_meta_pb);

auto rs_meta = std::make_shared<RowsetMeta>();

rs_meta->init_from_pb(meta_pb);

RowsetSharedPtr rowset;

// schema is nullptr implies using RowsetMeta.tablet_schema

Status s = RowsetFactory::create_rowset(nullptr, "", rs_meta, &rowset);

if (!s.ok()) {

LOG_WARNING("create rowset").tag("status", s);

return s;

}

rowsets.push_back(std::move(rowset));

}

if (!rowsets.empty()) {

// `rowsets.empty()` could happen after doing EMPTY_CUMULATIVE compaction. e.g.:

// BE has [0-1][2-11][12-12], [12-12] is delete predicate, cp is 2;

// after doing EMPTY_CUMULATIVE compaction, MS cp is 13, get_rowset will return [2-11][12-12].

bool version_overlap =

tablet->max_version_unlocked() >= rowsets.front()->start_version();

tablet->add_rowsets(std::move(rowsets), version_overlap, wlock,

options.warmup_delta_data ||

config::enable_warmup_immediately_on_new_rowset);

}

// Fill version holes

int64_t partition_max_version =

resp.has_partition_max_version() ? resp.partition_max_version() : -1;

RETURN_IF_ERROR(fill_version_holes(tablet, partition_max_version, wlock));

tablet->last_base_compaction_success_time_ms = stats.last_base_compaction_time_ms();

tablet->last_cumu_compaction_success_time_ms = stats.last_cumu_compaction_time_ms();

tablet->set_base_compaction_cnt(stats.base_compaction_cnt());

tablet->set_cumulative_compaction_cnt(stats.cumulative_compaction_cnt());

tablet->set_full_compaction_cnt(stats.full_compaction_cnt());

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();

}

}

bool CloudMetaMgr::sync_tablet_delete_bitmap_by_cache(CloudTablet* tablet,

std::ranges::range auto&& rs_metas,

DeleteBitmap* delete_bitmap) {

std::set<int64_t> txn_processed;

for (auto& rs_meta : rs_metas) {

auto txn_id = rs_meta.txn_id();

if (txn_processed.find(txn_id) != txn_processed.end()) {

continue;

}

txn_processed.insert(txn_id);

DeleteBitmapPtr tmp_delete_bitmap;

std::shared_ptr<PublishStatus> publish_status =

std::make_shared<PublishStatus>(PublishStatus::INIT);

CloudStorageEngine& engine = ExecEnv::GetInstance()->storage_engine().to_cloud();

Status status = engine.txn_delete_bitmap_cache().get_delete_bitmap(

txn_id, tablet->tablet_id(), &tmp_delete_bitmap, nullptr, &publish_status);

// CloudMetaMgr::sync_tablet_delete_bitmap_by_cache() is called after we sync rowsets from meta services.

// If the control flows reaches here, it's gauranteed that the rowsets is commited in meta services, so we can

// use the delete bitmap from cache directly if *publish_status == PublishStatus::SUCCEED without checking other

// stats(version or compaction stats)

if (status.ok() && *publish_status == PublishStatus::SUCCEED) {

// tmp_delete_bitmap contains sentinel marks, we should remove it before merge it to delete bitmap.

// Also, the version of delete bitmap key in tmp_delete_bitmap is DeleteBitmap::TEMP_VERSION_COMMON,

// we should replace it with the rowset's real version

DCHECK(rs_meta.start_version() == rs_meta.end_version());

int64_t rowset_version = rs_meta.start_version();

for (const auto& [delete_bitmap_key, bitmap_value] : tmp_delete_bitmap->delete_bitmap) {

// skip sentinel mark, which is used for delete bitmap correctness check

if (std::get<1>(delete_bitmap_key) != DeleteBitmap::INVALID_SEGMENT_ID) {

delete_bitmap->merge({std::get<0>(delete_bitmap_key),

std::get<1>(delete_bitmap_key), rowset_version},

bitmap_value);

}

}

engine.txn_delete_bitmap_cache().remove_unused_tablet_txn_info(txn_id,

tablet->tablet_id());

} else {

LOG_EVERY_N(INFO, 20)

<< "delete bitmap not found in cache, will sync rowset to get. tablet_id= "

<< tablet->tablet_id() << ", txn_id=" << txn_id << ", status=" << status;

return false;

}

}

return true;

}

Status CloudMetaMgr::_get_delete_bitmap_from_ms(GetDeleteBitmapRequest& req,

GetDeleteBitmapResponse& res) {

VLOG_DEBUG << "send GetDeleteBitmapRequest: " << req.ShortDebugString();

TEST_SYNC_POINT_CALLBACK("CloudMetaMgr::_get_delete_bitmap_from_ms", &req, &res);

auto st = retry_rpc(MetaServiceRPC::GET_DELETE_BITMAP, req, &res,

&MetaService_Stub::get_delete_bitmap,

{

.host_limiters = host_level_ms_rpc_rate_limiters_,

.backpressure_handler = ms_backpressure_handler_,

});

if (st.code() == ErrorCode::THRIFT_RPC_ERROR) {

return st;

}