nprasad2077/doris
1
0 Fork 0
mirror of https://github.com/apache/doris.git synced 2026-11-04 16:03:23 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
563b5ba7aacd581cfb18f0212c21f2c9c70f5d4c
doris/be/src/runtime/query_context.cpp
T
bobhan1 563b5ba7aa [feature](be) Add query read-ahead runtime ownership 
### What problem does this PR solve?

Issue Number: None

Related PR: None

Problem Summary: Concurrent range reads must be shared at BE scope while every scanner of one query shares cancellation and resident-buffer admission. Add a startup-owned FileRangeReadScheduler in Cloud mode, a lazily shared query context, cancellation propagation, and a mutable enable_query_read_ahead switch. Query and BE resident-buffer limits are configurable through read_ahead_max_bytes_per_query and read_ahead_max_bytes_per_be. The scheduler remains available while the switch is disabled so a runtime update can enable read-ahead for newly initialized reads without restarting the BE.

### Release note

Support runtime updates to enable_query_read_ahead and configurable query/BE read-ahead memory limits.

### Check List (For Author)

- Test: Unit Test
    - ./run-be-ut.sh --run --filter='FileRangeReadSchedulerTest.*:QueryContextReadAheadTest.*:SegmentReadAheadTest.*:ColumnReaderTest.*ReadAhead*' -j200
    - ./build.sh --be -j200
- Behavior changed: Yes. Cloud BEs can enable or disable query read-ahead at runtime; new reads observe the updated switch.
- Does this need documentation: Yes. The read-ahead design document lists these settings.
2026-10-29 16:27:15 +08:00

643 lines
27 KiB
C++
Raw Blame History

