This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.1 by this push:
new e2c2702dffa [Bug](runtime-filter) fix some rf error problems (#37155)
e2c2702dffa is described below
commit e2c2702dffae01baad5084d2355820fb2a358478
Author: Pxl <[email protected]>
AuthorDate: Thu Jul 4 20:03:46 2024 +0800
[Bug](runtime-filter) fix some rf error problems (#37155)
## Proposed changes
pick from #37273
---
be/src/exprs/runtime_filter.cpp | 17 +++++++++++++----
be/src/exprs/runtime_filter_slots.h | 9 +++++++++
be/src/pipeline/pipeline_fragment_context.h | 3 ---
.../pipeline/pipeline_x/pipeline_x_fragment_context.h | 12 ++----------
be/src/runtime/fragment_mgr.cpp | 8 +++++---
be/src/runtime/runtime_filter_mgr.cpp | 5 +++++
6 files changed, 34 insertions(+), 20 deletions(-)
diff --git a/be/src/exprs/runtime_filter.cpp b/be/src/exprs/runtime_filter.cpp
index 3f8a19f1b16..fb24cf1eb0d 100644
--- a/be/src/exprs/runtime_filter.cpp
+++ b/be/src/exprs/runtime_filter.cpp
@@ -1033,25 +1033,34 @@ Status IRuntimeFilter::publish(bool publish_local) {
class SyncSizeClosure : public AutoReleaseClosure<PSendFilterSizeRequest,
DummyBrpcCallback<PSendFilterSizeResponse>> {
std::shared_ptr<pipeline::Dependency> _dependency;
+ IRuntimeFilter* _filter;
using Base =
AutoReleaseClosure<PSendFilterSizeRequest,
DummyBrpcCallback<PSendFilterSizeResponse>>;
ENABLE_FACTORY_CREATOR(SyncSizeClosure);
void _process_if_rpc_failed() override {
((pipeline::CountedFinishDependency*)_dependency.get())->sub();
+ LOG(WARNING) << "sync filter size meet rpc error, filter=" <<
_filter->debug_string();
Base::_process_if_rpc_failed();
}
void _process_if_meet_error_status(const Status& status) override {
((pipeline::CountedFinishDependency*)_dependency.get())->sub();
- Base::_process_if_meet_error_status(status);
+ if (status.is<ErrorCode::END_OF_FILE>()) {
+ // rf merger backend may finished before rf's send_filter_size, we
just ignore filter in this case.
+ _filter->set_ignored();
+ } else {
+ LOG(WARNING) << "sync filter size meet error status, filter="
+ << _filter->debug_string();
+ Base::_process_if_meet_error_status(status);
+ }
}
public:
SyncSizeClosure(std::shared_ptr<PSendFilterSizeRequest> req,
std::shared_ptr<DummyBrpcCallback<PSendFilterSizeResponse>> callback,
- std::shared_ptr<pipeline::Dependency> dependency)
- : Base(req, callback), _dependency(std::move(dependency)) {}
+ std::shared_ptr<pipeline::Dependency> dependency,
IRuntimeFilter* filter)
+ : Base(req, callback), _dependency(std::move(dependency)),
_filter(filter) {}
};
Status IRuntimeFilter::send_filter_size(RuntimeState* state, uint64_t
local_filter_size) {
@@ -1094,7 +1103,7 @@ Status IRuntimeFilter::send_filter_size(RuntimeState*
state, uint64_t local_filt
auto request = std::make_shared<PSendFilterSizeRequest>();
auto callback =
DummyBrpcCallback<PSendFilterSizeResponse>::create_shared();
- auto closure = SyncSizeClosure::create_unique(request, callback,
_dependency);
+ auto closure = SyncSizeClosure::create_unique(request, callback,
_dependency, this);
auto* pquery_id = request->mutable_query_id();
pquery_id->set_hi(_state->query_id.hi());
pquery_id->set_lo(_state->query_id.lo());
diff --git a/be/src/exprs/runtime_filter_slots.h
b/be/src/exprs/runtime_filter_slots.h
index 0bf8a33f9f2..ebda4b56fcc 100644
--- a/be/src/exprs/runtime_filter_slots.h
+++ b/be/src/exprs/runtime_filter_slots.h
@@ -71,6 +71,9 @@ public:
// process ignore duplicate IN_FILTER
std::unordered_set<int> has_in_filter;
for (auto* filter : _runtime_filters) {
+ if (filter->get_ignored()) {
+ continue;
+ }
if (filter->get_real_type() != RuntimeFilterType::IN_FILTER) {
continue;
}
@@ -83,6 +86,9 @@ public:
// process ignore filter when it has IN_FILTER on same expr, and init
bloom filter size
for (auto* filter : _runtime_filters) {
+ if (filter->get_ignored()) {
+ continue;
+ }
if (filter->get_real_type() == RuntimeFilterType::IN_FILTER ||
!has_in_filter.contains(filter->expr_order())) {
continue;
@@ -95,6 +101,9 @@ public:
Status init_filters(RuntimeState* state, uint64_t local_hash_table_size) {
// process IN_OR_BLOOM_FILTER's real type
for (auto* filter : _runtime_filters) {
+ if (filter->get_ignored()) {
+ continue;
+ }
if (filter->type() == RuntimeFilterType::IN_OR_BLOOM_FILTER &&
get_real_size(filter, local_hash_table_size) >
state->runtime_filter_max_in_num()) {
RETURN_IF_ERROR(filter->change_to_bloom_filter());
diff --git a/be/src/pipeline/pipeline_fragment_context.h
b/be/src/pipeline/pipeline_fragment_context.h
index aaf48a9d06d..7eabb13b772 100644
--- a/be/src/pipeline/pipeline_fragment_context.h
+++ b/be/src/pipeline/pipeline_fragment_context.h
@@ -114,9 +114,6 @@ public:
virtual void clear_finished_tasks() {}
- virtual void add_merge_controller_handler(
- std::shared_ptr<RuntimeFilterMergeControllerEntity>& handler) {}
-
virtual Status send_report(bool);
Status update_status(Status status) {
diff --git a/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.h
b/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.h
index ea2212357a2..55866400374 100644
--- a/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.h
+++ b/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.h
@@ -57,6 +57,8 @@ class Dependency;
class PipelineXFragmentContext : public PipelineFragmentContext {
public:
+ ENABLE_FACTORY_CREATOR(PipelineXFragmentContext);
+
// Callback to report execution status of plan fragment.
// 'profile' is the cumulative profile, 'done' indicates whether the
execution
// is done or still continuing.
@@ -91,13 +93,6 @@ public:
}
}
- void add_merge_controller_handler(
- std::shared_ptr<RuntimeFilterMergeControllerEntity>& handler)
override {
- _merge_controller_handlers.emplace_back(handler);
- }
-
- // bool is_canceled() const { return _runtime_state->is_cancelled(); }
-
// Prepare global information including global states and the unique
operator tree shared by all pipeline tasks.
Status prepare(const doris::TPipelineFragmentParams& request) override;
@@ -179,9 +174,6 @@ private:
bool _need_local_merge = false;
- // It is used to manage the lifecycle of RuntimeFilterMergeController
- std::vector<std::shared_ptr<RuntimeFilterMergeControllerEntity>>
_merge_controller_handlers;
-
// TODO: remove the _sink and _multi_cast_stream_sink_senders to set both
// of it in pipeline task not the fragment_context
#ifdef __clang__
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index dba128003bc..fee9d51afba 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -1467,8 +1467,6 @@ Status FragmentMgr::apply_filterv2(const
PPublishFilterRequestV2* request,
Status FragmentMgr::send_filter_size(const PSendFilterSizeRequest* request) {
UniqueId queryid = request->query_id();
- std::shared_ptr<RuntimeFilterMergeControllerEntity> filter_controller;
- RETURN_IF_ERROR(_runtimefilter_controller.acquire(queryid,
&filter_controller));
std::shared_ptr<QueryContext> query_ctx;
{
@@ -1478,11 +1476,15 @@ Status FragmentMgr::send_filter_size(const
PSendFilterSizeRequest* request) {
std::lock_guard<std::mutex> lock(_lock);
auto iter = _query_ctx_map.find(query_id);
if (iter == _query_ctx_map.end()) {
- return Status::InvalidArgument("query-id: {}",
queryid.to_string());
+ return Status::EndOfFile("Query context (query-id: {}) not found,
maybe finished",
+ queryid.to_string());
}
query_ctx = iter->second;
}
+
+ std::shared_ptr<RuntimeFilterMergeControllerEntity> filter_controller;
+ RETURN_IF_ERROR(_runtimefilter_controller.acquire(queryid,
&filter_controller));
auto merge_status = filter_controller->send_filter_size(request);
return merge_status;
}
diff --git a/be/src/runtime/runtime_filter_mgr.cpp
b/be/src/runtime/runtime_filter_mgr.cpp
index dfc55edc7c3..77736e8407a 100644
--- a/be/src/runtime/runtime_filter_mgr.cpp
+++ b/be/src/runtime/runtime_filter_mgr.cpp
@@ -355,6 +355,7 @@ Status
RuntimeFilterMergeControllerEntity::send_filter_size(const PSendFilterSiz
auto* pquery_id = closure->request_->mutable_query_id();
pquery_id->set_hi(_state->query_id.hi());
pquery_id->set_lo(_state->query_id.lo());
+ closure->cntl_->set_timeout_ms(std::min(3600,
_state->execution_timeout) * 1000);
closure->request_->set_filter_id(filter_id);
closure->request_->set_filter_size(cnt_val->global_size);
@@ -475,6 +476,8 @@ Status RuntimeFilterMergeControllerEntity::merge(const
PMergeFilterRequest* requ
if (has_attachment) {
rpc_contexts[cur]->cntl.request_attachment().append(request_attachment);
}
+ rpc_contexts[cur]->cntl.set_timeout_ms(std::min(3600,
_state->execution_timeout) *
+ 1000);
rpc_contexts[cur]->cid = rpc_contexts[cur]->cntl.call_id();
// set fragment-id
for (size_t fid = 0; fid <
targets[cur].target_fragment_instance_ids.size();
@@ -548,6 +551,8 @@ Status RuntimeFilterMergeControllerEntity::merge(const
PMergeFilterRequest* requ
if (has_attachment) {
rpc_contexts[cur]->cntl.request_attachment().append(request_attachment);
}
+ rpc_contexts[cur]->cntl.set_timeout_ms(std::min(3600,
_state->execution_timeout) *
+ 1000);
rpc_contexts[cur]->cid = rpc_contexts[cur]->cntl.call_id();
// set fragment_instance_id
auto request_fragment_instance_id =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
