be/src/load/group_commit/group_commit_mgr.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 "load/group_commit/group_commit_mgr.h"

#include <gen_cpp/Types_types.h>
#include <glog/logging.h>

#include <chrono>

#include "cloud/config.h"
#include "common/compiler_util.h"
#include "common/config.h"
#include "common/status.h"
#include "exec/pipeline/dependency.h"
#include "runtime/cluster_info.h"
#include "runtime/exec_env.h"
#include "runtime/fragment_mgr.h"
#include "runtime/memory/mem_tracker_limiter.h"
#include "runtime/thread_context.h"
#include "service/backend_options.h"
#include "util/client_cache.h"
#include "util/debug_points.h"
#include "util/thrift_rpc_helper.h"
#include "util/time.h"

namespace doris {

bvar::Adder<uint64_t> group_commit_block_by_memory_counter("group_commit_block_by_memory_counter");

std::string LoadBlockQueue::_get_load_ids() {
    std::stringstream ss;
    ss << "[";
    for (auto& id : _load_ids_to_write_dep) {
        ss << id.first.to_string() << ", ";
    }
    ss << "]";
    return ss.str();
}

Status LoadBlockQueue::add_block(RuntimeState* runtime_state, std::shared_ptr<Block> block,
                                 bool write_wal, UniqueId& load_id) {
    DBUG_EXECUTE_IF("LoadBlockQueue.add_block.failed",
                    { return Status::InternalError("LoadBlockQueue.add_block.failed"); });
    DBUG_EXECUTE_IF("LoadBlockQueue.add_block.block_reuse_second", {
        int seq = _debug_add_block_seq.fetch_add(1);
        if (seq >= 1) {
            LOG(INFO) << "debug hold reuse 2nd+ add_block, label=" << label
                      << ", load_id=" << load_id.to_string();
            int64_t waited_ms = 0;
            while (waited_ms < 10000 && DebugPoints::instance()->is_enable(
                                                "LoadBlockQueue.add_block.block_reuse_second")) {
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
                waited_ms += 10;
            }
            LOG(INFO) << "debug release reuse 2nd+ add_block, label=" << label
                      << ", load_id=" << load_id.to_string() << ", waited_ms=" << waited_ms;
        }
    });
    std::unique_lock l(mutex);
    RETURN_IF_ERROR(status);
    if (UNLIKELY(runtime_state->is_cancelled())) {
        return runtime_state->cancel_reason();
    }
    RETURN_IF_ERROR(status);
    LOG(INFO) << "query_id: " << print_id(runtime_state->query_id())
              << ", add block rows=" << block->rows() << ", use group_commit label=" << label;
    DBUG_EXECUTE_IF("LoadBlockQueue.add_block.block", DBUG_BLOCK);
    if (block->rows() > 0) {
        if (!config::group_commit_wait_replay_wal_finish) {
            _block_queue.emplace_back(block);
            _data_bytes += block->bytes();
            size_t before_block_queues_bytes = _all_block_queues_bytes->load();
            _all_block_queues_bytes->fetch_add(block->bytes(), std::memory_order_relaxed);
            VLOG_DEBUG << "[Group Commit Debug] (LoadBlockQueue::add_block). "
                       << "Cur block rows=" << block->rows() << ", bytes=" << block->bytes()
                       << ". all block queues bytes from " << before_block_queues_bytes << " to  "
                       << _all_block_queues_bytes->load() << ", queue size=" << _block_queue.size()
                       << ". txn_id=" << txn_id << ", label=" << label
                       << ", instance_id=" << load_instance_id << ", load_ids=" << _get_load_ids();
        }
        if (write_wal || config::group_commit_wait_replay_wal_finish) {
            auto st = _v_wal_writer->write_wal(block.get());
            if (!st.ok()) {
                _cancel_without_lock(st);
                return st;
            }
        }
        if (!runtime_state->is_cancelled() && status.ok() &&
            _all_block_queues_bytes->load(std::memory_order_relaxed) >=
                    config::group_commit_queue_mem_limit) {
            group_commit_block_by_memory_counter << 1;
            auto dep_it = _load_ids_to_write_dep.find(load_id);
            DCHECK(dep_it != _load_ids_to_write_dep.end());
            if (dep_it != _load_ids_to_write_dep.end() && dep_it->second) {
                dep_it->second->block();
                VLOG_DEBUG << "block add_block for load_id=" << load_id << ", memory="
                           << _all_block_queues_bytes->load(std::memory_order_relaxed)
                           << ". inner load_id=" << load_instance_id << ", label=" << label;
            }
        }
    }
    if (!_need_commit.load()) {
        if (_data_bytes >= _group_commit_data_bytes) {
            VLOG_DEBUG << "group commit meets commit condition for data size, label=" << label
                       << ", instance_id=" << load_instance_id << ", data_bytes=" << _data_bytes;
            _need_commit.store(true);
            data_size_condition = true;
        }
        if (std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() -
                                                                  _start_time)
                    .count() >= _group_commit_interval_ms) {
            VLOG_DEBUG << "group commit meets commit condition for time interval, label=" << label
                       << ", instance_id=" << load_instance_id << ", data_bytes=" << _data_bytes;
            _need_commit.store(true);
        }
    }
    for (auto read_dep : _read_deps) {
        read_dep->set_ready();
        VLOG_DEBUG << "set ready for inner load_id=" << load_instance_id;
    }
    return Status::OK();
}

