be/src/exec/sink/writer/vtablet_writer.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 "exec/sink/writer/vtablet_writer.h"

#include <brpc/http_method.h>
#include <bthread/bthread.h>
#include <fmt/format.h>
#include <gen_cpp/DataSinks_types.h>
#include <gen_cpp/Descriptors_types.h>
#include <gen_cpp/Exprs_types.h>
#include <gen_cpp/FrontendService.h>
#include <gen_cpp/FrontendService_types.h>
#include <gen_cpp/HeartbeatService_types.h>
#include <gen_cpp/Metrics_types.h>
#include <gen_cpp/Types_types.h>
#include <gen_cpp/data.pb.h>
#include <gen_cpp/internal_service.pb.h>
#include <glog/logging.h>
#include <google/protobuf/stubs/common.h>
#include <sys/param.h>

#include <algorithm>
#include <initializer_list>
#include <memory>
#include <mutex>
#include <sstream>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>

#include "cloud/config.h"
#include "common/compiler_util.h" // IWYU pragma: keep
#include "common/config.h"
#include "common/logging.h"
#include "common/metrics/doris_metrics.h"
#include "common/object_pool.h"
#include "common/signal_handler.h"
#include "common/status.h"
#include "core/block/block.h"
#include "core/column/column.h"
#include "core/column/column_const.h"
#include "core/data_type/data_type.h"
#include "core/data_type/data_type_nullable.h"
#include "cpp/sync_point.h"
#include "exec/sink/autoinc_buffer.h"
#include "exec/sink/vrow_distribution.h"
#include "exec/sink/vtablet_block_convertor.h"
#include "exec/sink/vtablet_finder.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_fwd.h"
#include "runtime/descriptors.h"
#include "runtime/exec_env.h"
#include "runtime/memory/memory_reclamation.h"
#include "runtime/query_context.h"
#include "runtime/runtime_profile.h"
#include "runtime/runtime_state.h"
#include "runtime/thread_context.h"
#include "service/backend_options.h"
#include "storage/binlog.h"
#include "storage/tablet_info.h"
#include "util/brpc_closure.h"
#include "util/debug_points.h"
#include "util/defer_op.h"
#include "util/mem_info.h"
#include "util/network_util.h"
#include "util/proto_util.h"
#include "util/threadpool.h"
#include "util/thrift_rpc_helper.h"
#include "util/thrift_util.h"
#include "util/time.h"
#include "util/uid_util.h"