// 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/query_context.h"
#include <fmt/core.h>
#include <gen_cpp/FrontendService_types.h>
#include <gen_cpp/RuntimeProfile_types.h>
#include <gen_cpp/Types_types.h>
#include <glog/logging.h>
#include <algorithm>
#include <memory>
#include <mutex>
#include <utility>
#include <vector>
#include "cloud/config.h"
#include "common/logging.h"
#include "common/status.h"
#include "exec/operator/rec_cte_scan_operator.h"
#include "exec/pipeline/dependency.h"
#include "exec/pipeline/pipeline_fragment_context.h"
#include "exec/runtime_filter/runtime_filter_definitions.h"
#include "exec/spill/spill_file_manager.h"
#include "exprs/function/ai/ai_request_executor.h"
#include "io/cache/block_file_cache_factory.h"
#include "io/cache/remote_scan_cache_write_limiter.h"
#include "io/fs/file_range_read_scheduler.h"
#include "runtime/exec_env.h"
#include "runtime/fragment_mgr.h"
#include "runtime/memory/heap_profiler.h"
#include "runtime/runtime_query_statistics_mgr.h"
#include "runtime/runtime_state.h"
#include "runtime/thread_context.h"
#include "runtime/workload_group/workload_group.h"
#include "runtime/workload_group/workload_group_manager.h"
#include "runtime/workload_management/query_task_controller.h"
#include "storage/olap_common.h"
#include "util/mem_info.h"
#include "util/uid_util.h"
namespace doris {
class DelayReleaseToken : public Runnable {
ENABLE_FACTORY_CREATOR(DelayReleaseToken);
public:
DelayReleaseToken(std::unique_ptr<ThreadPoolToken>&& token) { token_ = std::move(token); }
~DelayReleaseToken() override = default;
void run() override {}
std::unique_ptr<ThreadPoolToken> token_;
};
const std::string toString(QuerySource queryType) {
switch (queryType) {
case QuerySource::INTERNAL_FRONTEND:
return "INTERNAL_FRONTEND";
case QuerySource::STREAM_LOAD:
return "STREAM_LOAD";
case QuerySource::GROUP_COMMIT_LOAD:
return "EXTERNAL_QUERY";
case QuerySource::ROUTINE_LOAD:
return "ROUTINE_LOAD";
case QuerySource::EXTERNAL_CONNECTOR:
return "EXTERNAL_CONNECTOR";
case QuerySource::EXTERNAL_FRONTEND:
return "EXTERNAL_FRONTEND";
default:
return "UNKNOWN";
}
}
std::shared_ptr<QueryContext> QueryContext::create(TUniqueId query_id, ExecEnv* exec_env,
const TQueryOptions& query_options,
TNetworkAddress coord_addr, bool is_nereids,
TNetworkAddress current_connect_fe,
QuerySource query_type) {
auto ctx = QueryContext::create_shared(query_id, exec_env, query_options, coord_addr,
is_nereids, current_connect_fe, query_type);
ctx->init_query_task_controller();
return ctx;
}
QueryContext::QueryContext(TUniqueId query_id, ExecEnv* exec_env,
const TQueryOptions& query_options, TNetworkAddress coord_addr,
bool is_nereids, TNetworkAddress current_connect_fe,
QuerySource query_source)
: _timeout_second(-1),
_query_id(std::move(query_id)),
_exec_env(exec_env),
_is_nereids(is_nereids),
_query_options(query_options),
_query_source(query_source) {
_init_resource_context();
SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(query_mem_tracker());
_query_watcher.start();
_execution_dependency = Dependency::create_unique(-1, -1, "ExecutionDependency", false);
_memory_sufficient_dependency =
Dependency::create_unique(-1, -1, "MemorySufficientDependency", true);
_runtime_filter_mgr = std::make_unique<RuntimeFilterMgr>(true);
_timeout_second = query_options.execution_timeout;
bool initialize_context_holder =
config::enable_file_cache && config::enable_file_cache_query_limit &&
query_options.__isset.enable_file_cache && query_options.enable_file_cache &&
query_options.__isset.file_cache_query_limit_percent &&
query_options.file_cache_query_limit_percent < 100;
// Initialize file cache context holders
if (initialize_context_holder) {
_query_context_holders = io::FileCacheFactory::instance()->get_query_context_holders(
_query_id, query_options.file_cache_query_limit_percent);
}
const bool initialize_remote_scan_cache_write_limiter =
config::is_cloud_mode() && config::enable_file_cache &&
query_options.__isset.file_cache_query_limit_bytes &&
query_options.file_cache_query_limit_bytes >= 0 &&
query_options.query_type == TQueryType::SELECT;
if (initialize_remote_scan_cache_write_limiter) {
_remote_scan_cache_write_limiter = std::make_unique<io::RemoteScanCacheWriteLimiter>(
_query_id, query_options.file_cache_query_limit_bytes);
}
bool is_query_type_valid = query_options.query_type == TQueryType::SELECT ||
query_options.query_type == TQueryType::LOAD ||
query_options.query_type == TQueryType::EXTERNAL;
DCHECK_EQ(is_query_type_valid, true);
this->coord_addr = coord_addr;
// current_connect_fe is used for report query statistics
this->current_connect_fe = current_connect_fe;
// external query has no current_connect_fe
if (query_options.query_type != TQueryType::EXTERNAL) {
bool is_report_fe_addr_valid =
!this->current_connect_fe.hostname.empty() && this->current_connect_fe.port != 0;
DCHECK_EQ(is_report_fe_addr_valid, true);
}
clock_gettime(CLOCK_MONOTONIC, &this->_query_arrival_timestamp);
DorisMetrics::instance()->query_ctx_cnt->increment(1);
_mem_arb = MemShareArbitrator::create_shared(
query_id, query_options.mem_limit,
query_options.__isset.max_scan_mem_ratio ? query_options.max_scan_mem_ratio : 1.0);
}
void QueryContext::_init_query_mem_tracker() {
// If user not set query limit, will use default 1TB memory limit. It is large enough to cover most cases.
constexpr int64_t DEFAULT_QUERY_MEM_LIMIT = 1LL << 60;
bool has_query_mem_limit = _query_options.__isset.mem_limit && (_query_options.mem_limit > 0);
int64_t user_set_mem_limit =
has_query_mem_limit ? _query_options.mem_limit : DEFAULT_QUERY_MEM_LIMIT;
int64_t adjusted_mem_limit = user_set_mem_limit;
if (adjusted_mem_limit > MemInfo::mem_limit()) {
VLOG_NOTICE << "Query memory limit "
<< PrettyPrinter::print(user_set_mem_limit, TUnit::BYTES)
<< " exceeds process memory limit of "
<< PrettyPrinter::print(MemInfo::mem_limit(), TUnit::BYTES)
<< ". Using process memory limit instead.";
adjusted_mem_limit = MemInfo::mem_limit();
}
// If the query is a pure load task(streamload, routine load, group commit), then it should not use
// memlimit per query to limit their memory usage.
if (is_pure_load_task()) {
adjusted_mem_limit = MemInfo::mem_limit();
}
std::shared_ptr<MemTrackerLimiter> query_mem_tracker;
if (_query_options.query_type == TQueryType::SELECT) {
query_mem_tracker = MemTrackerLimiter::create_shared(
MemTrackerLimiter::Type::QUERY, fmt::format("Query#Id={}", print_id(_query_id)),
adjusted_mem_limit);
} else if (_query_options.query_type == TQueryType::LOAD) {
query_mem_tracker = MemTrackerLimiter::create_shared(
MemTrackerLimiter::Type::LOAD, fmt::format("Load#Id={}", print_id(_query_id)),
adjusted_mem_limit);
} else if (_query_options.query_type == TQueryType::EXTERNAL) { // spark/flink/etc..
query_mem_tracker = MemTrackerLimiter::create_shared(
MemTrackerLimiter::Type::QUERY, fmt::format("External#Id={}", print_id(_query_id)),
adjusted_mem_limit);
} else {
LOG(FATAL) << "__builtin_unreachable";
__builtin_unreachable();
}
if (_query_options.__isset.is_report_success && _query_options.is_report_success) {
query_mem_tracker->enable_print_log_usage();
}
// If enable reserve memory, not enable check limit, because reserve memory will check it.
// If reserve enabled, even if the reserved memory size is smaller than the actual requested memory,
// and the query memory consumption is larger than the limit, we do not expect the query to fail
// after `check_limit` returns an error, but to run as long as possible,
// and will enter the paused state and try to spill when the query reserves next time.
// If the workload group or process runs out of memory, it will be forced to cancel.
query_mem_tracker->set_enable_check_limit(!(_query_options.__isset.enable_reserve_memory &&
_query_options.enable_reserve_memory));
_resource_ctx->memory_context()->set_mem_tracker(query_mem_tracker);
_resource_ctx->memory_context()->set_user_set_mem_limit(user_set_mem_limit);
}
void QueryContext::_init_resource_context() {
_resource_ctx = ResourceContext::create_shared();
_init_query_mem_tracker();
}
void QueryContext::init_query_task_controller() {
_resource_ctx->set_task_controller(QueryTaskController::create(shared_from_this()));
_resource_ctx->task_controller()->set_task_id(_query_id);
_resource_ctx->task_controller()->set_fe_addr(current_connect_fe);
_resource_ctx->task_controller()->set_query_type(_query_options.query_type);
#ifndef BE_TEST
_exec_env->runtime_query_statistics_mgr()->register_resource_context(print_id(_query_id),
_resource_ctx);
#endif
}
void QueryContext::record_spill_data_dir(SpillDataDir* data_dir) {
std::lock_guard lock(_spill_data_dirs_mutex);
_spill_data_dirs.emplace(data_dir);
}
QueryContext::~QueryContext() {
SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(query_mem_tracker());
{
std::lock_guard lock(_file_range_read_context_mutex);
if (_file_range_read_context != nullptr) {
_file_range_read_context->cancel();
}
}
// query mem tracker consumption is equal to 0, it means that after QueryContext is created,
// it is found that query already exists in _query_ctx_map, and query mem tracker is not used.
// query mem tracker consumption is not equal to 0 after use, because there is memory consumed
// on query mem tracker, released on other trackers.
std::string mem_tracker_msg;
if (query_mem_tracker()->peak_consumption() != 0) {
mem_tracker_msg = fmt::format(
"deregister query/load memory tracker, queryId={}, Limit={}, CurrUsed={}, "
"PeakUsed={}",
print_id(_query_id), PrettyPrinter::print_bytes(query_mem_tracker()->limit()),
PrettyPrinter::print_bytes(query_mem_tracker()->consumption()),
PrettyPrinter::print_bytes(query_mem_tracker()->peak_consumption()));
}
_resource_ctx->task_controller()->finish();
if (enable_profile()) {
_report_query_profile();
}
_runtime_filter_mgr.reset();
_execution_dependency.reset();
_runtime_predicates.clear();
file_scan_range_params_map.clear();
obj_pool.clear();
_merge_controller_handler.reset();
if (auto* spill_file_mgr = _exec_env->spill_file_mgr()) {
for (auto* data_dir : _spill_data_dirs) {
spill_file_mgr->delete_query_spill_directory(print_id(_query_id), data_dir);
}
}
DorisMetrics::instance()->query_ctx_cnt->increment(-1);
// fragment_mgr is nullptr in unittest
if (ExecEnv::GetInstance()->fragment_mgr()) {
ExecEnv::GetInstance()->fragment_mgr()->remove_query_context(this->_query_id);
}
// the only one msg shows query's end. any other msg should append to it if need.
timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
int64_t elapsed_ms = (now.tv_sec - _query_arrival_timestamp.tv_sec) * 1000LL +
(now.tv_nsec - _query_arrival_timestamp.tv_nsec) / 1000000LL;
LOG_INFO("Query {} deconstructed, elapsed_ms: {}, mem_tracker: {}", print_id(this->_query_id),
elapsed_ms, mem_tracker_msg);
}
void QueryContext::set_ready_to_execute(Status reason) {
set_execution_dependency_ready();
_exec_status.update(reason);
}
void QueryContext::set_ready_to_execute_only() {
set_execution_dependency_ready();
}
void QueryContext::set_execution_dependency_ready() {
_execution_dependency->set_ready();
}
void QueryContext::set_memory_sufficient(bool sufficient) {
if (sufficient) {
{
_memory_sufficient_dependency->set_ready();
_resource_ctx->task_controller()->reset_paused_reason();
}
} else {
_memory_sufficient_dependency->block();
_resource_ctx->task_controller()->add_paused_count();
}
}
void QueryContext::cancel(Status new_status) {
if (!_exec_status.update(new_status)) {
return;
}
{
std::lock_guard lock(_file_range_read_context_mutex);
if (_file_range_read_context != nullptr) {
_file_range_read_context->cancel();
}
}
// Tasks should be always runnable.
_execution_dependency->set_always_ready();
_memory_sufficient_dependency->set_always_ready();
if ((new_status.is<ErrorCode::MEM_LIMIT_EXCEEDED>() ||
new_status.is<ErrorCode::MEM_ALLOC_FAILED>()) &&
_query_options.__isset.dump_heap_profile_when_mem_limit_exceeded &&
_query_options.dump_heap_profile_when_mem_limit_exceeded) {
// if query is cancelled because of query mem limit exceeded, dump heap profile
// at the time of cancellation can get the most accurate memory usage for problem analysis
auto wg = workload_group();
auto log_str = fmt::format(
"Query {} canceled because of memory limit exceeded, dumping memory "
"detail profiles. wg: {}. {}",
print_id(_query_id), wg ? wg->debug_string() : "null",
doris::ProcessProfile::instance()->memory_profile()->process_memory_detail_str());
LOG_LONG_STRING(INFO, log_str);
std::string dot = HeapProfiler::instance()->dump_heap_profile_to_dot();
if (!dot.empty()) {
dot += "\n-------------------------------------------------------\n";
dot += "Copy the text after `digraph` in the above output to "
"http://www.webgraphviz.com to generate a dot graph.\n"
"after start heap profiler, if there is no operation, will print `No nodes "
"to "
"print`."
"If there are many errors: `addr2line: Dwarf Error`,"
"or other FAQ, reference doc: "
"https://doris.apache.org/community/developer-guide/debug-tool/#4-qa\n";
auto nest_log_str =
fmt::format("Query {}, dump heap profile to dot: {}", print_id(_query_id), dot);
LOG_LONG_STRING(INFO, nest_log_str);
}
}
set_ready_to_execute(new_status);
// Copy the fragment contexts under the map lock, then cancel them after releasing it. Fragment
// cancellation may take task-level locks and must not run while holding the query map lock.
std::vector<std::weak_ptr<PipelineFragmentContext>> ctx_to_cancel;
{
std::lock_guard<std::mutex> lock(_pipeline_map_write_lock);
for (auto& entry : _fragment_id_to_pipeline_ctx) {
ctx_to_cancel.push_back(entry.second);
}
}
for (auto& f_context : ctx_to_cancel) {
if (auto pipeline_ctx = f_context.lock()) {
pipeline_ctx->cancel(new_status);
}
}
}
std::shared_ptr<io::FileRangeReadContext> QueryContext::get_or_create_file_range_read_context(
io::FileRangeReadScheduler* scheduler) {
DORIS_CHECK(scheduler != nullptr);
std::lock_guard lock(_file_range_read_context_mutex);
if (_file_range_read_context == nullptr) {
_file_range_read_scheduler = scheduler;
_file_range_read_context = scheduler->create_context();
if (is_cancelled()) {
_file_range_read_context->cancel();
}
} else {
DORIS_CHECK(_file_range_read_scheduler == scheduler);
}
return _file_range_read_context;
}
void QueryContext::set_load_error_url(std::string error_url) {
std::lock_guard<std::mutex> lock(_error_url_lock);
_load_error_url = error_url;
}
std::string QueryContext::get_load_error_url() {
std::lock_guard<std::mutex> lock(_error_url_lock);
return _load_error_url;
}
void QueryContext::set_first_error_msg(std::string error_msg) {
std::lock_guard<std::mutex> lock(_error_url_lock);
_first_error_msg = error_msg;
}
std::string QueryContext::get_first_error_msg() {
std::lock_guard<std::mutex> lock(_error_url_lock);
return _first_error_msg;
}
std::string QueryContext::print_all_pipeline_context() {
std::vector<std::weak_ptr<PipelineFragmentContext>> ctx_to_print;
fmt::memory_buffer debug_string_buffer;
size_t i = 0;
{
fmt::format_to(debug_string_buffer, "{} pipeline fragment contexts in query {}. \n",
_fragment_id_to_pipeline_ctx.size(), print_id(_query_id));
{
std::lock_guard<std::mutex> lock(_pipeline_map_write_lock);
for (auto& [f_id, f_context] : _fragment_id_to_pipeline_ctx) {
ctx_to_print.push_back(f_context);
}
}
for (auto& f_context : ctx_to_print) {
if (auto pipeline_ctx = f_context.lock()) {
auto elapsed = pipeline_ctx->elapsed_time() / 1000000000.0;
fmt::format_to(debug_string_buffer,
"No.{} (elapse_second={}s, fragment_id={}) : {}\n", i, elapsed,
pipeline_ctx->get_fragment_id(), pipeline_ctx->debug_string());
i++;
}
}
}
return fmt::to_string(debug_string_buffer);
}
std::shared_ptr<AIExecutionStatistics> QueryContext::get_ai_statistics(int fragment_id) {
std::scoped_lock lock(_ai_statistics_mutex);
auto& statistics = _ai_statistics[fragment_id];
if (!statistics) {
statistics = std::make_shared<AIExecutionStatistics>();
}
return statistics;
}
std::shared_ptr<AIExecutionStatistics> QueryContext::find_ai_statistics(int fragment_id) const {
std::scoped_lock lock(_ai_statistics_mutex);
auto it = _ai_statistics.find(fragment_id);
return it == _ai_statistics.end() ? nullptr : it->second;
}
void QueryContext::set_pipeline_context(const int fragment_id,
std::shared_ptr<PipelineFragmentContext> pip_ctx) {
std::lock_guard<std::mutex> lock(_pipeline_map_write_lock);
// Use insert_or_assign instead of insert to support overwriting old entries
// when recursive CTE recreates PipelineFragmentContext between rounds.
_fragment_id_to_pipeline_ctx.insert_or_assign(fragment_id, pip_ctx);
}
doris::TaskScheduler* QueryContext::get_pipe_exec_scheduler() {
if (!_task_scheduler) {
throw Exception(Status::InternalError("task_scheduler is null"));
}
return _task_scheduler;
}
Status QueryContext::set_workload_group(WorkloadGroupPtr& wg) {
_resource_ctx->set_workload_group(wg);
// Should add query first, the workload group will not be deleted,
// then visit workload group's resource
// see task_group_manager::delete_workload_group_by_ids
RETURN_IF_ERROR(workload_group()->add_resource_ctx(_query_id, _resource_ctx));
workload_group()->get_query_scheduler(&_task_scheduler, &_scan_task_scheduler,
&_remote_scan_task_scheduler);
return Status::OK();
}
void QueryContext::add_fragment_profile(
int fragment_id, const std::vector<std::shared_ptr<TRuntimeProfileTree>>& pipeline_profiles,
std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
if (pipeline_profiles.empty()) {
std::string msg = fmt::format("Add pipeline profile failed, query {}, fragment {}",
print_id(this->_query_id), fragment_id);
LOG_ERROR(msg);
DCHECK(false) << msg;
return;
}
#ifndef NDEBUG
for (const auto& p : pipeline_profiles) {
DCHECK(p != nullptr) << fmt::format("Add pipeline profile failed, query {}, fragment {}",
print_id(this->_query_id), fragment_id);
}
#endif
std::lock_guard<std::mutex> l(_profile_mutex);
VLOG_ROW << fmt::format(
"Query add fragment profile, query {}, fragment {}, pipeline profile count {} ",
print_id(this->_query_id), fragment_id, pipeline_profiles.size());
_profile_map.insert(std::make_pair(fragment_id, pipeline_profiles));
if (load_channel_profile != nullptr) {
_load_channel_profile_map.insert(std::make_pair(fragment_id, load_channel_profile));
}
}
void QueryContext::_report_query_profile() {
std::lock_guard<std::mutex> lg(_profile_mutex);
for (auto& [fragment_id, fragment_profile] : _profile_map) {
// Recursive CTE stages reuse fragment IDs, but the saved profile is the first stage's
// detached snapshot. Refresh only AI counters from their owner, which accumulates all stages.
if (auto statistics = find_ai_statistics(fragment_id)) {
const int profile_level =
_query_options.__isset.profile_level ? _query_options.profile_level : 2;
statistics->update_profile_snapshot(*fragment_profile.front(), profile_level);
}
std::shared_ptr<TRuntimeProfileTree> load_channel_profile = nullptr;
if (_load_channel_profile_map.contains(fragment_id)) {
load_channel_profile = _load_channel_profile_map[fragment_id];
}
ExecEnv::GetInstance()->runtime_query_statistics_mgr()->register_fragment_profile(
_query_id, this->coord_addr, fragment_id, fragment_profile, load_channel_profile);
}
ExecEnv::GetInstance()->runtime_query_statistics_mgr()->trigger_profile_reporting();
}
std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>>
QueryContext::_collect_realtime_query_profile() {
std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>> res;
std::lock_guard<std::mutex> lock(_pipeline_map_write_lock);
for (const auto& [fragment_id, fragment_ctx_wptr] : _fragment_id_to_pipeline_ctx) {
if (auto fragment_ctx = fragment_ctx_wptr.lock()) {
if (fragment_ctx == nullptr) {
std::string msg =
fmt::format("PipelineFragmentContext is nullptr, query {} fragment_id: {}",
print_id(_query_id), fragment_id);
LOG_ERROR(msg);
DCHECK(false) << msg;
continue;
}
auto profile = fragment_ctx->collect_realtime_profile();
if (profile.empty()) {
std::string err_msg = fmt::format(
"Get nothing when collecting profile, query {}, fragment_id: {}",
print_id(_query_id), fragment_id);
LOG_ERROR(err_msg);
DCHECK(false) << err_msg;
continue;
}
res.insert(std::make_pair(fragment_id, profile));
}
}
return res;
}
TReportExecStatusParams QueryContext::get_realtime_exec_status() {
TReportExecStatusParams exec_status;
auto realtime_query_profile = _collect_realtime_query_profile();
std::vector<std::shared_ptr<TRuntimeProfileTree>> load_channel_profiles;
for (auto load_channel_profile : _load_channel_profile_map) {
if (load_channel_profile.second != nullptr) {
load_channel_profiles.push_back(load_channel_profile.second);
}
}
exec_status = RuntimeQueryStatisticsMgr::create_report_exec_status_params(
this->_query_id, std::move(realtime_query_profile), std::move(load_channel_profiles),
/*is_done=*/false);
return exec_status;
}
Status QueryContext::send_block_to_cte_scan(
const TUniqueId& instance_id, int node_id,
const google::protobuf::RepeatedPtrField<doris::PBlock>& pblocks, bool eos) {
std::unique_lock<std::mutex> l(_cte_scan_lock);
auto it = _cte_scan.find(std::make_pair(instance_id, node_id));
if (it == _cte_scan.end()) {
return Status::InternalError("RecCTEScan not found for instance {}, node {}",
print_id(instance_id), node_id);
}
for (const auto& pblock : pblocks) {
RETURN_IF_ERROR(it->second->add_block(pblock));
}
if (eos) {
it->second->set_ready();
}
return Status::OK();
}
void QueryContext::registe_cte_scan(const TUniqueId& instance_id, int node_id,
RecCTEScanLocalState* scan) {
std::unique_lock<std::mutex> l(_cte_scan_lock);
auto key = std::make_pair(instance_id, node_id);
DCHECK(!_cte_scan.contains(key)) << "Duplicate registe cte scan for instance "
<< print_id(instance_id) << ", node " << node_id;
_cte_scan.emplace(key, scan);
}
void QueryContext::deregiste_cte_scan(const TUniqueId& instance_id, int node_id) {
std::lock_guard<std::mutex> l(_cte_scan_lock);
auto key = std::make_pair(instance_id, node_id);
DCHECK(_cte_scan.contains(key)) << "Duplicate deregiste cte scan for instance "
<< print_id(instance_id) << ", node " << node_id;
_cte_scan.erase(key);
}
Status QueryContext::reset_global_rf(const google::protobuf::RepeatedField<int32_t>& filter_ids) {
if (_merge_controller_handler) {
return _merge_controller_handler->reset_global_rf(this, filter_ids);
}
return Status::OK();
}
void QueryContext::add_total_task_num(int delta) {
if (auto* qtc = dynamic_cast<QueryTaskController*>(_resource_ctx->task_controller())) {
qtc->add_total_task_num(delta);
}
}
void QueryContext::inc_finished_task_num() {
if (auto* qtc = dynamic_cast<QueryTaskController*>(_resource_ctx->task_controller())) {
qtc->inc_finished_task_num();
}
}
} // namespace doris
Reference in new issue View Git Blame Copy Permalink