Status LoadBlockQueue::get_block(RuntimeState* runtime_state, Block* block, bool* find_block,
                                 bool* eos, std::shared_ptr<Dependency> get_block_dep) {
    *find_block = false;
    *eos = false;
    std::unique_lock l(mutex);
    if (runtime_state->is_cancelled() || !status.ok()) {
        auto st = runtime_state->cancel_reason();
        _cancel_without_lock(st);
        return status;
    }
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(
                            std::chrono::steady_clock::now() - _start_time)
                            .count();
    if (!_need_commit.load() && duration >= _group_commit_interval_ms) {
        _need_commit.store(true);
    }
    if (_block_queue.empty()) {
        if (_need_commit.load() && duration >= 10 * _group_commit_interval_ms) {
            auto last_print_duration = std::chrono::duration_cast<std::chrono::milliseconds>(
                                               std::chrono::steady_clock::now() - _last_print_time)
                                               .count();
            if (last_print_duration >= 10000) {
                _last_print_time = std::chrono::steady_clock::now();
                LOG(INFO) << "find one group_commit need to commit, txn_id=" << txn_id
                          << ", label=" << label << ", instance_id=" << load_instance_id
                          << ", duration=" << duration << ", load_ids=" << _get_load_ids();
            }
        }
        VLOG_DEBUG << "get_block for inner load_id=" << load_instance_id << ", but queue is empty";
        if (!_need_commit.load()) {
            get_block_dep->block();
            VLOG_DEBUG << "block get_block for inner load_id=" << load_instance_id;
        }
    } else {
        const BlockData block_data = _block_queue.front();
        block->swap(*block_data.block);
        *find_block = true;
        _block_queue.pop_front();
        size_t before_block_queues_bytes = _all_block_queues_bytes->load();
        _all_block_queues_bytes->fetch_sub(block_data.block_bytes, std::memory_order_relaxed);
        VLOG_DEBUG << "[Group Commit Debug] (LoadBlockQueue::get_block). "
                   << "Cur block rows=" << block->rows() << ", bytes=" << block->bytes()
                   << ". all block queues bytes from " << before_block_queues_bytes << " to  "
                   << _all_block_queues_bytes->load() << ", queue size=" << _block_queue.size()
                   << ". txn_id=" << txn_id << ", label=" << label
                   << ", instance_id=" << load_instance_id << ", load_ids=" << _get_load_ids();
    }
    if (_block_queue.empty() && _need_commit.load() && _load_ids_to_write_dep.empty()) {
        *eos = true;
    } else {
        *eos = false;
    }
    if (_all_block_queues_bytes->load(std::memory_order_relaxed) <
        config::group_commit_queue_mem_limit) {
        for (auto& id : _load_ids_to_write_dep) {
            id.second->set_ready();
        }
        VLOG_DEBUG << "set ready for load_ids=" << _get_load_ids()
                   << ". inner load_id=" << load_instance_id << ", label=" << label;
    }
    return Status::OK();
}

Status LoadBlockQueue::remove_load_id(const UniqueId& load_id) {
    std::unique_lock l(mutex);
    if (_load_ids_to_write_dep.find(load_id) != _load_ids_to_write_dep.end()) {
        _load_ids_to_write_dep[load_id]->set_always_ready();
        _load_ids_to_write_dep.erase(load_id);
        for (auto read_dep : _read_deps) {
            read_dep->set_ready();
        }
        VLOG_DEBUG << "set ready for load_id=" << load_id << ", inner load_id=" << load_instance_id;
        return Status::OK();
    }
    return Status::NotFound<false>("load_id=" + load_id.to_string() +
                                   " not in block queue, label=" + label);
}

bool LoadBlockQueue::contain_load_id(const UniqueId& load_id) {
    std::unique_lock l(mutex);
    return _load_ids_to_write_dep.find(load_id) != _load_ids_to_write_dep.end();
}

Status LoadBlockQueue::add_load_id(const UniqueId& load_id,
                                   const std::shared_ptr<Dependency> put_block_dep) {
    std::unique_lock l(mutex);
    if (_need_commit.load() || !status.ok() || process_finish.load()) {
        return Status::InternalError<false>(
                "block queue cannot add load id, id=" + load_instance_id.to_string() +
                ", need_commit=" + (_need_commit.load() ? "true" : "false") +
                ", process_finish=" + (process_finish.load() ? "true" : "false") +
                ", queue_status=" + status.to_string());
    }
    _load_ids_to_write_dep[load_id] = put_block_dep;
    group_commit_load_count.fetch_add(1);
    return Status::OK();
}

void LoadBlockQueue::cancel(const Status& st) {
    DCHECK(!st.ok());
    std::unique_lock l(mutex);
    _cancel_without_lock(st);
}

