You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "mymeiyi (via GitHub)" <gi...@apache.org> on 2023/08/10 08:33:56 UTC
[GitHub] [doris] mymeiyi opened a new pull request, #22829: [feature](insert) Support group commit insert
mymeiyi opened a new pull request, #22829:
URL: https://github.com/apache/doris/pull/22829
## Proposed changes
Now `INSERT INTO ... VALUES ...` in Doris is a seperate load job, which will begin a transaction and write a rowset, this is costly for FE metadata, and BE io to compact many small rowsets.
This PR is to solve this problem by making multi inserts be loaded in one job.
Issue Number: close #xxx
<!--Describe your changes.-->
## Further comments
If this is a relatively large or complex change, kick off the discussion at [dev@doris.apache.org](mailto:dev@doris.apache.org) by explaining why you chose the solution you did and what alternatives you considered, etc...
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] dataroaring commented on a diff in pull request #22829: [feature](insert) Support group commit insert
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
dataroaring commented on code in PR #22829:
URL: https://github.com/apache/doris/pull/22829#discussion_r1297176506
##########
be/src/runtime/group_commit_mgr.cpp:
##########
@@ -0,0 +1,480 @@
+// 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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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_first_block_load_queue(
+ int64_t table_id, std::shared_ptr<vectorized::FutureBlock> block,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ DCHECK(table_id == _table_id);
+ DCHECK(block->first == true);
+ {
+ std::unique_lock l(_lock);
+ for (auto it = load_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == 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_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == nullptr) {
+ st = _create_group_commit_load(table_id, load_block_queue);
+ if (LIKELY(st.ok())) {
+ break;
+ }
+ }
+ }
+ RETURN_IF_ERROR(st);
+ if (load_block_queue->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<LoadBlockQueue>& load_block_queue) {
+ 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 schema_version = result.base_schema_version;
+ auto is_pipeline = result.__isset.pipeline_params;
+ auto& params = result.params;
+ auto& pipeline_params = result.pipeline_params;
+ int64_t db_id;
+ std::string label;
+ int64_t txn_id;
+ TUniqueId instance_id;
+ if (!is_pipeline) {
+ db_id = params.fragment.output_sink.olap_table_sink.db_id;
+ label = params.import_label;
+ txn_id = params.txn_conf.txn_id;
+ instance_id = params.params.fragment_instance_id;
+ } else {
+ db_id = pipeline_params.fragment.output_sink.olap_table_sink.db_id;
+ label = pipeline_params.import_label;
+ 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 << ", label=" << label
+ << ", txn_id=" << txn_id << ", instance_id=" << print_id(instance_id)
+ << ", is_pipeline=" << is_pipeline;
+ {
+ load_block_queue =
+ std::make_shared<LoadBlockQueue>(instance_id, label, txn_id, schema_version);
+ std::unique_lock l(_lock);
+ load_block_queues.emplace(instance_id, load_block_queue);
+ }
+ st = _exec_plan_fragment(db_id, table_id, label, txn_id, is_pipeline, params, pipeline_params);
+ if (!st.ok()) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, instance_id, st, true, nullptr);
+ }
+ 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,
+ bool prepare_failed, RuntimeState* state) {
+ {
+ std::lock_guard<doris::Mutex> l(_lock);
+ if (prepare_failed) {
+ auto it = load_block_queues.find(instance_id);
+ if (it != load_block_queues.end()) {
+ it->second->cancel(status);
+ }
+ }
+ load_block_queues.erase(instance_id);
+ }
+ TFinishGroupCommitRequest request;
+ request.__set_db_id(db_id);
+ request.__set_table_id(table_id);
+ request.__set_txn_id(txn_id);
+ request.__set_status(status.to_thrift());
+ if (state) {
+ 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 finish error, db_id=" << db_id << ", table_id=" << table_id
+ << ", label=" << label << ", txn_id=" << txn_id
+ << ", instance_id=" << print_id(instance_id)
+ << ", executor status=" << status.to_string()
+ << ", request commit status=" << st.to_string();
+ return st;
+ }
+ st = Status::create(result.status);
+ // TODO handle execute and commit error
+ 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);
+ if (prepare_failed) {
+ ss << ", prepare status=" << status.to_string();
+ } else {
+ ss << ", execute status=" << status.to_string();
+ }
+ ss << ", commit status=" << st.to_string();
+ if (state && !(state->get_error_log_file_path().empty())) {
+ ss << ", error_url=" << state->get_error_log_file_path();
+ }
+ LOG(INFO) << ss.str();
+ return st;
+}
+
+Status GroupCommitTable::_exec_plan_fragment(int64_t db_id, int64_t table_id,
+ const std::string& label, int64_t txn_id,
+ bool is_pipeline,
+ const TExecPlanFragmentParams& params,
+ const TPipelineFragmentParams& pipeline_params) {
+ auto finish_cb = [db_id, table_id, label, txn_id, this](RuntimeState* state, Status* status) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, state->fragment_instance_id(),
+ *status, false, state);
+ };
+ if (is_pipeline) {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(pipeline_params, finish_cb);
+ } else {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(params, finish_cb);
+ }
+}
+
+Status GroupCommitTable::get_load_block_queue(const TUniqueId& instance_id,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ 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;
+ return Status::OK();
+}
+
+GroupCommitMgr::GroupCommitMgr(ExecEnv* exec_env) : _exec_env(exec_env) {
+ ThreadPoolBuilder("InsertIntoGroupCommitThreadPool")
+ .set_min_threads(config::group_commit_insert_threads)
+ .set_max_threads(config::group_commit_insert_threads)
+ .build(&_insert_into_thread_pool);
+}
+
+GroupCommitMgr::~GroupCommitMgr() {}
+
+Status GroupCommitMgr::group_commit_insert(int64_t table_id, const TPlan& plan,
+ const TDescriptorTable& tdesc_tbl,
+ const TScanRangeParams& scan_range_params,
+ const PGroupCommitInsertRequest* request,
+ PGroupCommitInsertResponse* response) {
+ auto& nodes = plan.nodes;
+ DCHECK(nodes.size() > 0);
+ auto& plan_node = nodes.at(0);
+
+ TUniqueId pipe_id;
+ pipe_id.__set_hi(request->pipe_id().hi());
+ pipe_id.__set_lo(request->pipe_id().lo());
+
+ std::vector<std::shared_ptr<doris::vectorized::FutureBlock>> future_blocks;
+ {
+ std::shared_ptr<LoadBlockQueue> load_block_queue;
+ // new a stream load pipe, put it into stream load mgr, append row in thread pool
+ auto pipe = std::make_shared<io::StreamLoadPipe>(
+ io::kMaxPipeBufferedBytes /* max_buffered_bytes */, 64 * 1024 /* min_chunk_size */,
+ -1 /* total_length */, true /* use_proto */);
+ std::shared_ptr<StreamLoadContext> ctx = std::make_shared<StreamLoadContext>(_exec_env);
+ ctx->pipe = pipe;
+ RETURN_IF_ERROR(_exec_env->new_load_stream_mgr()->put(pipe_id, ctx));
+ std::unique_ptr<int, std::function<void(int*)>> remove_pipe_func((int*)0x01, [&](int*) {
+ if (load_block_queue != nullptr) {
+ load_block_queue->remove_block_id(pipe_id);
+ }
+ _exec_env->new_load_stream_mgr()->remove(pipe_id);
+ });
+ _insert_into_thread_pool->submit_func(
+ std::bind<void>(&GroupCommitMgr::_append_row, this, pipe, request));
+
+ std::unique_ptr<RuntimeState> runtime_state = RuntimeState::create_unique();
+ TQueryOptions query_options;
+ query_options.query_type = TQueryType::LOAD;
+ TQueryGlobals query_globals;
+ runtime_state->init(pipe_id, query_options, query_globals, _exec_env);
+ runtime_state->set_query_mem_tracker(std::make_shared<MemTrackerLimiter>(
+ MemTrackerLimiter::Type::LOAD, fmt::format("Load#Id={}", print_id(pipe_id)), -1));
+ DescriptorTbl* desc_tbl = nullptr;
+ RETURN_IF_ERROR(DescriptorTbl::create(runtime_state->obj_pool(), tdesc_tbl, &desc_tbl));
+ runtime_state->set_desc_tbl(desc_tbl);
+ auto file_scan_node =
+ vectorized::NewFileScanNode(runtime_state->obj_pool(), plan_node, *desc_tbl);
+ std::unique_ptr<int, std::function<void(int*)>> close_scan_node_func(
+ (int*)0x01, [&](int*) { file_scan_node.close(runtime_state.get()); });
+ // TFileFormatType::FORMAT_PROTO, TFileType::FILE_STREAM, set _range.load_id
+ RETURN_IF_ERROR(file_scan_node.init(plan_node, runtime_state.get()));
+ RETURN_IF_ERROR(file_scan_node.prepare(runtime_state.get()));
+ std::vector<TScanRangeParams> params_vector;
+ params_vector.emplace_back(scan_range_params);
+ file_scan_node.set_scan_ranges(params_vector);
+ RETURN_IF_ERROR(file_scan_node.open(runtime_state.get()));
+
+ std::unique_ptr<doris::vectorized::Block> _block =
+ doris::vectorized::Block::create_unique();
+ bool eof = false;
+ bool first = true;
+ while (!eof) {
+ // TODO what to do if read one block error
+ RETURN_IF_ERROR(file_scan_node.get_next(runtime_state.get(), _block.get(), &eof));
+ std::shared_ptr<doris::vectorized::FutureBlock> future_block =
+ std::make_shared<doris::vectorized::FutureBlock>();
+ future_block->swap(*(_block.get()));
+ future_block->set_info(request->base_schema_version(), pipe_id, first, eof);
+ // TODO what to do if add one block error
+ if (load_block_queue == nullptr) {
+ RETURN_IF_ERROR(_get_first_block_load_queue(request->db_id(), table_id,
+ future_block, load_block_queue));
+ response->set_label(load_block_queue->label);
+ response->set_txn_id(load_block_queue->txn_id);
+ }
+ RETURN_IF_ERROR(load_block_queue->add_block(future_block));
+ if (future_block->rows() > 0) {
+ future_blocks.emplace_back(future_block);
+ }
+ first = false;
+ }
+ if (!runtime_state->get_error_log_file_path().empty()) {
+ LOG(INFO) << "id=" << print_id(pipe_id)
+ << ", url=" << runtime_state->get_error_log_file_path()
+ << ", load rows=" << runtime_state->num_rows_load_total()
+ << ", filter rows=" << runtime_state->num_rows_load_filtered()
+ << ", unselect rows=" << runtime_state->num_rows_load_unselected()
+ << ", success rows=" << runtime_state->num_rows_load_success();
+ }
+ }
+ int64_t total_rows = 0;
+ int64_t loaded_rows = 0;
+ for (const auto& future_block : future_blocks) {
+ std::unique_lock<doris::Mutex> l(*(future_block->lock));
+ if (!std::get<0>(*(future_block->block_status))) {
Review Comment:
future_block->if_handled() is much more readable.
##########
be/src/runtime/group_commit_mgr.cpp:
##########
@@ -0,0 +1,480 @@
+// 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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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_first_block_load_queue(
+ int64_t table_id, std::shared_ptr<vectorized::FutureBlock> block,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ DCHECK(table_id == _table_id);
+ DCHECK(block->first == true);
+ {
+ std::unique_lock l(_lock);
+ for (auto it = load_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == 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_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == nullptr) {
+ st = _create_group_commit_load(table_id, load_block_queue);
+ if (LIKELY(st.ok())) {
+ break;
+ }
+ }
+ }
+ RETURN_IF_ERROR(st);
+ if (load_block_queue->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<LoadBlockQueue>& load_block_queue) {
+ 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 schema_version = result.base_schema_version;
+ auto is_pipeline = result.__isset.pipeline_params;
+ auto& params = result.params;
+ auto& pipeline_params = result.pipeline_params;
+ int64_t db_id;
+ std::string label;
+ int64_t txn_id;
+ TUniqueId instance_id;
+ if (!is_pipeline) {
+ db_id = params.fragment.output_sink.olap_table_sink.db_id;
+ label = params.import_label;
+ txn_id = params.txn_conf.txn_id;
+ instance_id = params.params.fragment_instance_id;
+ } else {
+ db_id = pipeline_params.fragment.output_sink.olap_table_sink.db_id;
+ label = pipeline_params.import_label;
+ 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 << ", label=" << label
+ << ", txn_id=" << txn_id << ", instance_id=" << print_id(instance_id)
+ << ", is_pipeline=" << is_pipeline;
+ {
+ load_block_queue =
+ std::make_shared<LoadBlockQueue>(instance_id, label, txn_id, schema_version);
+ std::unique_lock l(_lock);
+ load_block_queues.emplace(instance_id, load_block_queue);
+ }
+ st = _exec_plan_fragment(db_id, table_id, label, txn_id, is_pipeline, params, pipeline_params);
+ if (!st.ok()) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, instance_id, st, true, nullptr);
+ }
+ 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,
+ bool prepare_failed, RuntimeState* state) {
+ {
+ std::lock_guard<doris::Mutex> l(_lock);
+ if (prepare_failed) {
+ auto it = load_block_queues.find(instance_id);
+ if (it != load_block_queues.end()) {
+ it->second->cancel(status);
+ }
+ }
+ load_block_queues.erase(instance_id);
+ }
+ TFinishGroupCommitRequest request;
+ request.__set_db_id(db_id);
+ request.__set_table_id(table_id);
+ request.__set_txn_id(txn_id);
+ request.__set_status(status.to_thrift());
+ if (state) {
+ 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 finish error, db_id=" << db_id << ", table_id=" << table_id
+ << ", label=" << label << ", txn_id=" << txn_id
+ << ", instance_id=" << print_id(instance_id)
+ << ", executor status=" << status.to_string()
+ << ", request commit status=" << st.to_string();
+ return st;
+ }
+ st = Status::create(result.status);
+ // TODO handle execute and commit error
+ 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);
+ if (prepare_failed) {
+ ss << ", prepare status=" << status.to_string();
+ } else {
+ ss << ", execute status=" << status.to_string();
+ }
+ ss << ", commit status=" << st.to_string();
+ if (state && !(state->get_error_log_file_path().empty())) {
+ ss << ", error_url=" << state->get_error_log_file_path();
+ }
+ LOG(INFO) << ss.str();
+ return st;
+}
+
+Status GroupCommitTable::_exec_plan_fragment(int64_t db_id, int64_t table_id,
+ const std::string& label, int64_t txn_id,
+ bool is_pipeline,
+ const TExecPlanFragmentParams& params,
+ const TPipelineFragmentParams& pipeline_params) {
+ auto finish_cb = [db_id, table_id, label, txn_id, this](RuntimeState* state, Status* status) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, state->fragment_instance_id(),
+ *status, false, state);
+ };
+ if (is_pipeline) {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(pipeline_params, finish_cb);
+ } else {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(params, finish_cb);
+ }
+}
+
+Status GroupCommitTable::get_load_block_queue(const TUniqueId& instance_id,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ 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;
+ return Status::OK();
+}
+
+GroupCommitMgr::GroupCommitMgr(ExecEnv* exec_env) : _exec_env(exec_env) {
+ ThreadPoolBuilder("InsertIntoGroupCommitThreadPool")
+ .set_min_threads(config::group_commit_insert_threads)
+ .set_max_threads(config::group_commit_insert_threads)
+ .build(&_insert_into_thread_pool);
+}
+
+GroupCommitMgr::~GroupCommitMgr() {}
+
+Status GroupCommitMgr::group_commit_insert(int64_t table_id, const TPlan& plan,
+ const TDescriptorTable& tdesc_tbl,
+ const TScanRangeParams& scan_range_params,
+ const PGroupCommitInsertRequest* request,
+ PGroupCommitInsertResponse* response) {
+ auto& nodes = plan.nodes;
+ DCHECK(nodes.size() > 0);
+ auto& plan_node = nodes.at(0);
+
+ TUniqueId pipe_id;
+ pipe_id.__set_hi(request->pipe_id().hi());
+ pipe_id.__set_lo(request->pipe_id().lo());
+
+ std::vector<std::shared_ptr<doris::vectorized::FutureBlock>> future_blocks;
+ {
+ std::shared_ptr<LoadBlockQueue> load_block_queue;
+ // new a stream load pipe, put it into stream load mgr, append row in thread pool
+ auto pipe = std::make_shared<io::StreamLoadPipe>(
+ io::kMaxPipeBufferedBytes /* max_buffered_bytes */, 64 * 1024 /* min_chunk_size */,
+ -1 /* total_length */, true /* use_proto */);
+ std::shared_ptr<StreamLoadContext> ctx = std::make_shared<StreamLoadContext>(_exec_env);
+ ctx->pipe = pipe;
+ RETURN_IF_ERROR(_exec_env->new_load_stream_mgr()->put(pipe_id, ctx));
+ std::unique_ptr<int, std::function<void(int*)>> remove_pipe_func((int*)0x01, [&](int*) {
+ if (load_block_queue != nullptr) {
+ load_block_queue->remove_block_id(pipe_id);
+ }
+ _exec_env->new_load_stream_mgr()->remove(pipe_id);
+ });
+ _insert_into_thread_pool->submit_func(
+ std::bind<void>(&GroupCommitMgr::_append_row, this, pipe, request));
+
+ std::unique_ptr<RuntimeState> runtime_state = RuntimeState::create_unique();
+ TQueryOptions query_options;
+ query_options.query_type = TQueryType::LOAD;
+ TQueryGlobals query_globals;
+ runtime_state->init(pipe_id, query_options, query_globals, _exec_env);
+ runtime_state->set_query_mem_tracker(std::make_shared<MemTrackerLimiter>(
+ MemTrackerLimiter::Type::LOAD, fmt::format("Load#Id={}", print_id(pipe_id)), -1));
+ DescriptorTbl* desc_tbl = nullptr;
+ RETURN_IF_ERROR(DescriptorTbl::create(runtime_state->obj_pool(), tdesc_tbl, &desc_tbl));
+ runtime_state->set_desc_tbl(desc_tbl);
+ auto file_scan_node =
+ vectorized::NewFileScanNode(runtime_state->obj_pool(), plan_node, *desc_tbl);
+ std::unique_ptr<int, std::function<void(int*)>> close_scan_node_func(
+ (int*)0x01, [&](int*) { file_scan_node.close(runtime_state.get()); });
+ // TFileFormatType::FORMAT_PROTO, TFileType::FILE_STREAM, set _range.load_id
+ RETURN_IF_ERROR(file_scan_node.init(plan_node, runtime_state.get()));
+ RETURN_IF_ERROR(file_scan_node.prepare(runtime_state.get()));
+ std::vector<TScanRangeParams> params_vector;
+ params_vector.emplace_back(scan_range_params);
+ file_scan_node.set_scan_ranges(params_vector);
+ RETURN_IF_ERROR(file_scan_node.open(runtime_state.get()));
+
+ std::unique_ptr<doris::vectorized::Block> _block =
+ doris::vectorized::Block::create_unique();
+ bool eof = false;
+ bool first = true;
+ while (!eof) {
+ // TODO what to do if read one block error
+ RETURN_IF_ERROR(file_scan_node.get_next(runtime_state.get(), _block.get(), &eof));
+ std::shared_ptr<doris::vectorized::FutureBlock> future_block =
+ std::make_shared<doris::vectorized::FutureBlock>();
+ future_block->swap(*(_block.get()));
+ future_block->set_info(request->base_schema_version(), pipe_id, first, eof);
+ // TODO what to do if add one block error
+ if (load_block_queue == nullptr) {
+ RETURN_IF_ERROR(_get_first_block_load_queue(request->db_id(), table_id,
+ future_block, load_block_queue));
+ response->set_label(load_block_queue->label);
+ response->set_txn_id(load_block_queue->txn_id);
+ }
+ RETURN_IF_ERROR(load_block_queue->add_block(future_block));
+ if (future_block->rows() > 0) {
+ future_blocks.emplace_back(future_block);
+ }
+ first = false;
+ }
+ if (!runtime_state->get_error_log_file_path().empty()) {
+ LOG(INFO) << "id=" << print_id(pipe_id)
+ << ", url=" << runtime_state->get_error_log_file_path()
+ << ", load rows=" << runtime_state->num_rows_load_total()
+ << ", filter rows=" << runtime_state->num_rows_load_filtered()
+ << ", unselect rows=" << runtime_state->num_rows_load_unselected()
+ << ", success rows=" << runtime_state->num_rows_load_success();
+ }
+ }
+ int64_t total_rows = 0;
+ int64_t loaded_rows = 0;
+ for (const auto& future_block : future_blocks) {
+ std::unique_lock<doris::Mutex> l(*(future_block->lock));
+ if (!std::get<0>(*(future_block->block_status))) {
Review Comment:
future_block->if_handled() is much more readable.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1681917893
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1673127516
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] dataroaring commented on a diff in pull request #22829: [feature](insert) Support group commit insert
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
dataroaring commented on code in PR #22829:
URL: https://github.com/apache/doris/pull/22829#discussion_r1297181372
##########
be/src/vec/core/block.h:
##########
@@ -642,5 +645,25 @@ struct IteratorRowRef {
using BlockView = std::vector<IteratorRowRef>;
using BlockUPtr = std::unique_ptr<Block>;
+class FutureBlock : public Block {
Review Comment:
move to independent file, e.g. future_block.
##########
be/src/runtime/group_commit_mgr.cpp:
##########
@@ -0,0 +1,480 @@
+// 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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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_first_block_load_queue(
+ int64_t table_id, std::shared_ptr<vectorized::FutureBlock> block,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ DCHECK(table_id == _table_id);
+ DCHECK(block->first == true);
+ {
+ std::unique_lock l(_lock);
+ for (auto it = load_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == 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_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == nullptr) {
+ st = _create_group_commit_load(table_id, load_block_queue);
+ if (LIKELY(st.ok())) {
+ break;
+ }
+ }
+ }
+ RETURN_IF_ERROR(st);
+ if (load_block_queue->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<LoadBlockQueue>& load_block_queue) {
+ 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 schema_version = result.base_schema_version;
+ auto is_pipeline = result.__isset.pipeline_params;
+ auto& params = result.params;
+ auto& pipeline_params = result.pipeline_params;
+ int64_t db_id;
+ std::string label;
+ int64_t txn_id;
+ TUniqueId instance_id;
+ if (!is_pipeline) {
+ db_id = params.fragment.output_sink.olap_table_sink.db_id;
+ label = params.import_label;
+ txn_id = params.txn_conf.txn_id;
+ instance_id = params.params.fragment_instance_id;
+ } else {
+ db_id = pipeline_params.fragment.output_sink.olap_table_sink.db_id;
+ label = pipeline_params.import_label;
+ 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 << ", label=" << label
+ << ", txn_id=" << txn_id << ", instance_id=" << print_id(instance_id)
+ << ", is_pipeline=" << is_pipeline;
+ {
+ load_block_queue =
+ std::make_shared<LoadBlockQueue>(instance_id, label, txn_id, schema_version);
+ std::unique_lock l(_lock);
+ load_block_queues.emplace(instance_id, load_block_queue);
+ }
+ st = _exec_plan_fragment(db_id, table_id, label, txn_id, is_pipeline, params, pipeline_params);
+ if (!st.ok()) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, instance_id, st, true, nullptr);
+ }
+ 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,
+ bool prepare_failed, RuntimeState* state) {
+ {
+ std::lock_guard<doris::Mutex> l(_lock);
+ if (prepare_failed) {
+ auto it = load_block_queues.find(instance_id);
+ if (it != load_block_queues.end()) {
+ it->second->cancel(status);
+ }
+ }
+ load_block_queues.erase(instance_id);
+ }
+ TFinishGroupCommitRequest request;
+ request.__set_db_id(db_id);
+ request.__set_table_id(table_id);
+ request.__set_txn_id(txn_id);
+ request.__set_status(status.to_thrift());
+ if (state) {
+ 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 finish error, db_id=" << db_id << ", table_id=" << table_id
+ << ", label=" << label << ", txn_id=" << txn_id
+ << ", instance_id=" << print_id(instance_id)
+ << ", executor status=" << status.to_string()
+ << ", request commit status=" << st.to_string();
+ return st;
+ }
+ st = Status::create(result.status);
+ // TODO handle execute and commit error
+ 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);
+ if (prepare_failed) {
+ ss << ", prepare status=" << status.to_string();
+ } else {
+ ss << ", execute status=" << status.to_string();
+ }
+ ss << ", commit status=" << st.to_string();
+ if (state && !(state->get_error_log_file_path().empty())) {
+ ss << ", error_url=" << state->get_error_log_file_path();
+ }
+ LOG(INFO) << ss.str();
+ return st;
+}
+
+Status GroupCommitTable::_exec_plan_fragment(int64_t db_id, int64_t table_id,
+ const std::string& label, int64_t txn_id,
+ bool is_pipeline,
+ const TExecPlanFragmentParams& params,
+ const TPipelineFragmentParams& pipeline_params) {
+ auto finish_cb = [db_id, table_id, label, txn_id, this](RuntimeState* state, Status* status) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, state->fragment_instance_id(),
+ *status, false, state);
+ };
+ if (is_pipeline) {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(pipeline_params, finish_cb);
+ } else {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(params, finish_cb);
+ }
+}
+
+Status GroupCommitTable::get_load_block_queue(const TUniqueId& instance_id,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ 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;
+ return Status::OK();
+}
+
+GroupCommitMgr::GroupCommitMgr(ExecEnv* exec_env) : _exec_env(exec_env) {
+ ThreadPoolBuilder("InsertIntoGroupCommitThreadPool")
+ .set_min_threads(config::group_commit_insert_threads)
+ .set_max_threads(config::group_commit_insert_threads)
+ .build(&_insert_into_thread_pool);
+}
+
+GroupCommitMgr::~GroupCommitMgr() {}
+
+Status GroupCommitMgr::group_commit_insert(int64_t table_id, const TPlan& plan,
+ const TDescriptorTable& tdesc_tbl,
+ const TScanRangeParams& scan_range_params,
+ const PGroupCommitInsertRequest* request,
+ PGroupCommitInsertResponse* response) {
+ auto& nodes = plan.nodes;
+ DCHECK(nodes.size() > 0);
+ auto& plan_node = nodes.at(0);
+
+ TUniqueId pipe_id;
+ pipe_id.__set_hi(request->pipe_id().hi());
+ pipe_id.__set_lo(request->pipe_id().lo());
+
+ std::vector<std::shared_ptr<doris::vectorized::FutureBlock>> future_blocks;
+ {
+ std::shared_ptr<LoadBlockQueue> load_block_queue;
+ // new a stream load pipe, put it into stream load mgr, append row in thread pool
+ auto pipe = std::make_shared<io::StreamLoadPipe>(
+ io::kMaxPipeBufferedBytes /* max_buffered_bytes */, 64 * 1024 /* min_chunk_size */,
+ -1 /* total_length */, true /* use_proto */);
+ std::shared_ptr<StreamLoadContext> ctx = std::make_shared<StreamLoadContext>(_exec_env);
+ ctx->pipe = pipe;
+ RETURN_IF_ERROR(_exec_env->new_load_stream_mgr()->put(pipe_id, ctx));
+ std::unique_ptr<int, std::function<void(int*)>> remove_pipe_func((int*)0x01, [&](int*) {
+ if (load_block_queue != nullptr) {
+ load_block_queue->remove_block_id(pipe_id);
+ }
+ _exec_env->new_load_stream_mgr()->remove(pipe_id);
+ });
+ _insert_into_thread_pool->submit_func(
+ std::bind<void>(&GroupCommitMgr::_append_row, this, pipe, request));
+
+ std::unique_ptr<RuntimeState> runtime_state = RuntimeState::create_unique();
+ TQueryOptions query_options;
+ query_options.query_type = TQueryType::LOAD;
+ TQueryGlobals query_globals;
+ runtime_state->init(pipe_id, query_options, query_globals, _exec_env);
+ runtime_state->set_query_mem_tracker(std::make_shared<MemTrackerLimiter>(
+ MemTrackerLimiter::Type::LOAD, fmt::format("Load#Id={}", print_id(pipe_id)), -1));
+ DescriptorTbl* desc_tbl = nullptr;
+ RETURN_IF_ERROR(DescriptorTbl::create(runtime_state->obj_pool(), tdesc_tbl, &desc_tbl));
+ runtime_state->set_desc_tbl(desc_tbl);
+ auto file_scan_node =
+ vectorized::NewFileScanNode(runtime_state->obj_pool(), plan_node, *desc_tbl);
+ std::unique_ptr<int, std::function<void(int*)>> close_scan_node_func(
+ (int*)0x01, [&](int*) { file_scan_node.close(runtime_state.get()); });
+ // TFileFormatType::FORMAT_PROTO, TFileType::FILE_STREAM, set _range.load_id
+ RETURN_IF_ERROR(file_scan_node.init(plan_node, runtime_state.get()));
+ RETURN_IF_ERROR(file_scan_node.prepare(runtime_state.get()));
+ std::vector<TScanRangeParams> params_vector;
+ params_vector.emplace_back(scan_range_params);
+ file_scan_node.set_scan_ranges(params_vector);
+ RETURN_IF_ERROR(file_scan_node.open(runtime_state.get()));
+
+ std::unique_ptr<doris::vectorized::Block> _block =
+ doris::vectorized::Block::create_unique();
+ bool eof = false;
+ bool first = true;
+ while (!eof) {
+ // TODO what to do if read one block error
+ RETURN_IF_ERROR(file_scan_node.get_next(runtime_state.get(), _block.get(), &eof));
+ std::shared_ptr<doris::vectorized::FutureBlock> future_block =
+ std::make_shared<doris::vectorized::FutureBlock>();
+ future_block->swap(*(_block.get()));
+ future_block->set_info(request->base_schema_version(), pipe_id, first, eof);
+ // TODO what to do if add one block error
Review Comment:
3. Put the block into block queue.
##########
be/src/runtime/group_commit_mgr.cpp:
##########
@@ -0,0 +1,480 @@
+// 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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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_first_block_load_queue(
+ int64_t table_id, std::shared_ptr<vectorized::FutureBlock> block,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ DCHECK(table_id == _table_id);
+ DCHECK(block->first == true);
+ {
+ std::unique_lock l(_lock);
+ for (auto it = load_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == 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_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == nullptr) {
+ st = _create_group_commit_load(table_id, load_block_queue);
+ if (LIKELY(st.ok())) {
+ break;
+ }
+ }
+ }
+ RETURN_IF_ERROR(st);
+ if (load_block_queue->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<LoadBlockQueue>& load_block_queue) {
+ 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 schema_version = result.base_schema_version;
+ auto is_pipeline = result.__isset.pipeline_params;
+ auto& params = result.params;
+ auto& pipeline_params = result.pipeline_params;
+ int64_t db_id;
+ std::string label;
+ int64_t txn_id;
+ TUniqueId instance_id;
+ if (!is_pipeline) {
+ db_id = params.fragment.output_sink.olap_table_sink.db_id;
+ label = params.import_label;
+ txn_id = params.txn_conf.txn_id;
+ instance_id = params.params.fragment_instance_id;
+ } else {
+ db_id = pipeline_params.fragment.output_sink.olap_table_sink.db_id;
+ label = pipeline_params.import_label;
+ 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 << ", label=" << label
+ << ", txn_id=" << txn_id << ", instance_id=" << print_id(instance_id)
+ << ", is_pipeline=" << is_pipeline;
+ {
+ load_block_queue =
+ std::make_shared<LoadBlockQueue>(instance_id, label, txn_id, schema_version);
+ std::unique_lock l(_lock);
+ load_block_queues.emplace(instance_id, load_block_queue);
+ }
+ st = _exec_plan_fragment(db_id, table_id, label, txn_id, is_pipeline, params, pipeline_params);
+ if (!st.ok()) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, instance_id, st, true, nullptr);
+ }
+ 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,
+ bool prepare_failed, RuntimeState* state) {
+ {
+ std::lock_guard<doris::Mutex> l(_lock);
+ if (prepare_failed) {
+ auto it = load_block_queues.find(instance_id);
+ if (it != load_block_queues.end()) {
+ it->second->cancel(status);
+ }
+ }
+ load_block_queues.erase(instance_id);
+ }
+ TFinishGroupCommitRequest request;
+ request.__set_db_id(db_id);
+ request.__set_table_id(table_id);
+ request.__set_txn_id(txn_id);
+ request.__set_status(status.to_thrift());
+ if (state) {
+ 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 finish error, db_id=" << db_id << ", table_id=" << table_id
+ << ", label=" << label << ", txn_id=" << txn_id
+ << ", instance_id=" << print_id(instance_id)
+ << ", executor status=" << status.to_string()
+ << ", request commit status=" << st.to_string();
+ return st;
+ }
+ st = Status::create(result.status);
+ // TODO handle execute and commit error
+ 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);
+ if (prepare_failed) {
+ ss << ", prepare status=" << status.to_string();
+ } else {
+ ss << ", execute status=" << status.to_string();
+ }
+ ss << ", commit status=" << st.to_string();
+ if (state && !(state->get_error_log_file_path().empty())) {
+ ss << ", error_url=" << state->get_error_log_file_path();
+ }
+ LOG(INFO) << ss.str();
+ return st;
+}
+
+Status GroupCommitTable::_exec_plan_fragment(int64_t db_id, int64_t table_id,
+ const std::string& label, int64_t txn_id,
+ bool is_pipeline,
+ const TExecPlanFragmentParams& params,
+ const TPipelineFragmentParams& pipeline_params) {
+ auto finish_cb = [db_id, table_id, label, txn_id, this](RuntimeState* state, Status* status) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, state->fragment_instance_id(),
+ *status, false, state);
+ };
+ if (is_pipeline) {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(pipeline_params, finish_cb);
+ } else {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(params, finish_cb);
+ }
+}
+
+Status GroupCommitTable::get_load_block_queue(const TUniqueId& instance_id,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ 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;
+ return Status::OK();
+}
+
+GroupCommitMgr::GroupCommitMgr(ExecEnv* exec_env) : _exec_env(exec_env) {
+ ThreadPoolBuilder("InsertIntoGroupCommitThreadPool")
+ .set_min_threads(config::group_commit_insert_threads)
+ .set_max_threads(config::group_commit_insert_threads)
+ .build(&_insert_into_thread_pool);
+}
+
+GroupCommitMgr::~GroupCommitMgr() {}
+
+Status GroupCommitMgr::group_commit_insert(int64_t table_id, const TPlan& plan,
+ const TDescriptorTable& tdesc_tbl,
+ const TScanRangeParams& scan_range_params,
+ const PGroupCommitInsertRequest* request,
+ PGroupCommitInsertResponse* response) {
+ auto& nodes = plan.nodes;
+ DCHECK(nodes.size() > 0);
+ auto& plan_node = nodes.at(0);
+
+ TUniqueId pipe_id;
+ pipe_id.__set_hi(request->pipe_id().hi());
+ pipe_id.__set_lo(request->pipe_id().lo());
+
+ std::vector<std::shared_ptr<doris::vectorized::FutureBlock>> future_blocks;
+ {
Review Comment:
1. Prepare a pipe, then append rows to pipe, then scan node scans from the pipe, like stream load.
##########
be/src/runtime/group_commit_mgr.cpp:
##########
@@ -0,0 +1,480 @@
+// 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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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_first_block_load_queue(
+ int64_t table_id, std::shared_ptr<vectorized::FutureBlock> block,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ DCHECK(table_id == _table_id);
+ DCHECK(block->first == true);
+ {
+ std::unique_lock l(_lock);
+ for (auto it = load_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == 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_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == nullptr) {
+ st = _create_group_commit_load(table_id, load_block_queue);
+ if (LIKELY(st.ok())) {
+ break;
+ }
+ }
+ }
+ RETURN_IF_ERROR(st);
+ if (load_block_queue->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<LoadBlockQueue>& load_block_queue) {
+ 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 schema_version = result.base_schema_version;
+ auto is_pipeline = result.__isset.pipeline_params;
+ auto& params = result.params;
+ auto& pipeline_params = result.pipeline_params;
+ int64_t db_id;
+ std::string label;
+ int64_t txn_id;
+ TUniqueId instance_id;
+ if (!is_pipeline) {
+ db_id = params.fragment.output_sink.olap_table_sink.db_id;
+ label = params.import_label;
+ txn_id = params.txn_conf.txn_id;
+ instance_id = params.params.fragment_instance_id;
+ } else {
+ db_id = pipeline_params.fragment.output_sink.olap_table_sink.db_id;
+ label = pipeline_params.import_label;
+ 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 << ", label=" << label
+ << ", txn_id=" << txn_id << ", instance_id=" << print_id(instance_id)
+ << ", is_pipeline=" << is_pipeline;
+ {
+ load_block_queue =
+ std::make_shared<LoadBlockQueue>(instance_id, label, txn_id, schema_version);
+ std::unique_lock l(_lock);
+ load_block_queues.emplace(instance_id, load_block_queue);
+ }
+ st = _exec_plan_fragment(db_id, table_id, label, txn_id, is_pipeline, params, pipeline_params);
+ if (!st.ok()) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, instance_id, st, true, nullptr);
+ }
+ 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,
+ bool prepare_failed, RuntimeState* state) {
+ {
+ std::lock_guard<doris::Mutex> l(_lock);
+ if (prepare_failed) {
+ auto it = load_block_queues.find(instance_id);
+ if (it != load_block_queues.end()) {
+ it->second->cancel(status);
+ }
+ }
+ load_block_queues.erase(instance_id);
+ }
+ TFinishGroupCommitRequest request;
+ request.__set_db_id(db_id);
+ request.__set_table_id(table_id);
+ request.__set_txn_id(txn_id);
+ request.__set_status(status.to_thrift());
+ if (state) {
+ 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 finish error, db_id=" << db_id << ", table_id=" << table_id
+ << ", label=" << label << ", txn_id=" << txn_id
+ << ", instance_id=" << print_id(instance_id)
+ << ", executor status=" << status.to_string()
+ << ", request commit status=" << st.to_string();
+ return st;
+ }
+ st = Status::create(result.status);
+ // TODO handle execute and commit error
+ 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);
+ if (prepare_failed) {
+ ss << ", prepare status=" << status.to_string();
+ } else {
+ ss << ", execute status=" << status.to_string();
+ }
+ ss << ", commit status=" << st.to_string();
+ if (state && !(state->get_error_log_file_path().empty())) {
+ ss << ", error_url=" << state->get_error_log_file_path();
+ }
+ LOG(INFO) << ss.str();
+ return st;
+}
+
+Status GroupCommitTable::_exec_plan_fragment(int64_t db_id, int64_t table_id,
+ const std::string& label, int64_t txn_id,
+ bool is_pipeline,
+ const TExecPlanFragmentParams& params,
+ const TPipelineFragmentParams& pipeline_params) {
+ auto finish_cb = [db_id, table_id, label, txn_id, this](RuntimeState* state, Status* status) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, state->fragment_instance_id(),
+ *status, false, state);
+ };
+ if (is_pipeline) {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(pipeline_params, finish_cb);
+ } else {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(params, finish_cb);
+ }
+}
+
+Status GroupCommitTable::get_load_block_queue(const TUniqueId& instance_id,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ 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;
+ return Status::OK();
+}
+
+GroupCommitMgr::GroupCommitMgr(ExecEnv* exec_env) : _exec_env(exec_env) {
+ ThreadPoolBuilder("InsertIntoGroupCommitThreadPool")
+ .set_min_threads(config::group_commit_insert_threads)
+ .set_max_threads(config::group_commit_insert_threads)
+ .build(&_insert_into_thread_pool);
+}
+
+GroupCommitMgr::~GroupCommitMgr() {}
+
+Status GroupCommitMgr::group_commit_insert(int64_t table_id, const TPlan& plan,
+ const TDescriptorTable& tdesc_tbl,
+ const TScanRangeParams& scan_range_params,
+ const PGroupCommitInsertRequest* request,
+ PGroupCommitInsertResponse* response) {
+ auto& nodes = plan.nodes;
+ DCHECK(nodes.size() > 0);
+ auto& plan_node = nodes.at(0);
+
+ TUniqueId pipe_id;
+ pipe_id.__set_hi(request->pipe_id().hi());
+ pipe_id.__set_lo(request->pipe_id().lo());
+
+ std::vector<std::shared_ptr<doris::vectorized::FutureBlock>> future_blocks;
+ {
+ std::shared_ptr<LoadBlockQueue> load_block_queue;
+ // new a stream load pipe, put it into stream load mgr, append row in thread pool
+ auto pipe = std::make_shared<io::StreamLoadPipe>(
+ io::kMaxPipeBufferedBytes /* max_buffered_bytes */, 64 * 1024 /* min_chunk_size */,
+ -1 /* total_length */, true /* use_proto */);
+ std::shared_ptr<StreamLoadContext> ctx = std::make_shared<StreamLoadContext>(_exec_env);
+ ctx->pipe = pipe;
+ RETURN_IF_ERROR(_exec_env->new_load_stream_mgr()->put(pipe_id, ctx));
+ std::unique_ptr<int, std::function<void(int*)>> remove_pipe_func((int*)0x01, [&](int*) {
+ if (load_block_queue != nullptr) {
+ load_block_queue->remove_block_id(pipe_id);
+ }
+ _exec_env->new_load_stream_mgr()->remove(pipe_id);
+ });
+ _insert_into_thread_pool->submit_func(
+ std::bind<void>(&GroupCommitMgr::_append_row, this, pipe, request));
+
+ std::unique_ptr<RuntimeState> runtime_state = RuntimeState::create_unique();
+ TQueryOptions query_options;
+ query_options.query_type = TQueryType::LOAD;
+ TQueryGlobals query_globals;
+ runtime_state->init(pipe_id, query_options, query_globals, _exec_env);
+ runtime_state->set_query_mem_tracker(std::make_shared<MemTrackerLimiter>(
+ MemTrackerLimiter::Type::LOAD, fmt::format("Load#Id={}", print_id(pipe_id)), -1));
+ DescriptorTbl* desc_tbl = nullptr;
+ RETURN_IF_ERROR(DescriptorTbl::create(runtime_state->obj_pool(), tdesc_tbl, &desc_tbl));
+ runtime_state->set_desc_tbl(desc_tbl);
+ auto file_scan_node =
+ vectorized::NewFileScanNode(runtime_state->obj_pool(), plan_node, *desc_tbl);
+ std::unique_ptr<int, std::function<void(int*)>> close_scan_node_func(
+ (int*)0x01, [&](int*) { file_scan_node.close(runtime_state.get()); });
+ // TFileFormatType::FORMAT_PROTO, TFileType::FILE_STREAM, set _range.load_id
+ RETURN_IF_ERROR(file_scan_node.init(plan_node, runtime_state.get()));
+ RETURN_IF_ERROR(file_scan_node.prepare(runtime_state.get()));
+ std::vector<TScanRangeParams> params_vector;
+ params_vector.emplace_back(scan_range_params);
+ file_scan_node.set_scan_ranges(params_vector);
+ RETURN_IF_ERROR(file_scan_node.open(runtime_state.get()));
+
+ std::unique_ptr<doris::vectorized::Block> _block =
+ doris::vectorized::Block::create_unique();
+ bool eof = false;
+ bool first = true;
+ while (!eof) {
+ // TODO what to do if read one block error
+ RETURN_IF_ERROR(file_scan_node.get_next(runtime_state.get(), _block.get(), &eof));
+ std::shared_ptr<doris::vectorized::FutureBlock> future_block =
+ std::make_shared<doris::vectorized::FutureBlock>();
+ future_block->swap(*(_block.get()));
+ future_block->set_info(request->base_schema_version(), pipe_id, first, eof);
+ // TODO what to do if add one block error
+ if (load_block_queue == nullptr) {
+ RETURN_IF_ERROR(_get_first_block_load_queue(request->db_id(), table_id,
+ future_block, load_block_queue));
+ response->set_label(load_block_queue->label);
+ response->set_txn_id(load_block_queue->txn_id);
+ }
+ RETURN_IF_ERROR(load_block_queue->add_block(future_block));
+ if (future_block->rows() > 0) {
+ future_blocks.emplace_back(future_block);
+ }
+ first = false;
+ }
+ if (!runtime_state->get_error_log_file_path().empty()) {
+ LOG(INFO) << "id=" << print_id(pipe_id)
+ << ", url=" << runtime_state->get_error_log_file_path()
+ << ", load rows=" << runtime_state->num_rows_load_total()
+ << ", filter rows=" << runtime_state->num_rows_load_filtered()
+ << ", unselect rows=" << runtime_state->num_rows_load_unselected()
+ << ", success rows=" << runtime_state->num_rows_load_success();
+ }
+ }
+ int64_t total_rows = 0;
+ int64_t loaded_rows = 0;
+ for (const auto& future_block : future_blocks) {
+ std::unique_lock<doris::Mutex> l(*(future_block->lock));
+ if (!std::get<0>(*(future_block->block_status))) {
+ future_block->cv->wait(l);
Review Comment:
4 wait to wal
##########
be/src/runtime/group_commit_mgr.cpp:
##########
@@ -0,0 +1,480 @@
+// 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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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_first_block_load_queue(
+ int64_t table_id, std::shared_ptr<vectorized::FutureBlock> block,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ DCHECK(table_id == _table_id);
+ DCHECK(block->first == true);
+ {
+ std::unique_lock l(_lock);
+ for (auto it = load_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == 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_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == nullptr) {
+ st = _create_group_commit_load(table_id, load_block_queue);
+ if (LIKELY(st.ok())) {
+ break;
+ }
+ }
+ }
+ RETURN_IF_ERROR(st);
+ if (load_block_queue->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<LoadBlockQueue>& load_block_queue) {
+ 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 schema_version = result.base_schema_version;
+ auto is_pipeline = result.__isset.pipeline_params;
+ auto& params = result.params;
+ auto& pipeline_params = result.pipeline_params;
+ int64_t db_id;
+ std::string label;
+ int64_t txn_id;
+ TUniqueId instance_id;
+ if (!is_pipeline) {
+ db_id = params.fragment.output_sink.olap_table_sink.db_id;
+ label = params.import_label;
+ txn_id = params.txn_conf.txn_id;
+ instance_id = params.params.fragment_instance_id;
+ } else {
+ db_id = pipeline_params.fragment.output_sink.olap_table_sink.db_id;
+ label = pipeline_params.import_label;
+ 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 << ", label=" << label
+ << ", txn_id=" << txn_id << ", instance_id=" << print_id(instance_id)
+ << ", is_pipeline=" << is_pipeline;
+ {
+ load_block_queue =
+ std::make_shared<LoadBlockQueue>(instance_id, label, txn_id, schema_version);
+ std::unique_lock l(_lock);
+ load_block_queues.emplace(instance_id, load_block_queue);
+ }
+ st = _exec_plan_fragment(db_id, table_id, label, txn_id, is_pipeline, params, pipeline_params);
+ if (!st.ok()) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, instance_id, st, true, nullptr);
+ }
+ 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,
+ bool prepare_failed, RuntimeState* state) {
+ {
+ std::lock_guard<doris::Mutex> l(_lock);
+ if (prepare_failed) {
+ auto it = load_block_queues.find(instance_id);
+ if (it != load_block_queues.end()) {
+ it->second->cancel(status);
+ }
+ }
+ load_block_queues.erase(instance_id);
+ }
+ TFinishGroupCommitRequest request;
+ request.__set_db_id(db_id);
+ request.__set_table_id(table_id);
+ request.__set_txn_id(txn_id);
+ request.__set_status(status.to_thrift());
+ if (state) {
+ 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 finish error, db_id=" << db_id << ", table_id=" << table_id
+ << ", label=" << label << ", txn_id=" << txn_id
+ << ", instance_id=" << print_id(instance_id)
+ << ", executor status=" << status.to_string()
+ << ", request commit status=" << st.to_string();
+ return st;
+ }
+ st = Status::create(result.status);
+ // TODO handle execute and commit error
+ 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);
+ if (prepare_failed) {
+ ss << ", prepare status=" << status.to_string();
+ } else {
+ ss << ", execute status=" << status.to_string();
+ }
+ ss << ", commit status=" << st.to_string();
+ if (state && !(state->get_error_log_file_path().empty())) {
+ ss << ", error_url=" << state->get_error_log_file_path();
+ }
+ LOG(INFO) << ss.str();
+ return st;
+}
+
+Status GroupCommitTable::_exec_plan_fragment(int64_t db_id, int64_t table_id,
+ const std::string& label, int64_t txn_id,
+ bool is_pipeline,
+ const TExecPlanFragmentParams& params,
+ const TPipelineFragmentParams& pipeline_params) {
+ auto finish_cb = [db_id, table_id, label, txn_id, this](RuntimeState* state, Status* status) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, state->fragment_instance_id(),
+ *status, false, state);
+ };
+ if (is_pipeline) {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(pipeline_params, finish_cb);
+ } else {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(params, finish_cb);
+ }
+}
+
+Status GroupCommitTable::get_load_block_queue(const TUniqueId& instance_id,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ 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;
+ return Status::OK();
+}
+
+GroupCommitMgr::GroupCommitMgr(ExecEnv* exec_env) : _exec_env(exec_env) {
+ ThreadPoolBuilder("InsertIntoGroupCommitThreadPool")
+ .set_min_threads(config::group_commit_insert_threads)
+ .set_max_threads(config::group_commit_insert_threads)
+ .build(&_insert_into_thread_pool);
+}
+
+GroupCommitMgr::~GroupCommitMgr() {}
+
+Status GroupCommitMgr::group_commit_insert(int64_t table_id, const TPlan& plan,
Review Comment:
add some comments in this method.
##########
be/src/runtime/group_commit_mgr.cpp:
##########
@@ -0,0 +1,480 @@
+// 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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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 LoadBlockQueue::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_first_block_load_queue(
+ int64_t table_id, std::shared_ptr<vectorized::FutureBlock> block,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ DCHECK(table_id == _table_id);
+ DCHECK(block->first == true);
+ {
+ std::unique_lock l(_lock);
+ for (auto it = load_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == 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_block_queues.begin(); it != load_block_queues.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_block_queue = it->second;
+ break;
+ } else if (block->schema_version < it->second->schema_version) {
+ return Status::DataQualityError("schema version not match");
+ }
+ }
+ }
+ }
+ if (load_block_queue == nullptr) {
+ st = _create_group_commit_load(table_id, load_block_queue);
+ if (LIKELY(st.ok())) {
+ break;
+ }
+ }
+ }
+ RETURN_IF_ERROR(st);
+ if (load_block_queue->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<LoadBlockQueue>& load_block_queue) {
+ 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 schema_version = result.base_schema_version;
+ auto is_pipeline = result.__isset.pipeline_params;
+ auto& params = result.params;
+ auto& pipeline_params = result.pipeline_params;
+ int64_t db_id;
+ std::string label;
+ int64_t txn_id;
+ TUniqueId instance_id;
+ if (!is_pipeline) {
+ db_id = params.fragment.output_sink.olap_table_sink.db_id;
+ label = params.import_label;
+ txn_id = params.txn_conf.txn_id;
+ instance_id = params.params.fragment_instance_id;
+ } else {
+ db_id = pipeline_params.fragment.output_sink.olap_table_sink.db_id;
+ label = pipeline_params.import_label;
+ 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 << ", label=" << label
+ << ", txn_id=" << txn_id << ", instance_id=" << print_id(instance_id)
+ << ", is_pipeline=" << is_pipeline;
+ {
+ load_block_queue =
+ std::make_shared<LoadBlockQueue>(instance_id, label, txn_id, schema_version);
+ std::unique_lock l(_lock);
+ load_block_queues.emplace(instance_id, load_block_queue);
+ }
+ st = _exec_plan_fragment(db_id, table_id, label, txn_id, is_pipeline, params, pipeline_params);
+ if (!st.ok()) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, instance_id, st, true, nullptr);
+ }
+ 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,
+ bool prepare_failed, RuntimeState* state) {
+ {
+ std::lock_guard<doris::Mutex> l(_lock);
+ if (prepare_failed) {
+ auto it = load_block_queues.find(instance_id);
+ if (it != load_block_queues.end()) {
+ it->second->cancel(status);
+ }
+ }
+ load_block_queues.erase(instance_id);
+ }
+ TFinishGroupCommitRequest request;
+ request.__set_db_id(db_id);
+ request.__set_table_id(table_id);
+ request.__set_txn_id(txn_id);
+ request.__set_status(status.to_thrift());
+ if (state) {
+ 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 finish error, db_id=" << db_id << ", table_id=" << table_id
+ << ", label=" << label << ", txn_id=" << txn_id
+ << ", instance_id=" << print_id(instance_id)
+ << ", executor status=" << status.to_string()
+ << ", request commit status=" << st.to_string();
+ return st;
+ }
+ st = Status::create(result.status);
+ // TODO handle execute and commit error
+ 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);
+ if (prepare_failed) {
+ ss << ", prepare status=" << status.to_string();
+ } else {
+ ss << ", execute status=" << status.to_string();
+ }
+ ss << ", commit status=" << st.to_string();
+ if (state && !(state->get_error_log_file_path().empty())) {
+ ss << ", error_url=" << state->get_error_log_file_path();
+ }
+ LOG(INFO) << ss.str();
+ return st;
+}
+
+Status GroupCommitTable::_exec_plan_fragment(int64_t db_id, int64_t table_id,
+ const std::string& label, int64_t txn_id,
+ bool is_pipeline,
+ const TExecPlanFragmentParams& params,
+ const TPipelineFragmentParams& pipeline_params) {
+ auto finish_cb = [db_id, table_id, label, txn_id, this](RuntimeState* state, Status* status) {
+ _finish_group_commit_load(db_id, table_id, label, txn_id, state->fragment_instance_id(),
+ *status, false, state);
+ };
+ if (is_pipeline) {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(pipeline_params, finish_cb);
+ } else {
+ return _exec_env->fragment_mgr()->exec_plan_fragment(params, finish_cb);
+ }
+}
+
+Status GroupCommitTable::get_load_block_queue(const TUniqueId& instance_id,
+ std::shared_ptr<LoadBlockQueue>& load_block_queue) {
+ 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;
+ return Status::OK();
+}
+
+GroupCommitMgr::GroupCommitMgr(ExecEnv* exec_env) : _exec_env(exec_env) {
+ ThreadPoolBuilder("InsertIntoGroupCommitThreadPool")
+ .set_min_threads(config::group_commit_insert_threads)
+ .set_max_threads(config::group_commit_insert_threads)
+ .build(&_insert_into_thread_pool);
+}
+
+GroupCommitMgr::~GroupCommitMgr() {}
+
+Status GroupCommitMgr::group_commit_insert(int64_t table_id, const TPlan& plan,
+ const TDescriptorTable& tdesc_tbl,
+ const TScanRangeParams& scan_range_params,
+ const PGroupCommitInsertRequest* request,
+ PGroupCommitInsertResponse* response) {
+ auto& nodes = plan.nodes;
+ DCHECK(nodes.size() > 0);
+ auto& plan_node = nodes.at(0);
+
+ TUniqueId pipe_id;
+ pipe_id.__set_hi(request->pipe_id().hi());
+ pipe_id.__set_lo(request->pipe_id().lo());
+
+ std::vector<std::shared_ptr<doris::vectorized::FutureBlock>> future_blocks;
+ {
+ std::shared_ptr<LoadBlockQueue> load_block_queue;
+ // new a stream load pipe, put it into stream load mgr, append row in thread pool
+ auto pipe = std::make_shared<io::StreamLoadPipe>(
+ io::kMaxPipeBufferedBytes /* max_buffered_bytes */, 64 * 1024 /* min_chunk_size */,
+ -1 /* total_length */, true /* use_proto */);
+ std::shared_ptr<StreamLoadContext> ctx = std::make_shared<StreamLoadContext>(_exec_env);
+ ctx->pipe = pipe;
+ RETURN_IF_ERROR(_exec_env->new_load_stream_mgr()->put(pipe_id, ctx));
+ std::unique_ptr<int, std::function<void(int*)>> remove_pipe_func((int*)0x01, [&](int*) {
+ if (load_block_queue != nullptr) {
+ load_block_queue->remove_block_id(pipe_id);
+ }
+ _exec_env->new_load_stream_mgr()->remove(pipe_id);
+ });
+ _insert_into_thread_pool->submit_func(
+ std::bind<void>(&GroupCommitMgr::_append_row, this, pipe, request));
+
+ std::unique_ptr<RuntimeState> runtime_state = RuntimeState::create_unique();
+ TQueryOptions query_options;
+ query_options.query_type = TQueryType::LOAD;
+ TQueryGlobals query_globals;
+ runtime_state->init(pipe_id, query_options, query_globals, _exec_env);
+ runtime_state->set_query_mem_tracker(std::make_shared<MemTrackerLimiter>(
+ MemTrackerLimiter::Type::LOAD, fmt::format("Load#Id={}", print_id(pipe_id)), -1));
+ DescriptorTbl* desc_tbl = nullptr;
+ RETURN_IF_ERROR(DescriptorTbl::create(runtime_state->obj_pool(), tdesc_tbl, &desc_tbl));
+ runtime_state->set_desc_tbl(desc_tbl);
Review Comment:
2. FileScanNode consums data from the pipe.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] mymeiyi commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "mymeiyi (via GitHub)" <gi...@apache.org>.
mymeiyi commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1683843765
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1683678971
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] doris-robot commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1711047643
TeamCity be ut coverage result:
Function Coverage: 36.93% (7895/21378)
Line Coverage: 28.94% (63353/218947)
Region Coverage: 27.87% (32890/118008)
Branch Coverage: 24.46% (16891/69066)
Coverage Report: http://coverage.selectdb-in.cc/coverage/7ed2c95affefd4df90f63450a72b1b60f5bd031a_7ed2c95affefd4df90f63450a72b1b60f5bd031a/report/index.html
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1711221854
PR approved by at least one committer and no changes requested.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] mymeiyi commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "mymeiyi (via GitHub)" <gi...@apache.org>.
mymeiyi commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1711017156
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] mymeiyi commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "mymeiyi (via GitHub)" <gi...@apache.org>.
mymeiyi commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1708040721
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] dataroaring commented on a diff in pull request #22829: [feature](insert) Support group commit insert
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
dataroaring commented on code in PR #22829:
URL: https://github.com/apache/doris/pull/22829#discussion_r1297209672
##########
be/src/runtime/group_commit_mgr.h:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PaloInternalService_types.h>
+
+#include "common/status.h"
+#include "io/fs/stream_load_pipe.h"
+#include "util/lock.h"
+#include "util/threadpool.h"
+#include "util/thrift_util.h"
+#include "vec/core/block.h"
+
+namespace doris {
+class ExecEnv;
+class TPlan;
+class TDescriptorTable;
+class TUniqueId;
+class TExecPlanFragmentParams;
+class ObjectPool;
+class RuntimeState;
+class StreamLoadPipe;
+
+class LoadBlockQueue {
+public:
+ LoadBlockQueue(const UniqueId& load_instance_id, std::string& label, int64_t txn_id,
+ int64_t schema_version)
+ : load_instance_id(load_instance_id),
+ label(label),
+ txn_id(txn_id),
+ schema_version(schema_version),
+ _start_time(std::chrono::steady_clock::now()) {
+ _mutex = std::make_shared<doris::Mutex>();
+ _cv = std::make_shared<doris::ConditionVariable>();
+ };
+
+ Status add_block(std::shared_ptr<vectorized::FutureBlock> block);
+ Status get_block(vectorized::Block* block, bool* find_block, bool* eos);
+ void remove_block_id(const UniqueId& unique_id);
+ void cancel(const Status& st);
+
+ UniqueId load_instance_id;
+ std::string label;
+ int64_t txn_id;
+ int64_t schema_version;
+ bool need_commit = false;
+
+private:
+ std::chrono::steady_clock::time_point _start_time;
+
+ std::shared_ptr<doris::Mutex> _mutex;
+ std::shared_ptr<doris::ConditionVariable> _cv;
+ std::set<UniqueId> _block_unique_ids;
Review Comment:
rename it to _load_unique_ids.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] doris-robot commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1708165760
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 48.35 seconds
stream load tsv: 539 seconds loaded 74807831229 Bytes, about 132 MB/s
stream load json: 20 seconds loaded 2358488459 Bytes, about 112 MB/s
stream load orc: 64 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 31 seconds loaded 861443392 Bytes, about 26 MB/s
insert into select: 29.0 seconds inserted 10000000 Rows, about 344K ops/s
storage size: 17161712782 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] dataroaring merged pull request #22829: [feature](insert) Support group commit insert
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
dataroaring merged PR #22829:
URL: https://github.com/apache/doris/pull/22829
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] hello-stephen commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1685872992
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 48.56 seconds
stream load tsv: 552 seconds loaded 74807831229 Bytes, about 129 MB/s
stream load json: 20 seconds loaded 2358488459 Bytes, about 112 MB/s
stream load orc: 65 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 32 seconds loaded 861443392 Bytes, about 25 MB/s
insert into select: 31.3 seconds inserted 10000000 Rows, about 319K ops/s
storage size: 17162165047 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1672805716
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1682188446
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1711221903
PR approved by anyone and no changes requested.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1708047218
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] doris-robot commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1711062542
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 46.44 seconds
stream load tsv: 532 seconds loaded 74807831229 Bytes, about 134 MB/s
stream load json: 20 seconds loaded 2358488459 Bytes, about 112 MB/s
stream load orc: 64 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 31 seconds loaded 861443392 Bytes, about 26 MB/s
insert into select: 28.9 seconds inserted 10000000 Rows, about 346K ops/s
storage size: 17162198180 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] mymeiyi commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "mymeiyi (via GitHub)" <gi...@apache.org>.
mymeiyi commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1685763176
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1685750375
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1685601353
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1685604948
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] mymeiyi commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "mymeiyi (via GitHub)" <gi...@apache.org>.
mymeiyi commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1685602149
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1711020538
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] dataroaring commented on a diff in pull request #22829: [feature](insert) Support group commit insert
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1681910429
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] dataroaring commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
dataroaring commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1710994485
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] mymeiyi commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "mymeiyi (via GitHub)" <gi...@apache.org>.
mymeiyi commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1682190892
run buildall
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1683645747
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1683639559
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] hello-stephen commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1682295852
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 45.51 seconds
stream load tsv: 540 seconds loaded 74807831229 Bytes, about 132 MB/s
stream load json: 20 seconds loaded 2358488459 Bytes, about 112 MB/s
stream load orc: 65 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 31 seconds loaded 861443392 Bytes, about 26 MB/s
insert into select: 29.2 seconds inserted 10000000 Rows, about 342K ops/s
storage size: 17162150703 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1683618410
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] hello-stephen commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1685650975
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 48.04 seconds
stream load tsv: 542 seconds loaded 74807831229 Bytes, about 131 MB/s
stream load json: 21 seconds loaded 2358488459 Bytes, about 107 MB/s
stream load orc: 65 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 31 seconds loaded 861443392 Bytes, about 26 MB/s
insert into select: 29.3 seconds inserted 10000000 Rows, about 341K ops/s
storage size: 17162136301 Bytes
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #22829: [feature](insert) Support group commit insert
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22829:
URL: https://github.com/apache/doris/pull/22829#issuecomment-1708254936
clang-tidy review says "All clean, LGTM! :+1:"
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org