namespace doris {
class TExpr;

bvar::Adder<int64_t> g_sink_write_bytes;
bvar::PerSecond<bvar::Adder<int64_t>> g_sink_write_bytes_per_second("sink_throughput_byte",
                                                                    &g_sink_write_bytes, 60);
bvar::Adder<int64_t> g_sink_write_rows;
bvar::PerSecond<bvar::Adder<int64_t>> g_sink_write_rows_per_second("sink_throughput_row",
                                                                   &g_sink_write_rows, 60);
bvar::Adder<int64_t> g_sink_load_back_pressure_version_time_ms(
        "load_back_pressure_version_time_ms");

static const OlapTableIndexTablets* find_partition_index(const VOlapTablePartition& partition,
                                                         int64_t index_id) {
    for (const auto& index : partition.indexes) {
        if (index.index_id == index_id) {
            return &index;
        }
    }
    return nullptr;
}

static int64_t adaptive_bucket_be_id(const VOlapTablePartition& partition,
                                     const OlapTableIndexTablets* index) {
    if (index != nullptr && index->__isset.bucket_be_id) {
        return index->bucket_be_id > 0 ? index->bucket_be_id : BackendOptions::get_backend_id();
    }
    return partition.bucket_be_id > 0 ? partition.bucket_be_id : BackendOptions::get_backend_id();
}

static const std::vector<int32_t>& adaptive_local_bucket_seqs(const VOlapTablePartition& partition,
                                                              const OlapTableIndexTablets* index) {
    if (index != nullptr && index->__isset.local_bucket_seqs) {
        return index->local_bucket_seqs;
    }
    return partition.local_bucket_seqs;
}

static constexpr int64_t CLOSE_WAIT_EVENT_FALLBACK_MS = 1000;

Status IndexChannel::init(RuntimeState* state, const std::vector<TTabletWithPartition>& tablets,
                          bool incremental) {
    SCOPED_CONSUME_MEM_TRACKER(_index_channel_tracker.get());
    for (const auto& tablet : tablets) {
        // First find the location BEs of this tablet
        auto* tablet_locations = _parent->_location->find_tablet(tablet.tablet_id);
        if (tablet_locations == nullptr) {
            return Status::InternalError("unknown tablet, tablet_id={}", tablet.tablet_id);
        }
        std::vector<std::shared_ptr<VNodeChannel>> channels;
        // For tablet, deal with its' all replica (in some node).
        for (auto& replica_node_id : tablet_locations->node_ids) {
            std::shared_ptr<VNodeChannel> channel;
            auto it = _node_channels.find(replica_node_id);
            // when we prepare for TableSink or incremental open tablet, we need init
            if (it == _node_channels.end()) {
                // NodeChannel is not added to the _parent->_pool.
                // Because the deconstruction of NodeChannel may take a long time to wait rpc finish.
                // but the ObjectPool will hold a spin lock to delete objects.
                channel =
                        std::make_shared<VNodeChannel>(_parent, this, replica_node_id, incremental);
                _node_channels.emplace(replica_node_id, channel);
                // incremental opened new node. when close we have use two-stage close.
                if (incremental) {
                    _has_inc_node = true;
                }
                VLOG_CRITICAL << "init new node for instance " << _parent->_sender_id
                              << ", node id:" << replica_node_id << ", incremantal:" << incremental;
            } else {
                channel = it->second;
            }
            channel->add_tablet(tablet);
            if (_parent->_tablet_finder->is_adaptive_random_bucket() && config::is_cloud_mode()) {
                for (const auto* part : _parent->_vpartition->get_partitions()) {
                    if (part->id != tablet.partition_id) {
                        continue;
                    }
                    const auto* index = find_partition_index(*part, _index_id);
                    const auto bucket_be_id = adaptive_bucket_be_id(*part, index);
                    if (bucket_be_id != replica_node_id) {
                        continue;
                    }
                    _channels_by_partition.emplace(tablet.partition_id, channel);
                    break;
                }
            }
            if (_parent->_write_single_replica) {
                auto* slave_location = _parent->_slave_location->find_tablet(tablet.tablet_id);
                if (slave_location != nullptr) {
                    channel->add_slave_tablet_nodes(tablet.tablet_id, slave_location->node_ids);
                }
            }
            channels.push_back(channel);
            _tablets_by_channel[replica_node_id].insert(tablet.tablet_id);
        }
        _channels_by_tablet.emplace(tablet.tablet_id, std::move(channels));
    }
    for (auto& it : _node_channels) {
        RETURN_IF_ERROR(it.second->init(state));
    }
    if (_where_clause != nullptr) {
        RETURN_IF_ERROR(_where_clause->prepare(state, *_parent->_output_row_desc));
        RETURN_IF_ERROR(_where_clause->open(state));
    }

    return Status::OK();
}

void IndexChannel::mark_as_failed(const VNodeChannel* node_channel, const std::string& err,
                                  int64_t tablet_id) {
    DCHECK(node_channel != nullptr);
    LOG(INFO) << "mark node_id:" << node_channel->channel_info() << " tablet_id: " << tablet_id
              << " as failed, err: " << err;
    auto node_id = node_channel->node_id();
    const auto& it = _tablets_by_channel.find(node_id);
    if (it == _tablets_by_channel.end()) {
        return;
    }

    {
        std::lock_guard<std::mutex> l(_fail_lock);
        if (tablet_id == -1) {
            for (const auto the_tablet_id : it->second) {
                _failed_channels[the_tablet_id].insert(node_id);
                _failed_channels_msgs.emplace(the_tablet_id,
                                              err + ", host: " + node_channel->host());
                if (_failed_channels[the_tablet_id].size() > _max_failed_replicas(the_tablet_id)) {
                    _intolerable_failure_status = Status::Error<ErrorCode::INTERNAL_ERROR, false>(
                            _failed_channels_msgs[the_tablet_id]);
                }
            }
        } else {
            _failed_channels[tablet_id].insert(node_id);
            _failed_channels_msgs.emplace(tablet_id, err + ", host: " + node_channel->host());
            if (_failed_channels[tablet_id].size() > _max_failed_replicas(tablet_id)) {
                _intolerable_failure_status = Status::Error<ErrorCode::INTERNAL_ERROR, false>(
                        _failed_channels_msgs[tablet_id]);
            }
        }
    }
}

int IndexChannel::_max_failed_replicas(int64_t tablet_id) {
    auto [total_replicas_num, load_required_replicas_num] =
            _parent->_tablet_replica_info[tablet_id];
    int max_failed_replicas = total_replicas_num == 0
                                      ? (_parent->_num_replicas - 1) / 2
                                      : total_replicas_num - load_required_replicas_num;
    return max_failed_replicas;
}

int IndexChannel::_load_required_replicas_num(int64_t tablet_id) {
    auto [total_replicas_num, load_required_replicas_num] =
            _parent->_tablet_replica_info[tablet_id];
    if (total_replicas_num == 0) {
        return (_parent->_num_replicas + 1) / 2;
    }
    return load_required_replicas_num;
}

Status IndexChannel::check_intolerable_failure() {
    std::lock_guard<std::mutex> l(_fail_lock);
    return _intolerable_failure_status;
}

void IndexChannel::set_error_tablet_in_state(RuntimeState* state) {
    std::vector<TErrorTabletInfo> error_tablet_infos;

    {
        std::lock_guard<std::mutex> l(_fail_lock);
        for (const auto& it : _failed_channels_msgs) {
            TErrorTabletInfo error_info;
            error_info.__set_tabletId(it.first);
            error_info.__set_msg(it.second);
            error_tablet_infos.emplace_back(error_info);
        }
    }
    state->add_error_tablet_infos(error_tablet_infos);
}

void IndexChannel::set_tablets_received_rows(
        const std::vector<std::pair<int64_t, int64_t>>& tablets_received_rows, int64_t node_id) {
    for (const auto& [tablet_id, rows_num] : tablets_received_rows) {
        _tablets_received_rows[tablet_id].emplace_back(node_id, rows_num);
    }
}

void IndexChannel::set_tablets_filtered_rows(
        const std::vector<std::pair<int64_t, int64_t>>& tablets_filtered_rows, int64_t node_id) {
    for (const auto& [tablet_id, rows_num] : tablets_filtered_rows) {
        _tablets_filtered_rows[tablet_id].emplace_back(node_id, rows_num);
    }
}

Status IndexChannel::check_tablet_received_rows_consistency() {
    for (auto& tablet : _tablets_received_rows) {
        for (size_t i = 0; i < tablet.second.size(); i++) {
            VLOG_NOTICE << "check_tablet_received_rows_consistency, load_id: " << _parent->_load_id
                        << ", txn_id: " << std::to_string(_parent->_txn_id)
                        << ", tablet_id: " << tablet.first
                        << ", node_id: " << tablet.second[i].first
                        << ", rows_num: " << tablet.second[i].second;
            if (i == 0) {
                continue;
            }
            if (tablet.second[i].second != tablet.second[0].second) {
                return Status::InternalError(
                        "rows num written by multi replicas doest't match, load_id={}, txn_id={}, "
                        "tablt_id={}, node_id={}, rows_num={}, node_id={}, rows_num={}",
                        print_id(_parent->_load_id), _parent->_txn_id, tablet.first,
                        tablet.second[i].first, tablet.second[i].second, tablet.second[0].first,
                        tablet.second[0].second);
            }
        }
    }
    return Status::OK();
}

Status IndexChannel::check_tablet_filtered_rows_consistency() {
    for (auto& tablet : _tablets_filtered_rows) {
        for (size_t i = 0; i < tablet.second.size(); i++) {
            VLOG_NOTICE << "check_tablet_filtered_rows_consistency, load_id: " << _parent->_load_id
                        << ", txn_id: " << std::to_string(_parent->_txn_id)
                        << ", tablet_id: " << tablet.first
                        << ", node_id: " << tablet.second[i].first
                        << ", rows_num: " << tablet.second[i].second;
            if (i == 0) {
                continue;
            }
            if (tablet.second[i].second != tablet.second[0].second) {
                return Status::InternalError(
                        "rows num filtered by multi replicas doest't match, load_id={}, txn_id={}, "
                        "tablt_id={}, node_id={}, rows_num={}, node_id={}, rows_num={}",
                        print_id(_parent->_load_id), _parent->_txn_id, tablet.first,
                        tablet.second[i].first, tablet.second[i].second, tablet.second[0].first,
                        tablet.second[0].second);
            }
        }
    }
    return Status::OK();
}

static Status cancel_channel_and_check_intolerable_failure(Status status,
                                                           const std::string& err_msg,
                                                           IndexChannel& ich, VNodeChannel& nch) {
    LOG(WARNING) << nch.channel_info() << ", close channel failed, err: " << err_msg;
    ich.mark_as_failed(&nch, err_msg, -1);
    // cancel the node channel in best effort
    nch.cancel(err_msg);

    // check if index has intolerable failure
    if (Status index_st = ich.check_intolerable_failure(); !index_st.ok()) {
        status = std::move(index_st);
    } else if (Status receive_st = ich.check_tablet_received_rows_consistency(); !receive_st.ok()) {
        status = std::move(receive_st);
    } else if (Status filter_st = ich.check_tablet_filtered_rows_consistency(); !filter_st.ok()) {
        status = std::move(filter_st);
    }
    return status;
}

void IndexChannel::wait_for_close_event(int64_t observed_version, int64_t timeout_ms) {
    std::unique_lock<bthread::Mutex> lock(_close_wait_mutex);
    if (observed_version != close_wait_version()) {
        return;
    }
    static_cast<void>(_close_wait_cv.wait_for(lock, timeout_ms * 1000));
}

void IndexChannel::notify_close_wait() {
    _close_wait_version.fetch_add(1, std::memory_order_acq_rel);
    std::lock_guard<bthread::Mutex> lock(_close_wait_mutex);
    _close_wait_cv.notify_all();
}

Status IndexChannel::close_wait(
        RuntimeState* state, WriterStats* writer_stats,
        std::unordered_map<int64_t, AddBatchCounter>* node_add_batch_counter_map,
        std::unordered_set<int64_t> unfinished_node_channel_ids,
        bool need_wait_after_quorum_success) {
    DBUG_EXECUTE_IF("IndexChannel.close_wait.timeout",
                    { return Status::TimedOut("injected timeout"); });
    Status status = Status::OK();
    // 1. wait quorum success
    std::unordered_set<int64_t> need_finish_tablets;
    auto partition_ids = _parent->_tablet_finder->partition_ids();
    for (const auto& part : _parent->_vpartition->get_partitions()) {
        if (partition_ids.contains(part->id)) {
            for (const auto& index : part->indexes) {
                for (const auto& tablet_id : index.tablets) {
                    need_finish_tablets.insert(tablet_id);
                }
            }
        }
    }
    while (true) {
        int64_t close_wait_version = this->close_wait_version();
        RETURN_IF_ERROR(check_each_node_channel_close(
                &unfinished_node_channel_ids, node_add_batch_counter_map, writer_stats, status));
        bool quorum_success = _quorum_success(unfinished_node_channel_ids, need_finish_tablets);
        if (unfinished_node_channel_ids.empty() || quorum_success) {
            LOG(INFO) << "quorum_success: " << quorum_success
                      << ", is all finished: " << unfinished_node_channel_ids.empty()
                      << ", txn_id: " << _parent->_txn_id
                      << ", load_id: " << print_id(_parent->_load_id);
            break;
        }
        wait_for_close_event(close_wait_version, CLOSE_WAIT_EVENT_FALLBACK_MS);
    }

    // 2. wait for all node channel to complete as much as possible
    if (!unfinished_node_channel_ids.empty() && need_wait_after_quorum_success) {
        int64_t arrival_quorum_success_time = UnixMillis();
        int64_t max_wait_time_ms = _calc_max_wait_time_ms(unfinished_node_channel_ids);
        while (true) {
            int64_t close_wait_version = this->close_wait_version();
            RETURN_IF_ERROR(check_each_node_channel_close(&unfinished_node_channel_ids,
                                                          node_add_batch_counter_map, writer_stats,
                                                          status));
            if (unfinished_node_channel_ids.empty()) {
                break;
            }
            int64_t elapsed_ms = UnixMillis() - arrival_quorum_success_time;
            if (elapsed_ms > max_wait_time_ms ||
                _parent->_load_channel_timeout_s - elapsed_ms / 1000 <
                        config::quorum_success_remaining_timeout_seconds) {
                // cancel unfinished node channel
                std::stringstream unfinished_node_channel_host_str;
                for (auto& it : unfinished_node_channel_ids) {
                    unfinished_node_channel_host_str << _node_channels[it]->host() << ",";
                    _node_channels[it]->cancel("timeout");
                }
                LOG(WARNING) << "reach max wait time, max_wait_time_ms: " << max_wait_time_ms
                             << ", cancel unfinished node channel and finish close"
                             << ", load id: " << print_id(_parent->_load_id)
                             << ", txn_id: " << _parent->_txn_id << ", unfinished node channel: "
                             << unfinished_node_channel_host_str.str();
                break;
            }
            wait_for_close_event(close_wait_version, std::min(CLOSE_WAIT_EVENT_FALLBACK_MS,
                                                              max_wait_time_ms - elapsed_ms));
        }
    }
    return status;
}

Status IndexChannel::check_each_node_channel_close(
        std::unordered_set<int64_t>* unfinished_node_channel_ids,
        std::unordered_map<int64_t, AddBatchCounter>* node_add_batch_counter_map,
        WriterStats* writer_stats, Status status) {
    Status final_status = Status::OK();
    for (auto& it : _node_channels) {
        std::shared_ptr<VNodeChannel> node_channel = it.second;
        // If the node channel is not in the unfinished_node_channel_ids,
        // it means the node channel is already closed.
        if (!unfinished_node_channel_ids->contains(it.first)) {
            continue;
        }
        bool node_channel_closed = false;
        auto close_status = it.second->close_wait(_parent->_state, &node_channel_closed);
        if (node_channel_closed) {
            close_status = it.second->after_close_handle(_parent->_state, writer_stats,
                                                         node_add_batch_counter_map);
            unfinished_node_channel_ids->erase(it.first);
        }
        DBUG_EXECUTE_IF("IndexChannel.check_each_node_channel_close.close_status_not_ok",
                        { close_status = Status::InternalError("injected close status not ok"); });
        if (!close_status.ok()) {
            final_status = cancel_channel_and_check_intolerable_failure(
                    std::move(final_status), close_status.to_string(), *this, *it.second);
        }
    }

    return final_status;
}

bool IndexChannel::_quorum_success(const std::unordered_set<int64_t>& unfinished_node_channel_ids,
                                   const std::unordered_set<int64_t>& need_finish_tablets) {
    if (!config::enable_quorum_success_write) {
        return false;
    }
    if (need_finish_tablets.empty()) [[unlikely]] {
        return false;
    }

    // 1. collect all write tablets and finished tablets
    std::unordered_map<int64_t, int64_t> finished_tablets_replica;
    for (const auto& [node_id, node_channel] : _node_channels) {
        if (unfinished_node_channel_ids.contains(node_id) || !node_channel->check_status().ok()) {
            continue;
        }
        for (const auto& tablet_id : _tablets_by_channel[node_id]) {
            // Only count non-gap backends for quorum success.
            // Gap backends' success doesn't count toward majority write.
            auto gap_it = _parent->_tablet_version_gap_backends.find(tablet_id);
            if (gap_it == _parent->_tablet_version_gap_backends.end() ||
                gap_it->second.find(node_id) == gap_it->second.end()) {
                finished_tablets_replica[tablet_id]++;
            }
        }
    }

    // 2. check if quorum success
    for (const auto& tablet_id : need_finish_tablets) {
        if (finished_tablets_replica[tablet_id] < _load_required_replicas_num(tablet_id)) {
            return false;
        }
    }

    return true;
}

int64_t IndexChannel::_calc_max_wait_time_ms(
        const std::unordered_set<int64_t>& unfinished_node_channel_ids) {
    // 1. calculate avg speed of all unfinished node channel
    int64_t elapsed_ms = UnixMillis() - _start_time;
    int64_t total_bytes = 0;
    int finished_count = 0;
    for (const auto& [node_id, node_channel] : _node_channels) {
        if (unfinished_node_channel_ids.contains(node_id)) {
            continue;
        }
        total_bytes += node_channel->write_bytes();
        finished_count++;
    }
    // no data loaded in index channel, return 0
    if (total_bytes == 0 || finished_count == 0) {
        return 0;
    }
    // if elapsed_ms is equal to 0, explain the loaded data is too small
    if (elapsed_ms <= 0) {
        return config::quorum_success_min_wait_seconds * 1000;
    }
    double avg_speed =
            static_cast<double>(total_bytes) / (static_cast<double>(elapsed_ms) * finished_count);

    // 2. calculate max wait time of each unfinished node channel and return the max value
    int64_t max_wait_time_ms = 0;
    for (int64_t id : unfinished_node_channel_ids) {
        int64_t bytes = _node_channels[id]->write_bytes();
        int64_t wait =
                avg_speed > 0 ? static_cast<int64_t>(static_cast<double>(bytes) / avg_speed) : 0;
        max_wait_time_ms = std::max(max_wait_time_ms, wait);
    }

    // 3. calculate max wait time
    // introduce quorum_success_min_wait_seconds to avoid jitter of small load
    max_wait_time_ms -= UnixMillis() - _start_time;
    max_wait_time_ms =
            std::max(static_cast<int64_t>(static_cast<double>(max_wait_time_ms) *
                                          (1.0 + config::quorum_success_max_wait_multiplier)),
                     config::quorum_success_min_wait_seconds * 1000);

    return max_wait_time_ms;
}

static Status none_of(std::initializer_list<bool> vars) {
    bool none = std::none_of(vars.begin(), vars.end(), [](bool var) { return var; });
    Status st = Status::OK();
    if (!none) {
        std::string vars_str;
        std::for_each(vars.begin(), vars.end(),
                      [&vars_str](bool var) -> void { vars_str += (var ? "1/" : "0/"); });
        if (!vars_str.empty()) {
            vars_str.pop_back(); // 0/1/0/ -> 0/1/0
        }
        st = Status::Uninitialized(vars_str);
    }

    return st;
}

VNodeChannel::VNodeChannel(VTabletWriter* parent, IndexChannel* index_channel, int64_t node_id,
                           bool is_incremental)
        : _parent(parent),
          _index_channel(index_channel),
          _node_id(node_id),
          _is_incremental(is_incremental) {
    _cur_add_block_request = std::make_shared<PTabletWriterAddBlockRequest>();
    _node_channel_tracker = std::make_shared<MemTracker>(
            fmt::format("NodeChannel:indexID={}:threadId={}",
                        std::to_string(_index_channel->_index_id), ThreadContext::get_thread_id()));
    _load_mem_limit = MemInfo::mem_limit() * config::load_process_max_memory_limit_percent / 100;
}

VNodeChannel::~VNodeChannel() = default;

void VNodeChannel::clear_all_blocks() {
    std::lock_guard<std::mutex> lg(_pending_batches_lock);
    std::queue<AddBlockReq> empty;
    std::swap(_pending_blocks, empty);
    _cur_mutable_block.reset();
}

// we don't need to send tablet_writer_cancel rpc request when
// init failed, so set _is_closed to true.
// if "_cancelled" is set to true,
// no need to set _cancel_msg because the error will be
// returned directly via "TabletSink::prepare()" method.
Status VNodeChannel::init(RuntimeState* state) {
    if (_inited) {
        return Status::OK();
    }

    SCOPED_CONSUME_MEM_TRACKER(_node_channel_tracker.get());
    _task_exec_ctx = state->get_task_execution_context();
    _tuple_desc = _parent->_output_tuple_desc;
    _state = state;
    // get corresponding BE node.
    const auto* node = _parent->_nodes_info->find_node(_node_id);
    if (node == nullptr) {
        _cancelled = true;
        _is_closed = true;
        return Status::InternalError("unknown node id, id={}", _node_id);
    }
    _node_info = *node;

    _load_info = "load_id=" + print_id(_parent->_load_id) +
                 ", txn_id=" + std::to_string(_parent->_txn_id);

    _row_desc = std::make_unique<RowDescriptor>(_tuple_desc);
    _batch_size = state->batch_size();

    _stub = state->exec_env()->brpc_internal_client_cache()->get_client(_node_info.host,
                                                                        _node_info.brpc_port);
    if (_stub == nullptr) {
        _cancelled = true;
        _is_closed = true;
        return Status::InternalError("Get rpc stub failed, host={}, port={}, info={}",
                                     _node_info.host, _node_info.brpc_port, channel_info());
    }

    _rpc_timeout_ms = state->execution_timeout() * 1000;
    _timeout_watch.start();

    // Initialize _cur_add_block_request
    if (!_cur_add_block_request->has_id()) {
        *(_cur_add_block_request->mutable_id()) = _parent->_load_id;
    }
    _cur_add_block_request->set_index_id(_index_channel->_index_id);
    _cur_add_block_request->set_sender_id(_parent->_sender_id);
    _cur_add_block_request->set_backend_id(_node_id);
    _cur_add_block_request->set_eos(false);
    // Adaptive random bucket add-block RPCs carry partition ids because the receiver
    // chooses the current tablet from its local adaptive state.
    _cur_add_block_request->set_is_adaptive_random_bucket(
            _parent->_tablet_finder->is_adaptive_random_bucket());

    // add block closure
    // Has to using value to capture _task_exec_ctx because tablet writer may destroyed during callback.
    _send_block_callback = WriteBlockCallback<PTabletWriterAddBlockResult>::create_shared();
    _send_block_callback->addFailedHandler(
            [&, task_exec_ctx = _task_exec_ctx](const WriteBlockCallbackContext& ctx) {
                std::shared_ptr<TaskExecutionContext> ctx_lock = task_exec_ctx.lock();
                if (ctx_lock == nullptr) {
                    return;
                }
                _add_block_failed_callback(ctx);
            });

    _send_block_callback->addSuccessHandler(
            [&, task_exec_ctx = _task_exec_ctx](const PTabletWriterAddBlockResult& result,
                                                const WriteBlockCallbackContext& ctx) {
                std::shared_ptr<TaskExecutionContext> ctx_lock = task_exec_ctx.lock();
                if (ctx_lock == nullptr) {
                    return;
                }
                _add_block_success_callback(result, ctx);
            });

    _name = fmt::format("VNodeChannel[{}-{}]", _index_channel->_index_id, _node_id);
    // The node channel will send _batch_size rows of data each rpc. When the
    // number of tablets is large, the number of data rows received by each
    // tablet is small, TabletsChannel need to traverse each tablet for import.
    // so the import performance is poor. Therefore, we set _batch_size to
    // a relatively large value to improve the import performance.
    _batch_size = std::max(_batch_size, 8192);

    if (_state) {
        QueryContext* query_ctx = _state->get_query_ctx();
        if (query_ctx) {
            auto wg_ptr = query_ctx->workload_group();
            if (wg_ptr) {
                _wg_id = wg_ptr->id();
            }
        }
    }

    _inited = true;
    return Status::OK();
}

void VNodeChannel::_set_adaptive_random_bucket_open_request(PTabletWriterOpenRequest* request) {
    std::unordered_map<int64_t, std::vector<int64_t>> partition_to_ordered_tablets;
    std::unordered_map<int64_t, std::unordered_set<int64_t>> partition_to_local_tablets;
    for (const auto& tablet : _all_tablets) {
        partition_to_ordered_tablets[tablet.partition_id].push_back(tablet.tablet_id);
        partition_to_local_tablets[tablet.partition_id].insert(tablet.tablet_id);
    }
    std::unordered_map<int64_t, const VOlapTablePartition*> id_to_partition;
    for (const auto* part : _parent->_vpartition->get_partitions()) {
        id_to_partition.emplace(part->id, part);
    }
    for (const auto& [partition_id, ordered_tablets] : partition_to_ordered_tablets) {
        auto partition_it = id_to_partition.find(partition_id);
        if (partition_it == id_to_partition.end()) {
            LOG(WARNING) << "unknown partition for adaptive random bucket, load_id="
                         << _parent->_load_id << ", partition_id=" << partition_id;
            continue;
        }
        const auto* index_info =
                find_partition_index(*partition_it->second, _index_channel->_index_id);
        if (index_info == nullptr) {
            LOG(WARNING) << "unknown index for adaptive random bucket, load_id="
                         << _parent->_load_id << ", partition_id=" << partition_id
                         << ", index_id=" << _index_channel->_index_id;
            continue;
        }
        std::vector<int64_t> selected_ordered_tablets;
        const auto& local_bucket_seqs =
                adaptive_local_bucket_seqs(*partition_it->second, index_info);
        if (!local_bucket_seqs.empty()) {
            const auto& full_ordered_tablets = index_info->tablets;
            for (auto bucket_seq : local_bucket_seqs) {
                if (bucket_seq < 0 ||
                    bucket_seq >= cast_set<int32_t>(full_ordered_tablets.size())) {
                    LOG(WARNING) << "invalid local bucket seq, load_id=" << _parent->_load_id
                                 << ", partition_id=" << partition_id
                                 << ", bucket_seq=" << bucket_seq
                                 << ", full_ordered_tablets_size=" << full_ordered_tablets.size();
                    continue;
                }
                auto tablet_id = full_ordered_tablets[bucket_seq];
                if (!partition_to_local_tablets[partition_id].contains(tablet_id)) {
                    LOG(WARNING) << "skip non-local tablet selected by local bucket seq, load_id="
                                 << _parent->_load_id << ", partition_id=" << partition_id
                                 << ", bucket_seq=" << bucket_seq << ", tablet_id=" << tablet_id
                                 << ", node_id=" << _node_id;
                    continue;
                }
                selected_ordered_tablets.push_back(tablet_id);
            }
        } else {
            selected_ordered_tablets = ordered_tablets;
        }
        if (selected_ordered_tablets.empty()) {
            VLOG_DEBUG << "skip adaptive random bucket partition without selected local "
                          "tablet, load_id="
                       << _parent->_load_id << ", partition_id=" << partition_id
                       << ", node_id=" << _node_id;
            continue;
        }
        _adaptive_partition_compat_tablets[partition_id] = selected_ordered_tablets.front();
        auto* random_bucket_partition = request->add_random_bucket_partitions();
        random_bucket_partition->set_partition_id(partition_id);
        for (auto tablet_id : selected_ordered_tablets) {
            random_bucket_partition->add_ordered_tablet_ids(tablet_id);
        }
    }
}

void VNodeChannel::_open_internal(bool is_incremental) {
    if (_tablets_wait_open.empty()) {
        return;
    }
    SCOPED_CONSUME_MEM_TRACKER(_node_channel_tracker.get());
    auto request = std::make_shared<PTabletWriterOpenRequest>();
    request->mutable_id()->CopyFrom(_parent->_load_id);
    request->set_index_id(_index_channel->_index_id);
    request->set_txn_id(_parent->_txn_id);
    request->set_sender_id(_parent->_sender_id);
    request->mutable_schema()->CopyFrom(*_parent->_schema->to_protobuf());
    if (_parent->_t_sink.olap_table_sink.__isset.storage_vault_id) {
        request->set_storage_vault_id(_parent->_t_sink.olap_table_sink.storage_vault_id);
    }
    // Adaptive random bucket open RPCs initialize receiver-side selected-tablet state.
    request->set_is_adaptive_random_bucket(_parent->_tablet_finder->is_adaptive_random_bucket());
    std::set<int64_t> deduper;
    for (auto& tablet : _tablets_wait_open) {
        if (deduper.contains(tablet.tablet_id)) {
            continue;
        }
        auto* ptablet = request->add_tablets();
        ptablet->set_partition_id(tablet.partition_id);
        ptablet->set_tablet_id(tablet.tablet_id);
        deduper.insert(tablet.tablet_id);
        _all_tablets.push_back(std::move(tablet));
    }
    _tablets_wait_open.clear();

    request->set_num_senders(_parent->_num_senders);
    request->set_need_gen_rollup(false); // Useless but it is a required field in pb
    request->set_load_channel_timeout_s(_parent->_load_channel_timeout_s);
    request->set_is_high_priority(_parent->_is_high_priority);
    request->set_sender_ip(BackendOptions::get_localhost());
    request->set_is_vectorized(true);
    request->set_backend_id(_node_id);
    request->set_enable_profile(_state->enable_profile());
    request->set_is_incremental(is_incremental);
    request->set_txn_expiration(_parent->_txn_expiration);
    request->set_write_file_cache(_parent->_write_file_cache);

    if (_parent->_tablet_finder->is_adaptive_random_bucket()) {
        _set_adaptive_random_bucket_open_request(request.get());
    }

    if (_wg_id > 0) {
        request->set_workload_group_id(_wg_id);
    }

    auto open_callback = DummyBrpcCallback<PTabletWriterOpenResult>::create_shared();
    auto open_closure = AutoReleaseClosure<
            PTabletWriterOpenRequest,
            DummyBrpcCallback<PTabletWriterOpenResult>>::create_unique(request, open_callback);
    open_callback->cntl_->set_timeout_ms(config::tablet_writer_open_rpc_timeout_sec * 1000);
    if (config::tablet_writer_ignore_eovercrowded) {
        open_callback->cntl_->ignore_eovercrowded();
    }
    VLOG_DEBUG << fmt::format("txn {}: open NodeChannel to {}, incremental: {}, senders: {}",
                              _parent->_txn_id, _node_id, is_incremental, _parent->_num_senders);
    // the real transmission here. the corresponding BE's load mgr will open load channel for it.
    _stub->tablet_writer_open(open_closure->cntl_.get(), open_closure->request_.get(),
                              open_closure->response_.get(), open_closure.get());
    open_closure.release();
    _open_callbacks.push_back(open_callback);
}

void VNodeChannel::open() {
    _open_internal(false);
}

void VNodeChannel::incremental_open() {
    VLOG_DEBUG << "incremental opening node channel" << _node_id;
    _open_internal(true);
}

Status VNodeChannel::open_wait() {
    Status status;
    for (auto& open_callback : _open_callbacks) {
        // because of incremental open, we will wait multi times. so skip the closures which have been checked and set to nullptr in previous rounds
        if (open_callback == nullptr) {
            continue;
        }

        open_callback->join();
        SCOPED_CONSUME_MEM_TRACKER(_node_channel_tracker.get());
        if (open_callback->cntl_->Failed()) {
            if (!ExecEnv::GetInstance()->brpc_internal_client_cache()->available(
                        _stub, _node_info.host, _node_info.brpc_port)) {
                ExecEnv::GetInstance()->brpc_internal_client_cache()->erase(
                        open_callback->cntl_->remote_side());
            }
            _cancelled = true;
            auto error_code = open_callback->cntl_->ErrorCode();
            auto error_text = open_callback->cntl_->ErrorText();
            if (error_text.find("Reached timeout") != std::string::npos) {
                LOG(WARNING) << "failed to open tablet writer may caused by timeout. increase BE "
                                "config `tablet_writer_open_rpc_timeout_sec` if you are sure that "
                                "your table building and data are reasonable.";
            }
            return Status::Error<ErrorCode::INTERNAL_ERROR, false>(
                    "failed to open tablet writer, error={}, error_text={}, info={}",
                    berror(error_code), error_text, channel_info());
        }
        status = Status::create(open_callback->response_->status());

        if (!status.ok()) {
            _cancelled = true;
            return status;
        }
    }

    return status;
}

Status VNodeChannel::add_block(Block* block, const Payload* payload) {
    SCOPED_CONSUME_MEM_TRACKER(_node_channel_tracker.get());
    if (payload->row_part_tablet_ids == nullptr || payload->row_ids == nullptr ||
        payload->row_ids->empty()) {
        return Status::OK();
    }
    DCHECK_EQ(payload->row_ids->size(), payload->route_idxs.size());
    // If add_block() when _eos_is_produced==true, there must be sth wrong, we can only mark this channel as failed.
    auto st = none_of({_cancelled, _eos_is_produced});
    if (!st.ok()) {
        if (_cancelled) {
            std::lock_guard<std::mutex> l(_cancel_msg_lock);
            return Status::Error<ErrorCode::INTERNAL_ERROR, false>("add row failed. {}",
                                                                   _cancel_msg);
        } else {
            return std::move(st.prepend("already stopped, can't add row. cancelled/eos: "));
        }
    }

    // We use OlapTableSink mem_tracker which has the same ancestor of _plan node,
    // so in the ideal case, mem limit is a matter for _plan node.
    // But there is still some unfinished things, we do mem limit here temporarily.
    // _cancelled may be set by rpc callback, and it's possible that _cancelled might be set in any of the steps below.
    // It's fine to do a fake add_block() and return OK, because we will check _cancelled in next add_block() or mark_close().
    constexpr int64_t kBackPressureSleepMs = 10;
    auto* memtable_limiter = ExecEnv::GetInstance()->memtable_memory_limiter();
    while (true) {
        bool is_exceed_soft_mem_limit = GlobalMemoryArbitrator::is_exceed_soft_mem_limit();
        int64_t memtable_mem =
                (memtable_limiter != nullptr && memtable_limiter->mem_tracker() != nullptr)
                        ? memtable_limiter->mem_tracker()->consumption()
                        : 0;
        // Note: Memtable memory is not included in load memory statistics (MemoryProfile::load_current_usage())
        // for performance and memory control complexity reasons. Therefore, we explicitly add memtable memory
        // consumption here to ensure accurate back pressure decisions and prevent OOM during heavy loads.
        auto current_load_mem_value = MemoryProfile::load_current_usage() + memtable_mem;
        bool mem_limit_exceeded = is_exceed_soft_mem_limit ||
                                  current_load_mem_value > _load_mem_limit ||
                                  _pending_batches_bytes > _max_pending_batches_bytes;
        bool need_back_pressure = !_cancelled && !_state->is_cancelled() &&
                                  _pending_batches_num > 0 && mem_limit_exceeded;
        if (!need_back_pressure) {
            break;
        }
        SCOPED_RAW_TIMER(&_stat.mem_exceeded_block_ns);
        std::this_thread::sleep_for(std::chrono::milliseconds(kBackPressureSleepMs));
    }

    if (UNLIKELY(!_cur_mutable_block)) {
        _cur_mutable_block = MutableBlock::create_unique(block->clone_empty());
    }

    SCOPED_RAW_TIMER(&_stat.append_node_channel_ns);
    st = block->append_to_block_by_selector(_cur_mutable_block.get(), *payload->row_ids);
    if (!st.ok()) {
        _cancel_with_msg(fmt::format("{}, err: {}", channel_info(), st.to_string()));
        return st;
    }
    auto* row_part_tablet_ids = payload->row_part_tablet_ids;
    for (uint32_t route_idx : payload->route_idxs) {
        auto partition_id = row_part_tablet_ids->partition_ids[route_idx];
        _cur_add_block_request->add_partition_ids(partition_id);
        if (_parent->_tablet_finder->is_adaptive_random_bucket()) {
            auto tablet_it = _adaptive_partition_compat_tablets.find(partition_id);
            if (tablet_it == _adaptive_partition_compat_tablets.end()) {
                return Status::InternalError(
                        "{}, err: missing adaptive random bucket compatible tablet, "
                        "partition_id={}",
                        channel_info(), partition_id);
            }
            _cur_add_block_request->add_tablet_ids(tablet_it->second);
        } else {
            _cur_add_block_request->add_tablet_ids(row_part_tablet_ids->tablet_ids[route_idx]);
        }
    }
    for (auto row_binlog_lsn : payload->row_binlog_lsns) {
        _cur_add_block_request->add_row_binlog_lsns(row_binlog_lsn);
    }
    _write_bytes.fetch_add(_cur_mutable_block->bytes());

    if (_cur_mutable_block->rows() >= _batch_size ||
        _cur_mutable_block->bytes() > config::doris_scanner_row_bytes) {
        {
            SCOPED_ATOMIC_TIMER(&_queue_push_lock_ns);
            std::lock_guard<std::mutex> l(_pending_batches_lock);
            // To simplify the add_row logic, postpone adding block into req until the time of sending req
            _pending_batches_bytes += _cur_mutable_block->allocated_bytes();
            _cur_add_block_request->set_eos(
                    false); // for multi-add, only when marking close we set it eos.
            // Copy the request to tmp request to add to pend block queue
            auto tmp_add_block_request = std::make_shared<PTabletWriterAddBlockRequest>();
            *tmp_add_block_request = *_cur_add_block_request;
            _pending_blocks.emplace(std::move(_cur_mutable_block), tmp_add_block_request);
            _pending_batches_num++;
            VLOG_DEBUG << "VTabletWriter:" << _parent << " VNodeChannel:" << this
                       << " pending_batches_bytes:" << _pending_batches_bytes
                       << " jobid:" << std::to_string(_state->load_job_id())
                       << " loadinfo:" << _load_info;
        }
        _cur_mutable_block = MutableBlock::create_unique(block->clone_empty());
        _cur_add_block_request->clear_tablet_ids();
        _cur_add_block_request->clear_partition_ids();
        _cur_add_block_request->clear_row_binlog_lsns();
    }

    return Status::OK();
}

static void injection_full_gc_fn() {
    MemoryReclamation::revoke_process_memory("injection_full_gc_fn");
}

int VNodeChannel::try_send_and_fetch_status(RuntimeState* state,
                                            std::unique_ptr<ThreadPoolToken>& thread_pool_token) {
    DBUG_EXECUTE_IF("VNodeChannel.try_send_and_fetch_status_full_gc", {
        std::thread t(injection_full_gc_fn);
        t.join();
    });

    if (_cancelled || _send_finished) { // not run
        return 0;
    }

    auto load_back_pressure_version_wait_time_ms = _load_back_pressure_version_wait_time_ms.load();
    if (UNLIKELY(load_back_pressure_version_wait_time_ms > 0)) {
        std::this_thread::sleep_for(
                std::chrono::milliseconds(load_back_pressure_version_wait_time_ms));
        _load_back_pressure_version_block_ms.fetch_add(
                load_back_pressure_version_wait_time_ms); // already in milliseconds
        _load_back_pressure_version_wait_time_ms = 0;
    }

    // set closure for sending block.
    if (!_send_block_callback->try_set_in_flight()) {
        // There is packet in flight, skip.
        return _send_finished ? 0 : 1;
    }

    // We are sure that try_send_batch is not running
    if (_pending_batches_num > 0) {
        auto s = thread_pool_token->submit_func([this, state] { try_send_pending_block(state); });
        if (!s.ok()) {
            _cancel_with_msg("submit send_batch task to send_batch_thread_pool failed");
            // sending finished. clear in flight
            _send_block_callback->clear_in_flight();
        }
        // in_flight is cleared in closure::Run
    } else {
        // sending finished. clear in flight
        _send_block_callback->clear_in_flight();
    }
    return _send_finished ? 0 : 1;
}

void VNodeChannel::_cancel_with_msg(const std::string& msg) {
    LOG(WARNING) << "cancel node channel " << channel_info() << ", error message: " << msg;
    {
        std::lock_guard<std::mutex> l(_cancel_msg_lock);
        if (_cancel_msg.empty()) {
            _cancel_msg = msg;
        }
    }
    _cancelled = true;
    _index_channel->notify_close_wait();
}

void VNodeChannel::_refresh_back_pressure_version_wait_time(
        const ::google::protobuf::RepeatedPtrField<::doris::PTabletLoadRowsetInfo>&
                tablet_load_infos) {
    int64_t max_rowset_num_gap = 0;
    // if any one tablet is under high load pressure, we would make the whole procedure
    // sleep to prevent the corresponding BE return -235
    std::for_each(
            tablet_load_infos.begin(), tablet_load_infos.end(),
            [&max_rowset_num_gap](auto& load_info) {
                int64_t cur_rowset_num = load_info.current_rowset_nums();
                int64_t high_load_point = load_info.max_config_rowset_nums() *
                                          (config::load_back_pressure_version_threshold / 100);
                DCHECK(cur_rowset_num > high_load_point);
                max_rowset_num_gap = std::max(max_rowset_num_gap, cur_rowset_num - high_load_point);
            });
    // to slow down the high load pressure
    // we would use the rowset num gap to calculate one sleep time
    // for example:
    // if the max tablet version is 2000, there are 3 BE
    // A: ====================  1800
    // B: ===================   1700
    // C: ==================    1600
    //    ==================    1600
    //                      ^
    //                      the high load point
    // then then max gap is 1800 - (max tablet version * config::load_back_pressure_version_threshold / 100) = 200,
    // we would make the whole send procesure sleep
    // 1200ms for compaction to be done toe reduce the high pressure
    auto max_time = config::max_load_back_pressure_version_wait_time_ms;
    if (UNLIKELY(max_rowset_num_gap > 0)) {
        _load_back_pressure_version_wait_time_ms.store(
                std::min(max_rowset_num_gap + 1000, max_time));
        LOG(INFO) << "try to back pressure version, wait time(ms): "
                  << _load_back_pressure_version_wait_time_ms
                  << ", load id: " << print_id(_parent->_load_id)
                  << ", max_rowset_num_gap: " << max_rowset_num_gap;
    }
}

void VNodeChannel::try_send_pending_block(RuntimeState* state) {
    SCOPED_ATTACH_TASK(state);
    SCOPED_CONSUME_MEM_TRACKER(_node_channel_tracker);
    SCOPED_ATOMIC_TIMER(&_actual_consume_ns);
    signal::set_signal_task_id(_parent->_load_id);
    AddBlockReq send_block;
    {
        std::lock_guard<std::mutex> l(_pending_batches_lock);
        DCHECK(!_pending_blocks.empty());
        send_block = std::move(_pending_blocks.front());
        _pending_blocks.pop();
        _pending_batches_num--;
        _pending_batches_bytes -= send_block.first->allocated_bytes();
    }

    auto mutable_block = std::move(send_block.first);
    auto request = std::move(send_block.second); // doesn't need to be saved in heap

    // tablet_ids has already set when add row
    request->set_packet_seq(_next_packet_seq);
    auto block = mutable_block->to_block();
    int request_rows = request->is_adaptive_random_bucket() && !request->eos()
                               ? request->partition_ids_size()
                               : request->tablet_ids_size();
    if (block.rows() != request_rows) {
        cancel(
                fmt::format("{}, err: invalid add block request row count, block rows: {}, "
                            "request rows: {}, adaptive_random_bucket: {}, eos: {}",
                            channel_info(), block.rows(), request_rows,
                            request->is_adaptive_random_bucket(), request->eos()));
        _send_block_callback->clear_in_flight();
        return;
    }
    if (block.rows() > 0) {
        SCOPED_ATOMIC_TIMER(&_serialize_batch_ns);
        size_t uncompressed_bytes = 0, compressed_bytes = 0;
        int64_t compressed_time = 0;
        Status st = block.serialize(state->be_exec_version(), request->mutable_block(),
                                    &uncompressed_bytes, &compressed_bytes, &compressed_time,
                                    state->fragement_transmission_compression_type(),
                                    _parent->_transfer_large_data_by_brpc);
        TEST_INJECTION_POINT_CALLBACK("VNodeChannel::try_send_block", &st);
        if (!st.ok()) {
            cancel(fmt::format("{}, err: {}", channel_info(), st.to_string()));
            _send_block_callback->clear_in_flight();
            return;
        }
        if (double(compressed_bytes) >= double(config::brpc_max_body_size) * 0.95F) {
            LOG(WARNING) << "send block too large, this rpc may failed. send size: "
                         << compressed_bytes << ", threshold: " << config::brpc_max_body_size
                         << ", " << channel_info();
        }
    }

    auto remain_ms = _rpc_timeout_ms - _timeout_watch.elapsed_time() / NANOS_PER_MILLIS;
    if (UNLIKELY(remain_ms < config::min_load_rpc_timeout_ms)) {
        if (remain_ms <= 0 && !request->eos()) {
            cancel(fmt::format("{}, err: load timeout after {} ms", channel_info(),
                               _rpc_timeout_ms));
            _send_block_callback->clear_in_flight();
            return;
        } else {
            remain_ms = config::min_load_rpc_timeout_ms;
        }
    }

    _send_block_callback->reset();
    _send_block_callback->cntl_->set_timeout_ms(remain_ms);
    if (config::tablet_writer_ignore_eovercrowded) {
        _send_block_callback->cntl_->ignore_eovercrowded();
    }

    if (request->eos()) {
        if (!request->is_adaptive_random_bucket() || !request->has_block()) {
            for (auto pid : _parent->_tablet_finder->partition_ids()) {
                request->add_partition_ids(pid);
            }
        }

        request->set_write_single_replica(_parent->_write_single_replica);
        if (_parent->_write_single_replica) {
            for (auto& _slave_tablet_node : _slave_tablet_nodes) {
                PSlaveTabletNodes slave_tablet_nodes;
                for (auto node_id : _slave_tablet_node.second) {
                    const auto* node = _parent->_nodes_info->find_node(node_id);
                    DBUG_EXECUTE_IF("VNodeChannel.try_send_pending_block.slave_node_not_found", {
                        LOG(WARNING) << "trigger "
                                        "VNodeChannel.try_send_pending_block.slave_node_not_found "
                                        "debug point will set node to nullptr";
                        node = nullptr;
                    });
                    if (node == nullptr) {
                        LOG(WARNING) << "slave node not found, node_id=" << node_id;
                        cancel(fmt::format("slave node not found, node_id={}", node_id));
                        _send_block_callback->clear_in_flight();
                        return;
                    }
                    PNodeInfo* pnode = slave_tablet_nodes.add_slave_nodes();
                    pnode->set_id(node->id);
                    pnode->set_option(node->option);
                    pnode->set_host(node->host);
                    pnode->set_async_internal_port(node->brpc_port);
                }
                request->mutable_slave_tablet_nodes()->insert(
                        {_slave_tablet_node.first, slave_tablet_nodes});
            }
        }

        // eos request must be the last request-> it's a signal makeing callback function to set _add_batch_finished true.
        // end_mark makes is_last_rpc true when rpc finished and call callbacks.
        _send_block_callback->end_mark();
        _send_finished = true;
        CHECK(_pending_batches_num == 0) << _pending_batches_num;
    }

    auto send_block_closure = AutoReleaseClosure<
            PTabletWriterAddBlockRequest,
            WriteBlockCallback<PTabletWriterAddBlockResult>>::create_unique(request,
                                                                            _send_block_callback);
    if (_parent->_transfer_large_data_by_brpc && request->has_block() &&
        request->block().has_column_values() && request->ByteSizeLong() > MIN_HTTP_BRPC_SIZE) {
        Status st = request_embed_attachment_contain_blockv2(send_block_closure->request_.get(),
                                                             send_block_closure);
        if (!st.ok()) {
            cancel(fmt::format("{}, err: {}", channel_info(), st.to_string()));
            _send_block_callback->clear_in_flight();
            return;
        }

        std::string host = _node_info.host;
        auto dns_cache = ExecEnv::GetInstance()->dns_cache();
        if (dns_cache == nullptr) {
            LOG(WARNING) << "DNS cache is not initialized, skipping hostname resolve";
        } else if (!is_valid_ip(_node_info.host)) {
            Status status = dns_cache->get(_node_info.host, &host);
            if (!status.ok()) {
                LOG(WARNING) << "failed to get ip from host " << _node_info.host << ": "
                             << status.to_string();
                cancel(fmt::format("failed to get ip from host {}", _node_info.host));
                _send_block_callback->clear_in_flight();
                return;
            }
        }
        //format an ipv6 address
        std::string brpc_url = get_brpc_http_url(host, _node_info.brpc_port);
        std::shared_ptr<PBackendService_Stub> _brpc_http_stub =
                _state->exec_env()->brpc_internal_client_cache()->get_new_client_no_cache(brpc_url,
                                                                                          "http");
        if (_brpc_http_stub == nullptr) {
            cancel(fmt::format("{}, failed to open brpc http client to {}", channel_info(),
                               brpc_url));
            _send_block_callback->clear_in_flight();
            return;
        }
        _send_block_callback->cntl_->http_request().uri() =
                brpc_url + "/PInternalServiceImpl/tablet_writer_add_block_by_http";
        _send_block_callback->cntl_->http_request().set_method(brpc::HTTP_METHOD_POST);
        _send_block_callback->cntl_->http_request().set_content_type("application/json");

        {
            _brpc_http_stub->tablet_writer_add_block_by_http(
                    send_block_closure->cntl_.get(), nullptr, send_block_closure->response_.get(),
                    send_block_closure.get());
            send_block_closure.release();
        }
    } else {
        _send_block_callback->cntl_->http_request().Clear();
        {
            _stub->tablet_writer_add_block(
                    send_block_closure->cntl_.get(), send_block_closure->request_.get(),
                    send_block_closure->response_.get(), send_block_closure.get());
            send_block_closure.release();
        }
    }

    _next_packet_seq++;
}

void VNodeChannel::_add_block_success_callback(const PTabletWriterAddBlockResult& result,
                                               const WriteBlockCallbackContext& ctx) {
    std::lock_guard<std::mutex> l(this->_closed_lock);
    if (this->_is_closed) {
        // if the node channel is closed, no need to call the following logic,
        // and notice that _index_channel may already be destroyed.
        return;
    }
    SCOPED_ATTACH_TASK(_state);
    Status status(Status::create(result.status()));
    if (status.ok()) {
        _refresh_back_pressure_version_wait_time(result.tablet_load_rowset_num_infos());
        // if has error tablet, handle them first
        for (const auto& error : result.tablet_errors()) {
            _index_channel->mark_as_failed(this, "tablet error: " + error.msg(), error.tablet_id());
        }

        Status st = _index_channel->check_intolerable_failure();
        if (!st.ok()) {
            _cancel_with_msg(st.to_string());
        } else if (ctx._is_last_rpc) {
            for (const auto& tablet : result.tablet_vec()) {
                DBUG_EXECUTE_IF("VNodeChannel.add_block_success_callback.incomplete_commit_info", {
                    auto target_tablet_id = dp->param<int64_t>("tablet_id", -1);
                    if (tablet.tablet_id() == target_tablet_id) {
                        LOG(INFO) << "skip tablet info: " << tablet.tablet_id()
                                  << ", backend_id: " << _node_id;
                        continue;
                    }
                });
                TTabletCommitInfo commit_info;
                commit_info.tabletId = tablet.tablet_id();
                commit_info.backendId = _node_id;
                _tablet_commit_infos.emplace_back(std::move(commit_info));
                if (tablet.has_received_rows()) {
                    _tablets_received_rows.emplace_back(tablet.tablet_id(), tablet.received_rows());
                }
                if (tablet.has_num_rows_filtered()) {
                    _tablets_filtered_rows.emplace_back(tablet.tablet_id(),
                                                        tablet.num_rows_filtered());
                }
                VLOG_CRITICAL << "master replica commit info: tabletId=" << tablet.tablet_id()
                              << ", backendId=" << _node_id
                              << ", master node id: " << this->node_id()
                              << ", host: " << this->host() << ", txn_id=" << _parent->_txn_id;
            }
            if (_parent->_write_single_replica) {
                for (const auto& tablet_slave_node_ids : result.success_slave_tablet_node_ids()) {
                    for (auto slave_node_id : tablet_slave_node_ids.second.slave_node_ids()) {
                        TTabletCommitInfo commit_info;
                        commit_info.tabletId = tablet_slave_node_ids.first;
                        commit_info.backendId = slave_node_id;
                        _tablet_commit_infos.emplace_back(std::move(commit_info));
                        VLOG_CRITICAL
                                << "slave replica commit info: tabletId="
                                << tablet_slave_node_ids.first << ", backendId=" << slave_node_id
                                << ", master node id: " << this->node_id()
                                << ", host: " << this->host() << ", txn_id=" << _parent->_txn_id;
                    }
                }
            }
            _add_batches_finished = true;
            _index_channel->notify_close_wait();
        }
    } else {
        _cancel_with_msg(fmt::format("{}, add batch req success but status isn't ok, err: {}",
                                     channel_info(), status.to_string()));
    }

    if (result.has_execution_time_us()) {
        _add_batch_counter.add_batch_execution_time_us += result.execution_time_us();
        _add_batch_counter.add_batch_wait_execution_time_us += result.wait_execution_time_us();
        _add_batch_counter.add_batch_num++;
    }
    if (result.has_load_channel_profile()) {
        TRuntimeProfileTree tprofile;
        const auto* buf = (const uint8_t*)result.load_channel_profile().data();
        auto len = cast_set<uint32_t>(result.load_channel_profile().size());
        auto st = deserialize_thrift_msg(buf, &len, false, &tprofile);
        if (st.ok()) {
            _state->load_channel_profile()->update(tprofile);
        } else {
            LOG(WARNING) << "load channel TRuntimeProfileTree deserialize failed, errmsg=" << st;
        }
    }
}

void VNodeChannel::_add_block_failed_callback(const WriteBlockCallbackContext& ctx) {
    std::lock_guard<std::mutex> l(this->_closed_lock);
    if (this->_is_closed) {
        // if the node channel is closed, no need to call `mark_as_failed`,
        // and notice that _index_channel may already be destroyed.
        return;
    }
    SCOPED_ATTACH_TASK(_state);
    // If rpc failed, mark all tablets on this node channel as failed
    _index_channel->mark_as_failed(this,
                                   fmt::format("rpc failed, error code:{}, error text:{}",
                                               _send_block_callback->cntl_->ErrorCode(),
                                               _send_block_callback->cntl_->ErrorText()),
                                   -1);
    if (_send_block_callback->cntl_->ErrorText().find("Reached timeout") != std::string::npos) {
        LOG(WARNING) << "rpc failed may caused by timeout. increase BE config "
                        "`min_load_rpc_timeout_ms` of to avoid this if you are sure that your "
                        "table building and data are reasonable.";
    }
    Status st = _index_channel->check_intolerable_failure();
    if (!st.ok()) {
        _cancel_with_msg(fmt::format("{}, err: {}", channel_info(), st.to_string()));
    } else if (ctx._is_last_rpc) {
        // if this is last rpc, will must set _add_batches_finished. otherwise, node channel's close_wait
        // will be blocked.
        _add_batches_finished = true;
        _index_channel->notify_close_wait();
    }
}

// When _cancelled is true, we still need to send a tablet_writer_cancel
// rpc request to truly release the load channel
void VNodeChannel::cancel(const std::string& cancel_msg) {
    if (_is_closed) {
        // skip the channels that have been canceled or close_wait.
        return;
    }
    SCOPED_CONSUME_MEM_TRACKER(_node_channel_tracker.get());
    // set _is_closed to true finally
    Defer set_closed {[&]() {
        std::lock_guard<std::mutex> l(_closed_lock);
        _is_closed = true;
    }};
    // we don't need to wait last rpc finished, cause closure's release/reset will join.
    // But do we need brpc::StartCancel(call_id)?
    _cancel_with_msg(cancel_msg);
    // if not inited, _stub will be nullptr, skip sending cancel rpc
    if (!_inited) {
        return;
    }

    auto request = std::make_shared<PTabletWriterCancelRequest>();
    request->mutable_id()->CopyFrom(_parent->_load_id);
    request->set_index_id(_index_channel->_index_id);
    request->set_sender_id(_parent->_sender_id);
    request->set_cancel_reason(cancel_msg);

    // cancel is already in post-processing, so error status could be ignored. so not keeping cancel_callback is acceptable.
    auto cancel_callback = DummyBrpcCallback<PTabletWriterCancelResult>::create_shared();
    auto closure = AutoReleaseClosure<
            PTabletWriterCancelRequest,
            DummyBrpcCallback<PTabletWriterCancelResult>>::create_unique(request, cancel_callback);

    auto remain_ms = _rpc_timeout_ms - _timeout_watch.elapsed_time() / NANOS_PER_MILLIS;
    if (UNLIKELY(remain_ms < config::min_load_rpc_timeout_ms)) {
        remain_ms = config::min_load_rpc_timeout_ms;
    }
    cancel_callback->cntl_->set_timeout_ms(remain_ms);
    if (config::tablet_writer_ignore_eovercrowded) {
        closure->cntl_->ignore_eovercrowded();
    }
    _stub->tablet_writer_cancel(closure->cntl_.get(), closure->request_.get(),
                                closure->response_.get(), closure.get());
    closure.release();
}

Status VNodeChannel::close_wait(RuntimeState* state, bool* is_closed) {
    DBUG_EXECUTE_IF("VNodeChannel.close_wait_full_gc", {
        std::thread t(injection_full_gc_fn);
        t.join();
    });
    SCOPED_CONSUME_MEM_TRACKER(_node_channel_tracker.get());

    *is_closed = true;

    auto st = none_of({_cancelled, !_eos_is_produced});
    if (!st.ok()) {
        if (_cancelled) {
            std::lock_guard<std::mutex> l(_cancel_msg_lock);
            return Status::Error<ErrorCode::INTERNAL_ERROR, false>("wait close failed. {}",
                                                                   _cancel_msg);
        } else {
            return std::move(
                    st.prepend("already stopped, skip waiting for close. cancelled/!eos: "));
        }
    }

    DBUG_EXECUTE_IF("VNodeChannel.close_wait.cancelled", {
        _cancelled = true;
        _cancel_msg = "injected cancel";
    });

    if (state->is_cancelled()) {
        _cancel_with_msg(state->cancel_reason().to_string());
    }

    // Waiting for finished until _add_batches_finished changed by rpc's finished callback.
    // it may take a long time, so we couldn't set a timeout
    // For pipeline engine, the close is called in async writer's process block method,
    // so that it will not block pipeline thread.
    if (!_add_batches_finished && !_cancelled && !state->is_cancelled()) {
        *is_closed = false;
        return Status::OK();
    }
    VLOG_CRITICAL << _parent->_sender_id << " close wait finished";
    return Status::OK();
}

Status VNodeChannel::after_close_handle(
        RuntimeState* state, WriterStats* writer_stats,
        std::unordered_map<int64_t, AddBatchCounter>* node_add_batch_counter_map) {
    Status st = Status::Error<ErrorCode::INTERNAL_ERROR, false>(get_cancel_msg());
    _close_time_ms = UnixMillis() - _close_time_ms;

    if (_add_batches_finished) {
        _close_check();
        _state->add_tablet_commit_infos(_tablet_commit_infos);

        _index_channel->set_error_tablet_in_state(state);
        _index_channel->set_tablets_received_rows(_tablets_received_rows, _node_id);
        _index_channel->set_tablets_filtered_rows(_tablets_filtered_rows, _node_id);

        std::lock_guard<std::mutex> l(_closed_lock);
        // only when normal close, we set _is_closed to true.
        // otherwise, we will set it to true in cancel().
        _is_closed = true;
        st = Status::OK();
    }

    time_report(node_add_batch_counter_map, writer_stats);
    return st;
}

Status VNodeChannel::check_status() {
    return none_of({_cancelled, !_eos_is_produced});
}

void VNodeChannel::_close_check() {
    std::lock_guard<std::mutex> lg(_pending_batches_lock);
    CHECK(_pending_blocks.empty()) << name();
    CHECK(_cur_mutable_block == nullptr) << name();
}

void VNodeChannel::mark_close(bool hang_wait) {
    auto st = none_of({_cancelled, _eos_is_produced});
    if (!st.ok()) {
        return;
    }

    bool need_adaptive_random_bucket_eos = _cur_add_block_request->is_adaptive_random_bucket();
    {
        std::lock_guard<std::mutex> l(_pending_batches_lock);
        if (!_cur_mutable_block) [[unlikely]] {
            // never had a block arrived. add a dummy block
            _cur_mutable_block = MutableBlock::create_unique();
        }
        if (need_adaptive_random_bucket_eos && _cur_mutable_block->rows() > 0) {
            _cur_add_block_request->set_eos(false);
            auto tmp_add_block_request =
                    std::make_shared<PTabletWriterAddBlockRequest>(*_cur_add_block_request);
            _pending_blocks.emplace(std::move(_cur_mutable_block), tmp_add_block_request);
            _pending_batches_num++;
            _cur_add_block_request->clear_tablet_ids();
            _cur_add_block_request->clear_partition_ids();
            _cur_mutable_block = MutableBlock::create_unique();
        }
        _cur_add_block_request->set_eos(true);
        _cur_add_block_request->set_hang_wait(hang_wait);
        auto tmp_add_block_request =
                std::make_shared<PTabletWriterAddBlockRequest>(*_cur_add_block_request);
        // when prepare to close, add block to queue so that try_send_pending_block thread will send it.
        _pending_blocks.emplace(std::move(_cur_mutable_block), tmp_add_block_request);
        _pending_batches_num++;
        DCHECK(_pending_blocks.back().second->eos());
        _close_time_ms = UnixMillis();
        LOG(INFO) << channel_info()
                  << " mark closed, left pending batch size: " << _pending_blocks.size()
                  << " hang_wait: " << hang_wait;
    }

    _eos_is_produced = true;
}

VTabletWriter::VTabletWriter(const TDataSink& t_sink, const VExprContextSPtrs& output_exprs,
                             std::shared_ptr<Dependency> dep, std::shared_ptr<Dependency> fin_dep)
        : AsyncResultWriter(output_exprs, dep, fin_dep), _t_sink(t_sink) {
    _transfer_large_data_by_brpc = config::transfer_large_data_by_brpc;
}

void VTabletWriter::_send_batch_process() {
    SCOPED_TIMER(_non_blocking_send_timer);
    SCOPED_ATTACH_TASK(_state);
    SCOPED_CONSUME_MEM_TRACKER(_mem_tracker);

    int sleep_time = int(config::olap_table_sink_send_interval_microseconds *
                         (_vpartition->is_auto_partition()
                                  ? config::olap_table_sink_send_interval_auto_partition_factor
                                  : 1));

    while (true) {
        // incremental open will temporarily make channels into abnormal state. stop checking when this.
        std::unique_lock<bthread::Mutex> l(_stop_check_channel);

        int running_channels_num = 0;
        int opened_nodes = 0;
        for (const auto& index_channel : _channels) {
            index_channel->for_each_node_channel([&running_channels_num,
                                                  this](const std::shared_ptr<VNodeChannel>& ch) {
                // if this channel all completed(cancelled), got 0. else 1.
                running_channels_num +=
                        ch->try_send_and_fetch_status(_state, this->_send_batch_thread_pool_token);
            });
            opened_nodes += index_channel->num_node_channels();
        }

        // auto partition table may have no node channel temporarily. wait to open.
        if (opened_nodes != 0 && running_channels_num == 0) {
            LOG(INFO) << "All node channels are stopped(maybe finished/offending/cancelled), "
                         "sender thread exit. "
                      << print_id(_load_id);
            return;
        }

        // for auto partition tables, there's a situation: we haven't open any node channel but decide to cancel the task.
        // then the judge in front will never be true because opened_nodes won't increase. so we have to specially check wether we called close.
        // we must RECHECK opened_nodes below, after got closed signal, because it may changed. Think of this:
        //      checked opened_nodes = 0 ---> new block arrived ---> task finished, close() was called ---> we got _try_close here
        // if we don't check again, we may lose the last package.
        if (_try_close.load(std::memory_order_acquire)) {
            opened_nodes = 0;
            std::ranges::for_each(_channels,
                                  [&opened_nodes](const std::shared_ptr<IndexChannel>& ich) {
                                      opened_nodes += ich->num_node_channels();
                                  });
            if (opened_nodes == 0) {
                LOG(INFO) << "No node channel have ever opened but now we have to close. sender "
                             "thread exit. "
                          << print_id(_load_id);
                return;
            }
        }
        bthread_usleep(sleep_time);
    }
}

static void* periodic_send_batch(void* writer) {
    auto* tablet_writer = (VTabletWriter*)(writer);
    tablet_writer->_send_batch_process();
    return nullptr;
}

Status VTabletWriter::open(doris::RuntimeState* state, doris::RuntimeProfile* profile) {
    RETURN_IF_ERROR(_init(state, profile));
    signal::set_signal_task_id(_load_id);
    SCOPED_TIMER(profile->total_time_counter());
    SCOPED_TIMER(_open_timer);
    SCOPED_CONSUME_MEM_TRACKER(_mem_tracker.get());

    fmt::memory_buffer buf;
    for (const auto& index_channel : _channels) {
        fmt::format_to(buf, "index id:{}", index_channel->_index_id);
        index_channel->for_each_node_channel(
                [](const std::shared_ptr<VNodeChannel>& ch) { ch->open(); });
    }
    VLOG_DEBUG << "list of open index id = " << fmt::to_string(buf);

    for (const auto& index_channel : _channels) {
        index_channel->set_start_time(UnixMillis());
        index_channel->for_each_node_channel([&index_channel](
                                                     const std::shared_ptr<VNodeChannel>& ch) {
            auto st = ch->open_wait();
            if (!st.ok()) {
                // The open() phase is mainly to generate DeltaWriter instances on the nodes corresponding to each node channel.
                // This phase will not fail due to a single tablet.
                // Therefore, if the open() phase fails, all tablets corresponding to the node need to be marked as failed.
                index_channel->mark_as_failed(
                        ch.get(),
                        fmt::format("{}, open failed, err: {}", ch->channel_info(), st.to_string()),
                        -1);
            }
        });

        RETURN_IF_ERROR(index_channel->check_intolerable_failure());
    }
    _send_batch_thread_pool_token = state->exec_env()->send_batch_thread_pool()->new_token(
            ThreadPool::ExecutionMode::CONCURRENT, _send_batch_parallelism);

    // start to send batch continually. this must be called after _init
    if (bthread_start_background(&_sender_thread, nullptr, periodic_send_batch, (void*)this) != 0) {
        return Status::Error<ErrorCode::INTERNAL_ERROR>("bthread_start_backgroud failed");
    }
    return Status::OK();
}

Status VTabletWriter::on_partitions_created(TCreatePartitionResult* result) {
    // add new tablet locations. it will use by address. so add to pool
    auto* new_locations = _pool->add(new std::vector<TTabletLocation>(result->tablets));
    _location->add_locations(*new_locations);
    if (_write_single_replica) {
        auto* slave_locations = _pool->add(new std::vector<TTabletLocation>(result->slave_tablets));
        _slave_location->add_locations(*slave_locations);
    }

    // update new node info
    _nodes_info->add_nodes(result->nodes);

    // incremental open node channel
    RETURN_IF_ERROR(_incremental_open_node_channel(result->partitions));

    return Status::OK();
}

static Status on_partitions_created(void* writer, TCreatePartitionResult* result) {
    return static_cast<VTabletWriter*>(writer)->on_partitions_created(result);
}

Status VTabletWriter::_init_row_distribution() {
    _row_distribution.init({.state = _state,
                            .block_convertor = _block_convertor.get(),
                            .tablet_finder = _tablet_finder.get(),
                            .vpartition = _vpartition,
                            .add_partition_request_timer = _add_partition_request_timer,
                            .txn_id = _txn_id,
                            .pool = _pool,
                            .location = _location,
                            .vec_output_expr_ctxs = &_vec_output_expr_ctxs,
                            .schema = _schema,
                            .caller = this,
                            .write_single_replica = _write_single_replica,
                            .create_partition_callback = &::doris::on_partitions_created});

    return _row_distribution.open(_output_row_desc);
}

Status VTabletWriter::_init(RuntimeState* state, RuntimeProfile* profile) {
    DCHECK(_t_sink.__isset.olap_table_sink);
    _pool = state->obj_pool();
    auto& table_sink = _t_sink.olap_table_sink;
    _load_id.set_hi(table_sink.load_id.hi);
    _load_id.set_lo(table_sink.load_id.lo);
    _txn_id = table_sink.txn_id;
    _num_replicas = table_sink.num_replicas;
    _tuple_desc_id = table_sink.tuple_id;
    _write_file_cache = table_sink.write_file_cache;
    _schema.reset(new OlapTableSchemaParam());
    RETURN_IF_ERROR(_schema->init(table_sink.schema));
    bool has_row_binlog = std::any_of(_schema->indexes().begin(), _schema->indexes().end(),
                                      [](const auto* index) { return index->row_binlog_id > 0; });
    if (has_row_binlog) {
        _row_binlog_lsn_buffer = GlobalAutoIncBuffers::GetInstance()->get_auto_inc_buffer(
                _schema->db_id(), _schema->table_id(), kBinlogLsnAutoIncId);
    }
    _schema->set_timestamp_ms(state->timestamp_ms());
    _schema->set_nano_seconds(state->nano_seconds());
    _schema->set_timezone(state->timezone());
    _location = _pool->add(new OlapTableLocationParam(table_sink.location));
    _nodes_info = _pool->add(new DorisNodesInfo(table_sink.nodes_info));
    if (table_sink.__isset.write_single_replica && table_sink.write_single_replica) {
        _write_single_replica = true;
        _slave_location = _pool->add(new OlapTableLocationParam(table_sink.slave_location));
        if (!config::enable_single_replica_load) {
            return Status::InternalError("single replica load is disabled on BE.");
        }
    }

    if (config::is_cloud_mode() &&
        (!table_sink.__isset.txn_timeout_s || table_sink.txn_timeout_s <= 0)) {
        return Status::InternalError("The txn_timeout_s of TDataSink is invalid");
    }
    _txn_expiration = ::time(nullptr) + table_sink.txn_timeout_s;

    if (table_sink.__isset.load_channel_timeout_s) {
        _load_channel_timeout_s = table_sink.load_channel_timeout_s;
    } else {
        _load_channel_timeout_s = config::streaming_load_rpc_max_alive_time_sec;
    }
    if (table_sink.__isset.send_batch_parallelism && table_sink.send_batch_parallelism > 1) {
        _send_batch_parallelism = table_sink.send_batch_parallelism;
    }
    // If distributed column list is empty, the table uses random distribution.
    // Mode priority (highest to lowest):
    //   1. FIND_TABLET_EVERY_SINK: load_to_single_tablet=true (legacy single-tablet mode).
    //   2. FIND_TABLET_RANDOM_BUCKET: FE set enable_adaptive_random_bucket on the sink,
    //      meaning enable_adaptive_random_bucket_load is ON. Using a sink-level flag (mirroring
    //      load_to_single_tablet) ensures the mode is fixed correctly when the initial
    //      partition list is empty (e.g. auto-partition tables on first load).
    //   3. FIND_TABLET_EVERY_BATCH: default round-robin per batch.
    auto find_tablet_mode = OlapTabletFinder::FindTabletMode::FIND_TABLET_EVERY_ROW;
    if (table_sink.partition.distributed_columns.empty()) {
        if (table_sink.__isset.load_to_single_tablet && table_sink.load_to_single_tablet) {
            find_tablet_mode = OlapTabletFinder::FindTabletMode::FIND_TABLET_EVERY_SINK;
        } else if (table_sink.__isset.enable_adaptive_random_bucket &&
                   table_sink.enable_adaptive_random_bucket && config::is_cloud_mode()) {
            find_tablet_mode = OlapTabletFinder::FindTabletMode::FIND_TABLET_RANDOM_BUCKET;
        } else {
            find_tablet_mode = OlapTabletFinder::FindTabletMode::FIND_TABLET_EVERY_BATCH;
        }
    }
    _vpartition = _pool->add(new doris::VOlapTablePartitionParam(_schema, table_sink.partition));
    _tablet_finder = std::make_unique<OlapTabletFinder>(_vpartition, find_tablet_mode);
    RETURN_IF_ERROR(_vpartition->init());

    _state = state;
    _operator_profile = profile;

    _sender_id = state->per_fragment_instance_idx();
    _num_senders = state->num_per_fragment_instances();
    _is_high_priority =
            (state->execution_timeout() <= config::load_task_high_priority_threshold_second);
    DBUG_EXECUTE_IF("VTabletWriter._init.is_high_priority", { _is_high_priority = true; });
    // profile must add to state's object pool
    _mem_tracker =
            std::make_shared<MemTracker>("OlapTableSink:" + std::to_string(state->load_job_id()));
    SCOPED_TIMER(profile->total_time_counter());
    SCOPED_CONSUME_MEM_TRACKER(_mem_tracker.get());

    // get table's tuple descriptor
    _output_tuple_desc = state->desc_tbl().get_tuple_descriptor(_tuple_desc_id);
    if (_output_tuple_desc == nullptr) {
        LOG(WARNING) << "unknown destination tuple descriptor, id=" << _tuple_desc_id;
        return Status::InternalError("unknown destination tuple descriptor");
    }

    if (!_vec_output_expr_ctxs.empty() &&
        _output_tuple_desc->slots().size() != _vec_output_expr_ctxs.size()) {
        LOG(WARNING) << "output tuple slot num should be equal to num of output exprs, "
                     << "output_tuple_slot_num " << _output_tuple_desc->slots().size()
                     << " output_expr_num " << _vec_output_expr_ctxs.size();
        return Status::InvalidArgument(
                "output_tuple_slot_num {} should be equal to output_expr_num {}",
                _output_tuple_desc->slots().size(), _vec_output_expr_ctxs.size());
    }

    _block_convertor = std::make_unique<OlapTableBlockConvertor>(_output_tuple_desc);
    // if partition_type is OLAP_TABLE_SINK_HASH_PARTITIONED, we handle the processing of auto_increment column
    // on exchange node rather than on TabletWriter
    _block_convertor->init_autoinc_info(
            _schema->db_id(), _schema->table_id(), _state->batch_size(),
            _schema->is_fixed_partial_update() && !_schema->auto_increment_coulumn().empty(),
            _schema->auto_increment_column_unique_id());
    _output_row_desc = _pool->add(new RowDescriptor(_output_tuple_desc));

    // add all counter
    _input_rows_counter = ADD_COUNTER(profile, "RowsRead", TUnit::UNIT);
    _output_rows_counter = ADD_COUNTER(profile, "RowsProduced", TUnit::UNIT);
    _filtered_rows_counter = ADD_COUNTER(profile, "RowsFiltered", TUnit::UNIT);
    _send_data_timer = ADD_TIMER(profile, "SendDataTime");
    _wait_mem_limit_timer = ADD_CHILD_TIMER(profile, "WaitMemLimitTime", "SendDataTime");
    _row_distribution_timer = ADD_CHILD_TIMER(profile, "RowDistributionTime", "SendDataTime");
    _filter_timer = ADD_CHILD_TIMER(profile, "FilterTime", "SendDataTime");
    _where_clause_timer = ADD_CHILD_TIMER(profile, "WhereClauseTime", "SendDataTime");
    _append_node_channel_timer = ADD_CHILD_TIMER(profile, "AppendNodeChannelTime", "SendDataTime");
    _add_partition_request_timer =
            ADD_CHILD_TIMER(profile, "AddPartitionRequestTime", "SendDataTime");
    _validate_data_timer = ADD_TIMER(profile, "ValidateDataTime");
    _open_timer = ADD_TIMER(profile, "OpenTime");
    _close_timer = ADD_TIMER(profile, "CloseWaitTime");
    _non_blocking_send_timer = ADD_TIMER(profile, "NonBlockingSendTime");
    _non_blocking_send_work_timer =
            ADD_CHILD_TIMER(profile, "NonBlockingSendWorkTime", "NonBlockingSendTime");
    _serialize_batch_timer =
            ADD_CHILD_TIMER(profile, "SerializeBatchTime", "NonBlockingSendWorkTime");
    _total_add_batch_exec_timer = ADD_TIMER(profile, "TotalAddBatchExecTime");
    _max_add_batch_exec_timer = ADD_TIMER(profile, "MaxAddBatchExecTime");
    _total_wait_exec_timer = ADD_TIMER(profile, "TotalWaitExecTime");
    _max_wait_exec_timer = ADD_TIMER(profile, "MaxWaitExecTime");
    _add_batch_number = ADD_COUNTER(profile, "NumberBatchAdded", TUnit::UNIT);
    _num_node_channels = ADD_COUNTER(profile, "NumberNodeChannels", TUnit::UNIT);
    _load_back_pressure_version_time_ms = ADD_TIMER(profile, "LoadBackPressureVersionTimeMs");

#ifdef DEBUG
    // check: tablet ids should be unique
    {
        std::unordered_set<int64_t> tablet_ids;
        const auto& partitions = _vpartition->get_partitions();
        for (int i = 0; i < _schema->indexes().size(); ++i) {
            for (const auto& partition : partitions) {
                for (const auto& tablet : partition->indexes[i].tablets) {
                    CHECK(tablet_ids.count(tablet) == 0) << "found duplicate tablet id: " << tablet;
                    tablet_ids.insert(tablet);
                }
            }
        }
    }
#endif

    // open all channels
    const auto& partitions = _vpartition->get_partitions();
    for (int i = 0; i < _schema->indexes().size(); ++i) {
        // collect all tablets belong to this rollup
        std::vector<TTabletWithPartition> tablets;
        auto* index = _schema->indexes()[i];
        for (const auto& part : partitions) {
            for (const auto& tablet : part->indexes[i].tablets) {
                TTabletWithPartition tablet_with_partition;
                tablet_with_partition.partition_id = part->id;
                tablet_with_partition.tablet_id = tablet;
                tablets.emplace_back(std::move(tablet_with_partition));
                _build_tablet_replica_info(tablet, part);
            }
        }
        if (tablets.empty() && !_vpartition->is_auto_partition()) {
            LOG(WARNING) << "load job:" << state->load_job_id() << " index: " << index->index_id
                         << " would open 0 tablet";
        }
        _channels.emplace_back(new IndexChannel(this, index->index_id, index->where_clause));
        _index_id_to_channel[index->index_id] = _channels.back();
        RETURN_IF_ERROR(_channels.back()->init(state, tablets));
    }

    RETURN_IF_ERROR(_init_row_distribution());

    _inited = true;
    return Status::OK();
}

Status VTabletWriter::_incremental_open_node_channel(
        const std::vector<TOlapTablePartition>& partitions) {
    // do what we did in prepare() for partitions. indexes which don't change when we create new partition is orthogonal to partitions.
    std::unique_lock<bthread::Mutex> _l(_stop_check_channel);
    for (int i = 0; i < _schema->indexes().size(); ++i) {
        const OlapTableIndexSchema* index = _schema->indexes()[i];
        std::vector<TTabletWithPartition> tablets;
        for (const auto& t_part : partitions) {
            VOlapTablePartition* part = nullptr;
            RETURN_IF_ERROR(_vpartition->generate_partition_from(t_part, part));
            for (const auto& tablet : part->indexes[i].tablets) {
                TTabletWithPartition tablet_with_partition;
                tablet_with_partition.partition_id = part->id;
                tablet_with_partition.tablet_id = tablet;
                tablets.emplace_back(std::move(tablet_with_partition));
                _build_tablet_replica_info(tablet, part);
            }
            DCHECK(!tablets.empty()) << "incremental open got nothing!";
        }
        // update and reinit for existing channels.
        std::shared_ptr<IndexChannel> channel = _index_id_to_channel[index->index_id];
        DCHECK(channel != nullptr);
        RETURN_IF_ERROR(channel->init(_state, tablets, true)); // add tablets into it
    }

    fmt::memory_buffer buf;
    for (auto& channel : _channels) {
        // incremental open new partition's tablet on storage side
        channel->for_each_node_channel(
                [](const std::shared_ptr<VNodeChannel>& ch) { ch->incremental_open(); });
        fmt::format_to(buf, "index id:{}", channel->_index_id);
        VLOG_DEBUG << "list of open index id = " << fmt::to_string(buf);

        channel->for_each_node_channel([&channel](const std::shared_ptr<VNodeChannel>& ch) {
            auto st = ch->open_wait();
            if (!st.ok()) {
                // The open() phase is mainly to generate DeltaWriter instances on the nodes corresponding to each node channel.
                // This phase will not fail due to a single tablet.
                // Therefore, if the open() phase fails, all tablets corresponding to the node need to be marked as failed.
                channel->mark_as_failed(
                        ch.get(),
                        fmt::format("{}, open failed, err: {}", ch->channel_info(), st.to_string()),
                        -1);
            }
        });

        RETURN_IF_ERROR(channel->check_intolerable_failure());
    }

    return Status::OK();
}

void VTabletWriter::_build_tablet_replica_info(const int64_t tablet_id,
                                               VOlapTablePartition* partition) {
    if (partition != nullptr) {
        int total_replicas_num =
                partition->total_replica_num == 0 ? _num_replicas : partition->total_replica_num;
        int load_required_replicas_num = partition->load_required_replica_num == 0
                                                 ? (_num_replicas + 1) / 2
                                                 : partition->load_required_replica_num;
        _tablet_replica_info.emplace(
                tablet_id, std::make_pair(total_replicas_num, load_required_replicas_num));
        // Copy version gap backends info for this tablet
        if (auto it = partition->tablet_version_gap_backends.find(tablet_id);
            it != partition->tablet_version_gap_backends.end()) {
            _tablet_version_gap_backends[tablet_id] = it->second;
        }
    } else {
        _tablet_replica_info.emplace(tablet_id,
                                     std::make_pair(_num_replicas, (_num_replicas + 1) / 2));
    }
}

void VTabletWriter::_cancel_all_channel(Status status) {
    for (const auto& index_channel : _channels) {
        index_channel->for_each_node_channel([&status](const std::shared_ptr<VNodeChannel>& ch) {
            ch->cancel(status.to_string());
        });
    }
    LOG(INFO) << fmt::format(
            "close olap table sink. load_id={}, txn_id={}, canceled all node channels due to "
            "error: {}",
            print_id(_load_id), _txn_id, status);
}

Status VTabletWriter::_send_new_partition_batch() {
    if (_row_distribution.need_deal_batching()) { // maybe try_close more than 1 time
        RETURN_IF_ERROR(_row_distribution.automatic_create_partition());

        Block tmp_block = _row_distribution._batching_block->to_block(); // Borrow out, for lval ref

        // these order is unique.
        //  1. clear batching stats(and flag goes true) so that we won't make a new batching process in dealing batched block.
        //  2. deal batched block
        //  3. now reuse the column of lval block. cuz write doesn't real adjust it. it generate a new block from that.
        _row_distribution.clear_batching_stats();
        Defer recover_batching_block([&]() {
            _row_distribution._batching_block->set_mutable_columns(
                    std::move(tmp_block).mutate_columns());
            _row_distribution._batching_block->clear_column_data();
        });
        RETURN_IF_ERROR(this->write(_state, tmp_block));
        _row_distribution._deal_batched = false;
    }
    return Status::OK();
}

void VTabletWriter::_do_try_close(RuntimeState* state, const Status& exec_status) {
    SCOPED_TIMER(_close_timer);
    Status status = exec_status;

    // must before set _try_close
    if (status.ok()) {
        SCOPED_TIMER(_operator_profile->total_time_counter());
        _row_distribution._deal_batched = true;
        status = _send_new_partition_batch();
    }

    _try_close.store(true, std::memory_order_release); // will stop periodic thread
    if (status.ok()) {
        // BE id -> add_batch method counter
        std::unordered_map<int64_t, AddBatchCounter> node_add_batch_counter_map;

        // only if status is ok can we call this _profile->total_time_counter().
        // if status is not ok, this sink may not be prepared, so that _profile is null
        SCOPED_TIMER(_operator_profile->total_time_counter());
        for (const auto& index_channel : _channels) {
            // two-step mark close. first we send close_origin to recievers to close all originly exist TabletsChannel.
            // when they all closed, we are sure all Writer of instances called _do_try_close. that means no new channel
            // will be opened. the refcount of recievers will be monotonically decreasing. then we are safe to close all
            // our channels.
            if (index_channel->has_incremental_node_channel()) {
                if (!status.ok()) {
                    break;
                }
                VLOG_TRACE << _sender_id << " first stage close start " << _txn_id;
                index_channel->for_init_node_channel(
                        [&index_channel, &status, this](const std::shared_ptr<VNodeChannel>& ch) {
                            if (!status.ok() || ch->is_closed()) {
                                return;
                            }
                            VLOG_DEBUG << index_channel->_parent->_sender_id << "'s " << ch->host()
                                       << "mark close1 for inits " << _txn_id;
                            ch->mark_close(true);
                            if (ch->is_cancelled()) {
                                status = cancel_channel_and_check_intolerable_failure(
                                        std::move(status), ch->get_cancel_msg(), *index_channel,
                                        *ch);
                            }
                        });
                if (!status.ok()) {
                    break;
                }
                // Do not need to wait after quorum success,
                // for first-stage close_wait only ensure incremental node channels load has been completed,
                // unified waiting in the second-stage close_wait.
                status = index_channel->close_wait(_state, nullptr, nullptr,
                                                   index_channel->init_node_channel_ids(), false);
                if (!status.ok()) {
                    break;
                }
                VLOG_DEBUG << _sender_id << " first stage finished. closeing inc nodes " << _txn_id;
                index_channel->for_inc_node_channel(
                        [&index_channel, &status, this](const std::shared_ptr<VNodeChannel>& ch) {
                            if (!status.ok() || ch->is_closed()) {
                                return;
                            }
                            // only first try close, all node channels will mark_close()
                            VLOG_DEBUG << index_channel->_parent->_sender_id << "'s " << ch->host()
                                       << "mark close2 for inc " << _txn_id;
                            ch->mark_close();
                            if (ch->is_cancelled()) {
                                status = cancel_channel_and_check_intolerable_failure(
                                        std::move(status), ch->get_cancel_msg(), *index_channel,
                                        *ch);
                            }
                        });
            } else { // not has_incremental_node_channel
                VLOG_TRACE << _sender_id << " has no incremental channels " << _txn_id;
                index_channel->for_each_node_channel(
                        [&index_channel, &status](const std::shared_ptr<VNodeChannel>& ch) {
                            if (!status.ok() || ch->is_closed()) {
                                return;
                            }
                            // only first try close, all node channels will mark_close()
                            ch->mark_close();
                            if (ch->is_cancelled()) {
                                status = cancel_channel_and_check_intolerable_failure(
                                        std::move(status), ch->get_cancel_msg(), *index_channel,
                                        *ch);
                            }
                        });
            }
        } // end for index channels
    }

    if (!status.ok()) {
        _cancel_all_channel(status);
        _close_status = status;
    }
}

Status VTabletWriter::close(Status exec_status) {
    if (!_inited) {
        DCHECK(!exec_status.ok());
        _cancel_all_channel(exec_status);
        _close_status = exec_status;
        return _close_status;
    }

    SCOPED_TIMER(_close_timer);
    SCOPED_TIMER(_operator_profile->total_time_counter());

    // will make the last batch of request-> close_wait will wait this finished.
    _do_try_close(_state, exec_status);
    TEST_INJECTION_POINT("VOlapTableSink::close");

    DBUG_EXECUTE_IF("VTabletWriter.close.sleep", {
        auto sleep_sec = dp->param<int32_t>("sleep_sec", 1);
        auto token = dp->param<std::string>("token", "");
        LOG(INFO) << "hit debug point VTabletWriter.close.sleep, token=" << token;
        std::this_thread::sleep_for(std::chrono::seconds(sleep_sec));
    });
    DBUG_EXECUTE_IF("VTabletWriter.close.close_status_not_ok",
                    { _close_status = Status::InternalError("injected close status not ok"); });

    // If _close_status is not ok, all nodes have been canceled in try_close.
    if (_close_status.ok()) {
        auto status = Status::OK();
        // BE id -> add_batch method counter
        std::unordered_map<int64_t, AddBatchCounter> node_add_batch_counter_map;
        WriterStats writer_stats;

        for (const auto& index_channel : _channels) {
            if (!status.ok()) {
                break;
            }
            int64_t add_batch_exec_time = 0;
            int64_t wait_exec_time = 0;
            status = index_channel->close_wait(_state, &writer_stats, &node_add_batch_counter_map,
                                               index_channel->each_node_channel_ids(), true);

            // Due to the non-determinism of compaction, the rowsets of each replica may be different from each other on different
            // BE nodes. The number of rows filtered in SegmentWriter depends on the historical rowsets located in the correspoding
            // BE node. So we check the number of rows filtered on each succeccful BE to ensure the consistency of the current load
            if (status.ok() && !_write_single_replica && _schema->is_strict_mode() &&
                _schema->is_partial_update()) {
                if (Status st = index_channel->check_tablet_filtered_rows_consistency(); !st.ok()) {
                    status = st;
                } else {
                    _state->set_num_rows_filtered_in_strict_mode_partial_update(
                            index_channel->num_rows_filtered());
                }
            }

            writer_stats.num_node_channels += index_channel->num_node_channels();
            writer_stats.max_add_batch_exec_time_ns =
                    std::max(add_batch_exec_time, writer_stats.max_add_batch_exec_time_ns);
            writer_stats.max_wait_exec_time_ns =
                    std::max(wait_exec_time, writer_stats.max_wait_exec_time_ns);
        } // end for index channels

        if (status.ok()) {
            // TODO need to be improved
            LOG(INFO) << "total mem_exceeded_block_ns="
                      << writer_stats.channel_stat.mem_exceeded_block_ns
                      << ", total queue_push_lock_ns=" << writer_stats.queue_push_lock_ns
                      << ", total actual_consume_ns=" << writer_stats.actual_consume_ns
                      << ", load id=" << print_id(_load_id) << ", txn_id=" << _txn_id;

            COUNTER_SET(_input_rows_counter, _number_input_rows);
            COUNTER_SET(_output_rows_counter, _number_output_rows);
            COUNTER_SET(_filtered_rows_counter,
                        _block_convertor->num_filtered_rows() +
                                _tablet_finder->num_filtered_rows() +
                                _state->num_rows_filtered_in_strict_mode_partial_update());
            COUNTER_SET(_send_data_timer, _send_data_ns);
            COUNTER_SET(_row_distribution_timer, (int64_t)_row_distribution_watch.elapsed_time());
            COUNTER_SET(_filter_timer, _filter_ns);
            COUNTER_SET(_append_node_channel_timer,
                        writer_stats.channel_stat.append_node_channel_ns);
            COUNTER_SET(_where_clause_timer, writer_stats.channel_stat.where_clause_ns);
            COUNTER_SET(_wait_mem_limit_timer, writer_stats.channel_stat.mem_exceeded_block_ns);
            COUNTER_SET(_validate_data_timer, _block_convertor->validate_data_ns());
            COUNTER_SET(_serialize_batch_timer, writer_stats.serialize_batch_ns);
            COUNTER_SET(_non_blocking_send_work_timer, writer_stats.actual_consume_ns);
            COUNTER_SET(_total_add_batch_exec_timer, writer_stats.total_add_batch_exec_time_ns);
            COUNTER_SET(_max_add_batch_exec_timer, writer_stats.max_add_batch_exec_time_ns);
            COUNTER_SET(_total_wait_exec_timer, writer_stats.total_wait_exec_time_ns);
            COUNTER_SET(_max_wait_exec_timer, writer_stats.max_wait_exec_time_ns);
            COUNTER_SET(_add_batch_number, writer_stats.total_add_batch_num);
            COUNTER_SET(_num_node_channels, writer_stats.num_node_channels);
            COUNTER_SET(_load_back_pressure_version_time_ms,
                        writer_stats.load_back_pressure_version_time_ms);
            g_sink_load_back_pressure_version_time_ms
                    << writer_stats.load_back_pressure_version_time_ms;

            // _number_input_rows don't contain num_rows_load_filtered and num_rows_load_unselected in scan node
            int64_t num_rows_load_total = _number_input_rows + _state->num_rows_load_filtered() +
                                          _state->num_rows_load_unselected();
            _state->set_num_rows_load_total(num_rows_load_total);
            _state->update_num_rows_load_filtered(
                    _block_convertor->num_filtered_rows() + _tablet_finder->num_filtered_rows() +
                    _state->num_rows_filtered_in_strict_mode_partial_update());
            _state->update_num_rows_load_unselected(
                    _tablet_finder->num_immutable_partition_filtered_rows());

            if (_state->enable_profile() && _state->profile_level() >= 2) {
                // Output detailed profiling info for auto-partition requests
                _row_distribution.output_profile_info(_operator_profile);
            }

            // print log of add batch time of all node, for tracing load performance easily
            std::stringstream ss;
            ss << "finished to close olap table sink. load_id=" << print_id(_load_id)
               << ", txn_id=" << _txn_id
               << ", node add batch time(ms)/wait execution time(ms)/close time(ms)/num: ";
            for (auto const& pair : node_add_batch_counter_map) {
                ss << "{" << pair.first << ":(" << (pair.second.add_batch_execution_time_us / 1000)
                   << ")(" << (pair.second.add_batch_wait_execution_time_us / 1000) << ")("
                   << pair.second.close_wait_time_ms << ")(" << pair.second.add_batch_num << ")} ";
            }
            LOG(INFO) << ss.str();
        } else {
            _cancel_all_channel(status);
        }
        _close_status = status;
    }

    // Sender join() must put after node channels mark_close/cancel.
    // But there is no specific sequence required between sender join() & close_wait().
    if (_sender_thread) {
        bthread_join(_sender_thread, nullptr);
        // We have to wait all task in _send_batch_thread_pool_token finished,
        // because it is difficult to handle concurrent problem if we just
        // shutdown it.
        _send_batch_thread_pool_token->wait();
    }

    // We clear NodeChannels' batches here, cuz NodeChannels' batches destruction will use
    // OlapTableSink::_mem_tracker and its parents.
    // But their destructions are after OlapTableSink's.
    for (const auto& index_channel : _channels) {
        index_channel->for_each_node_channel(
                [](const std::shared_ptr<VNodeChannel>& ch) { ch->clear_all_blocks(); });
    }
    return _close_status;
}

Status VTabletWriter::_generate_one_index_channel_payload(
        RowPartTabletIds& row_part_tablet_id, int32_t index_idx,
        ChannelDistributionPayload& channel_payload) {
    auto& row_ids = row_part_tablet_id.row_ids;
    auto& partition_ids = row_part_tablet_id.partition_ids;
    auto& tablet_ids = row_part_tablet_id.tablet_ids;

    size_t row_cnt = row_ids.size();
    bool has_row_binlog = _schema->indexes()[index_idx]->row_binlog_id > 0;
    std::vector<int64_t> row_binlog_lsns;
    if (has_row_binlog && row_cnt > 0) {
        DCHECK(_row_binlog_lsn_buffer != nullptr);
        RETURN_IF_ERROR(allocate_binlog_lsn(_row_binlog_lsn_buffer, row_cnt, row_binlog_lsns));
    }

    for (size_t i = 0; i < row_ids.size(); i++) {
        if (_tablet_finder->is_adaptive_random_bucket() && config::is_cloud_mode()) {
            auto partition_it = _channels[index_idx]->_channels_by_partition.find(partition_ids[i]);
            if (partition_it == _channels[index_idx]->_channels_by_partition.end()) {
                return Status::InternalError(
                        "unknown partition channel, load_id={}, index_id={}, partition_id={}",
                        print_id(_load_id), _channels[index_idx]->_index_id, partition_ids[i]);
            }
            auto payload_it =
                    channel_payload.find(partition_it->second.get()); // <VNodeChannel*, Payload>
            if (payload_it == channel_payload.end()) {
                auto [tmp_it, _] = channel_payload.emplace(
                        partition_it->second.get(),
                        Payload {std::make_unique<IColumn::Selector>(), &row_part_tablet_id,
                                 std::vector<uint32_t>(), std::vector<int64_t>()});
                payload_it = tmp_it;
                payload_it->second.row_ids->reserve(row_cnt);
                payload_it->second.route_idxs.reserve(row_cnt);
                if (has_row_binlog) {
                    payload_it->second.row_binlog_lsns.reserve(row_cnt);
                }
            }
            payload_it->second.row_ids->push_back(row_ids[i]);
            payload_it->second.route_idxs.push_back(cast_set<uint32_t>(i));
            if (has_row_binlog) {
                payload_it->second.row_binlog_lsns.push_back(row_binlog_lsns[i]);
            }
            continue;
        }

        // (tablet_id, VNodeChannel) where this tablet locate
        auto it = _channels[index_idx]->_channels_by_tablet.find(tablet_ids[i]);
        if (it == _channels[index_idx]->_channels_by_tablet.end()) {
            return Status::InternalError("unknown tablet, load_id={}, index_id={}, tablet_id={}",
                                         print_id(_load_id), _channels[index_idx]->_index_id,
                                         tablet_ids[i]);
        }

        std::vector<std::shared_ptr<VNodeChannel>>& tablet_locations = it->second;
        for (const auto& locate_node : tablet_locations) {
            auto payload_it = channel_payload.find(locate_node.get()); // <VNodeChannel*, Payload>
            if (payload_it == channel_payload.end()) {
                auto [tmp_it, _] = channel_payload.emplace(
                        locate_node.get(),
                        Payload {std::make_unique<IColumn::Selector>(), &row_part_tablet_id,
                                 std::vector<uint32_t>(), std::vector<int64_t>()});
                payload_it = tmp_it;
                payload_it->second.row_ids->reserve(row_cnt);
                payload_it->second.route_idxs.reserve(row_cnt);
                if (has_row_binlog) {
                    payload_it->second.row_binlog_lsns.reserve(row_cnt);
                }
            }
            payload_it->second.row_ids->push_back(row_ids[i]);
            payload_it->second.route_idxs.push_back(cast_set<uint32_t>(i));
            if (has_row_binlog) {
                payload_it->second.row_binlog_lsns.push_back(row_binlog_lsns[i]);
            }
        }
    }
    return Status::OK();
}

Status VTabletWriter::_generate_index_channels_payloads(
        std::vector<RowPartTabletIds>& row_part_tablet_ids,
        ChannelDistributionPayloadVec& payload) {
    for (int i = 0; i < _schema->indexes().size(); i++) {
        RETURN_IF_ERROR(_generate_one_index_channel_payload(row_part_tablet_ids[i], i, payload[i]));
    }
    return Status::OK();
}

Status VTabletWriter::write(RuntimeState* state, doris::Block& input_block) {
    SCOPED_CONSUME_MEM_TRACKER(_mem_tracker.get());
    Status status = Status::OK();

    DCHECK(_state);
    DCHECK(_state->query_options().__isset.dry_run_query);
    if (_state->query_options().dry_run_query) {
        return status;
    }

    // check out of limit
    RETURN_IF_ERROR(_send_new_partition_batch());

    const bool is_replaying_batched_block = _row_distribution._deal_batched;
    auto rows = input_block.rows();
    auto bytes = input_block.bytes();
    if (UNLIKELY(rows == 0)) {
        return status;
    }
    SCOPED_TIMER(_operator_profile->total_time_counter());
    SCOPED_RAW_TIMER(&_send_data_ns);

    std::shared_ptr<Block> block;
    _number_input_rows += rows;
    // update incrementally so that FE can get the progress.
    // the real 'num_rows_load_total' will be set when sink being closed.
    _state->update_num_rows_load_total(rows);
    _state->update_num_bytes_load_total(bytes);
    if (!is_replaying_batched_block) {
        DorisMetrics::instance()->load_rows->increment(rows);
        DorisMetrics::instance()->load_bytes->increment(bytes);
    }

    _row_distribution_watch.start();
    RETURN_IF_ERROR(_row_distribution.generate_rows_distribution(
            input_block, block, _row_part_tablet_ids, _number_input_rows));

    ChannelDistributionPayloadVec channel_to_payload;

    channel_to_payload.resize(_channels.size());
    Status generate_payload_status =
            _generate_index_channels_payloads(_row_part_tablet_ids, channel_to_payload);
    _row_distribution_watch.stop();
    RETURN_IF_ERROR(generate_payload_status);

    // Add block to node channel
    for (size_t i = 0; i < _channels.size(); i++) {
        for (const auto& entry : channel_to_payload[i]) {
            // if this node channel is already failed, this add_row will be skipped
            // entry.second is a [row -> tablet] mapping
            auto st = entry.first->add_block(block.get(), &entry.second);
            if (!st.ok()) {
                _channels[i]->mark_as_failed(entry.first, st.to_string());
            }
        }
    }

    // check intolerable failure
    for (const auto& index_channel : _channels) {
        RETURN_IF_ERROR(index_channel->check_intolerable_failure());
    }

    g_sink_write_bytes << bytes;
    g_sink_write_rows << rows;
    return Status::OK();
}

} // namespace doris

Coverage Report

Created: 2026-07-17 19:12