void LoadBlockQueue::_cancel_without_lock(const Status& st) {
    LOG(INFO) << "cancel group_commit, instance_id=" << load_instance_id << ", label=" << label
              << ", status=" << st.to_string();
    status =
            Status::Cancelled("cancel group_commit, label=" + label + ", status=" + st.to_string());
    size_t before_block_queues_bytes = _all_block_queues_bytes->load();
    while (!_block_queue.empty()) {
        const BlockData& block_data = _block_queue.front();
        _all_block_queues_bytes->fetch_sub(block_data.block_bytes, std::memory_order_relaxed);
        _block_queue.pop_front();
    }
    VLOG_DEBUG << "[Group Commit Debug] (LoadBlockQueue::_cancel_without_block). "
               << "all block queues bytes from " << before_block_queues_bytes << " to "
               << _all_block_queues_bytes->load() << ", queue size=" << _block_queue.size()
               << ", txn_id=" << txn_id << ", label=" << label
               << ", instance_id=" << load_instance_id << ", load_ids=" << _get_load_ids();
    for (auto& id : _load_ids_to_write_dep) {
        id.second->set_always_ready();
    }
    for (auto read_dep : _read_deps) {
        read_dep->set_ready();
    }
    VLOG_DEBUG << "set ready for load_ids=" << _get_load_ids()
               << ", inner load_id=" << load_instance_id;
}

Status GroupCommitTable::get_first_block_load_queue(
        int64_t table_id, int64_t base_schema_version, int64_t index_size, const UniqueId& load_id,
        std::shared_ptr<LoadBlockQueue>& load_block_queue, int be_exe_version,
        TGroupCommitMode::type group_commit_mode, std::shared_ptr<Dependency> create_plan_dep,
        std::shared_ptr<Dependency> put_block_dep) {
    DCHECK(table_id == _table_id);
    std::unique_lock l(_lock);
    auto try_to_get_matched_queue = [&](bool& need_create_plan) -> Status {
        need_create_plan = false;
        for (const auto& [_, inner_block_queue] : _load_block_queues) {
            if (inner_block_queue->contain_load_id(load_id)) {
                load_block_queue = inner_block_queue;
                return Status::OK();
            }
        }
        RETURN_IF_ERROR(_check_wal_backlog(group_commit_mode));
        for (const auto& [_, inner_block_queue] : _load_block_queues) {
            if (!inner_block_queue->need_commit()) {
                if (base_schema_version == inner_block_queue->schema_version &&
                    index_size == inner_block_queue->index_size) {
                    if (inner_block_queue->add_load_id(load_id, put_block_dep).ok()) {
                        load_block_queue = inner_block_queue;
                        return Status::OK();
                    }
                } else {
                    return Status::DataQualityError<false>(
                            "schema version not match, maybe a schema change is in process. "
                            "Please retry this load manually.");
                }
            }
        }
        need_create_plan = true;
        return Status::OK();
    };

    bool need_create_plan = false;
    RETURN_IF_ERROR(try_to_get_matched_queue(need_create_plan));
    if (!need_create_plan) {
        return Status::OK();
    }
    create_plan_dep->block();
    _create_plan_be_exe_version = be_exe_version;
    if (_create_plan_deps.empty()) {
        _create_plan_start_time_ms = MonotonicMillis();
    }
    _create_plan_deps.emplace(load_id, std::make_tuple(create_plan_dep, put_block_dep,
                                                       base_schema_version, index_size));
    [[maybe_unused]] auto submit_st = _submit_create_group_commit_load();
    RETURN_IF_ERROR(try_to_get_matched_queue(need_create_plan));
    if (need_create_plan) {
        return Status::InternalError<false>("can not get a block queue for table_id: " +
                                            std::to_string(_table_id) + _create_plan_failed_reason);
    }
    return Status::OK();
}

Status GroupCommitTable::_check_wal_backlog(TGroupCommitMode::type group_commit_mode) {
    int32_t max_wal_num = config::group_commit_max_wal_num_per_table;
    if (group_commit_mode != TGroupCommitMode::ASYNC_MODE || max_wal_num <= 0) {
        return Status::OK();
    }
    size_t wal_num = _exec_env->wal_mgr()->get_wal_queue_size(_table_id);
    if (wal_num < static_cast<size_t>(max_wal_num)) {
        return Status::OK();
    }
    std::string failed_reason = _exec_env->wal_mgr()->get_last_replay_wal_failed_reason(_table_id);
    if (failed_reason.empty()) {
        return Status::OK();
    }
    return Status::Error<ErrorCode::EXCEEDED_LIMIT>(
            "Too many group commit async WALs for table {} on be host {}. wal num={}, limit={}, "
            "last replay wal failed reason: {}",
            _table_id, BackendOptions::get_localhost(), wal_num, max_wal_num, failed_reason);
}

Status GroupCommitTable::submit_create_group_commit_load() {
    std::unique_lock l(_lock);
    if (_create_plan_deps.empty()) {
        return Status::OK();
    }
    return _submit_create_group_commit_load();
}

