be/src/cloud/cloud_internal_service.cpp

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

#include <bthread/countdown_event.h>

#include <algorithm>
#include <chrono>
#include <limits>
#include <list>
#include <memory>
#include <optional>
#include <thread>
#include <unordered_map>

#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet.h"
#include "cloud/cloud_tablet_mgr.h"
#include "cloud/cloud_warm_up_manager.h"
#include "cloud/cloud_warmup_metrics.h"
#include "cloud/config.h"
#include "io/cache/block_file_cache.h"
#include "io/cache/block_file_cache_downloader.h"
#include "io/cache/block_file_cache_factory.h"
#include "runtime/thread_context.h"
#include "runtime/workload_management/io_throttle.h"
#include "util/async_io.h"
#include "util/bvar_windowed_adder.h"
#include "util/debug_points.h"

namespace doris {
#include "common/compile_check_avoid_begin.h"

bvar::Adder<uint64_t> g_file_cache_get_by_peer_num("file_cache_get_by_peer_num");
bvar::Adder<uint64_t> g_file_cache_get_by_peer_blocks_num("file_cache_get_by_peer_blocks_num");
bvar::Adder<uint64_t> g_file_cache_get_by_peer_success_num("file_cache_get_by_peer_success_num");
bvar::Adder<uint64_t> g_file_cache_get_by_peer_failed_num("file_cache_get_by_peer_failed_num");
bvar::LatencyRecorder g_file_cache_get_by_peer_server_latency(
        "file_cache_get_by_peer_server_latency");
bvar::LatencyRecorder g_file_cache_get_by_peer_read_cache_file_latency(
        "file_cache_get_by_peer_read_cache_file_latency");
bvar::LatencyRecorder g_cloud_internal_service_get_file_cache_meta_by_tablet_id_latency(
        "cloud_internal_service_get_file_cache_meta_by_tablet_id_latency");

CloudInternalServiceImpl::CloudInternalServiceImpl(CloudStorageEngine& engine, ExecEnv* exec_env)
        : PInternalService(exec_env), _engine(engine) {}

CloudInternalServiceImpl::~CloudInternalServiceImpl() = default;

void CloudInternalServiceImpl::alter_vault_sync(google::protobuf::RpcController* controller,
                                                const doris::PAlterVaultSyncRequest* request,
                                                PAlterVaultSyncResponse* response,
                                                google::protobuf::Closure* done) {
    LOG(INFO) << "alter be to sync vault info from Meta Service";
    // If the vaults containing hdfs vault then it would try to create hdfs connection using jni
    // which would acuiqre one thread local jniEnv. But bthread context can't guarantee that the brpc
    // worker thread wouldn't do bthread switch between worker threads.
    bool ret = _heavy_work_pool.try_offer([this, done]() {
        brpc::ClosureGuard closure_guard(done);
        _engine.sync_storage_vault();
    });
    if (!ret) {
        brpc::ClosureGuard closure_guard(done);
        LOG(WARNING) << "fail to offer alter_vault_sync request to the work pool, pool="
                     << _heavy_work_pool.get_info();
    }
}

FileCacheType cache_type_to_pb(io::FileCacheType type) {
    switch (type) {
    case io::FileCacheType::TTL:
        return FileCacheType::TTL;
    case io::FileCacheType::INDEX:
        return FileCacheType::INDEX;
    case io::FileCacheType::NORMAL:
        return FileCacheType::NORMAL;
    default:
        DCHECK(false);
    }
    return FileCacheType::NORMAL;
}

static int64_t current_unix_time_us() {
    return std::chrono::duration_cast<std::chrono::microseconds>(
                   std::chrono::system_clock::now().time_since_epoch())
            .count();
}

static std::optional<int64_t> warm_up_rowset_cross_host_latency_us(int64_t start_unix_ts_us,
                                                                   int64_t end_unix_ts_us) {
    // The start timestamp is generated by the caller BE. Mixed-version callers may omit it, and
    // system clocks across BEs are not guaranteed to be ordered.
    if (start_unix_ts_us <= 0 || end_unix_ts_us < start_unix_ts_us) {
        return std::nullopt;
    }
    return end_unix_ts_us - start_unix_ts_us;
}

static void add_file_cache_block_meta_to_response(
        PGetFileCacheMetaResponse* resp, int64_t tablet_id, const std::string& rowset_id,
        int32_t segment_id, const std::string& file_name,
        const std::tuple<int64_t, int64_t, io::FileCacheType, int64_t>& tuple,
        const RowsetSharedPtr& rowset, bool is_index) {
    FileCacheBlockMeta* meta = resp->add_file_cache_block_metas();
    meta->set_tablet_id(tablet_id);
    meta->set_rowset_id(rowset_id);
    meta->set_segment_id(segment_id);
    meta->set_file_name(file_name);

    if (!is_index) {
        // .dat
        meta->set_file_size(rowset->rowset_meta()->segment_file_size(segment_id));
        meta->set_file_type(doris::FileType::SEGMENT_FILE);
    } else {
        // .idx
        const auto& idx_file_info = rowset->rowset_meta()->inverted_index_file_info(segment_id);
        meta->set_file_size(idx_file_info.has_index_size() ? idx_file_info.index_size() : -1);
        meta->set_file_type(doris::FileType::INVERTED_INDEX_FILE);
    }

    meta->set_offset(std::get<0>(tuple));
    meta->set_size(std::get<1>(tuple));
    meta->set_cache_type(cache_type_to_pb(std::get<2>(tuple)));
    meta->set_expiration_time(std::get<3>(tuple));
}

static void process_segment_file_cache_meta(PGetFileCacheMetaResponse* resp,
                                            const RowsetSharedPtr& rowset, int64_t tablet_id,
                                            const std::string& rowset_id, int32_t segment_id,
                                            bool is_index) {
    const char* extension = is_index ? ".idx" : ".dat";
    std::string file_name = fmt::format("{}_{}{}", rowset_id, segment_id, extension);
    auto cache_key = io::BlockFileCache::hash(file_name);
    auto* cache = io::FileCacheFactory::instance()->get_by_path(cache_key);
    if (!cache) return;
    auto segments_meta = cache->get_hot_blocks_meta(cache_key);
    for (const auto& tuple : segments_meta) {
        add_file_cache_block_meta_to_response(resp, tablet_id, rowset_id, segment_id, file_name,
                                              tuple, rowset, is_index);
    }
}

void CloudInternalServiceImpl::get_file_cache_meta_by_tablet_id(
        google::protobuf::RpcController* controller [[maybe_unused]],
        const PGetFileCacheMetaRequest* request, PGetFileCacheMetaResponse* response,
        google::protobuf::Closure* done) {
    brpc::ClosureGuard closure_guard(done);
    if (!config::enable_file_cache) {
        LOG_WARNING("try to access tablet file cache meta, but file cache not enabled");
        return;
    }
    auto begin_ts = std::chrono::duration_cast<std::chrono::microseconds>(
                            std::chrono::steady_clock::now().time_since_epoch())
                            .count();
    std::ostringstream tablet_ids_stream;
    int count = 0;
    for (const auto& tablet_id : request->tablet_ids()) {
        tablet_ids_stream << tablet_id << ", ";
        count++;
        if (count >= 10) {
            break;
        }
    }
    LOG(INFO) << "warm up get meta from this be, tablets num=" << request->tablet_ids().size()
              << ", first 10 tablet_ids=[ " << tablet_ids_stream.str() << " ]";
    for (const auto& tablet_id : request->tablet_ids()) {
        auto res = _engine.tablet_mgr().get_tablet(tablet_id);
        if (!res.has_value()) {
            LOG(ERROR) << "failed to get tablet: " << tablet_id
                       << " err msg: " << res.error().msg();
            continue;
        }
        CloudTabletSPtr tablet = std::move(res.value());
        auto st = tablet->sync_rowsets();
        if (!st) {
            // just log failed, try it best
            LOG(WARNING) << "failed to sync rowsets: " << tablet_id
                         << " err msg: " << st.to_string();
        }
        auto rowsets = tablet->get_snapshot_rowset();

        for (const RowsetSharedPtr& rowset : rowsets) {
            std::string rowset_id = rowset->rowset_id().to_string();
            for (int32_t segment_id = 0; segment_id < rowset->num_segments(); ++segment_id) {
                process_segment_file_cache_meta(response, rowset, tablet_id, rowset_id, segment_id,
                                                false);
                process_segment_file_cache_meta(response, rowset, tablet_id, rowset_id, segment_id,
                                                true);
            }
        }
    }
    auto end_ts = std::chrono::duration_cast<std::chrono::microseconds>(
                          std::chrono::steady_clock::now().time_since_epoch())
                          .count();
    g_cloud_internal_service_get_file_cache_meta_by_tablet_id_latency << (end_ts - begin_ts);
    LOG(INFO) << "get file cache meta by tablet ids = [ " << tablet_ids_stream.str() << " ] took "
              << end_ts - begin_ts << " us";
    VLOG_DEBUG << "get file cache meta by tablet id request=" << request->DebugString()
               << ", response=" << response->DebugString();
}

namespace {
// Helper functions for fetch_peer_data

Status handle_peer_file_range_request(const std::string& path, PFetchPeerDataResponse* response) {
    // Read specific range [file_offset, file_offset+file_size) across cached blocks
    auto datas = io::FileCacheFactory::instance()->get_cache_data_by_path(path);
    for (auto& cb : datas) {
        *(response->add_datas()) = std::move(cb);
    }
    return Status::OK();
}

void set_error_response(PFetchPeerDataResponse* response, const std::string& error_msg) {
    response->mutable_status()->add_error_msgs(error_msg);
    response->mutable_status()->set_status_code(TStatusCode::INTERNAL_ERROR);
}

Status read_file_block(const std::shared_ptr<io::FileBlock>& file_block, size_t file_size,
                       doris::CacheBlockPB* output) {
    std::string data;
    // ATTN: calculate the rightmost boundary value of the block, due to inaccurate current block meta information.
    // see CachedRemoteFileReader::read_at_impl for more details.
    // Ensure file_size >= file_block->offset() to avoid underflow
    if (file_size < file_block->offset()) {
        LOG(WARNING) << "file_size (" << file_size << ") < file_block->offset("
                     << file_block->offset() << ")";
        return Status::InternalError<false>("file_size less than block offset");
    }
    size_t read_size = std::min(static_cast<size_t>(file_size - file_block->offset()),
                                file_block->range().size());
    data.resize(read_size);

    auto begin_read_file_ts = std::chrono::duration_cast<std::chrono::microseconds>(
                                      std::chrono::steady_clock::now().time_since_epoch())
                                      .count();

    SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->s3_file_buffer_tracker());
    Slice slice(data.data(), data.size());
    Status read_st = file_block->read(slice, /*read_offset=*/0);

