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

#include <brpc/callback.h>
#include <butil/endpoint.h>
#include <fmt/format.h>
#include <gen_cpp/data.pb.h>
#include <gen_cpp/internal_service.pb.h>
#include <gen_cpp/olap_file.pb.h>
#include <gen_cpp/types.pb.h>
#include <glog/logging.h>
#include <stddef.h>
#include <stdint.h>

#include <algorithm>
#include <cstdint>
#include <memory>
#include <ostream>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

#include "bthread/countdown_event.h"
#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet.h"
#include "cloud/cloud_tablet_mgr.h"
#include "cloud/config.h"
#include "common/config.h"
#include "common/consts.h"
#include "common/exception.h"
#include "exec/tablet_info.h" // DorisNodesInfo
#include "olap/olap_common.h"
#include "olap/rowset/beta_rowset.h"
#include "olap/storage_engine.h"
#include "olap/tablet_fwd.h"
#include "olap/tablet_manager.h"
#include "olap/tablet_schema.h"
#include "olap/utils.h"
#include "runtime/descriptors.h"
#include "runtime/exec_env.h"      // ExecEnv
#include "runtime/fragment_mgr.h"  // FragmentMgr
#include "runtime/runtime_state.h" // RuntimeState
#include "runtime/types.h"
#include "util/brpc_client_cache.h" // BrpcClientCache
#include "util/defer_op.h"
#include "vec/columns/column.h"
#include "vec/columns/column_nullable.h"
#include "vec/columns/column_string.h"
#include "vec/common/assert_cast.h"
#include "vec/common/string_ref.h"
#include "vec/core/block.h" // Block
#include "vec/data_types/data_type_factory.hpp"
#include "vec/data_types/data_type_struct.h"
#include "vec/data_types/serde/data_type_serde.h"
#include "vec/exec/format/orc/vorc_reader.h"
#include "vec/exec/format/parquet/vparquet_reader.h"
#include "vec/exec/scan/file_scanner.h"
#include "vec/functions/function_helpers.h"
#include "vec/jsonb/serialize.h"