Status GroupCommitTable::_submit_create_group_commit_load() {
    if (_is_creating_plan_fragment) {
        return Status::OK();
    }

    int64_t timeout_ms = config::group_commit_create_plan_timeout_ms;
    if (timeout_ms > 0 && !_create_plan_deps.empty()) {
        int64_t now_ms = MonotonicMillis();
        if (_create_plan_start_time_ms > 0 && now_ms - _create_plan_start_time_ms > timeout_ms) {
            std::string last_create_plan_failed_reason = _create_plan_failed_reason;
            _create_plan_failed_reason =
                    ". group commit create plan timeout after " + std::to_string(timeout_ms) + "ms";
            if (!last_create_plan_failed_reason.empty()) {
                _create_plan_failed_reason +=
                        ", last create plan error: " + last_create_plan_failed_reason;
            }
            for (const auto& [id, load_info] : _create_plan_deps) {
                std::get<0>(load_info)->set_ready();
            }
            _create_plan_deps.clear();
            _create_plan_start_time_ms = 0;
            return Status::OK();
        }
    }

    auto mem_tracker = _group_commit_mgr->group_commit_mem_tracker();
    int be_exe_version = _create_plan_be_exe_version;
    _is_creating_plan_fragment = true;
    auto submit_st = _thread_pool->submit_func([&, be_exe_version, mem_tracker] {
        std::shared_ptr<LoadBlockQueue> created_load_block_queue;
        Status create_group_commit_st = Status::OK();
        std::string create_plan_failed_reason;
        Defer defer {[&]() {
            bool need_resubmit = !create_group_commit_st.ok();
            std::unique_lock l(_lock);
            _is_creating_plan_fragment = false;
            _create_plan_failed_reason = create_plan_failed_reason;
            if (created_load_block_queue && create_group_commit_st.ok() &&
                !created_load_block_queue->need_commit()) {
                std::vector<UniqueId> success_load_ids;
                for (const auto& [id, load_info] : _create_plan_deps) {
                    auto create_dep = std::get<0>(load_info);
                    auto put_dep = std::get<1>(load_info);
                    if (created_load_block_queue->schema_version == std::get<2>(load_info) &&
                        created_load_block_queue->index_size == std::get<3>(load_info)) {
                        auto st = created_load_block_queue->add_load_id(id, put_dep);
                        if (!st.ok()) {
                            LOG(WARNING) << "failed to add pending load_id into created "
                                            "group commit queue, load_id="
                                         << id << ", label=" << created_load_block_queue->label
                                         << ", status=" << st.to_string();
                            need_resubmit = true;
                        } else {
                            create_dep->set_ready();
                            success_load_ids.emplace_back(id);
                        }
                    } else if (created_load_block_queue->schema_version > std::get<2>(load_info) ||
                               (created_load_block_queue->schema_version ==
                                        std::get<2>(load_info) &&
                                created_load_block_queue->index_size != std::get<3>(load_info))) {
                        // schema version mismatch:
                        //   1. the schema version of created load block queue is newer than the load request
                        //   2. the index size is not equal
                        // set ready for the load request to let it fail
                        create_dep->set_ready();
                        success_load_ids.emplace_back(id);
                    }
                }
                for (const auto& id : success_load_ids) {
                    _create_plan_deps.erase(id);
                }
                if (_create_plan_deps.empty()) {
                    _create_plan_start_time_ms = 0;
                }
            }
            if (!_create_plan_deps.empty()) {
                need_resubmit = true;
            }
            if (need_resubmit && _group_commit_mgr) {
                LOG(INFO) << "resubmit create group commit load task for table: " << _table_id;
                _group_commit_mgr->add_need_create_plan_table(_table_id);
            }
        }};
        create_group_commit_st =
                _create_group_commit_load(be_exe_version, mem_tracker, created_load_block_queue);
        if (!create_group_commit_st.ok()) {
            LOG(WARNING) << "create group commit load error: "
                         << create_group_commit_st.to_string();
            create_plan_failed_reason = ". create group commit load error: " +
                                        create_group_commit_st.to_string().substr(0, 300);
        } else {
            create_plan_failed_reason = "";
        }
    });
    if (!submit_st.ok()) {
        _is_creating_plan_fragment = false;
        _create_plan_failed_reason =
                ". create group commit load error: submit create group commit load task failed: " +
                submit_st.to_string().substr(0, 300);
        for (const auto& [id, load_info] : _create_plan_deps) {
            std::get<0>(load_info)->set_ready();
        }
        _create_plan_deps.clear();
        LOG(WARNING) << "submit create group commit load task for table: " << _table_id
                     << ", error: " << submit_st.to_string();
    }
    return submit_st;
}

void GroupCommitTable::remove_load_id(const UniqueId& load_id) {
    std::unique_lock l(_lock);
    if (_create_plan_deps.find(load_id) != _create_plan_deps.end()) {
        _create_plan_deps.erase(load_id);
        return;
    }
    for (const auto& [_, inner_block_queue] : _load_block_queues) {
        if (inner_block_queue->remove_load_id(load_id).ok()) {
            return;
        }
    }
}