    auto end_read_file_ts = std::chrono::duration_cast<std::chrono::microseconds>(
                                    std::chrono::steady_clock::now().time_since_epoch())
                                    .count();
    g_file_cache_get_by_peer_read_cache_file_latency << (end_read_file_ts - begin_read_file_ts);

    if (read_st.ok()) {
        output->set_block_offset(static_cast<int64_t>(file_block->offset()));
        output->set_block_size(static_cast<int64_t>(read_size));
        output->set_data(std::move(data));
        return Status::OK();
    } else {
        g_file_cache_get_by_peer_failed_num << 1;
        LOG(WARNING) << "read cache block failed: " << read_st;
        return read_st;
    }
}

Status handle_peer_file_cache_block_request(const PFetchPeerDataRequest* request,
                                            PFetchPeerDataResponse* response) {
    const auto& path = request->path();
    auto hash = io::BlockFileCache::hash(path);
    auto* cache = io::FileCacheFactory::instance()->get_by_path(hash);
    if (cache == nullptr) {
        g_file_cache_get_by_peer_failed_num << 1;
        set_error_response(response, "can't get file cache instance");
        return Status::InternalError<false>("can't get file cache instance");
    }

    io::CacheContext ctx {};
    io::ReadStatistics local_stats;
    ctx.stats = &local_stats;

    for (const auto& cb_req : request->cache_req()) {
        size_t offset = static_cast<size_t>(std::max<int64_t>(0, cb_req.block_offset()));
        size_t size = static_cast<size_t>(std::max<int64_t>(0, cb_req.block_size()));
        auto holder = cache->get_or_set(hash, offset, size, ctx);

        for (auto& fb : holder.file_blocks) {
            if (fb->state() != io::FileBlock::State::DOWNLOADED) {
                g_file_cache_get_by_peer_failed_num << 1;
                LOG(WARNING) << "read cache block failed, state=" << fb->state();
                set_error_response(response, "read cache file error");
                return Status::InternalError<false>("cache block not downloaded");
            }

            g_file_cache_get_by_peer_blocks_num << 1;
            doris::CacheBlockPB* out = response->add_datas();
            Status read_status = read_file_block(fb, request->file_size(), out);
            if (!read_status.ok()) {
                set_error_response(response, "read cache file error");
                return read_status;
            }
        }
    }

    return Status::OK();
}
} // namespace