namespace doris {

Status RowIDFetcher::init() {
    DorisNodesInfo nodes_info;
    nodes_info.setNodes(_fetch_option.t_fetch_opt.nodes_info);
    for (auto [node_id, node_info] : nodes_info.nodes_info()) {
        auto client = ExecEnv::GetInstance()->brpc_internal_client_cache()->get_client(
                node_info.host, node_info.brpc_port);
        if (!client) {
            LOG(WARNING) << "Get rpc stub failed, host=" << node_info.host
                         << ", port=" << node_info.brpc_port;
            return Status::InternalError("RowIDFetcher failed to init rpc client, host={}, port={}",
                                         node_info.host, node_info.brpc_port);
        }
        _stubs.push_back(client);
    }
    return Status::OK();
}

PMultiGetRequest RowIDFetcher::_init_fetch_request(const vectorized::ColumnString& row_locs) const {
    PMultiGetRequest mget_req;
    _fetch_option.desc->to_protobuf(mget_req.mutable_desc());
    for (SlotDescriptor* slot : _fetch_option.desc->slots()) {
        // ignore rowid
        if (slot->col_name() == BeConsts::ROWID_COL) {
            continue;
        }
        slot->to_protobuf(mget_req.add_slots());
    }
    for (size_t i = 0; i < row_locs.size(); ++i) {
        PRowLocation row_loc;
        StringRef row_id_rep = row_locs.get_data_at(i);
        // TODO: When transferring data between machines with different byte orders (endianness),
        // not performing proper handling may lead to issues in parsing and exchanging the data.
        auto location = reinterpret_cast<const GlobalRowLoacation*>(row_id_rep.data);
        row_loc.set_tablet_id(location->tablet_id);
        row_loc.set_rowset_id(location->row_location.rowset_id.to_string());
        row_loc.set_segment_id(location->row_location.segment_id);
        row_loc.set_ordinal_id(location->row_location.row_id);
        *mget_req.add_row_locs() = std::move(row_loc);
    }
    // Set column desc
    for (const TColumn& tcolumn : _fetch_option.t_fetch_opt.column_desc) {
        TabletColumn column(tcolumn);
        column.to_schema_pb(mget_req.add_column_desc());
    }
    PUniqueId& query_id = *mget_req.mutable_query_id();
    query_id.set_hi(_fetch_option.runtime_state->query_id().hi);
    query_id.set_lo(_fetch_option.runtime_state->query_id().lo);
    mget_req.set_be_exec_version(_fetch_option.runtime_state->be_exec_version());
    mget_req.set_fetch_row_store(_fetch_option.t_fetch_opt.fetch_row_store);
    return mget_req;
}

static void fetch_callback(bthread::CountdownEvent* counter) {
    Defer __defer([&] { counter->signal(); });
}

Status RowIDFetcher::_merge_rpc_results(const PMultiGetRequest& request,
                                        const std::vector<PMultiGetResponse>& rsps,
                                        const std::vector<brpc::Controller>& cntls,
                                        vectorized::Block* output_block,
                                        std::vector<PRowLocation>* rows_id) const {
    output_block->clear();
    for (const auto& cntl : cntls) {
        if (cntl.Failed()) {
            LOG(WARNING) << "Failed to fetch meet rpc error:" << cntl.ErrorText()
                         << ", host:" << cntl.remote_side();
            return Status::InternalError(cntl.ErrorText());
        }
    }
    vectorized::DataTypeSerDeSPtrs serdes;
    std::unordered_map<uint32_t, uint32_t> col_uid_to_idx;
    std::vector<std::string> default_values;
    default_values.resize(_fetch_option.desc->slots().size());
    auto merge_function = [&](const PMultiGetResponse& resp) {
        Status st(Status::create(resp.status()));
        if (!st.ok()) {
            LOG(WARNING) << "Failed to fetch " << st.to_string();
            return st;
        }
        for (const PRowLocation& row_id : resp.row_locs()) {
            rows_id->push_back(row_id);
        }
        // Merge binary rows
        if (request.fetch_row_store()) {
            CHECK(resp.row_locs().size() == resp.binary_row_data_size());
            if (output_block->is_empty_column()) {
                *output_block = vectorized::Block(_fetch_option.desc->slots(), 1);
            }
            if (serdes.empty() && col_uid_to_idx.empty()) {
                serdes = vectorized::create_data_type_serdes(_fetch_option.desc->slots());
                for (int i = 0; i < _fetch_option.desc->slots().size(); ++i) {
                    col_uid_to_idx[_fetch_option.desc->slots()[i]->col_unique_id()] = i;
                    default_values[i] = _fetch_option.desc->slots()[i]->col_default_value();
                }
            }
            for (int i = 0; i < resp.binary_row_data_size(); ++i) {
                RETURN_IF_ERROR(vectorized::JsonbSerializeUtil::jsonb_to_block(
                        serdes, resp.binary_row_data(i).data(), resp.binary_row_data(i).size(),
                        col_uid_to_idx, *output_block, default_values, {}));
            }
            return Status::OK();
        }
        // Merge partial blocks
        vectorized::Block partial_block;
        RETURN_IF_ERROR(partial_block.deserialize(resp.block()));
        if (partial_block.is_empty_column()) {
            return Status::OK();
        }
        CHECK(resp.row_locs().size() == partial_block.rows());
        if (output_block->is_empty_column()) {
            output_block->swap(partial_block);
        } else if (partial_block.columns() != output_block->columns()) {
            return Status::Error<ErrorCode::INTERNAL_ERROR>(
                    "Merge block not match, self:[{}], input:[{}], ", output_block->dump_types(),
                    partial_block.dump_types());
        } else {
            for (int i = 0; i < output_block->columns(); ++i) {
                output_block->get_by_position(i).column->assume_mutable()->insert_range_from(
                        *partial_block.get_by_position(i)
                                 .column->convert_to_full_column_if_const()
                                 .get(),
                        0, partial_block.rows());
            }
        }
        return Status::OK();
    };

    for (const auto& resp : rsps) {
        RETURN_IF_ERROR(merge_function(resp));
    }
    return Status::OK();
}

bool _has_char_type(const vectorized::DataTypePtr& type) {
    switch (type->get_primitive_type()) {
    case TYPE_CHAR: {
        return true;
    }
    case TYPE_ARRAY: {
        const auto* arr_type =
                assert_cast<const vectorized::DataTypeArray*>(remove_nullable(type).get());
        return _has_char_type(arr_type->get_nested_type());
    }
    case TYPE_MAP: {
        const auto* map_type =
                assert_cast<const vectorized::DataTypeMap*>(remove_nullable(type).get());
        return _has_char_type(map_type->get_key_type()) ||
               _has_char_type(map_type->get_value_type());
    }
    case TYPE_STRUCT: {
        const auto* struct_type =
                assert_cast<const vectorized::DataTypeStruct*>(remove_nullable(type).get());
        return std::any_of(
                struct_type->get_elements().begin(), struct_type->get_elements().end(),
                [&](const vectorized::DataTypePtr& dt) -> bool { return _has_char_type(dt); });
    }
    default:
        return false;
    }
}

Status RowIDFetcher::fetch(const vectorized::ColumnPtr& column_row_ids,
                           vectorized::Block* res_block) {
    CHECK(!_stubs.empty());
    PMultiGetRequest mget_req = _init_fetch_request(assert_cast<const vectorized::ColumnString&>(
            *vectorized::remove_nullable(column_row_ids).get()));
    std::vector<PMultiGetResponse> resps(_stubs.size());
    std::vector<brpc::Controller> cntls(_stubs.size());
    bthread::CountdownEvent counter(_stubs.size());
    for (size_t i = 0; i < _stubs.size(); ++i) {
        cntls[i].set_timeout_ms(config::fetch_rpc_timeout_seconds * 1000);
        auto callback = brpc::NewCallback(fetch_callback, &counter);
        _stubs[i]->multiget_data(&cntls[i], &mget_req, &resps[i], callback);
    }
    counter.wait();

    // Merge
    std::vector<PRowLocation> rows_locs;
    rows_locs.reserve(rows_locs.size());
    RETURN_IF_ERROR(_merge_rpc_results(mget_req, resps, cntls, res_block, &rows_locs));
    if (rows_locs.size() < column_row_ids->size()) {
        return Status::InternalError("Miss matched return row loc count {}, expected {}, input {}",
                                     rows_locs.size(), res_block->rows(), column_row_ids->size());
    }
    // Final sort by row_ids sequence, since row_ids is already sorted if need
    std::map<GlobalRowLoacation, size_t> positions;
    for (size_t i = 0; i < rows_locs.size(); ++i) {
        RowsetId rowset_id;
        rowset_id.init(rows_locs[i].rowset_id());
        GlobalRowLoacation grl(rows_locs[i].tablet_id(), rowset_id, rows_locs[i].segment_id(),
                               rows_locs[i].ordinal_id());
        positions[grl] = i;
    };
    // TODO remove this warning code
    if (positions.size() < rows_locs.size()) {
        LOG(WARNING) << "contains duplicated row entry";
    }
    vectorized::IColumn::Permutation permutation;
    permutation.reserve(column_row_ids->size());
    for (size_t i = 0; i < column_row_ids->size(); ++i) {
        auto location =
                reinterpret_cast<const GlobalRowLoacation*>(column_row_ids->get_data_at(i).data);
        permutation.push_back(positions[*location]);
    }
    for (size_t i = 0; i < res_block->columns(); ++i) {
        res_block->get_by_position(i).column =
                res_block->get_by_position(i).column->permute(permutation, permutation.size());
    }
    // Check row consistency
    RETURN_IF_CATCH_EXCEPTION(res_block->check_number_of_rows());
    // shrink for char type
    std::vector<size_t> char_type_idx;
    for (size_t i = 0; i < _fetch_option.desc->slots().size(); i++) {
        const auto& column_desc = _fetch_option.desc->slots()[i];
        const auto type = column_desc->type();
        if (_has_char_type(type)) {
            char_type_idx.push_back(i);
        }
    }
    res_block->shrink_char_type_column_suffix_zero(char_type_idx);
    VLOG_DEBUG << "dump block:" << res_block->dump_data(0, 10);
    return Status::OK();
}

struct IteratorKey {
    int64_t tablet_id;
    RowsetId rowset_id;
    uint64_t segment_id;
    int slot_id;