Status GroupCommitTable::_create_group_commit_load(
        int be_exe_version, const std::shared_ptr<MemTrackerLimiter>& mem_tracker,
        std::shared_ptr<LoadBlockQueue>& created_load_block_queue) {
    Status st = Status::OK();
    TStreamLoadPutResult result;
    std::string label;
    int64_t txn_id;
    TUniqueId instance_id;
    {
        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(mem_tracker);
        UniqueId load_id = UniqueId::gen_uid();
        TUniqueId tload_id;
        tload_id.__set_hi(load_id.hi);
        tload_id.__set_lo(load_id.lo);
        std::regex reg("-");
        label = "group_commit_" + std::regex_replace(load_id.to_string(), reg, "_");
        std::stringstream ss;
        ss << "insert into doris_internal_table_id(" << _table_id << ") WITH LABEL " << label
           << " select * from group_commit(\"table_id\"=\"" << _table_id << "\")";
        TStreamLoadPutRequest request;
        request.__set_load_sql(ss.str());
        request.__set_loadId(tload_id);
        request.__set_label(label);
        request.__set_token("group_commit"); // this is a fake, fe not check it now
        request.__set_max_filter_ratio(1.0);
        request.__set_strictMode(false);
        request.__set_partial_update(false);
        // this is an internal interface, use admin to pass the auth check
        request.__set_user("admin");
        if (_exec_env->cluster_info()->backend_id != 0) {
            request.__set_backend_id(_exec_env->cluster_info()->backend_id);
        } else {
            LOG(WARNING) << "_exec_env->cluster_info not set backend_id";
        }
        TNetworkAddress master_addr = _exec_env->cluster_info()->master_fe_addr;
        st = ThriftRpcHelper::rpc<FrontendServiceClient>(
                master_addr.hostname, master_addr.port,
                [&result, &request](FrontendServiceConnection& client) {
                    client->streamLoadPut(result, request);
                },
                10000L);
        if (!st.ok()) {
            LOG(WARNING) << "create group commit load rpc error, st=" << st.to_string();
            return st;
        }
        st = Status::create<false>(result.status);
        if (st.ok() && !result.__isset.pipeline_params) {
            st = Status::InternalError("Non-pipeline is disabled!");
        }
        if (!st.ok()) {
            LOG(WARNING) << "create group commit load error, st=" << st.to_string();
            return st;
        }
        auto& pipeline_params = result.pipeline_params;
        auto schema_version = pipeline_params.fragment.output_sink.olap_table_sink.schema.version;
        auto index_size =
                pipeline_params.fragment.output_sink.olap_table_sink.schema.indexes.size();
        DCHECK(pipeline_params.fragment.output_sink.olap_table_sink.db_id == _db_id);
        txn_id = pipeline_params.txn_conf.txn_id;
        DCHECK(pipeline_params.local_params.size() == 1);
        instance_id = pipeline_params.local_params[0].fragment_instance_id;
        VLOG_DEBUG << "create plan fragment, db_id=" << _db_id << ", table=" << _table_id
                   << ", schema version=" << schema_version << ", index size=" << index_size
                   << ", label=" << label << ", txn_id=" << txn_id
                   << ", instance_id=" << print_id(instance_id);
        {
            auto load_block_queue = std::make_shared<LoadBlockQueue>(
                    instance_id, label, txn_id, schema_version, index_size, _all_block_queues_bytes,
                    result.wait_internal_group_commit_finish, result.group_commit_interval_ms,
                    result.group_commit_data_bytes);
            RETURN_IF_ERROR(load_block_queue->create_wal(
                    _db_id, _table_id, txn_id, label, _exec_env->wal_mgr(),
                    pipeline_params.fragment.output_sink.olap_table_sink.schema.slot_descs,
                    be_exe_version));
            created_load_block_queue = load_block_queue;

            std::unique_lock l(_lock);
            _load_block_queues.emplace(instance_id, load_block_queue);
            std::vector<UniqueId> success_load_ids;
            for (const auto& [id, load_info] : _create_plan_deps) {
                auto create_dep = std::get<0>(load_info);
                auto put_dep = std::get<1>(load_info);
                if (load_block_queue->schema_version == std::get<2>(load_info) &&
                    load_block_queue->index_size == std::get<3>(load_info)) {
                    if (load_block_queue->add_load_id(id, put_dep).ok()) {
                        create_dep->set_ready();
                        success_load_ids.emplace_back(id);
                    }
                } else if (load_block_queue->schema_version > std::get<2>(load_info) ||
                           (load_block_queue->schema_version == std::get<2>(load_info) &&
                            load_block_queue->index_size != std::get<3>(load_info))) {
                    // schema version mismatch:
                    //   1. the schema version of created load block queue is newer than the load request
                    //   2. the index size is not equal
                    // set ready for the load request to let it fail
                    create_dep->set_ready();
                    success_load_ids.emplace_back(id);
                }
            }
            for (const auto& id : success_load_ids) {
                _create_plan_deps.erase(id);
            }
            if (_create_plan_deps.empty()) {
                _create_plan_start_time_ms = 0;
            }
        }
    }
    st = _exec_plan_fragment(_db_id, _table_id, label, txn_id, result.pipeline_params);
    if (!st.ok()) {
        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(mem_tracker);
        auto finish_st = _finish_group_commit_load(_db_id, _table_id, label, txn_id, instance_id,
                                                   st, nullptr);
        if (!finish_st.ok()) {
            LOG(WARNING) << "finish group commit error, label=" << label
                         << ", st=" << finish_st.to_string();
        }
    }
    return st;
}