void CloudInternalServiceImpl::fetch_peer_data(google::protobuf::RpcController* controller
                                               [[maybe_unused]],
                                               const PFetchPeerDataRequest* request,
                                               PFetchPeerDataResponse* response,
                                               google::protobuf::Closure* done) {
    bool ret = _heavy_work_pool.try_offer([request, response, done]() {
        brpc::ClosureGuard closure_guard(done);
        g_file_cache_get_by_peer_num << 1;

        if (!config::enable_file_cache) {
            LOG_WARNING("try to access file cache data, but file cache not enabled");
            return;
        }

        auto begin_ts = std::chrono::duration_cast<std::chrono::microseconds>(
                                std::chrono::steady_clock::now().time_since_epoch())
                                .count();

        const auto type = request->type();
        const auto& path = request->path();
        response->mutable_status()->set_status_code(TStatusCode::OK);

        Status status = Status::OK();
        if (type == PFetchPeerDataRequest_Type_PEER_FILE_RANGE) {
            status = handle_peer_file_range_request(path, response);
        } else if (type == PFetchPeerDataRequest_Type_PEER_FILE_CACHE_BLOCK) {
            status = handle_peer_file_cache_block_request(request, response);
        }

        if (!status.ok()) {
            LOG(WARNING) << "fetch peer data failed: " << status.to_string();
            set_error_response(response, status.to_string());
        }

        DBUG_EXECUTE_IF("CloudInternalServiceImpl::fetch_peer_data_slower", {
            int st_us = dp->param<int>("sleep", 1000);
            LOG_WARNING("CloudInternalServiceImpl::fetch_peer_data_slower").tag("sleep", st_us);
            bthread_usleep(st_us);
        });

        auto end_ts = std::chrono::duration_cast<std::chrono::microseconds>(
                              std::chrono::steady_clock::now().time_since_epoch())
                              .count();
        g_file_cache_get_by_peer_server_latency << (end_ts - begin_ts);
        g_file_cache_get_by_peer_success_num << 1;

        VLOG_DEBUG << "fetch cache request=" << request->DebugString()
                   << ", response=" << response->DebugString();
    });

    if (!ret) {
        brpc::ClosureGuard closure_guard(done);
        LOG(WARNING) << "fail to offer fetch peer data request to the work pool, pool="
                     << _heavy_work_pool.get_info();
    }
}

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_submitted_segment_num(
        "file_cache_event_driven_warm_up_submitted_segment_num");
bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_finished_segment_num(
        "file_cache_event_driven_warm_up_finished_segment_num");
bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_failed_segment_num(
        "file_cache_event_driven_warm_up_failed_segment_num");
bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_submitted_segment_size(
        "file_cache_event_driven_warm_up_submitted_segment_size");
bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_finished_segment_size(
        "file_cache_event_driven_warm_up_finished_segment_size");
bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_failed_segment_size(
        "file_cache_event_driven_warm_up_failed_segment_size");
bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_submitted_index_num(
        "file_cache_event_driven_warm_up_submitted_index_num");
bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_finished_index_num(
        "file_cache_event_driven_warm_up_finished_index_num");
bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_failed_index_num(
        "file_cache_event_driven_warm_up_failed_index_num");
bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_submitted_index_size(
        "file_cache_event_driven_warm_up_submitted_index_size");
bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_finished_index_size(
        "file_cache_event_driven_warm_up_finished_index_size");
bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_failed_index_size(
        "file_cache_event_driven_warm_up_failed_index_size");
bvar::Status<int64_t> g_file_cache_warm_up_rowset_last_handle_unix_ts(
        "file_cache_warm_up_rowset_last_handle_unix_ts", 0);
bvar::Status<int64_t> g_file_cache_warm_up_rowset_last_finish_unix_ts(
        "file_cache_warm_up_rowset_last_finish_unix_ts", 0);
bvar::LatencyRecorder g_file_cache_warm_up_rowset_latency("file_cache_warm_up_rowset_latency");
bvar::LatencyRecorder g_file_cache_warm_up_rowset_request_to_handle_latency(
        "file_cache_warm_up_rowset_request_to_handle_latency");
bvar::LatencyRecorder g_file_cache_warm_up_rowset_handle_to_finish_latency(
        "file_cache_warm_up_rowset_handle_to_finish_latency");
bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_slow_count(
        "file_cache_warm_up_rowset_slow_count");
bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_request_to_handle_slow_count(
        "file_cache_warm_up_rowset_request_to_handle_slow_count");
bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_handle_to_finish_slow_count(
        "file_cache_warm_up_rowset_handle_to_finish_slow_count");
bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_wait_for_compaction_num(
        "file_cache_warm_up_rowset_wait_for_compaction_num");
bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_wait_for_compaction_timeout_num(
        "file_cache_warm_up_rowset_wait_for_compaction_timeout_num");

// Per-job windowed metrics for target BE
// bvar::Window enforces MAX_SECONDS_LIMIT = 3600, so the longest window is 1h.
static constexpr int WINDOW_5M = 300;
static constexpr int WINDOW_30M = 1800;
static constexpr int WINDOW_1H = 3600;