    // unordered map std::equal_to
    bool operator==(const IteratorKey& rhs) const {
        return tablet_id == rhs.tablet_id && rowset_id == rhs.rowset_id &&
               segment_id == rhs.segment_id && slot_id == rhs.slot_id;
    }
};

struct HashOfIteratorKey {
    size_t operator()(const IteratorKey& key) const {
        size_t seed = 0;
        seed = HashUtil::hash64(&key.tablet_id, sizeof(key.tablet_id), seed);
        seed = HashUtil::hash64(&key.rowset_id.hi, sizeof(key.rowset_id.hi), seed);
        seed = HashUtil::hash64(&key.rowset_id.mi, sizeof(key.rowset_id.mi), seed);
        seed = HashUtil::hash64(&key.rowset_id.lo, sizeof(key.rowset_id.lo), seed);
        seed = HashUtil::hash64(&key.segment_id, sizeof(key.segment_id), seed);
        seed = HashUtil::hash64(&key.slot_id, sizeof(key.slot_id), seed);
        return seed;
    }
};

struct IteratorItem {
    std::unique_ptr<ColumnIterator> iterator;
    // for holding the reference of segment to avoid use after release
    SegmentSharedPtr segment;
};

Status RowIdStorageReader::read_by_rowids(const PMultiGetRequest& request,
                                          PMultiGetResponse* response) {
    // read from storage engine row id by row id
    OlapReaderStatistics stats;
    vectorized::Block result_block;
    int64_t acquire_tablet_ms = 0;
    int64_t acquire_rowsets_ms = 0;
    int64_t acquire_segments_ms = 0;
    int64_t lookup_row_data_ms = 0;

    // init desc
    std::vector<SlotDescriptor> slots;
    slots.reserve(request.slots().size());
    for (const auto& pslot : request.slots()) {
        slots.push_back(SlotDescriptor(pslot));
    }

    // init read schema
    TabletSchema full_read_schema;
    for (const ColumnPB& column_pb : request.column_desc()) {
        full_read_schema.append_column(TabletColumn(column_pb));
    }

    std::unordered_map<IteratorKey, IteratorItem, HashOfIteratorKey> iterator_map;
    // read row by row
    for (size_t i = 0; i < request.row_locs_size(); ++i) {
        const auto& row_loc = request.row_locs(i);
        MonotonicStopWatch watch;
        watch.start();
        BaseTabletSPtr tablet = scope_timer_run(
                [&]() {
                    auto res = ExecEnv::get_tablet(row_loc.tablet_id(), nullptr, true);
                    return !res.has_value() ? nullptr
                                            : std::dynamic_pointer_cast<BaseTablet>(res.value());
                },
                &acquire_tablet_ms);
        RowsetId rowset_id;
        rowset_id.init(row_loc.rowset_id());
        if (!tablet) {
            continue;
        }
        // We ensured it's rowset is not released when init Tablet reader param, rowset->update_delayed_expired_timestamp();
        BetaRowsetSharedPtr rowset = std::static_pointer_cast<BetaRowset>(scope_timer_run(
                [&]() {
                    return ExecEnv::GetInstance()->storage_engine().get_quering_rowset(rowset_id);
                },
                &acquire_rowsets_ms));
        if (!rowset) {
            LOG(INFO) << "no such rowset " << rowset_id;
            continue;
        }
        size_t row_size = 0;
        Defer _defer([&]() {
            LOG_EVERY_N(INFO, 100)
                    << "multiget_data single_row, cost(us):" << watch.elapsed_time() / 1000
                    << ", row_size:" << row_size;
            *response->add_row_locs() = row_loc;
        });
        // TODO: supoort session variable enable_page_cache and disable_file_cache if necessary.
        SegmentCacheHandle segment_cache;
        RETURN_IF_ERROR(scope_timer_run(
                [&]() {
                    return SegmentLoader::instance()->load_segments(rowset, &segment_cache, true);
                },
                &acquire_segments_ms));
        // find segment
        auto it = std::find_if(segment_cache.get_segments().cbegin(),
                               segment_cache.get_segments().cend(),
                               [&row_loc](const segment_v2::SegmentSharedPtr& seg) {
                                   return seg->id() == row_loc.segment_id();
                               });
        if (it == segment_cache.get_segments().end()) {
            continue;
        }
        segment_v2::SegmentSharedPtr segment = *it;
        GlobalRowLoacation row_location(row_loc.tablet_id(), rowset->rowset_id(),
                                        row_loc.segment_id(), row_loc.ordinal_id());
        // fetch by row store, more effcient way
        if (request.fetch_row_store()) {
            CHECK(tablet->tablet_schema()->has_row_store_for_all_columns());
            RowLocation loc(rowset_id, segment->id(), row_loc.ordinal_id());
            string* value = response->add_binary_row_data();
            RETURN_IF_ERROR(scope_timer_run(
                    [&]() { return tablet->lookup_row_data({}, loc, rowset, stats, *value); },
                    &lookup_row_data_ms));
            row_size = value->size();
            continue;
        }

        // fetch by column store
        if (result_block.is_empty_column()) {
            result_block = vectorized::Block(slots, request.row_locs().size());
        }
        VLOG_DEBUG << "Read row location "
                   << fmt::format("{}, {}, {}, {}", row_location.tablet_id,
                                  row_location.row_location.rowset_id.to_string(),
                                  row_location.row_location.segment_id,
                                  row_location.row_location.row_id);
        for (int x = 0; x < slots.size(); ++x) {
            auto row_id = static_cast<segment_v2::rowid_t>(row_loc.ordinal_id());
            vectorized::MutableColumnPtr column =
                    result_block.get_by_position(x).column->assume_mutable();
            IteratorKey iterator_key {.tablet_id = tablet->tablet_id(),
                                      .rowset_id = rowset_id,
                                      .segment_id = row_loc.segment_id(),
                                      .slot_id = slots[x].id()};
            IteratorItem& iterator_item = iterator_map[iterator_key];
            if (iterator_item.segment == nullptr) {
                // hold the reference
                iterator_map[iterator_key].segment = segment;
            }
            segment = iterator_item.segment;
            RETURN_IF_ERROR(segment->seek_and_read_by_rowid(full_read_schema, &slots[x], row_id,
                                                            column, stats, iterator_item.iterator));
        }
    }
    // serialize block if not empty
    if (!result_block.is_empty_column()) {
        VLOG_DEBUG << "dump block:" << result_block.dump_data(0, 10)
                   << ", be_exec_version:" << request.be_exec_version();
        [[maybe_unused]] size_t compressed_size = 0;
        [[maybe_unused]] size_t uncompressed_size = 0;
        int be_exec_version = request.has_be_exec_version() ? request.be_exec_version() : 0;
        RETURN_IF_ERROR(result_block.serialize(be_exec_version, response->mutable_block(),
                                               &uncompressed_size, &compressed_size,
                                               segment_v2::CompressionTypePB::LZ4));
    }

    LOG(INFO) << "Query stats: "
              << fmt::format(
                         "hit_cached_pages:{}, total_pages_read:{}, compressed_bytes_read:{}, "
                         "io_latency:{}ns, "
                         "uncompressed_bytes_read:{},"
                         "bytes_read:{},"
                         "acquire_tablet_ms:{}, acquire_rowsets_ms:{}, acquire_segments_ms:{}, "
                         "lookup_row_data_ms:{}",
                         stats.cached_pages_num, stats.total_pages_num, stats.compressed_bytes_read,
                         stats.io_ns, stats.uncompressed_bytes_read, stats.bytes_read,
                         acquire_tablet_ms, acquire_rowsets_ms, acquire_segments_ms,
                         lookup_row_data_ms);
    return Status::OK();
}

Status RowIdStorageReader::read_by_rowids(const PMultiGetRequestV2& request,
                                          PMultiGetResponseV2* response) {
    if (request.request_block_descs_size()) {
        auto tquery_id = ((UniqueId)request.query_id()).to_thrift();
        std::vector<vectorized::Block> result_blocks(request.request_block_descs_size());

        OlapReaderStatistics stats;
        int64_t acquire_tablet_ms = 0;
        int64_t acquire_rowsets_ms = 0;
        int64_t acquire_segments_ms = 0;
        int64_t lookup_row_data_ms = 0;

        int64_t external_init_reader_ms = 0;
        int64_t external_get_block_ms = 0;

        // Add counters for different file mapping types
        std::unordered_map<FileMappingType, int64_t> file_type_counts;

        auto id_file_map =
                ExecEnv::GetInstance()->get_id_manager()->get_id_file_map(request.query_id());
        // if id_file_map is null, means the BE not have scan range, just return ok
        if (!id_file_map) {
            // padding empty block to response
            for (int i = 0; i < request.request_block_descs_size(); ++i) {
                response->add_blocks();
            }
            return Status::OK();
        }

        for (int i = 0; i < request.request_block_descs_size(); ++i) {
            const auto& request_block_desc = request.request_block_descs(i);
            if (request_block_desc.row_id_size() >= 1) {
                // Since this block belongs to the same table, we only need to take the first type for judgment.
                auto first_file_id = request_block_desc.file_id(0);
                auto first_file_mapping = id_file_map->get_file_mapping(first_file_id);
                if (!first_file_mapping) {
                    return Status::InternalError(
                            "Backend:{} file_mapping not found, query_id: {}, file_id: {}",
                            BackendOptions::get_localhost(), print_id(request.query_id()),
                            first_file_id);
                }
                file_type_counts[first_file_mapping->type] += request_block_desc.row_id_size();

                // prepare slots to build block
                std::vector<SlotDescriptor> slots;
                slots.reserve(request_block_desc.slots_size());
                for (const auto& pslot : request_block_desc.slots()) {
                    slots.push_back(SlotDescriptor(pslot));
                }
                // prepare block char vector shrink for char type
                std::vector<size_t> char_type_idx;
                for (int j = 0; j < slots.size(); ++j) {
                    auto slot = slots[j];
                    if (_has_char_type(slot.type())) {
                        char_type_idx.push_back(j);
                    }
                }

                if (first_file_mapping->type == FileMappingType::INTERNAL) {
                    RETURN_IF_ERROR(read_batch_doris_format_row(
                            request_block_desc, id_file_map, slots, tquery_id, result_blocks[i],
                            stats, &acquire_tablet_ms, &acquire_rowsets_ms, &acquire_segments_ms,
                            &lookup_row_data_ms));
                } else {
                    RETURN_IF_ERROR(read_batch_external_row(
                            request_block_desc, id_file_map, slots, first_file_mapping, tquery_id,
                            result_blocks[i], &external_init_reader_ms, &external_get_block_ms));
                }

                // after read the block, shrink char type block
                result_blocks[i].shrink_char_type_column_suffix_zero(char_type_idx);
            }

            [[maybe_unused]] size_t compressed_size = 0;
            [[maybe_unused]] size_t uncompressed_size = 0;
            int be_exec_version = request.has_be_exec_version() ? request.be_exec_version() : 0;
            RETURN_IF_ERROR(result_blocks[i].serialize(
                    be_exec_version, response->add_blocks()->mutable_block(), &uncompressed_size,
                    &compressed_size, segment_v2::CompressionTypePB::LZ4));
        }

        // Build file type statistics string
        std::string file_type_stats;
        for (const auto& [type, count] : file_type_counts) {
            if (!file_type_stats.empty()) {
                file_type_stats += ", ";
            }
            file_type_stats += fmt::format("{}:{}", type, count);
        }

        LOG(INFO) << "Query stats: "
                  << fmt::format(
                             "Internal table:"
                             "hit_cached_pages:{}, total_pages_read:{}, compressed_bytes_read:{}, "
                             "io_latency:{}ns, uncompressed_bytes_read:{}, bytes_read:{}, "
                             "acquire_tablet_ms:{}, acquire_rowsets_ms:{}, acquire_segments_ms:{}, "
                             "lookup_row_data_ms:{}, file_types:[{}]; "
                             "External table : init_reader_ms:{}, get_block_ms:{}",
                             stats.cached_pages_num, stats.total_pages_num,
                             stats.compressed_bytes_read, stats.io_ns,
                             stats.uncompressed_bytes_read, stats.bytes_read, acquire_tablet_ms,
                             acquire_rowsets_ms, acquire_segments_ms, lookup_row_data_ms,
                             file_type_stats, external_init_reader_ms, external_get_block_ms);
    }

    if (request.has_gc_id_map() && request.gc_id_map()) {
        ExecEnv::GetInstance()->get_id_manager()->remove_id_file_map(request.query_id());
    }

    return Status::OK();
}

Status RowIdStorageReader::read_batch_doris_format_row(
        const PRequestBlockDesc& request_block_desc, std::shared_ptr<IdFileMap> id_file_map,
        std::vector<SlotDescriptor>& slots, const TUniqueId& query_id,
        vectorized::Block& result_block, OlapReaderStatistics& stats, int64_t* acquire_tablet_ms,
        int64_t* acquire_rowsets_ms, int64_t* acquire_segments_ms, int64_t* lookup_row_data_ms) {
    if (result_block.is_empty_column()) [[likely]] {
        result_block = vectorized::Block(slots, request_block_desc.row_id_size());
    }

    TabletSchema full_read_schema;
    for (const ColumnPB& column_pb : request_block_desc.column_descs()) {
        full_read_schema.append_column(TabletColumn(column_pb));
    }
    std::unordered_map<IteratorKey, IteratorItem, HashOfIteratorKey> iterator_map;
    std::string row_store_buffer;
    RowStoreReadStruct row_store_read_struct(row_store_buffer);
    if (request_block_desc.fetch_row_store()) {
        for (int i = 0; i < request_block_desc.slots_size(); ++i) {
            row_store_read_struct.serdes.emplace_back(slots[i].get_data_type_ptr()->get_serde());
            row_store_read_struct.col_uid_to_idx[slots[i].col_unique_id()] = i;
            row_store_read_struct.default_values.emplace_back(slots[i].col_default_value());
        }
    }

    for (size_t j = 0; j < request_block_desc.row_id_size(); ++j) {
        auto file_id = request_block_desc.file_id(j);
        auto file_mapping = id_file_map->get_file_mapping(file_id);
        if (!file_mapping) {
            return Status::InternalError(
                    "Backend:{} file_mapping not found, query_id: {}, file_id: {}",
                    BackendOptions::get_localhost(), print_id(query_id), file_id);
        }

        RETURN_IF_ERROR(read_doris_format_row(
                id_file_map, file_mapping, request_block_desc.row_id(j), slots, full_read_schema,
                row_store_read_struct, stats, acquire_tablet_ms, acquire_rowsets_ms,
                acquire_segments_ms, lookup_row_data_ms, iterator_map, result_block));
    }
    return Status::OK();
}

Status RowIdStorageReader::read_batch_external_row(const PRequestBlockDesc& request_block_desc,
                                                   std::shared_ptr<IdFileMap> id_file_map,
                                                   std::vector<SlotDescriptor>& slots,
                                                   std::shared_ptr<FileMapping> first_file_mapping,
                                                   const TUniqueId& query_id,
                                                   vectorized::Block& result_block,
                                                   int64_t* init_reader_ms, int64_t* get_block_ms) {
    TFileScanRangeParams rpc_scan_params;
    TupleDescriptor tuple_desc(request_block_desc.desc(), false);
    std::unordered_map<std::string, int> colname_to_slot_id;
    std::unique_ptr<RuntimeState> runtime_state = nullptr;
    std::unique_ptr<RuntimeProfile> runtime_profile;
    runtime_profile = std::make_unique<RuntimeProfile>("ExternalRowIDFetcher");

    std::unique_ptr<vectorized::FileScanner> vfile_scanner_ptr = nullptr;

    {
        if (result_block.is_empty_column()) [[likely]] {
            result_block = vectorized::Block(slots, request_block_desc.row_id_size());
        }

        auto& external_info = first_file_mapping->get_external_file_info();
        int plan_node_id = external_info.plan_node_id;
        const auto& first_scan_range_desc = external_info.scan_range_desc;

        DCHECK(id_file_map->get_external_scan_params().contains(plan_node_id));
        const auto* old_scan_params = &(id_file_map->get_external_scan_params().at(plan_node_id));
        rpc_scan_params = *old_scan_params;

        rpc_scan_params.required_slots.clear();
        rpc_scan_params.column_idxs.clear();
        rpc_scan_params.slot_name_to_schema_pos.clear();

        std::set partition_name_set(first_scan_range_desc.columns_from_path_keys.begin(),
                                    first_scan_range_desc.columns_from_path_keys.end());
        for (auto slot_idx = 0; slot_idx < slots.size(); ++slot_idx) {
            auto& slot = slots[slot_idx];
            tuple_desc.add_slot(&slot);
            colname_to_slot_id.emplace(slot.col_name(), slot.id());
            TFileScanSlotInfo slot_info;
            slot_info.slot_id = slot.id();
            auto column_idx = request_block_desc.column_idxs(slot_idx);
            if (partition_name_set.contains(slot.col_name())) {
                //This is partition column.
                slot_info.is_file_slot = false;
            } else {
                rpc_scan_params.column_idxs.emplace_back(column_idx);
                slot_info.is_file_slot = true;
            }
            rpc_scan_params.default_value_of_src_slot.emplace(slot.id(), TExpr {});
            rpc_scan_params.required_slots.emplace_back(slot_info);
            rpc_scan_params.slot_name_to_schema_pos.emplace(slot.col_name(), column_idx);
        }

        const auto& query_options = id_file_map->get_query_options();
        const auto& query_globals = id_file_map->get_query_globals();
        /*
         * The scan stage needs the information in query_options to generate different behaviors according to the specific variables:
         *  query_options.hive_parquet_use_column_names, query_options.truncate_char_or_varchar_columns,query_globals.time_zone ...
         *
         * To ensure the same behavior as the scan stage, I get query_options query_globals from id_file_map, then create runtime_state
         * and pass it to vfile_scanner so that the runtime_state information is the same as the scan stage and the behavior is also consistent.
         */
        runtime_state = RuntimeState::create_unique(query_id, -1, query_options, query_globals,
                                                    ExecEnv::GetInstance());

        vfile_scanner_ptr = vectorized::FileScanner::create_unique(
                runtime_state.get(), runtime_profile.get(), &rpc_scan_params, &colname_to_slot_id,
                &tuple_desc);

        RETURN_IF_ERROR(vfile_scanner_ptr->prepare_for_read_one_line(first_scan_range_desc));
    }

    for (size_t j = 0; j < request_block_desc.row_id_size(); ++j) {
        auto file_id = request_block_desc.file_id(j);
        auto file_mapping = id_file_map->get_file_mapping(file_id);
        if (!file_mapping) {
            return Status::InternalError(
                    "Backend:{} file_mapping not found, query_id: {}, file_id: {}",
                    BackendOptions::get_localhost(), print_id(query_id), file_id);
        }

        auto& external_info = file_mapping->get_external_file_info();
        auto& scan_range_desc = external_info.scan_range_desc;

        // Clear to avoid reading iceberg position delete file...
        scan_range_desc.table_format_params.iceberg_params = TIcebergFileDesc {};

        // Clear to avoid reading hive transactional delete delta file...
        scan_range_desc.table_format_params.transactional_hive_params = TTransactionalHiveDesc {};

        RETURN_IF_ERROR(vfile_scanner_ptr->read_one_line_from_range(
                scan_range_desc, request_block_desc.row_id(j), &result_block, external_info,
                init_reader_ms, get_block_ms));
    }
    return Status::OK();
}

Status RowIdStorageReader::read_doris_format_row(
        const std::shared_ptr<IdFileMap>& id_file_map,
        const std::shared_ptr<FileMapping>& file_mapping, int64_t row_id,
        std::vector<SlotDescriptor>& slots, const TabletSchema& full_read_schema,
        RowStoreReadStruct& row_store_read_struct, OlapReaderStatistics& stats,
        int64_t* acquire_tablet_ms, int64_t* acquire_rowsets_ms, int64_t* acquire_segments_ms,
        int64_t* lookup_row_data_ms,
        std::unordered_map<IteratorKey, IteratorItem, HashOfIteratorKey>& iterator_map,
        vectorized::Block& result_block) {
    auto [tablet_id, rowset_id, segment_id] = file_mapping->get_doris_format_info();
    BaseTabletSPtr tablet = scope_timer_run(
            [&]() {
                auto res = ExecEnv::get_tablet(tablet_id);
                return !res.has_value() ? nullptr
                                        : std::dynamic_pointer_cast<BaseTablet>(res.value());
            },
            acquire_tablet_ms);
    if (!tablet) {
        return Status::InternalError(
                "Backend:{} tablet not found, tablet_id: {}, rowset_id: {}, segment_id: {}, "
                "row_id: {}",
                BackendOptions::get_localhost(), tablet_id, rowset_id.to_string(), segment_id,
                row_id);
    }

    BetaRowsetSharedPtr rowset = std::static_pointer_cast<BetaRowset>(
            scope_timer_run([&]() { return id_file_map->get_temp_rowset(tablet_id, rowset_id); },
                            acquire_rowsets_ms));
    if (!rowset) {
        return Status::InternalError(
                "Backend:{} rowset_id not found, tablet_id: {}, rowset_id: {}, segment_id: {}, "
                "row_id: {}",
                BackendOptions::get_localhost(), tablet_id, rowset_id.to_string(), segment_id,
                row_id);
    }

    SegmentCacheHandle segment_cache;
    RETURN_IF_ERROR(scope_timer_run(
            [&]() {
                return SegmentLoader::instance()->load_segments(rowset, &segment_cache, true);
            },
            acquire_segments_ms));

    auto it =
            std::find_if(segment_cache.get_segments().cbegin(), segment_cache.get_segments().cend(),
                         [segment_id](const segment_v2::SegmentSharedPtr& seg) {
                             return seg->id() == segment_id;
                         });
    if (it == segment_cache.get_segments().end()) {
        return Status::InternalError(
                "Backend:{} segment not found, tablet_id: {}, rowset_id: {}, segment_id: {}, "
                "row_id: {}",
                BackendOptions::get_localhost(), tablet_id, rowset_id.to_string(), segment_id,
                row_id);
    }
    segment_v2::SegmentSharedPtr segment = *it;

    // if row_store_read_struct not empty, means the line we should read from row_store
    if (!row_store_read_struct.default_values.empty()) {
        CHECK(tablet->tablet_schema()->has_row_store_for_all_columns());
        RowLocation loc(rowset_id, segment->id(), row_id);
        row_store_read_struct.row_store_buffer.clear();
        RETURN_IF_ERROR(scope_timer_run(
                [&]() {
                    return tablet->lookup_row_data({}, loc, rowset, stats,
                                                   row_store_read_struct.row_store_buffer);
                },
                lookup_row_data_ms));

        RETURN_IF_ERROR(vectorized::JsonbSerializeUtil::jsonb_to_block(
                row_store_read_struct.serdes, row_store_read_struct.row_store_buffer.data(),
                row_store_read_struct.row_store_buffer.size(), row_store_read_struct.col_uid_to_idx,
                result_block, row_store_read_struct.default_values, {}));
    } else {
        for (int x = 0; x < slots.size(); ++x) {
            vectorized::MutableColumnPtr column =
                    result_block.get_by_position(x).column->assume_mutable();
            IteratorKey iterator_key {.tablet_id = tablet_id,
                                      .rowset_id = rowset_id,
                                      .segment_id = segment_id,
                                      .slot_id = slots[x].id()};
            IteratorItem& iterator_item = iterator_map[iterator_key];
            if (iterator_item.segment == nullptr) {
                iterator_map[iterator_key].segment = segment;
            }
            segment = iterator_item.segment;
            RETURN_IF_ERROR(segment->seek_and_read_by_rowid(full_read_schema, &slots[x], row_id,
                                                            column, stats, iterator_item.iterator));
        }
    }

    return Status::OK();
}

} // namespace doris