Status GroupCommitTable::_finish_group_commit_load(int64_t db_id, int64_t table_id,
                                                   const std::string& label, int64_t txn_id,
                                                   const TUniqueId& instance_id, Status& status,
                                                   RuntimeState* state) {
    Status st;
    Status result_status;
    DBUG_EXECUTE_IF("LoadBlockQueue._finish_group_commit_load.err_status", {
        status = Status::InternalError("LoadBlockQueue._finish_group_commit_load.err_status");
    });
    DBUG_EXECUTE_IF("LoadBlockQueue._finish_group_commit_load.load_error",
                    { status = Status::InternalError("load_error"); });
    if (status.ok()) {
        DBUG_EXECUTE_IF("LoadBlockQueue._finish_group_commit_load.commit_error", {
            status = Status::InternalError("LoadBlockQueue._finish_group_commit_load.commit_error");
        });
        // commit txn
        TLoadTxnCommitRequest request;
        // deprecated and should be removed in 3.1, use token instead
        request.__set_auth_code(0);
        request.__set_token(_exec_env->cluster_info()->curr_auth_token);
        request.__set_db_id(db_id);
        request.__set_table_id(table_id);
        request.__set_txnId(txn_id);
        request.__set_thrift_rpc_timeout_ms(config::txn_commit_rpc_timeout_ms);
        request.__set_groupCommit(true);
        request.__set_receiveBytes(state->num_bytes_load_total());
        if (_exec_env->cluster_info()->backend_id != 0) {
            request.__set_backendId(_exec_env->cluster_info()->backend_id);
        } else {
            LOG(WARNING) << "_exec_env->cluster_info not set backend_id";
        }
        if (state) {
            request.__set_commitInfos(state->tablet_commit_infos());
        }
        TLoadTxnCommitResult result;
        TNetworkAddress master_addr = _exec_env->cluster_info()->master_fe_addr;
        int retry_times = 0;
        while (retry_times < config::mow_stream_load_commit_retry_times) {
            st = ThriftRpcHelper::rpc<FrontendServiceClient>(
                    master_addr.hostname, master_addr.port,
                    [&request, &result](FrontendServiceConnection& client) {
                        client->loadTxnCommit(result, request);
                    },
                    config::txn_commit_rpc_timeout_ms);
            if (st.ok()) {
                result_status = Status::create(result.status);
                // DELETE_BITMAP_LOCK_ERROR will be retried
                if (result_status.ok() ||
                    !result_status.is<ErrorCode::DELETE_BITMAP_LOCK_ERROR>()) {
                    break;
                }
                LOG_WARNING("Failed to commit txn on group commit")
                        .tag("label", label)
                        .tag("txn_id", txn_id)
                        .tag("retry_times", retry_times)
                        .error(result_status);
            } else {
                LOG_WARNING("Failed to commit txn on group commit")
                        .tag("label", label)
                        .tag("txn_id", txn_id)
                        .tag("retry_times", retry_times)
                        .error(st);
            }
            retry_times++;
        }
        DBUG_EXECUTE_IF("LoadBlockQueue._finish_group_commit_load.commit_success_and_rpc_error",
                        { result_status = Status::InternalError("commit_success_and_rpc_error"); });
    } else {
        // abort txn
        TLoadTxnRollbackRequest request;
        // deprecated and should be removed in 3.1, use token instead
        request.__set_auth_code(0);
        request.__set_token(_exec_env->cluster_info()->curr_auth_token);
        request.__set_db_id(db_id);
        request.__set_txnId(txn_id);
        request.__set_reason(status.to_string());
        TLoadTxnRollbackResult result;
        TNetworkAddress master_addr = _exec_env->cluster_info()->master_fe_addr;
        st = ThriftRpcHelper::rpc<FrontendServiceClient>(
                master_addr.hostname, master_addr.port,
                [&request, &result](FrontendServiceConnection& client) {
                    client->loadTxnRollback(result, request);
                });
        if (st.ok()) {
            result_status = Status::create<false>(result.status);
        }
        DBUG_EXECUTE_IF("LoadBlockQueue._finish_group_commit_load.err_status", {
            std ::string msg = "abort txn";
            LOG(INFO) << "debug promise set: " << msg;
            ExecEnv::GetInstance()->group_commit_mgr()->debug_promise.set_value(
                    Status ::InternalError(msg));
        });
    }
    std::shared_ptr<LoadBlockQueue> load_block_queue;
    {
        std::lock_guard<std::mutex> l(_lock);
        auto it = _load_block_queues.find(instance_id);
        if (it != _load_block_queues.end()) {
            load_block_queue = it->second;
            if (!status.ok()) {
                load_block_queue->cancel(status);
            }
            //close wal
            RETURN_IF_ERROR(load_block_queue->close_wal());
            // notify sync mode loads
            {
                std::unique_lock l2(load_block_queue->mutex);
                load_block_queue->process_finish = true;
                for (auto dep : load_block_queue->dependencies) {
                    dep->set_always_ready();
                }
            }
        }
        _load_block_queues.erase(instance_id);
    }
    if (!load_block_queue) {
        LOG(WARNING) << "finish group commit can not find load block queue, label=" << label
                     << ", txn_id=" << txn_id << ", instance_id=" << print_id(instance_id);
    }
    // status: exec_plan_fragment result
    // st: commit txn rpc status
    // result_status: commit txn result
    DBUG_EXECUTE_IF("LoadBlockQueue._finish_group_commit_load.err_st", {
        st = Status::InternalError("LoadBlockQueue._finish_group_commit_load.err_st");
    });
    if (status.ok() && st.ok() &&
        (result_status.ok() || result_status.is<ErrorCode::PUBLISH_TIMEOUT>())) {
        if (!config::group_commit_wait_replay_wal_finish) {
            auto delete_st = _exec_env->wal_mgr()->delete_wal(table_id, txn_id);
            if (!delete_st.ok()) {
                LOG(WARNING) << "fail to delete wal " << txn_id << ", st=" << delete_st.to_string();
            }
        }
    } else {
        std::string wal_path;
        RETURN_IF_ERROR(_exec_env->wal_mgr()->get_wal_path(txn_id, wal_path));
        RETURN_IF_ERROR(_exec_env->wal_mgr()->add_recover_wal(db_id, table_id, txn_id, wal_path));
    }
    std::stringstream ss;
    ss << "finish group commit, db_id=" << db_id << ", table_id=" << table_id << ", label=" << label
       << ", txn_id=" << txn_id << ", instance_id=" << print_id(instance_id)
       << ", exec_plan_fragment status=" << status.to_string()
       << ", commit/abort txn rpc status=" << st.to_string()
       << ", commit/abort txn status=" << result_status.to_string();
    if (load_block_queue) {
        ss << ", this group commit includes " << load_block_queue->group_commit_load_count
           << " loads, flush because meet "
           << (load_block_queue->data_size_condition ? "data size " : "time ") << "condition";
    } else {
        ss << ", load block queue is missing when finishing group commit";
    }
    ss << ", wal space info:" << ExecEnv::GetInstance()->wal_mgr()->get_wal_dirs_info_string();
    if (state) {
        if (!state->get_error_log_file_path().empty()) {
            ss << ", error_url=" << state->get_error_log_file_path();
        }
        if (!state->get_first_error_msg().empty()) {
            ss << ", first_error_msg=" << state->get_first_error_msg();
        }
        ss << ", rows=" << state->num_rows_load_success();
    }
    LOG(INFO) << ss.str();
    DBUG_EXECUTE_IF("LoadBlockQueue._finish_group_commit_load.get_wal_back_pressure_msg", {
        if (dp->param<int64_t>("table_id", -1) == table_id) {
            std ::string msg = _exec_env->wal_mgr()->get_wal_dirs_info_string();
            LOG(INFO) << "table_id" << std::to_string(table_id) << " set debug promise: " << msg;
            ExecEnv::GetInstance()->group_commit_mgr()->debug_promise.set_value(
                    Status ::InternalError(msg));
        }
    };);
    return st;
}