MBvarWindowedAdder g_warmup_ed_finish_segment_num("warmup_ed_finish_segment_num", {"job_id"},
                                                  {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);
MBvarWindowedAdder g_warmup_ed_finish_segment_size("warmup_ed_finish_segment_size", {"job_id"},
                                                   {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);
MBvarWindowedAdder g_warmup_ed_finish_index_num("warmup_ed_finish_index_num", {"job_id"},
                                                {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);
MBvarWindowedAdder g_warmup_ed_finish_index_size("warmup_ed_finish_index_size", {"job_id"},
                                                 {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);
MBvarWindowedAdder g_warmup_ed_fail_segment_num("warmup_ed_fail_segment_num", {"job_id"},
                                                {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);
MBvarWindowedAdder g_warmup_ed_fail_segment_size("warmup_ed_fail_segment_size", {"job_id"},
                                                 {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);
MBvarWindowedAdder g_warmup_ed_fail_index_num("warmup_ed_fail_index_num", {"job_id"},
                                              {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);
MBvarWindowedAdder g_warmup_ed_fail_index_size("warmup_ed_fail_index_size", {"job_id"},
                                               {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);
bvar::MultiDimension<bvar::Status<int64_t>> g_warmup_ed_last_finish_ts({"job_id"});

void update_warmup_ed_last_finish_ts(const std::string& job_id_str) {
    auto* finish_ts = g_warmup_ed_last_finish_ts.get_stats(std::list<std::string> {job_id_str});
    if (finish_ts) {
        finish_ts->set_value(std::chrono::duration_cast<std::chrono::milliseconds>(
                                     std::chrono::system_clock::now().time_since_epoch())
                                     .count());
    }
}

void record_warmup_ed_finish_segment(const std::string& job_id_str, int64_t segment_size) {
    g_warmup_ed_finish_segment_num.put({job_id_str}, 1);
    g_warmup_ed_finish_segment_size.put({job_id_str}, segment_size);
    update_warmup_ed_last_finish_ts(job_id_str);
}

void record_warmup_ed_finish_index(const std::string& job_id_str, int64_t idx_size) {
    g_warmup_ed_finish_index_num.put({job_id_str}, 1);
    g_warmup_ed_finish_index_size.put({job_id_str}, idx_size);
    update_warmup_ed_last_finish_ts(job_id_str);
}

void record_warmup_ed_fail_segment(const std::string& job_id_str, int64_t segment_size) {
    g_warmup_ed_fail_segment_num.put({job_id_str}, 1);
    g_warmup_ed_fail_segment_size.put({job_id_str}, segment_size);
}

void record_warmup_ed_fail_index(const std::string& job_id_str, int64_t idx_size) {
    g_warmup_ed_fail_index_num.put({job_id_str}, 1);
    g_warmup_ed_fail_index_size.put({job_id_str}, idx_size);
}

void record_warmup_ed_skipped_rowset_as_finished(RowsetMeta& rs_meta,
                                                 const std::string& job_id_str) {
    auto schema_ptr = rs_meta.tablet_schema();
    bool has_inverted_index = schema_ptr->has_inverted_index() || schema_ptr->has_ann_index();
    auto idx_version = schema_ptr->get_inverted_index_storage_format();
    for (int64_t segment_id = 0; segment_id < rs_meta.num_segments(); segment_id++) {
        record_warmup_ed_finish_segment(job_id_str, rs_meta.segment_file_size(segment_id));

        if (!has_inverted_index) {
            continue;
        }
        auto&& inverted_index_info = rs_meta.inverted_index_file_info(segment_id);
        if (idx_version == InvertedIndexStorageFormatPB::V1) {
            std::unordered_map<int64_t, int64_t> index_size_map;
            for (const auto& info : inverted_index_info.index_info()) {
                if (info.index_file_size() != -1) {
                    index_size_map[info.index_id()] = info.index_file_size();
                } else {
                    VLOG_DEBUG << "Invalid index_file_size for segment_id " << segment_id
                               << ", index_id " << info.index_id();
                }
            }
            for (const auto& index : schema_ptr->inverted_indexes()) {
                record_warmup_ed_finish_index(job_id_str, index_size_map[index->index_id()]);
            }
        } else { // InvertedIndexStorageFormatPB::V2
            int64_t idx_size = 0;
            if (inverted_index_info.has_index_size()) {
                idx_size = inverted_index_info.index_size();
            } else {
                VLOG_DEBUG << "index_size is not set for segment " << segment_id;
            }
            record_warmup_ed_finish_index(job_id_str, idx_size);
        }
    }
}

void handle_segment_download_done(Status st, int64_t tablet_id, const RowsetId& rowset_id,
                                  int64_t segment_id, std::shared_ptr<CloudTablet> tablet,
                                  std::shared_ptr<bthread::CountdownEvent> wait, Version version,
                                  int64_t segment_size, int64_t request_ts, int64_t handle_ts,
                                  std::string job_id_str, int64_t upstream_trigger_ts_ms) {
    DBUG_EXECUTE_IF("CloudInternalServiceImpl::warm_up_rowset.download_segment", {
        auto sleep_time = dp->param<int>("sleep", 3);
        LOG_INFO("[verbose] block download for rowset={}, version={}, sleep={}",
                 rowset_id.to_string(), version.to_string(), sleep_time);
        std::this_thread::sleep_for(std::chrono::seconds(sleep_time));
    });
    DBUG_EXECUTE_IF(
            "CloudInternalServiceImpl::warm_up_rowset.download_segment.inject_"
            "error",
            {
                st = Status::InternalError("injected error");
                LOG_INFO("[verbose] inject error, tablet={}, rowset={}, st={}", tablet_id,
                         rowset_id.to_string(), st.to_string());
            });
    if (st.ok()) {
        g_file_cache_event_driven_warm_up_finished_segment_num << 1;
        g_file_cache_event_driven_warm_up_finished_segment_size << segment_size;
        record_warmup_ed_finish_segment(job_id_str, segment_size);
        int64_t now_ts = current_unix_time_us();
        g_file_cache_warm_up_rowset_last_finish_unix_ts.set_value(now_ts);
        auto rowset_latency_us = warm_up_rowset_cross_host_latency_us(request_ts, now_ts);
        if (rowset_latency_us.has_value()) {
            g_file_cache_warm_up_rowset_latency << *rowset_latency_us;
        }
        g_file_cache_warm_up_rowset_handle_to_finish_latency << (now_ts - handle_ts);
        if (rowset_latency_us.has_value() &&
            *rowset_latency_us > config::warm_up_rowset_slow_log_ms * 1000) {
            g_file_cache_warm_up_rowset_slow_count << 1;
            LOG(INFO) << "warm up rowset took " << *rowset_latency_us
                      << " us, tablet_id: " << tablet_id << ", rowset_id: " << rowset_id.to_string()
                      << ", segment_id: " << segment_id;
        }
        if (now_ts - handle_ts > config::warm_up_rowset_slow_log_ms * 1000) {
            g_file_cache_warm_up_rowset_handle_to_finish_slow_count << 1;
            LOG(INFO) << "warm up rowset (handle to finish) took " << now_ts - handle_ts
                      << " us, tablet_id: " << tablet_id << ", rowset_id: " << rowset_id.to_string()
                      << ", segment_id: " << segment_id;
        }
    } else {
        g_file_cache_event_driven_warm_up_failed_segment_num << 1;
        g_file_cache_event_driven_warm_up_failed_segment_size << segment_size;
        record_warmup_ed_fail_segment(job_id_str, segment_size);
        LOG(WARNING) << "download segment failed, tablet_id: " << tablet_id
                     << " rowset_id: " << rowset_id.to_string() << ", error: " << st;
    }
    if (tablet->complete_rowset_segment_warmup(WarmUpTriggerSource::EVENT_DRIVEN, rowset_id, st, 1,
                                               0)
                .trigger_source == WarmUpTriggerSource::EVENT_DRIVEN) {
        VLOG_DEBUG << "warmup rowset " << version.to_string() << "(" << rowset_id.to_string()
                   << ") completed";
    }
    g_warmup_ed_downstream_progress_tracker.record_task_done(job_id_str, upstream_trigger_ts_ms);
    if (wait) {
        wait->signal();
    }
}

void handle_inverted_index_download_done(Status st, int64_t tablet_id, const RowsetId& rowset_id,
                                         int64_t segment_id, std::string index_path,
                                         std::shared_ptr<CloudTablet> tablet,
                                         std::shared_ptr<bthread::CountdownEvent> wait,
                                         Version version, uint64_t idx_size, int64_t request_ts,
                                         int64_t handle_ts, std::string job_id_str,
                                         int64_t upstream_trigger_ts_ms) {
    DBUG_EXECUTE_IF("CloudInternalServiceImpl::warm_up_rowset.download_inverted_idx", {
        auto sleep_time = dp->param<int>("sleep", 3);
        LOG_INFO(
                "[verbose] block download for rowset={}, inverted index "
                "file={}, sleep={}",
                rowset_id.to_string(), index_path, sleep_time);
        std::this_thread::sleep_for(std::chrono::seconds(sleep_time));
    });
    if (st.ok()) {
        g_file_cache_event_driven_warm_up_finished_index_num << 1;
        g_file_cache_event_driven_warm_up_finished_index_size << idx_size;
        record_warmup_ed_finish_index(job_id_str, static_cast<int64_t>(idx_size));
        int64_t now_ts = current_unix_time_us();
        g_file_cache_warm_up_rowset_last_finish_unix_ts.set_value(now_ts);
        auto rowset_latency_us = warm_up_rowset_cross_host_latency_us(request_ts, now_ts);
        if (rowset_latency_us.has_value()) {
            g_file_cache_warm_up_rowset_latency << *rowset_latency_us;
        }
        g_file_cache_warm_up_rowset_handle_to_finish_latency << (now_ts - handle_ts);
        if (rowset_latency_us.has_value() &&
            *rowset_latency_us > config::warm_up_rowset_slow_log_ms * 1000) {
            g_file_cache_warm_up_rowset_slow_count << 1;
            LOG(INFO) << "warm up rowset took " << *rowset_latency_us
                      << " us, tablet_id: " << tablet_id << ", rowset_id: " << rowset_id.to_string()
                      << ", segment_id: " << segment_id;
        }
        if (now_ts - handle_ts > config::warm_up_rowset_slow_log_ms * 1000) {
            g_file_cache_warm_up_rowset_handle_to_finish_slow_count << 1;
            LOG(INFO) << "warm up rowset (handle to finish) took " << now_ts - handle_ts
                      << " us, tablet_id: " << tablet_id << ", rowset_id: " << rowset_id.to_string()
                      << ", segment_id: " << segment_id;
        }
    } else {
        g_file_cache_event_driven_warm_up_failed_index_num << 1;
        g_file_cache_event_driven_warm_up_failed_index_size << idx_size;
        record_warmup_ed_fail_index(job_id_str, static_cast<int64_t>(idx_size));
        LOG(WARNING) << "download inverted index failed, tablet_id: " << tablet_id
                     << " rowset_id: " << rowset_id << ", error: " << st;
    }
    if (tablet->complete_rowset_segment_warmup(WarmUpTriggerSource::EVENT_DRIVEN, rowset_id, st, 0,
                                               1)
                .trigger_source == WarmUpTriggerSource::EVENT_DRIVEN) {
        VLOG_DEBUG << "warmup rowset " << version.to_string() << "(" << rowset_id.to_string()
                   << ") completed";
    }
    g_warmup_ed_downstream_progress_tracker.record_task_done(job_id_str, upstream_trigger_ts_ms);
    if (wait) {
        wait->signal();
    }
}

void CloudInternalServiceImpl::warm_up_rowset(google::protobuf::RpcController* controller
                                              [[maybe_unused]],
                                              const PWarmUpRowsetRequest* request,
                                              PWarmUpRowsetResponse* response,
                                              google::protobuf::Closure* done) {
    brpc::ClosureGuard closure_guard(done);
    std::shared_ptr<bthread::CountdownEvent> wait = nullptr;
    timespec due_time;
    if (request->has_sync_wait_timeout_ms() && request->sync_wait_timeout_ms() > 0) {
        g_file_cache_warm_up_rowset_wait_for_compaction_num << 1;
        wait = std::make_shared<bthread::CountdownEvent>(0);
        VLOG_DEBUG << "sync_wait_timeout: " << request->sync_wait_timeout_ms() << " ms";
        due_time = butil::milliseconds_from_now(request->sync_wait_timeout_ms());
    }

    // Extract job_id from request (0 if not set, for backward compatibility)
    std::string job_id_str = std::to_string(request->has_job_id() ? request->job_id() : 0);
    int64_t upstream_trigger_ts_ms =
            request->has_upstream_trigger_ts_ms() ? request->upstream_trigger_ts_ms() : 0;

    for (auto& rs_meta_pb : request->rowset_metas()) {
        RowsetMeta rs_meta;
        rs_meta.init_from_pb(rs_meta_pb);
        auto storage_resource = rs_meta.remote_storage_resource();
        if (!storage_resource) {
            LOG(WARNING) << storage_resource.error();
            continue;
        }
        int64_t tablet_id = rs_meta.tablet_id();
        auto rowset_id = rs_meta.rowset_id();
        bool local_only = !(request->has_skip_existence_check() && request->skip_existence_check());
        auto res = _engine.tablet_mgr().get_tablet(tablet_id, /* warmup_data = */ false,
                                                   /* sync_delete_bitmap = */ true,
                                                   /* sync_stats = */ nullptr,
                                                   /* local_only = */ local_only);
        if (!res.has_value()) {
            LOG_WARNING("Warm up error ").tag("tablet_id", tablet_id).error(res.error());
            if (res.error().msg().find("local_only=true") != std::string::npos ||
                res.error().msg().find("force_use_only_cached=true") != std::string::npos) {
                res.error().set_code(ErrorCode::TABLE_NOT_FOUND);
            }
            res.error().to_protobuf(response->mutable_status());
            continue;
        }
        auto tablet = res.value();
        auto tablet_meta = tablet->tablet_meta();

        int64_t handle_ts = current_unix_time_us();
        g_file_cache_warm_up_rowset_last_handle_unix_ts.set_value(handle_ts);
        int64_t request_ts = request->has_unix_ts_us() ? request->unix_ts_us() : 0;
        auto request_to_handle_latency_us =
                warm_up_rowset_cross_host_latency_us(request_ts, handle_ts);
        if (request_to_handle_latency_us.has_value()) {
            g_file_cache_warm_up_rowset_request_to_handle_latency << *request_to_handle_latency_us;
        }
        if (request_to_handle_latency_us.has_value() &&
            *request_to_handle_latency_us > config::warm_up_rowset_slow_log_ms * 1000) {
            g_file_cache_warm_up_rowset_request_to_handle_slow_count << 1;
            LOG(INFO) << "warm up rowset (request to handle) took " << *request_to_handle_latency_us
                      << " us, tablet_id: " << rs_meta.tablet_id()
                      << ", rowset_id: " << rowset_id.to_string();
        }
        int64_t expiration_time = tablet_meta->ttl_seconds();

        if (!tablet->add_rowset_warmup_state(rs_meta, WarmUpTriggerSource::EVENT_DRIVEN)) {
            LOG(INFO) << "found duplicate warmup task for rowset " << rowset_id.to_string()
                      << ", skip it";
            g_warmup_ed_downstream_progress_tracker.record_task_done(job_id_str,
                                                                     upstream_trigger_ts_ms);
            record_warmup_ed_skipped_rowset_as_finished(rs_meta, job_id_str);
            continue;
        }
        if (rs_meta.num_segments() == 0) {
            g_warmup_ed_downstream_progress_tracker.record_task_done(job_id_str,
                                                                     upstream_trigger_ts_ms);
        }

        for (int64_t segment_id = 0; segment_id < rs_meta.num_segments(); segment_id++) {
            if (!config::file_cache_enable_only_warm_up_idx) {
                auto segment_size = rs_meta.segment_file_size(segment_id);

                // Use rs_meta.fs() instead of storage_resource.value()->fs to support packed files.
                // PackedFileSystem wrapper in rs_meta.fs() handles the index_map lookup and
                // reads from the correct packed file.
                io::DownloadFileMeta download_meta {
                        .path = storage_resource.value()->remote_segment_path(rs_meta, segment_id),
                        .file_size = segment_size,
                        .offset = 0,
                        .download_size = segment_size,
                        .file_system = rs_meta.fs(),
                        .ctx = {.is_index_data = false,
                                .expiration_time = expiration_time,
                                .is_dryrun = config::enable_reader_dryrun_when_download_file_cache,
                                .is_warmup = true},
                        .download_done =
                                [=, version = rs_meta.version()](Status st) {
                                    handle_segment_download_done(
                                            st, tablet_id, rowset_id, segment_id, tablet, wait,
                                            version, segment_size, request_ts, handle_ts,
                                            job_id_str, upstream_trigger_ts_ms);
                                },
                        .tablet_id = tablet_id};

                g_file_cache_event_driven_warm_up_submitted_segment_num << 1;
                g_file_cache_event_driven_warm_up_submitted_segment_size << segment_size;
                if (wait) {
                    wait->add_count();
                }
                g_warmup_ed_downstream_progress_tracker.record_task_submit(job_id_str,
                                                                           upstream_trigger_ts_ms);

                _engine.file_cache_block_downloader().submit_download_task(download_meta);
            }

            // Use rs_meta.fs() to support packed files for inverted index download.
            auto download_inverted_index = [&, tablet, job_id_str](std::string index_path,
                                                                   uint64_t idx_size) {
                io::DownloadFileMeta download_meta {
                        .path = io::Path(index_path),
                        .file_size = static_cast<int64_t>(idx_size),
                        .file_system = rs_meta.fs(),
                        .ctx = {.is_index_data = false, // DORIS-20877
                                .expiration_time = expiration_time,
                                .is_dryrun = config::enable_reader_dryrun_when_download_file_cache,
                                .is_warmup = true},
                        .download_done =
                                [=, version = rs_meta.version()](Status st) {
                                    handle_inverted_index_download_done(
                                            st, tablet_id, rowset_id, segment_id, index_path,
                                            tablet, wait, version, idx_size, request_ts, handle_ts,
                                            job_id_str, upstream_trigger_ts_ms);
                                },
                        .tablet_id = tablet_id,
                };
                g_file_cache_event_driven_warm_up_submitted_index_num << 1;
                g_file_cache_event_driven_warm_up_submitted_index_size << idx_size;
                tablet->update_rowset_warmup_state_inverted_idx_num(
                        WarmUpTriggerSource::EVENT_DRIVEN, rowset_id, 1);
                if (wait) {
                    wait->add_count();
                }
                g_warmup_ed_downstream_progress_tracker.record_task_submit(job_id_str,
                                                                           upstream_trigger_ts_ms);
                _engine.file_cache_block_downloader().submit_download_task(download_meta);
            };

            // inverted index
            auto schema_ptr = rs_meta.tablet_schema();
            auto idx_version = schema_ptr->get_inverted_index_storage_format();

            if (schema_ptr->has_inverted_index() || schema_ptr->has_ann_index()) {
                if (idx_version == InvertedIndexStorageFormatPB::V1) {
                    auto&& inverted_index_info = rs_meta.inverted_index_file_info(segment_id);
                    std::unordered_map<int64_t, int64_t> index_size_map;
                    for (const auto& info : inverted_index_info.index_info()) {
                        if (info.index_file_size() != -1) {
                            index_size_map[info.index_id()] = info.index_file_size();
                        } else {
                            VLOG_DEBUG << "Invalid index_file_size for segment_id " << segment_id
                                       << ", index_id " << info.index_id();
                        }
                    }
                    for (const auto& index : schema_ptr->inverted_indexes()) {
                        auto idx_path = storage_resource.value()->remote_idx_v1_path(
                                rs_meta, segment_id, index->index_id(), index->get_index_suffix());
                        download_inverted_index(idx_path, index_size_map[index->index_id()]);
                    }
                } else { // InvertedIndexStorageFormatPB::V2
                    auto&& inverted_index_info = rs_meta.inverted_index_file_info(segment_id);
                    int64_t idx_size = 0;
                    if (inverted_index_info.has_index_size()) {
                        idx_size = inverted_index_info.index_size();
                    } else {
                        VLOG_DEBUG << "index_size is not set for segment " << segment_id;
                    }
                    auto idx_path =
                            storage_resource.value()->remote_idx_v2_path(rs_meta, segment_id);
                    download_inverted_index(idx_path, idx_size);
                }
            }
        }
    }
    if (wait && wait->timed_wait(due_time)) {
        g_file_cache_warm_up_rowset_wait_for_compaction_timeout_num << 1;
        LOG_WARNING("the time spent warming up {} rowsets exceeded {} ms",
                    request->rowset_metas().size(), request->sync_wait_timeout_ms());
    }
}

bvar::Adder<uint64_t> g_file_cache_recycle_cache_finished_segment_num(
        "file_cache_recycle_cache_finished_segment_num");
bvar::Adder<uint64_t> g_file_cache_recycle_cache_finished_index_num(
        "file_cache_recycle_cache_finished_index_num");

void CloudInternalServiceImpl::recycle_cache(google::protobuf::RpcController* controller
                                             [[maybe_unused]],
                                             const PRecycleCacheRequest* request,
                                             PRecycleCacheResponse* response,
                                             google::protobuf::Closure* done) {
    brpc::ClosureGuard closure_guard(done);

    if (!config::enable_file_cache) {
        return;
    }
    for (const auto& meta : request->cache_metas()) {
        for (int64_t segment_id = 0; segment_id < meta.num_segments(); segment_id++) {
            auto file_key = Segment::file_cache_key(meta.rowset_id(), segment_id);
            auto* file_cache = io::FileCacheFactory::instance()->get_by_path(file_key);
            file_cache->remove_if_cached_async(file_key);
            g_file_cache_recycle_cache_finished_segment_num << 1;
        }

        // inverted index
        for (const auto& file_name : meta.index_file_names()) {
            auto file_key = io::BlockFileCache::hash(file_name);
            auto* file_cache = io::FileCacheFactory::instance()->get_by_path(file_key);
            file_cache->remove_if_cached_async(file_key);
            g_file_cache_recycle_cache_finished_index_num << 1;
        }
    }
}

#include "common/compile_check_avoid_end.h"
} // namespace doris

Coverage Report

Created: 2026-06-17 03:24