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

#include "runtime/fragment_mgr.h"

#include "runtime/memory/mem_tracker_limiter.h"

#include "runtime/thread_context.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,

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 = [&]() -> Status {

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

}

}

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.");

}

}

}

return Status::InternalError<false>("can not get a block queue for table_id: " +

std::to_string(_table_id) + _create_plan_failed_reason);

};

if (try_to_get_matched_queue().ok()) {

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

}

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,

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,

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