Status GroupCommitTable::_exec_plan_fragment(int64_t db_id, int64_t table_id,
                                             const std::string& label, int64_t txn_id,
                                             const TPipelineFragmentParams& pipeline_params) {
    auto finish_cb = [db_id, table_id, label, txn_id, this](RuntimeState* state, Status* status) {
        DCHECK(state);
        auto finish_st = _finish_group_commit_load(db_id, table_id, label, txn_id,
                                                   state->fragment_instance_id(), *status, state);
        if (!finish_st.ok()) {
            LOG(WARNING) << "finish group commit error, label=" << label
                         << ", st=" << finish_st.to_string();
        }
    };

    TPipelineFragmentParamsList mocked;
    return _exec_env->fragment_mgr()->exec_plan_fragment(
            pipeline_params, QuerySource::GROUP_COMMIT_LOAD, finish_cb, mocked);
}

Status GroupCommitTable::get_load_block_queue(const TUniqueId& instance_id,
                                              std::shared_ptr<LoadBlockQueue>& load_block_queue,
                                              std::shared_ptr<Dependency> get_block_dep) {
    std::unique_lock l(_lock);
    auto it = _load_block_queues.find(instance_id);
    if (it == _load_block_queues.end()) {
        return Status::InternalError("group commit load instance " + print_id(instance_id) +
                                     " not found");
    }
    load_block_queue = it->second;
    load_block_queue->append_read_dependency(get_block_dep);
    return Status::OK();
}

GroupCommitMgr::GroupCommitMgr(ExecEnv* exec_env) : _exec_env(exec_env) {
    static_cast<void>(ThreadPoolBuilder("GroupCommitThreadPool")
                              .set_min_threads(1)
                              .set_max_threads(config::group_commit_insert_threads)
                              .build(&_thread_pool));
    _all_block_queues_bytes = std::make_shared<std::atomic_size_t>(0);
    _group_commit_mem_tracker =
            MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::LOAD, "GroupCommit");
    _create_plan_thread = std::thread(&GroupCommitMgr::_create_plan_worker, this);
}

GroupCommitMgr::~GroupCommitMgr() {
    stop();
    LOG(INFO) << "GroupCommitMgr is destoried";
}

void GroupCommitMgr::stop() {
    {
        std::lock_guard<std::mutex> l(_need_create_plan_lock);
        if (_stopped) {
            return;
        }
        _stopped = true;
    }
    _need_create_plan_cv.notify_all();
    if (_create_plan_thread.joinable()) {
        _create_plan_thread.join();
    }
    _thread_pool->shutdown();
    LOG(INFO) << "GroupCommitMgr is stopped";
}

void GroupCommitMgr::add_need_create_plan_table(int64_t table_id) {
    {
        std::lock_guard<std::mutex> l(_need_create_plan_lock);
        if (_stopped) {
            return;
        }
        _need_create_plan_tables.insert(table_id);
    }
    _need_create_plan_cv.notify_one();
}

