This is an automated email from the ASF dual-hosted git repository.
Mryange pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 50fd7fd48c8 [fix](be) Fix recursive CTE fragment count (#68234)
50fd7fd48c8 is described below
commit 50fd7fd48c8cac5a775f98bf0e8dc7fb31150fd1
Author: Mryange <[email protected]>
AuthorDate: Tue Sep 22 15:31:41 2026 +0800
[fix](be) Fix recursive CTE fragment count (#68234)
After a recursive CTE pipeline fragment was rebuilt, the Backend
`currentFragmentNum` metric could become negative. Root cause: removing
the old pipeline fragment context decremented the executing fragment
count, while inserting the rebuilt context did not increment it. This
change centralizes fragment count updates, increments the count after a
rebuilt context is successfully prepared and inserted, and adds a
focused BE unit test for the REBUILD lifecycle.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/agent/task_worker_pool.cpp | 18 ++-----
be/src/exec/pipeline/pipeline_fragment_context.cpp | 4 ++
be/src/runtime/fragment_mgr.cpp | 29 +++++++----
be/src/runtime/fragment_mgr.h | 5 +-
.../fragment_mgr_cross_cluster_cancel_test.cpp | 60 ++++++++++++++++++++++
be/test/vec/spill/spill_file_test.cpp | 2 +-
6 files changed, 89 insertions(+), 29 deletions(-)
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 59f9be925f8..0fe9b9fc5bd 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -504,11 +504,7 @@ void add_task_count(const TAgentTaskRequest& task, int n) {
// cloud auto stop need sc jobs, a tablet's sc can also be considered
a fragment
if (n > 0) {
// only count fragment when task is actually starting
- doris::g_fragment_executing_count << 1;
- int64_t now = duration_cast<std::chrono::milliseconds>(
-
std::chrono::system_clock::now().time_since_epoch())
- .count();
- g_fragment_last_active_time.set_value(now);
+ increment_fragment_executing_count();
}
return;
}
@@ -2302,11 +2298,7 @@ void alter_tablet_callback(StorageEngine& engine, const
TAgentTaskRequest& req)
alter_tablet(engine, req, signature, task_type, &finish_task_request);
finish_task(finish_task_request);
}
- doris::g_fragment_executing_count << -1;
- int64_t now = duration_cast<std::chrono::milliseconds>(
- std::chrono::system_clock::now().time_since_epoch())
- .count();
- g_fragment_last_active_time.set_value(now);
+ decrement_fragment_executing_count();
remove_task_info(req.task_type, req.signature);
}
@@ -2328,11 +2320,7 @@ void alter_cloud_tablet_callback(CloudStorageEngine&
engine, const TAgentTaskReq
alter_cloud_tablet(engine, req, signature, task_type,
&finish_task_request);
finish_task(finish_task_request);
}
- doris::g_fragment_executing_count << -1;
- int64_t now = duration_cast<std::chrono::milliseconds>(
- std::chrono::system_clock::now().time_since_epoch())
- .count();
- g_fragment_last_active_time.set_value(now);
+ decrement_fragment_executing_count();
// Clean up alter_version before remove_task_info to avoid race:
// remove_task_info allows same-signature re-submit, whose
pre_submit_callback
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp
b/be/src/exec/pipeline/pipeline_fragment_context.cpp
index 5fa5c4c2932..81091aa2421 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp
@@ -356,6 +356,10 @@ Status
PipelineFragmentContext::_build_and_prepare_full_pipeline(ThreadPool* thr
}
Status PipelineFragmentContext::prepare(ThreadPool* thread_pool) {
+ DBUG_EXECUTE_IF("fault_inject::PipelineFragmentContext::prepare.skip", {
+ _prepared = true;
+ return Status::OK();
+ });
if (_prepared) {
return Status::InternalError("Already prepared");
}
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index a26e2c157d8..23666c7258b 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -106,6 +106,22 @@ bvar::Status<uint64_t> g_fragment_last_active_time(
std::chrono::system_clock::now().time_since_epoch())
.count());
+void increment_fragment_executing_count() {
+ g_fragment_executing_count << 1;
+ int64_t now = duration_cast<std::chrono::milliseconds>(
+ std::chrono::system_clock::now().time_since_epoch())
+ .count();
+ g_fragment_last_active_time.set_value(now);
+}
+
+void decrement_fragment_executing_count() {
+ g_fragment_executing_count << -1;
+ int64_t now = duration_cast<std::chrono::milliseconds>(
+ std::chrono::system_clock::now().time_since_epoch())
+ .count();
+ g_fragment_last_active_time.set_value(now);
+}
+
uint64_t get_fragment_executing_count() {
return g_fragment_executing_count.get_value();
}
@@ -425,11 +441,7 @@ Status FragmentMgr::start_query_execution(const
PExecPlanFragmentStartRequest* r
void FragmentMgr::remove_pipeline_context(std::pair<TUniqueId, int> key) {
if (_pipeline_map.erase(key)) {
- int64_t now = duration_cast<std::chrono::milliseconds>(
-
std::chrono::system_clock::now().time_since_epoch())
- .count();
- g_fragment_executing_count << -1;
- g_fragment_last_active_time.set_value(now);
+ decrement_fragment_executing_count();
}
}
@@ -675,11 +687,7 @@ Status FragmentMgr::exec_plan_fragment(const
TPipelineFragmentParams& params,
DBUG_EXECUTE_IF("FragmentMgr.exec_plan_fragment.failed",
{ return
Status::Aborted("FragmentMgr.exec_plan_fragment.failed"); });
{
- int64_t now = duration_cast<std::chrono::milliseconds>(
-
std::chrono::system_clock::now().time_since_epoch())
- .count();
- g_fragment_executing_count << 1;
- g_fragment_last_active_time.set_value(now);
+ increment_fragment_executing_count();
// (query_id, fragment_id) is executed only on one BE, locks
_pipeline_map.
auto res = _pipeline_map.find({params.query_id, params.fragment_id});
@@ -1375,6 +1383,7 @@ Status FragmentMgr::rerun_fragment(const
std::shared_ptr<brpc::ClosureGuard>& gu
// Insert new PFC into _pipeline_map (old one was removed)
_pipeline_map.insert({info.params.query_id, info.params.fragment_id},
context);
+ increment_fragment_executing_count();
// Update QueryContext mapping (must support overwrite)
q_ctx->set_pipeline_context(info.params.fragment_id, context);
diff --git a/be/src/runtime/fragment_mgr.h b/be/src/runtime/fragment_mgr.h
index 5a5bf64fd8b..9eb09eced21 100644
--- a/be/src/runtime/fragment_mgr.h
+++ b/be/src/runtime/fragment_mgr.h
@@ -48,9 +48,6 @@ class IOBufAsZeroCopyInputStream;
}
namespace doris {
-extern bvar::Adder<uint64_t> g_fragment_executing_count;
-extern bvar::Status<uint64_t> g_fragment_last_active_time;
-
class PipelineFragmentContext;
class QueryContext;
class DescriptorTbl;
@@ -271,4 +268,6 @@ private:
uint64_t get_fragment_executing_count();
uint64_t get_fragment_last_active_time();
+void increment_fragment_executing_count();
+void decrement_fragment_executing_count();
} // namespace doris
diff --git a/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
b/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
index dd705635b6e..0372474aa03 100644
--- a/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
+++ b/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
@@ -20,6 +20,7 @@
#include <gtest/gtest.h>
#include "common/config.h"
+#include "exec/pipeline/pipeline_fragment_context.h"
#include "runtime/descriptor_helper.h"
#include "runtime/exec_env.h"
#include "runtime/fragment_mgr.h"
@@ -27,6 +28,7 @@
#include "runtime/index_policy/index_policy_mgr.h"
#include "runtime/workload_group/workload_group_manager.h"
#include "storage/id_manager.h"
+#include "util/debug_points.h"
#include "util/defer_op.h"
namespace doris {
@@ -226,6 +228,64 @@ TEST(FragmentMgrRerunnableParamsTest,
StopReleasesLastQueryContextRefOutsideLock
delete fragment_mgr;
}
+TEST_F(FragmentMgrCrossClusterCancelTest,
RebuildRestoresFragmentExecutingCount) {
+ auto* fragment_mgr = _exec_env.fragment_mgr();
+ const bool previous_enable_debug_points = config::enable_debug_points;
+ constexpr auto debug_point_name =
"fault_inject::PipelineFragmentContext::prepare.skip";
+ config::enable_debug_points = true;
+ DebugPoints::instance()->add(debug_point_name);
+
+ TUniqueId query_id;
+ query_id.__set_hi(505);
+ query_id.__set_lo(606);
+ constexpr int fragment_id = 1;
+ Defer cleanup([&] {
+ DebugPoints::instance()->remove(debug_point_name);
+ config::enable_debug_points = previous_enable_debug_points;
+ fragment_mgr->remove_pipeline_context({query_id, fragment_id});
+ fragment_mgr->remove_query_context(query_id);
+ });
+
+ TPipelineFragmentParams params;
+ params.__set_query_id(query_id);
+ params.__set_fragment_id(fragment_id);
+ params.__set_need_notify_close(true);
+ params.__set_is_simplified_param(false);
+ TNetworkAddress coord;
+ coord.hostname = "fe-rebuild";
+ coord.port = 9030;
+ params.__set_coord(coord);
+ params.__set_is_nereids(true);
+ params.__set_current_connect_fe(coord);
+ params.__set_fragment_num_on_host(1);
+ params.__set_query_options(_make_min_query_options(/*fe_process_uuid*/
789));
+ params.__set_desc_tbl(_make_min_desc_tbl());
+
+ std::shared_ptr<QueryContext> query_ctx;
+ TPipelineFragmentParamsList parent;
+ ASSERT_TRUE(fragment_mgr
+ ->_get_or_create_query_ctx(params, parent,
QuerySource::INTERNAL_FRONTEND,
+ query_ctx)
+ .ok());
+ ASSERT_NE(query_ctx, nullptr);
+ {
+ std::lock_guard lock(fragment_mgr->_rerunnable_params_lock);
+ auto& info = fragment_mgr->_rerunnable_params_map[{query_id,
fragment_id}];
+ info.params = params;
+ info.query_ctx = query_ctx;
+ }
+
+ const auto count_before_rebuild = get_fragment_executing_count();
+ auto st =
+ fragment_mgr->rerun_fragment({}, query_id, fragment_id,
PRerunFragmentParams::REBUILD);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ EXPECT_EQ(fragment_mgr->_pipeline_map.num_items(), 1);
+ EXPECT_EQ(get_fragment_executing_count(), count_before_rebuild + 1);
+
+ fragment_mgr->remove_pipeline_context({query_id, fragment_id});
+ EXPECT_EQ(get_fragment_executing_count(), count_before_rebuild);
+}
+
TEST(FragmentMgrExternalScanTest, SelectedColumnsFollowOutputExpressionOrder) {
TDescriptorTableBuilder desc_tbl_builder;
TTupleDescriptorBuilder tuple_builder;
diff --git a/be/test/vec/spill/spill_file_test.cpp
b/be/test/vec/spill/spill_file_test.cpp
index c4d4f140635..042273392d2 100644
--- a/be/test/vec/spill/spill_file_test.cpp
+++ b/be/test/vec/spill/spill_file_test.cpp
@@ -1295,7 +1295,7 @@ TEST_F(SpillFileTest,
FinalCloseReleasesRerunnableQueryContextAndDeletesSpillDir
auto context = std::make_shared<PipelineFragmentContext>(
query_id, params, query_ctx, exec_env, [](RuntimeState*,
Status*) {});
fragment_mgr->_pipeline_map.insert({query_id, fragment_id}, context);
- g_fragment_executing_count << 1;
+ increment_fragment_executing_count();
query_ctx->set_pipeline_context(fragment_id, context);
{
std::lock_guard lock(fragment_mgr->_rerunnable_params_lock);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
