This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 089c3658fd5 branch-4.1: [fix](doris catalog) FragmentMgr should not
cancel virtual doris cluster query #62135 (#62213)
089c3658fd5 is described below
commit 089c3658fd556941ef74966f13e24b6f6ac1441e
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Apr 9 11:03:27 2026 +0800
branch-4.1: [fix](doris catalog) FragmentMgr should not cancel virtual
doris cluster query #62135 (#62213)
Cherry-picked from #62135
Co-authored-by: HonestManXin <[email protected]>
---
be/src/runtime/fragment_mgr.cpp | 18 ++-
be/src/runtime/query_context.cpp | 2 +
be/src/runtime/query_context.h | 3 +-
.../fragment_mgr_cross_cluster_cancel_test.cpp | 156 +++++++++++++++++++++
4 files changed, 175 insertions(+), 4 deletions(-)
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index ae22def3012..8f1edbd00df 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -764,9 +764,16 @@ Status FragmentMgr::_get_or_create_query_ctx(const
TPipelineFragmentParams& para
// This may be a first fragment request of the query.
// Create the query fragments context.
- query_ctx = QueryContext::create(query_id, _exec_env,
params.query_options,
- params.coord,
params.is_nereids,
-
params.current_connect_fe, query_source);
+ // Cross-cluster query: coordinator FE may not belong
to local cluster.
+ // In that case, cancel_worker() should not cancel it
based on local FE liveness.
+ QuerySource actual_query_source = query_source;
+ if (query_source == QuerySource::INTERNAL_FRONTEND &&
+
!_exec_env->get_running_frontends().contains(params.coord)) {
+ actual_query_source =
QuerySource::EXTERNAL_FRONTEND;
+ }
+ query_ctx = QueryContext::create(
+ query_id, _exec_env, params.query_options,
params.coord,
+ params.is_nereids, params.current_connect_fe,
actual_query_source);
SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(query_ctx->query_mem_tracker());
RETURN_IF_ERROR(DescriptorTbl::create(
&(query_ctx->obj_pool), params.desc_tbl,
&(query_ctx->desc_tbl)));
@@ -1119,6 +1126,11 @@ void FragmentMgr::_collect_invalid_queries(
-> Status {
for (const auto& it : map) {
if (auto q_ctx = it.second.lock()) {
+ // Cross-cluster query: coordinator FE is not in local
`running_fes`,
+ // we should not cancel it based on local coordinator liveness.
+ if (q_ctx->get_query_source() ==
QuerySource::EXTERNAL_FRONTEND) {
+ continue;
+ }
q_contexts.push_back(q_ctx);
const int64_t fe_process_uuid = q_ctx->get_fe_process_uuid();
diff --git a/be/src/runtime/query_context.cpp b/be/src/runtime/query_context.cpp
index 198666346c0..e00a37aadb1 100644
--- a/be/src/runtime/query_context.cpp
+++ b/be/src/runtime/query_context.cpp
@@ -73,6 +73,8 @@ const std::string toString(QuerySource queryType) {
return "ROUTINE_LOAD";
case QuerySource::EXTERNAL_CONNECTOR:
return "EXTERNAL_CONNECTOR";
+ case QuerySource::EXTERNAL_FRONTEND:
+ return "EXTERNAL_FRONTEND";
default:
return "UNKNOWN";
}
diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h
index c4ed6cafe85..d5e559890c0 100644
--- a/be/src/runtime/query_context.h
+++ b/be/src/runtime/query_context.h
@@ -70,7 +70,8 @@ enum class QuerySource {
STREAM_LOAD,
GROUP_COMMIT_LOAD,
ROUTINE_LOAD,
- EXTERNAL_CONNECTOR
+ EXTERNAL_CONNECTOR,
+ EXTERNAL_FRONTEND
};
const std::string toString(QuerySource query_source);
diff --git a/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
b/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
new file mode 100644
index 00000000000..91c242f4b8f
--- /dev/null
+++ b/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
@@ -0,0 +1,156 @@
+// 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 <gen_cpp/PaloInternalService_types.h>
+#include <gtest/gtest.h>
+
+#include "runtime/descriptor_helper.h"
+#include "runtime/exec_env.h"
+#include "runtime/fragment_mgr.h"
+#include "runtime/workload_group/workload_group_manager.h"
+
+namespace doris {
+
+class FragmentMgrCrossClusterCancelTest : public testing::Test {
+public:
+ void SetUp() override {
+ // Make frontends list deterministic for this ExecEnv instance.
+ _exec_env.update_frontends({});
+
+ _exec_env._workload_group_manager = new WorkloadGroupMgr();
+ // Ensure there is a "normal" workload group, otherwise
WorkloadGroupMgr::get_group() will throw.
+ WorkloadGroupInfo normal_wg_info {.id = 1, .name = "normal"};
+
_exec_env._workload_group_manager->get_or_create_workload_group(normal_wg_info);
+
+ _exec_env._fragment_mgr = new FragmentMgr(&_exec_env);
+ }
+
+ void TearDown() override {
+ if (_exec_env._fragment_mgr != nullptr) {
+ _exec_env._fragment_mgr->stop();
+ }
+ delete _exec_env._fragment_mgr;
+ _exec_env._fragment_mgr = nullptr;
+ delete _exec_env._workload_group_manager;
+ _exec_env._workload_group_manager = nullptr;
+ _exec_env.update_frontends({});
+ }
+
+protected:
+ static TDescriptorTable _make_min_desc_tbl() {
+ TDescriptorTableBuilder dtb;
+ TTupleDescriptorBuilder tuple_builder;
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(TYPE_INT)
+ .nullable(true)
+ .column_name("c1")
+ .column_pos(1)
+ .build());
+ tuple_builder.build(&dtb);
+ return dtb.desc_tbl();
+ }
+
+ static TQueryOptions _make_min_query_options(int64_t fe_process_uuid) {
+ TQueryOptions query_options;
+ query_options.__set_query_type(TQueryType::SELECT);
+ query_options.__set_execution_timeout(60);
+ query_options.__set_query_timeout(60);
+ query_options.__set_mem_limit(64L * 1024 * 1024);
+ query_options.__set_fe_process_uuid(fe_process_uuid);
+ return query_options;
+ }
+
+ ExecEnv _exec_env;
+};
+
+TEST_F(FragmentMgrCrossClusterCancelTest,
+ MarkQuerySourceAsExternalFrontendWhenCoordinatorNotLocal) {
+ auto* fragment_mgr = _exec_env.fragment_mgr();
+ ASSERT_NE(fragment_mgr, nullptr);
+
+ TUniqueId query_id;
+ query_id.__set_hi(1);
+ query_id.__set_lo(2);
+
+ TNetworkAddress coord;
+ coord.hostname = "fe-a";
+ coord.port = 9030;
+
+ TPipelineFragmentParams params;
+ params.__set_query_id(query_id);
+ params.__set_is_simplified_param(false);
+ params.__set_coord(coord);
+ params.__set_is_nereids(false);
+ params.__set_current_connect_fe(coord);
+ params.__set_fragment_num_on_host(1);
+ params.__set_query_options(_make_min_query_options(/*fe_process_uuid*/
123));
+ params.__set_desc_tbl(_make_min_desc_tbl());
+
+ std::shared_ptr<QueryContext> query_ctx;
+ TPipelineFragmentParamsList parent;
+ auto st = fragment_mgr->_get_or_create_query_ctx(params, parent,
QuerySource::INTERNAL_FRONTEND,
+ query_ctx);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ ASSERT_NE(query_ctx, nullptr);
+ EXPECT_EQ(query_ctx->get_query_source(), QuerySource::EXTERNAL_FRONTEND);
+}
+
+TEST_F(FragmentMgrCrossClusterCancelTest,
CancelWorkerInvalidQueryDetectionSkipsExternalFrontend) {
+ auto* fragment_mgr = _exec_env.fragment_mgr();
+ ASSERT_NE(fragment_mgr, nullptr);
+
+ TUniqueId query_id;
+ query_id.__set_hi(3);
+ query_id.__set_lo(4);
+
+ TNetworkAddress coord;
+ coord.hostname = "fe-b";
+ coord.port = 9030;
+
+ TPipelineFragmentParams params;
+ params.__set_query_id(query_id);
+ params.__set_is_simplified_param(false);
+ params.__set_coord(coord);
+ params.__set_is_nereids(false);
+ params.__set_current_connect_fe(coord);
+ params.__set_fragment_num_on_host(1);
+ params.__set_query_options(_make_min_query_options(/*fe_process_uuid*/
456));
+ params.__set_desc_tbl(_make_min_desc_tbl());
+
+ std::shared_ptr<QueryContext> query_ctx;
+ TPipelineFragmentParamsList parent;
+ auto st = fragment_mgr->_get_or_create_query_ctx(params, parent,
QuerySource::INTERNAL_FRONTEND,
+ query_ctx);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ ASSERT_NE(query_ctx, nullptr);
+ ASSERT_EQ(query_ctx->get_query_source(), QuerySource::EXTERNAL_FRONTEND);
+
+ std::vector<TUniqueId> queries_lost_coordinator;
+ std::vector<TUniqueId> queries_pipeline_task_leak;
+ std::map<int64_t, std::unordered_set<TUniqueId>>
running_queries_on_all_fes;
+ std::map<TNetworkAddress, FrontendInfo> running_fes;
+ timespec ts;
+ ts.tv_sec = 0;
+ ts.tv_nsec = 0;
+
+ fragment_mgr->_collect_invalid_queries(queries_lost_coordinator,
queries_pipeline_task_leak,
+ running_queries_on_all_fes,
running_fes, ts);
+ EXPECT_TRUE(queries_lost_coordinator.empty());
+ EXPECT_TRUE(queries_pipeline_task_leak.empty());
+}
+
+} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