void GroupCommitMgr::_create_plan_worker() {
    SCOPED_INIT_THREAD_CONTEXT();
    while (true) {
        std::set<int64_t> need_create_plan_tables;
        {
            std::unique_lock<std::mutex> l(_need_create_plan_lock);
            _need_create_plan_cv.wait(
                    l, [this] { return _stopped || !_need_create_plan_tables.empty(); });
            if (_stopped && _need_create_plan_tables.empty()) {
                return;
            }
            need_create_plan_tables.swap(_need_create_plan_tables);
        }
        for (const auto table_id : need_create_plan_tables) {
            std::shared_ptr<GroupCommitTable> group_commit_table;
            {
                std::lock_guard<std::mutex> l(_lock);
                auto it = _table_map.find(table_id);
                if (it == _table_map.end()) {
                    continue;
                }
                group_commit_table = it->second;
            }
            auto st = group_commit_table->submit_create_group_commit_load();
            if (!st.ok()) {
                LOG(WARNING) << "submit create group commit load task from worker for table: "
                             << table_id << ", error: " << st.to_string();
            }
        }
    }
}

Status GroupCommitMgr::get_first_block_load_queue(
        int64_t db_id, int64_t table_id, int64_t base_schema_version, int64_t index_size,
        const UniqueId& load_id, std::shared_ptr<LoadBlockQueue>& load_block_queue,
        int be_exe_version, TGroupCommitMode::type group_commit_mode,
        std::shared_ptr<Dependency> create_plan_dep, std::shared_ptr<Dependency> put_block_dep) {
    std::shared_ptr<GroupCommitTable> group_commit_table;
    {
        std::lock_guard wlock(_lock);
        if (_table_map.find(table_id) == _table_map.end()) {
            _table_map.emplace(table_id, std::make_shared<GroupCommitTable>(
                                                 _exec_env, _thread_pool.get(), db_id, table_id,
                                                 _all_block_queues_bytes, this));
        }
        group_commit_table = _table_map[table_id];
    }
    RETURN_IF_ERROR(group_commit_table->get_first_block_load_queue(
            table_id, base_schema_version, index_size, load_id, load_block_queue, be_exe_version,
            group_commit_mode, create_plan_dep, put_block_dep));
    return Status::OK();
}

Status GroupCommitMgr::get_load_block_queue(int64_t table_id, const TUniqueId& instance_id,
                                            std::shared_ptr<LoadBlockQueue>& load_block_queue,
                                            std::shared_ptr<Dependency> get_block_dep) {
    std::shared_ptr<GroupCommitTable> group_commit_table;
    {
        std::lock_guard<std::mutex> l(_lock);
        auto it = _table_map.find(table_id);
        if (it == _table_map.end()) {
            return Status::NotFound("table_id: " + std::to_string(table_id) +
                                    ", instance_id: " + print_id(instance_id) + " dose not exist");
        }
        group_commit_table = it->second;
    }
    return group_commit_table->get_load_block_queue(instance_id, load_block_queue, get_block_dep);
}

void GroupCommitMgr::remove_load_id(int64_t table_id, const UniqueId& load_id) {
    std::lock_guard wlock(_lock);
    if (_table_map.find(table_id) != _table_map.end()) {
        _table_map.find(table_id)->second->remove_load_id(load_id);
    }
}

Status LoadBlockQueue::create_wal(int64_t db_id, int64_t tb_id, int64_t wal_id,
                                  const std::string& import_label, WalManager* wal_manager,
                                  std::vector<TSlotDescriptor>& slot_desc, int be_exe_version) {
    std::string real_label = config::group_commit_wait_replay_wal_finish
                                     ? import_label + "_test_wait"
                                     : import_label;
    RETURN_IF_ERROR(ExecEnv::GetInstance()->wal_mgr()->create_wal_path(
            db_id, tb_id, wal_id, real_label, _wal_base_path, WAL_VERSION));
    _v_wal_writer = std::make_shared<VWalWriter>(db_id, tb_id, wal_id, real_label, wal_manager,
                                                 slot_desc, be_exe_version);
    return _v_wal_writer->init();
}

Status LoadBlockQueue::close_wal() {
    if (_v_wal_writer != nullptr) {
        RETURN_IF_ERROR(_v_wal_writer->close());
    }
    return Status::OK();
}

void LoadBlockQueue::append_dependency(std::shared_ptr<Dependency> finish_dep) {
    std::lock_guard<std::mutex> lock(mutex);
    // If not finished, dependencies should be blocked.
    if (!process_finish) {
        finish_dep->block();
        dependencies.push_back(finish_dep);
    }
}

void LoadBlockQueue::append_read_dependency(std::shared_ptr<Dependency> read_dep) {
    std::lock_guard<std::mutex> lock(mutex);
    _read_deps.push_back(read_dep);
}

bool LoadBlockQueue::has_enough_wal_disk_space(size_t estimated_wal_bytes) {
    DBUG_EXECUTE_IF("LoadBlockQueue.has_enough_wal_disk_space.low_space", { return false; });
    auto* wal_mgr = ExecEnv::GetInstance()->wal_mgr();
    size_t available_bytes = 0;
    {
        Status st = wal_mgr->get_wal_dir_available_size(_wal_base_path, &available_bytes);
        if (!st.ok()) {
            LOG(WARNING) << "get wal dir available size failed, st=" << st.to_string();
        }
    }
    if (estimated_wal_bytes < available_bytes) {
        Status st =
                wal_mgr->update_wal_dir_estimated_wal_bytes(_wal_base_path, estimated_wal_bytes, 0);
        if (!st.ok()) {
            LOG(WARNING) << "update wal dir estimated_wal_bytes failed, reason: " << st.to_string();
        }
        return true;
    } else {
        return false;
    }
}

} // namespace doris

Coverage Report

Created: 2026-08-06 14:04