dataroaring commented on code in PR #22829: URL: https://github.com/apache/doris/pull/22829#discussion_r1295968980
########## be/src/runtime/group_commit_mgr.cpp: ########## @@ -0,0 +1,434 @@ +// 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 "runtime/group_commit_mgr.h" + +#include <gen_cpp/Descriptors_types.h> +#include <gen_cpp/FrontendService.h> +#include <gen_cpp/HeartbeatService.h> +#include <gen_cpp/PaloInternalService_types.h> +#include <gen_cpp/Types_types.h> + +#include "client_cache.h" +#include "common/object_pool.h" +#include "exec/data_sink.h" +#include "io/fs/stream_load_pipe.h" +#include "runtime/exec_env.h" +#include "runtime/fragment_mgr.h" +#include "runtime/runtime_state.h" +#include "runtime/stream_load/new_load_stream_mgr.h" +#include "runtime/stream_load/stream_load_context.h" +#include "util/thrift_rpc_helper.h" +#include "vec/exec/scan/new_file_scan_node.h" + +namespace doris { + +class TPlan; +class FragmentExecState; + +Status LoadInstanceInfo::add_block(std::shared_ptr<vectorized::FutureBlock> block) { + DCHECK(block->schema_version == schema_version); + std::unique_lock l(*_mutex); + RETURN_IF_ERROR(_status); + if (block->rows() > 0) { + _block_queue.push_back(block); + } + if (block->eos) { + _block_unique_ids.erase(block->unique_id); + } else if (block->first) { + _block_unique_ids.emplace(block->unique_id); + } + _cv->notify_one(); + return Status::OK(); +} + +Status LoadInstanceInfo::get_block(vectorized::Block* block, bool* find_block, bool* eos) { + *find_block = false; + *eos = false; + std::unique_lock l(*_mutex); + if (!need_commit) { + auto left_seconds = 10 - std::chrono::duration_cast<std::chrono::seconds>( + std::chrono::steady_clock::now() - _start_time) + .count(); + if (left_seconds <= 0) { + need_commit = true; + } + } + while (_status.ok() && _block_queue.empty() && + (!need_commit || (need_commit && !_block_unique_ids.empty()))) { + // TODO make 10s as a config + auto left_seconds = 10; + if (!need_commit) { + left_seconds = 10 - std::chrono::duration_cast<std::chrono::seconds>( + std::chrono::steady_clock::now() - _start_time) + .count(); + if (left_seconds <= 0) { + need_commit = true; + break; + } + } +#if !defined(USE_BTHREAD_SCANNER) + _cv->wait_for(l, std::chrono::seconds(left_seconds)); +#else + _cv->wait_for(l, left_seconds * 1000000); +#endif + } + if (!_block_queue.empty()) { + auto& future_block = _block_queue.front(); + auto* fblock = static_cast<vectorized::FutureBlock*>(block); + fblock->swap_future_block(future_block); + *find_block = true; + _block_queue.pop_front(); + } + if (_block_queue.empty()) { + if (need_commit && _block_unique_ids.empty()) { + *eos = true; + } else { + *eos = false; + } + } + return Status::OK(); +} + +void LoadInstanceInfo::remove_block_id(const UniqueId& unique_id) { + std::unique_lock l(*_mutex); + if (_block_unique_ids.find(unique_id) != _block_unique_ids.end()) { + _block_unique_ids.erase(unique_id); + _cv->notify_one(); + } +} + +void LoadInstanceInfo::cancel(const Status& st) { + DCHECK(!st.ok()); + std::unique_lock l(*_mutex); + _status = st; + while (!_block_queue.empty()) { + { + auto& future_block = _block_queue.front(); + auto block_status = std::make_tuple<bool, Status, int64_t, int64_t>( + true, Status(st), future_block->rows(), 0); + std::unique_lock<doris::Mutex> l0(*(future_block->lock)); + block_status.swap(*(future_block->block_status)); + future_block->cv->notify_all(); + } + _block_queue.pop_front(); + } +} + +Status GroupCommitTable::get_block_load_instance_info( + int64_t table_id, std::shared_ptr<vectorized::FutureBlock> block, + std::shared_ptr<LoadInstanceInfo>& load_instance_info) { + DCHECK(table_id == _table_id); + DCHECK(block->first == true); + { + std::unique_lock l(_lock); + for (auto it = load_instance_infos.begin(); it != load_instance_infos.end(); ++it) { + // TODO if block schema version is less than fragment schema version, return error + if (!it->second->need_commit && it->second->schema_version == block->schema_version) { + if (block->schema_version == it->second->schema_version) { + load_instance_info = it->second; + break; + } else if (block->schema_version < it->second->schema_version) { + return Status::DataQualityError("schema version not match"); + } + } + } + } + if (load_instance_info == nullptr) { + Status st = Status::OK(); + for (int i = 0; i < 3; ++i) { + std::unique_lock l(_request_fragment_mutex); + // check if there is a re-usefully fragment + { + std::unique_lock l1(_lock); + for (auto it = load_instance_infos.begin(); it != load_instance_infos.end(); ++it) { + // TODO if block schema version is less than fragment schema version, return error + if (!it->second->need_commit) { + if (block->schema_version == it->second->schema_version) { + load_instance_info = it->second; + break; + } else if (block->schema_version < it->second->schema_version) { + return Status::DataQualityError("schema version not match"); + } + } + } + } + if (load_instance_info == nullptr) { + st = _create_group_commit_load(table_id, load_instance_info); + if (LIKELY(st.ok())) { + break; + } + } + } + RETURN_IF_ERROR(st); + if (load_instance_info->schema_version != block->schema_version) { + // TODO check this is the first block + return Status::DataQualityError("schema version not match"); + } + } + return Status::OK(); +} + +Status GroupCommitTable::_create_group_commit_load( + int64_t table_id, std::shared_ptr<LoadInstanceInfo>& load_instance_info) { + TRequestGroupCommitFragmentRequest request; + request.__set_db_id(_db_id); + request.__set_table_id(table_id); + TRequestGroupCommitFragmentResult result; + TNetworkAddress master_addr = _exec_env->master_info()->network_address; + RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>( + master_addr.hostname, master_addr.port, + [&result, &request](FrontendServiceConnection& client) { + client->requestGroupCommitFragment(result, request); + }, + 10000L)); + Status st = Status::create(result.status); + if (!st.ok()) { + LOG(WARNING) << "create group commit load error, st=" << st.to_string(); + } + RETURN_IF_ERROR(st); + auto& params = result.params; + auto db_id = result.params.fragment.output_sink.olap_table_sink.db_id; + auto instance_id = params.params.fragment_instance_id; + auto schema_version = result.base_schema_version; + VLOG_DEBUG << "create plan fragment, table=" << table_id + << ", instance_id=" << print_id(instance_id) + << ", schema version=" << schema_version; + { + load_instance_info = std::make_shared<LoadInstanceInfo>(instance_id, schema_version); + std::unique_lock l(_lock); + load_instance_infos.emplace(instance_id, load_instance_info); + } + st = _exe_plan_fragment(db_id, table_id, params); + if (!st.ok()) { + LOG(WARNING) << "execute plan fragment error, st=" << st + << ", instance_id=" << print_id(instance_id); + std::unique_lock l(_lock); + load_instance_infos.erase(instance_id); + // should notify all blocks + load_instance_info->cancel(st); + } + return st; +} + +Status GroupCommitTable::_exe_plan_fragment(int64_t db_id, int64_t table_id, + const TExecPlanFragmentParams& params) { + auto st = _exec_env->fragment_mgr()->exec_plan_fragment( + params, [db_id, table_id, params, this](RuntimeState* state, Status* status) { + auto& instance_id = state->fragment_instance_id(); + { + std::lock_guard<doris::Mutex> l(_lock); + load_instance_infos.erase(instance_id); + } + TFinishGroupCommitRequest request; + request.__set_db_id(db_id); + request.__set_table_id(table_id); + request.__set_txn_id(params.txn_conf.txn_id); + request.__set_status(status->to_thrift()); + request.__set_commit_infos(state->tablet_commit_infos()); + TFinishGroupCommitResult result; + // plan this load + TNetworkAddress master_addr = _exec_env->master_info()->network_address; + Status st; + st = ThriftRpcHelper::rpc<FrontendServiceClient>( + master_addr.hostname, master_addr.port, + [&result, &request](FrontendServiceConnection& client) { + client->finishGroupCommit(result, request); + }, + 10000L); + if (!st.ok()) { + LOG(WARNING) << "request commit error, table_id=" << table_id + << ", executor status=" << status->to_string() + << ", request commit status=" << st.to_string() + << ", instance_id=" << print_id(instance_id) + << ", rows=" << state->num_rows_load_total(); + return; + } + st = Status::create(result.status); + // TODO handle execute and commit error + LOG(INFO) << "commit group commit, table_id=" << table_id + << ", instance_id=" << print_id(instance_id) + << ", txn_id=" << params.txn_conf.txn_id + << ", execute status=" << status->to_string() + << ", commit status=" << st.to_string() + << ", url=" << state->get_error_log_file_path(); + }); + return st; +} + +Status GroupCommitTable::get_load_instance_info( + const TUniqueId& instance_id, std::shared_ptr<LoadInstanceInfo>& load_instance_info) { + std::unique_lock l(_lock); + auto it = load_instance_infos.find(instance_id); + if (it == load_instance_infos.end()) { + return Status::InternalError("instance " + print_id(instance_id) + " not found"); + } + load_instance_info = it->second; + return Status::OK(); +} + +GroupCommitMgr::GroupCommitMgr(ExecEnv* exec_env) : _exec_env(exec_env) { + ThreadPoolBuilder("InsertIntoGroupCommitThreadPool") + .set_min_threads(10) + .set_max_threads(10) Review Comment: make it as a config. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For additional commands, e-mail: commits-h...@doris.apache.org
