This is an automated email from the ASF dual-hosted git repository. BiteTheDDDDt pushed a commit to branch opt_perf_4.1 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 286c9f6702cea4651c56d67a1cea1c9ca02f6866 Author: Copilot <[email protected]> AuthorDate: Thu Jun 25 17:26:43 2026 +0800 [feature](runtime-filter) Add adaptive global runtime filter tree publish Issue Number: N/A Related PR: N/A Problem Summary: Global runtime filters are currently published from the merge coordinator directly to every scan target. When a merged filter is large and the cluster has many BEs, the coordinator sends many large duplicate RPC attachments. This change adds adaptive tree publish for global runtime filters so each relay node forwards a bounded number of copies based on runtime_filter_tree_publish_max_send_bytes. Add runtime_filter_tree_publish_max_send_bytes session variable to control adaptive global runtime filter tree publish. Set it to 0 to disable tree publish. - Test: - Manual test: git diff --cached --check - Manual test: build-support/check-format.sh - Manual test: DORIS_HOME=/mnt/disk8/xiaolei/codespaces/doris2 DORIS_THIRDPARTY=/mnt/disk8/xiaolei/codespaces/doris2/thirdparty ninja -C be/build_Release src/exec/CMakeFiles/Exec.dir/runtime_filter/runtime_filter_mgr.cpp.o src/service/CMakeFiles/Service.dir/internal_service.cpp.o - Manual test: ./build.sh --fe - Behavior changed: Yes. Large global runtime filters can be published through an adaptive relay tree instead of direct coordinator fanout. - Does this need documentation: No (cherry picked from commit 75171cb7c77f903c96542ea473e4083fe0ecd070) --- be/src/exec/runtime_filter/runtime_filter_mgr.cpp | 254 +++++++++++++++++++-- be/src/exec/runtime_filter/runtime_filter_mgr.h | 31 ++- be/src/service/internal_service.cpp | 23 +- .../runtime_filter/runtime_filter_mgr_test.cpp | 110 +++++++++ .../java/org/apache/doris/qe/SessionVariable.java | 26 +++ gensrc/proto/internal_service.proto | 9 +- gensrc/thrift/PaloInternalService.thrift | 1 + 7 files changed, 426 insertions(+), 28 deletions(-) diff --git a/be/src/exec/runtime_filter/runtime_filter_mgr.cpp b/be/src/exec/runtime_filter/runtime_filter_mgr.cpp index e52439d9c26..29b7aaa3e2f 100644 --- a/be/src/exec/runtime_filter/runtime_filter_mgr.cpp +++ b/be/src/exec/runtime_filter/runtime_filter_mgr.cpp @@ -25,6 +25,9 @@ #include <gen_cpp/internal_service.pb.h> #include <gen_cpp/types.pb.h> +#include <algorithm> +#include <cstdlib> +#include <cstring> #include <mutex> #include <ostream> #include <string> @@ -44,9 +47,190 @@ #include "runtime/thread_context.h" #include "util/brpc_client_cache.h" #include "util/brpc_closure.h" +#include "util/uid_util.h" namespace doris { #include "common/compile_check_begin.h" + +namespace { + +bool can_tree_publish(const std::vector<TRuntimeFilterTargetParamsV2>& targets) { + return std::all_of(targets.begin(), targets.end(), [](const auto& target) { + return target.__isset.target_fragment_ids && !target.target_fragment_ids.empty(); + }); +} + +std::vector<RuntimeFilterPublishTarget> build_runtime_filter_publish_targets( + const std::vector<TRuntimeFilterTargetParamsV2>& targets) { + std::vector<RuntimeFilterPublishTarget> publish_targets; + publish_targets.reserve(targets.size()); + for (const auto& target : targets) { + DORIS_CHECK(target.__isset.target_fragment_ids); + DORIS_CHECK(!target.target_fragment_ids.empty()); + RuntimeFilterPublishTarget publish_target; + publish_target.addr.set_hostname(target.target_fragment_instance_addr.hostname); + publish_target.addr.set_port(target.target_fragment_instance_addr.port); + publish_target.fragment_ids = target.target_fragment_ids; + publish_targets.emplace_back(std::move(publish_target)); + } + return publish_targets; +} + +Status send_runtime_filter_relay_rpc(const RuntimeFilterPublishTask& task, + const butil::IOBuf& request_attachment, int timeout_ms, + std::weak_ptr<QueryContext> query_ctx) { + std::shared_ptr<PBackendService_Stub> stub( + ExecEnv::GetInstance()->brpc_internal_client_cache()->get_client(task.receiver.addr)); + if (stub == nullptr) { + LOG(WARNING) << "Failed to init runtime filter relay rpc to " + << task.receiver.addr.hostname() << ":" << task.receiver.addr.port(); + return Status::InternalError("Failed to init runtime filter relay rpc to {}:{}", + task.receiver.addr.hostname(), task.receiver.addr.port()); + } + + auto closure = + AutoReleaseClosure<PPublishFilterRequestV2, DummyBrpcCallback<PPublishFilterResponse>>:: + create_unique(std::make_shared<PPublishFilterRequestV2>(task.request), + DummyBrpcCallback<PPublishFilterResponse>::create_shared(), + query_ctx); + if (!request_attachment.empty()) { + closure->cntl_->request_attachment().append(request_attachment); + } + closure->cntl_->set_timeout_ms(timeout_ms); + if (config::execution_ignore_eovercrowded) { + closure->cntl_->ignore_eovercrowded(); + } + stub->apply_filterv2(closure->cntl_.get(), closure->request_.get(), closure->response_.get(), + closure.get()); + closure.release(); + return Status::OK(); +} + +void set_request_direct_publish_target(const TRuntimeFilterTargetParamsV2& target, + PPublishFilterRequestV2* request) { + if (target.__isset.target_fragment_ids && !target.target_fragment_ids.empty()) { + for (const auto& target_fragment_id : target.target_fragment_ids) { + request->add_fragment_ids(target_fragment_id); + } + } else { + for (const auto& target_fragment_instance_id : target.target_fragment_instance_ids) { + PUniqueId* id = request->add_fragment_instance_ids(); + id->set_hi(target_fragment_instance_id.hi); + id->set_lo(target_fragment_instance_id.lo); + } + } +} + +} // namespace + +std::vector<std::vector<RuntimeFilterPublishTarget>> split_runtime_filter_publish_targets( + const std::vector<RuntimeFilterPublishTarget>& targets, int fanout) { + DORIS_CHECK(!targets.empty()); + DORIS_CHECK(fanout > 0); + size_t slice_count = std::min(targets.size(), static_cast<size_t>(fanout)); + std::vector<std::vector<RuntimeFilterPublishTarget>> slices; + slices.reserve(slice_count); + for (size_t offset = 0; offset < targets.size();) { + size_t remaining_targets = targets.size() - offset; + size_t remaining_slices = slice_count - slices.size(); + size_t slice_size = (remaining_targets + remaining_slices - 1) / remaining_slices; + slices.emplace_back(targets.begin() + offset, targets.begin() + offset + slice_size); + offset += slice_size; + } + return slices; +} + +std::vector<RuntimeFilterPublishTask> build_runtime_filter_publish_tasks( + const PPublishFilterRequestV2& base_request, + const std::vector<RuntimeFilterPublishTarget>& targets, int fanout) { + std::vector<RuntimeFilterPublishTask> tasks; + auto slices = split_runtime_filter_publish_targets(targets, fanout); + PPublishFilterRequestV2 request_template = base_request; + request_template.clear_fragment_ids(); + request_template.clear_fragment_instance_ids(); + request_template.clear_forward_targets(); + tasks.reserve(slices.size()); + for (const auto& slice : slices) { + DORIS_CHECK(!slice.empty()); + RuntimeFilterPublishTask task; + task.receiver = slice.front(); + task.request = request_template; + + for (int32_t fragment_id : task.receiver.fragment_ids) { + task.request.add_fragment_ids(fragment_id); + } + for (size_t i = 1; i < slice.size(); ++i) { + DORIS_CHECK(!slice[i].fragment_ids.empty()); + PPublishFilterForwardTarget* forward_target = task.request.add_forward_targets(); + forward_target->mutable_target_addr()->CopyFrom(slice[i].addr); + for (int32_t fragment_id : slice[i].fragment_ids) { + forward_target->add_fragment_ids(fragment_id); + } + } + tasks.emplace_back(std::move(task)); + } + return tasks; +} + +int calculate_tree_publish_fanout(int64_t serialized_filter_size, size_t target_count, + int64_t max_send_bytes) { + DORIS_CHECK(serialized_filter_size > 0); + DORIS_CHECK(max_send_bytes >= 0); + if (max_send_bytes == 0 || target_count <= 1) { + return 0; + } + + const int64_t direct_target_limit = max_send_bytes / serialized_filter_size; + if (target_count <= static_cast<size_t>(direct_target_limit)) { + return 0; + } + + return static_cast<int>(std::max<int64_t>(1, direct_target_limit)); +} + +Status forward_runtime_filter(const PPublishFilterRequestV2& request, + const butil::IOBuf& request_attachment, + std::weak_ptr<QueryContext> query_ctx) { + if (request.forward_targets().empty()) { + return Status::OK(); + } + + std::vector<RuntimeFilterPublishTarget> targets; + targets.reserve(request.forward_targets_size()); + for (const auto& forward_target : request.forward_targets()) { + DORIS_CHECK(forward_target.has_target_addr()); + DORIS_CHECK(forward_target.fragment_ids_size() > 0); + RuntimeFilterPublishTarget target; + target.addr.CopyFrom(forward_target.target_addr()); + target.fragment_ids.assign(forward_target.fragment_ids().begin(), + forward_target.fragment_ids().end()); + targets.emplace_back(std::move(target)); + } + + DORIS_CHECK(request.has_tree_publish_fanout()); + int fanout = request.tree_publish_fanout(); + DORIS_CHECK(fanout > 0); + DORIS_CHECK(request.has_publish_rpc_timeout_ms()); + int timeout_ms = request.publish_rpc_timeout_ms(); + auto tasks = build_runtime_filter_publish_tasks(request, targets, fanout); + VLOG_NOTICE << "Runtime filter relay publish filter_id=" << request.filter_id() + << ", forward_targets=" << targets.size() << ", child_rpc_count=" << tasks.size() + << ", fanout=" << fanout; + auto st = Status::OK(); + for (const auto& task : tasks) { + Status rpc_st = + send_runtime_filter_relay_rpc(task, request_attachment, timeout_ms, query_ctx); + if (!rpc_st.ok()) { + LOG(WARNING) << "Failed to forward runtime filter, query_id=" + << print_id(request.query_id()) << ", filter_id=" << request.filter_id() + << ", target=" << task.receiver.addr.hostname() << ":" + << task.receiver.addr.port() << ", status=" << rpc_st; + st = std::move(rpc_st); + } + } + return st; +} + RuntimeFilterMgr::RuntimeFilterMgr(const bool is_global) : _is_global(is_global), _tracker(std::make_unique<MemTracker>( @@ -406,17 +590,20 @@ Status RuntimeFilterMergeControllerEntity::merge(std::shared_ptr<QueryContext> q query_ctx->ignore_runtime_filter_error() ? std::weak_ptr<QueryContext> {} : query_ctx, - merge_time, request->query_id(), query_ctx->execution_timeout()); + merge_time, request->query_id(), query_ctx->execution_timeout(), + query_ctx->query_options()); } return Status::OK(); } -Status RuntimeFilterMergeControllerEntity::_send_rf_to_target(GlobalMergeContext& cnt_val, - std::weak_ptr<QueryContext> ctx, - int64_t merge_time, - PUniqueId query_id, - int execution_timeout) { - DCHECK_GT(cnt_val.targetv2_info.size(), 0); +Status RuntimeFilterMergeControllerEntity::_send_rf_to_target( + GlobalMergeContext& cnt_val, std::weak_ptr<QueryContext> ctx, int64_t merge_time, + PUniqueId query_id, int execution_timeout, const TQueryOptions& query_options) { + if (cnt_val.targetv2_info.empty()) { + return Status::InternalError( + "_send_rf_to_target called with empty targetv2_info, filter: {}", + cnt_val.merger ? cnt_val.merger->debug_string() : "unknown"); + } if (cnt_val.done) { return Status::InternalError("Runtime filter has been sent", @@ -446,6 +633,41 @@ Status RuntimeFilterMergeControllerEntity::_send_rf_to_target(GlobalMergeContext } std::vector<TRuntimeFilterTargetParamsV2>& targets = cnt_val.targetv2_info; + int timeout_ms = get_execution_rpc_timeout_ms(execution_timeout); + apply_request.set_merge_time(merge_time); + *apply_request.mutable_query_id() = query_id; + const int64_t serialized_filter_size = + static_cast<int64_t>(apply_request.ByteSizeLong()) + std::max(0, len); + int fanout = 0; + if (can_tree_publish(targets) && + query_options.__isset.runtime_filter_tree_publish_max_send_bytes) { + int64_t max_send_bytes = query_options.runtime_filter_tree_publish_max_send_bytes; + DORIS_CHECK(max_send_bytes >= 0); + fanout = calculate_tree_publish_fanout(serialized_filter_size, targets.size(), + max_send_bytes); + } + const bool use_tree_publish = fanout > 0 && targets.size() > static_cast<size_t>(fanout); + + if (use_tree_publish) { + apply_request.set_tree_publish_fanout(fanout); + apply_request.set_publish_rpc_timeout_ms(timeout_ms); + auto publish_targets = build_runtime_filter_publish_targets(targets); + auto tasks = build_runtime_filter_publish_tasks(apply_request, publish_targets, fanout); + VLOG_NOTICE << "Runtime filter tree publish filter_id=" << apply_request.filter_id() + << ", serialized_bytes=" << serialized_filter_size + << ", target_count=" << targets.size() << ", child_rpc_count=" << tasks.size() + << ", fanout=" << fanout; + auto st = Status::OK(); + for (const auto& task : tasks) { + Status rpc_st = + send_runtime_filter_relay_rpc(task, request_attachment, timeout_ms, ctx); + if (!rpc_st.ok()) { + st = std::move(rpc_st); + } + } + return st; + } + auto st = Status::OK(); for (auto& target : targets) { auto closure = AutoReleaseClosure<PPublishFilterRequestV2, @@ -453,30 +675,16 @@ Status RuntimeFilterMergeControllerEntity::_send_rf_to_target(GlobalMergeContext create_unique(std::make_shared<PPublishFilterRequestV2>(apply_request), DummyBrpcCallback<PPublishFilterResponse>::create_shared(), ctx); - closure->request_->set_merge_time(merge_time); - *closure->request_->mutable_query_id() = query_id; if (has_attachment) { closure->cntl_->request_attachment().append(request_attachment); } - closure->cntl_->set_timeout_ms(get_execution_rpc_timeout_ms(execution_timeout)); + closure->cntl_->set_timeout_ms(timeout_ms); if (config::execution_ignore_eovercrowded) { closure->cntl_->ignore_eovercrowded(); } - // set fragment-id - if (target.__isset.target_fragment_ids) { - for (auto& target_fragment_id : target.target_fragment_ids) { - closure->request_->add_fragment_ids(target_fragment_id); - } - } else { - // FE not upgraded yet. - for (auto& target_fragment_instance_id : target.target_fragment_instance_ids) { - PUniqueId* cur_id = closure->request_->add_fragment_instance_ids(); - cur_id->set_hi(target_fragment_instance_id.hi); - cur_id->set_lo(target_fragment_instance_id.lo); - } - } + set_request_direct_publish_target(target, closure->request_.get()); std::shared_ptr<PBackendService_Stub> stub( ExecEnv::GetInstance()->brpc_internal_client_cache()->get_client( diff --git a/be/src/exec/runtime_filter/runtime_filter_mgr.h b/be/src/exec/runtime_filter/runtime_filter_mgr.h index d680e339f61..94058c2c547 100644 --- a/be/src/exec/runtime_filter/runtime_filter_mgr.h +++ b/be/src/exec/runtime_filter/runtime_filter_mgr.h @@ -23,6 +23,7 @@ #include <gen_cpp/internal_service.pb.h> #include <glog/logging.h> +#include <cstddef> #include <cstdint> #include <map> #include <memory> @@ -35,8 +36,9 @@ #include "util/uid_util.h" namespace butil { +class IOBuf; class IOBufAsZeroCopyInputStream; -} +} // namespace butil namespace doris { #include "common/compile_check_begin.h" @@ -52,6 +54,30 @@ class RuntimeFilterWrapper; class QueryContext; class ExecEnv; +struct RuntimeFilterPublishTarget { + PNetworkAddress addr; + std::vector<int32_t> fragment_ids; +}; + +struct RuntimeFilterPublishTask { + RuntimeFilterPublishTarget receiver; + PPublishFilterRequestV2 request; +}; + +std::vector<std::vector<RuntimeFilterPublishTarget>> split_runtime_filter_publish_targets( + const std::vector<RuntimeFilterPublishTarget>& targets, int fanout); + +std::vector<RuntimeFilterPublishTask> build_runtime_filter_publish_tasks( + const PPublishFilterRequestV2& base_request, + const std::vector<RuntimeFilterPublishTarget>& targets, int fanout); + +int calculate_tree_publish_fanout(int64_t serialized_filter_size, size_t target_count, + int64_t max_send_bytes); + +Status forward_runtime_filter(const PPublishFilterRequestV2& request, + const butil::IOBuf& request_attachment, + std::weak_ptr<QueryContext> query_ctx); + struct LocalMergeContext { std::shared_ptr<RuntimeFilterMerger> merger; std::vector<std::shared_ptr<RuntimeFilterProducer>> producers; @@ -184,7 +210,8 @@ private: const int producer_size); Status _send_rf_to_target(GlobalMergeContext& cnt_val, std::weak_ptr<QueryContext> ctx, - int64_t merge_time, PUniqueId query_id, int execution_timeout); + int64_t merge_time, PUniqueId query_id, int execution_timeout, + const TQueryOptions& query_options); // protect _filter_map AnnotatedSharedMutex _filter_map_mutex; diff --git a/be/src/service/internal_service.cpp b/be/src/service/internal_service.cpp index 050692f1488..47c84b56323 100644 --- a/be/src/service/internal_service.cpp +++ b/be/src/service/internal_service.cpp @@ -67,6 +67,7 @@ #include "exec/connector/vjdbc_connector.h" #include "exec/exchange/vdata_stream_mgr.h" #include "exec/rowid_fetcher.h" +#include "exec/runtime_filter/runtime_filter_mgr.h" #include "exec/sink/writer/varrow_flight_result_writer.h" #include "exec/sink/writer/vmysql_result_writer.h" #include "exprs/function/dictionary_factory.h" @@ -95,6 +96,7 @@ #include "runtime/exec_env.h" #include "runtime/fold_constant_executor.h" #include "runtime/fragment_mgr.h" +#include "runtime/query_context.h" #include "runtime/result_block_buffer.h" #include "runtime/result_buffer_mgr.h" #include "runtime/runtime_profile.h" @@ -1492,8 +1494,11 @@ void PInternalService::apply_filterv2(::google::protobuf::RpcController* control bool ret = _light_work_pool.try_offer([this, controller, request, response, done]() { signal::SignalTaskIdKeeper keeper(request->query_id()); brpc::ClosureGuard closure_guard(done); - auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment(); - butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment); + const butil::IOBuf& request_attachment = + static_cast<brpc::Controller*>(controller)->request_attachment(); + butil::IOBuf apply_attachment = request_attachment; + butil::IOBuf forward_attachment = request_attachment; + butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(apply_attachment); VLOG_NOTICE << "rpc apply_filterv2 recv"; Status st; try { @@ -1504,6 +1509,20 @@ void PInternalService::apply_filterv2(::google::protobuf::RpcController* control if (!st.ok()) { LOG(WARNING) << "apply filter meet error: " << st.to_string(); } + std::weak_ptr<QueryContext> forward_ctx; + if (auto query_ctx = _exec_env->fragment_mgr()->get_query_ctx( + UniqueId(request->query_id()).to_thrift())) { + if (!query_ctx->ignore_runtime_filter_error()) { + forward_ctx = query_ctx; + } + } + Status forward_st = forward_runtime_filter(*request, forward_attachment, forward_ctx); + if (!forward_st.ok()) { + LOG(WARNING) << "forward runtime filter meet error: " << forward_st.to_string(); + if (st.ok()) { + st = std::move(forward_st); + } + } st.to_protobuf(response->mutable_status()); }); if (!ret) { diff --git a/be/test/exec/runtime_filter/runtime_filter_mgr_test.cpp b/be/test/exec/runtime_filter/runtime_filter_mgr_test.cpp index 5a50e762c25..5f69f411cf1 100644 --- a/be/test/exec/runtime_filter/runtime_filter_mgr_test.cpp +++ b/be/test/exec/runtime_filter/runtime_filter_mgr_test.cpp @@ -20,12 +20,37 @@ #include <glog/logging.h> #include <gtest/gtest.h> +#include <string> + #include "exec/pipeline/thrift_builder.h" #include "exec/runtime_filter/runtime_filter_producer.h" #include "runtime/query_context.h" #include "testutil/mock/mock_runtime_state.h" namespace doris { +namespace { + +RuntimeFilterPublishTarget make_publish_target(int index) { + RuntimeFilterPublishTarget target; + target.addr.set_hostname("host" + std::to_string(index)); + target.addr.set_port(9000 + index); + target.fragment_ids.push_back(index); + return target; +} + +std::vector<int32_t> flatten_fragment_ids( + const std::vector<std::vector<RuntimeFilterPublishTarget>>& slices) { + std::vector<int32_t> fragment_ids; + for (const auto& slice : slices) { + for (const auto& target : slice) { + fragment_ids.insert(fragment_ids.end(), target.fragment_ids.begin(), + target.fragment_ids.end()); + } + } + return fragment_ids; +} + +} // namespace class RuntimeFilterMgrTest : public testing::Test { public: @@ -186,4 +211,89 @@ TEST_F(RuntimeFilterMgrTest, TestRuntimeFilterMergeControllerEntity) { } } +TEST_F(RuntimeFilterMgrTest, SplitRuntimeFilterPublishTargets) { + std::vector<RuntimeFilterPublishTarget> targets; + for (int i = 0; i < 10; ++i) { + targets.push_back(make_publish_target(i)); + } + + auto slices = split_runtime_filter_publish_targets(targets, 3); + ASSERT_EQ(slices.size(), 3); + EXPECT_EQ(slices[0].size(), 4); + EXPECT_EQ(slices[1].size(), 3); + EXPECT_EQ(slices[2].size(), 3); + EXPECT_EQ(flatten_fragment_ids(slices), (std::vector<int32_t> {0, 1, 2, 3, 4, 5, 6, 7, 8, 9})); + + slices = split_runtime_filter_publish_targets(targets, 20); + ASSERT_EQ(slices.size(), targets.size()); + for (const auto& slice : slices) { + EXPECT_EQ(slice.size(), 1); + } + + slices = split_runtime_filter_publish_targets(targets, 1); + ASSERT_EQ(slices.size(), 1); + EXPECT_EQ(slices[0].size(), targets.size()); +} + +TEST_F(RuntimeFilterMgrTest, CalculateTreePublishFanout) { + constexpr int64_t MB = 1024L * 1024L; + constexpr int64_t max_send_bytes = 256L * MB; + constexpr size_t target_count = 48; + + EXPECT_EQ(calculate_tree_publish_fanout(1L * MB, target_count, max_send_bytes), 0); + EXPECT_EQ(calculate_tree_publish_fanout(4L * MB, target_count, max_send_bytes), 0); + EXPECT_EQ(calculate_tree_publish_fanout(8L * MB, target_count, max_send_bytes), 32); + EXPECT_EQ(calculate_tree_publish_fanout(16L * MB, target_count, max_send_bytes), 16); + EXPECT_EQ(calculate_tree_publish_fanout(32L * MB, target_count, max_send_bytes), 8); + EXPECT_EQ(calculate_tree_publish_fanout(64L * MB, target_count, max_send_bytes), 4); + EXPECT_EQ(calculate_tree_publish_fanout(128L * MB, target_count, max_send_bytes), 2); + + EXPECT_EQ(calculate_tree_publish_fanout(4L * MB, target_count, 0), 0); + EXPECT_EQ(calculate_tree_publish_fanout(256L * MB, target_count, max_send_bytes), 1); + EXPECT_EQ(calculate_tree_publish_fanout(1L * MB, 1024, max_send_bytes), 256); +} + +TEST_F(RuntimeFilterMgrTest, BuildRuntimeFilterPublishTasks) { + PPublishFilterRequestV2 base_request; + base_request.set_filter_id(10); + base_request.mutable_query_id()->set_hi(1); + base_request.mutable_query_id()->set_lo(2); + base_request.set_filter_type(PFilterType::BLOOM_FILTER); + base_request.set_tree_publish_fanout(2); + base_request.set_publish_rpc_timeout_ms(3000); + base_request.add_fragment_ids(999); + auto* stale_forward_target = base_request.add_forward_targets(); + stale_forward_target->mutable_target_addr()->set_hostname("stale"); + stale_forward_target->mutable_target_addr()->set_port(1); + stale_forward_target->add_fragment_ids(999); + + std::vector<RuntimeFilterPublishTarget> targets; + for (int i = 0; i < 5; ++i) { + targets.push_back(make_publish_target(i)); + } + + auto tasks = build_runtime_filter_publish_tasks(base_request, targets, 2); + ASSERT_EQ(tasks.size(), 2); + + EXPECT_EQ(tasks[0].receiver.addr.hostname(), "host0"); + EXPECT_EQ(tasks[0].request.fragment_ids_size(), 1); + EXPECT_EQ(tasks[0].request.fragment_ids(0), 0); + ASSERT_EQ(tasks[0].request.forward_targets_size(), 2); + EXPECT_EQ(tasks[0].request.forward_targets(0).target_addr().hostname(), "host1"); + EXPECT_EQ(tasks[0].request.forward_targets(0).fragment_ids(0), 1); + EXPECT_EQ(tasks[0].request.forward_targets(1).target_addr().hostname(), "host2"); + EXPECT_EQ(tasks[0].request.forward_targets(1).fragment_ids(0), 2); + + EXPECT_EQ(tasks[1].receiver.addr.hostname(), "host3"); + EXPECT_EQ(tasks[1].request.fragment_ids_size(), 1); + EXPECT_EQ(tasks[1].request.fragment_ids(0), 3); + ASSERT_EQ(tasks[1].request.forward_targets_size(), 1); + EXPECT_EQ(tasks[1].request.forward_targets(0).target_addr().hostname(), "host4"); + EXPECT_EQ(tasks[1].request.forward_targets(0).fragment_ids(0), 4); + + EXPECT_EQ(tasks[0].request.filter_id(), base_request.filter_id()); + EXPECT_EQ(tasks[0].request.tree_publish_fanout(), 2); + EXPECT_EQ(tasks[0].request.publish_rpc_timeout_ms(), 3000); +} + } // namespace doris diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index e3752257797..34e82825c48 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -261,6 +261,8 @@ public class SessionVariable implements Serializable, Writable { "runtime_filter_broadcast_join_producer_num"; public static final String ENABLE_SYNC_RUNTIME_FILTER_SIZE = "enable_sync_runtime_filter_size"; + public static final String RUNTIME_FILTER_TREE_PUBLISH_MAX_SEND_BYTES = + "runtime_filter_tree_publish_max_send_bytes"; public static final String ENABLE_PARALLEL_RESULT_SINK = "enable_parallel_result_sink"; @@ -1749,6 +1751,10 @@ public class SessionVariable implements Serializable, Writable { + "The legacy Coordinator path keeps the existing behavior."}) private int runtimeFilterBroadcastJoinProducerNum = 3; + @VariableMgr.VarAttr(name = RUNTIME_FILTER_TREE_PUBLISH_MAX_SEND_BYTES, needForward = true, fuzzy = true, + checker = "checkRuntimeFilterTreePublishMaxSendBytes") + private long runtimeFilterTreePublishMaxSendBytes = 256L * 1024L * 1024L; + @VariableMgr.VarAttr(name = "runtime_filter_max_build_row_count", needForward = true, fuzzy = false) public long runtimeFilterMaxBuildRowCount = 64L * 1024L * 1024L; @@ -3784,6 +3790,9 @@ public class SessionVariable implements Serializable, Writable { this.enableParallelScan = random.nextInt(2) == 0; this.enableRuntimeFilterPrune = (randomInt % 10) == 0; this.enableRuntimeFilterPartitionPrune = (randomInt % 2) == 0; + this.runtimeFilterTreePublishMaxSendBytes = + Util.getRandomLong(0, 64L * 1024L * 1024L, 128L * 1024L * 1024L, + 256L * 1024L * 1024L); switch (randomInt) { case 0: @@ -4743,6 +4752,10 @@ public class SessionVariable implements Serializable, Writable { this.runtimeFilterBroadcastJoinProducerNum = runtimeFilterBroadcastJoinProducerNum; } + public long getRuntimeFilterTreePublishMaxSendBytes() { + return runtimeFilterTreePublishMaxSendBytes; + } + public void setEnableLocalShuffle(boolean enableLocalShuffle) { this.enableLocalShuffle = enableLocalShuffle; } @@ -5532,6 +5545,7 @@ public class SessionVariable implements Serializable, Writable { tResult.setRuntimeBloomFilterMinSize(runtimeBloomFilterMinSize); tResult.setRuntimeBloomFilterMaxSize(runtimeBloomFilterMaxSize); tResult.setRuntimeFilterWaitInfinitely(runtimeFilterWaitInfinitely); + tResult.setRuntimeFilterTreePublishMaxSendBytes(runtimeFilterTreePublishMaxSendBytes); tResult.setEnableFuzzyBlockableTask(enableFuzzyBlockableTask); if (cpuResourceLimit > 0) { @@ -6489,6 +6503,18 @@ public class SessionVariable implements Serializable, Writable { } } + public void checkRuntimeFilterTreePublishMaxSendBytes(String maxSendBytes) { + long value = Long.valueOf(maxSendBytes); + if (value < 0) { + UnsupportedOperationException exception = + new UnsupportedOperationException( + "runtime_filter_tree_publish_max_send_bytes can not be set to " + + maxSendBytes + ", it must be greater or equal to 0"); + LOG.warn("Check runtime_filter_tree_publish_max_send_bytes failed", exception); + throw exception; + } + } + public void checkSqlConvertorFeatures(String features) { if (Strings.isNullOrEmpty(features)) { return; diff --git a/gensrc/proto/internal_service.proto b/gensrc/proto/internal_service.proto index 4a41ac9ed4a..569abbc4ad8 100644 --- a/gensrc/proto/internal_service.proto +++ b/gensrc/proto/internal_service.proto @@ -672,8 +672,16 @@ message PPublishFilterRequestV2 { optional uint64 local_merge_time = 13; optional bool disabled = 14; optional uint32 stage = 15; + repeated PPublishFilterForwardTarget forward_targets = 16; + optional int32 tree_publish_fanout = 17; + optional int32 publish_rpc_timeout_ms = 18; }; +message PPublishFilterForwardTarget { + optional PNetworkAddress target_addr = 1; + repeated int32 fragment_ids = 2; +} + message PPublishFilterResponse { required PStatus status = 1; }; @@ -1243,4 +1251,3 @@ service PBackendService { rpc fetch_peer_data(PFetchPeerDataRequest) returns (PFetchPeerDataResponse); rpc request_cdc_client(PRequestCdcClientRequest) returns (PRequestCdcClientResult); }; - diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift index 71b318f6f0c..9ab7b0a6ce6 100644 --- a/gensrc/thrift/PaloInternalService.thrift +++ b/gensrc/thrift/PaloInternalService.thrift @@ -496,6 +496,7 @@ struct TQueryOptions { 219: optional i64 ann_index_candidate_rows_threshold = 0 // Candidate row ratio threshold against segment rows. Existing default is 0.3. 220: optional double ann_index_candidate_rows_percent_threshold = 0.3 + 221: optional i64 runtime_filter_tree_publish_max_send_bytes = 268435456 // For cloud, to control if the content would be written into file cache // In write path, to control if the content would be written into file cache. // In read path, read from file cache or remote storage when execute query. --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
