You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/03/28 14:45:49 UTC
(doris) branch branch-4.1 updated: [Chore](pick) rec cte be part (#61806)
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 db38dc9c58f [Chore](pick) rec cte be part (#61806)
db38dc9c58f is described below
commit db38dc9c58f0f8a2943689c6fc2f36838274a006
Author: Pxl <xl...@selectdb.com>
AuthorDate: Sat Mar 28 22:45:43 2026 +0800
[Chore](pick) rec cte be part (#61806)
pick #58916 #59701 #59768 #59792 #60307 #60618 #61130
---
be/src/exec/exchange/vdata_stream_sender.cpp | 1 -
be/src/exec/operator/exchange_source_operator.cpp | 5 +
be/src/exec/operator/exchange_source_operator.h | 19 ++
be/src/exec/operator/operator.cpp | 10 +
be/src/exec/operator/operator.h | 7 +
.../partitioned_hash_join_sink_operator.cpp | 4 +
.../exec/operator/rec_cte_anchor_sink_operator.cpp | 56 ++++
.../exec/operator/rec_cte_anchor_sink_operator.h | 127 ++++++++
be/src/exec/operator/rec_cte_scan_operator.h | 86 ++++++
be/src/exec/operator/rec_cte_sink_operator.cpp | 55 ++++
be/src/exec/operator/rec_cte_sink_operator.h | 103 +++++++
be/src/exec/operator/rec_cte_source_operator.cpp | 85 ++++++
be/src/exec/operator/rec_cte_source_operator.h | 330 +++++++++++++++++++++
be/src/exec/operator/scan_operator.cpp | 35 ++-
be/src/exec/operator/scan_operator.h | 6 +-
be/src/exec/operator/sort_sink_operator.cpp | 3 +-
be/src/exec/operator/sort_sink_operator.h | 2 +-
be/src/exec/operator/spill_sort_sink_operator.cpp | 2 +-
be/src/exec/operator/union_sink_operator.h | 18 ++
be/src/exec/pipeline/pipeline.h | 2 +-
be/src/exec/pipeline/pipeline_fragment_context.cpp | 302 ++++++++++++++-----
be/src/exec/pipeline/pipeline_fragment_context.h | 57 +++-
be/src/exec/pipeline/pipeline_task.cpp | 19 +-
be/src/exec/pipeline/rec_cte_shared_state.h | 179 +++++++++++
be/src/exec/runtime_filter/runtime_filter.cpp | 6 +-
be/src/exec/runtime_filter/runtime_filter.h | 7 +
.../exec/runtime_filter/runtime_filter_consumer.h | 12 +-
be/src/exec/runtime_filter/runtime_filter_merger.h | 9 +-
be/src/exec/runtime_filter/runtime_filter_mgr.cpp | 84 +++++-
be/src/exec/runtime_filter/runtime_filter_mgr.h | 26 +-
.../runtime_filter/runtime_filter_producer.cpp | 10 +
be/src/exec/scan/scanner_context.h | 4 -
be/src/runtime/fragment_mgr.cpp | 191 +++++++++++-
be/src/runtime/fragment_mgr.h | 37 ++-
be/src/runtime/query_context.cpp | 47 ++-
be/src/runtime/query_context.h | 28 +-
be/src/runtime/runtime_predicate.h | 1 +
be/src/runtime/runtime_state.cpp | 44 ++-
be/src/runtime/runtime_state.h | 22 +-
be/src/runtime/task_execution_context.cpp | 32 ++
be/src/runtime/task_execution_context.h | 23 +-
be/src/service/internal_service.cpp | 56 ++++
be/src/service/internal_service.h | 10 +
.../runtime_filter_consumer_test.cpp | 20 +-
.../runtime_filter/runtime_filter_mgr_test.cpp | 12 +-
gensrc/proto/internal_service.proto | 45 +++
gensrc/thrift/PaloInternalService.thrift | 4 +-
gensrc/thrift/PlanNodes.thrift | 30 +-
48 files changed, 2083 insertions(+), 190 deletions(-)
diff --git a/be/src/exec/exchange/vdata_stream_sender.cpp b/be/src/exec/exchange/vdata_stream_sender.cpp
index 213a2aec59c..99dcc4e5772 100644
--- a/be/src/exec/exchange/vdata_stream_sender.cpp
+++ b/be/src/exec/exchange/vdata_stream_sender.cpp
@@ -276,7 +276,6 @@ Status Channel::close(RuntimeState* state) {
return Status::OK();
}
_closed = true;
-
if (!_need_close) {
return Status::OK();
}
diff --git a/be/src/exec/operator/exchange_source_operator.cpp b/be/src/exec/operator/exchange_source_operator.cpp
index 0b54291fcf3..aa28619c587 100644
--- a/be/src/exec/operator/exchange_source_operator.cpp
+++ b/be/src/exec/operator/exchange_source_operator.cpp
@@ -223,4 +223,9 @@ Status ExchangeSourceOperatorX::close(RuntimeState* state) {
_is_closed = true;
return OperatorX<ExchangeLocalState>::close(state);
}
+
+Status ExchangeSourceOperatorX::reset(RuntimeState* state) {
+ auto& local_state = get_local_state(state);
+ return local_state.reset(state);
+}
} // namespace doris
diff --git a/be/src/exec/operator/exchange_source_operator.h b/be/src/exec/operator/exchange_source_operator.h
index 5c273dca7b4..b973e4923bd 100644
--- a/be/src/exec/operator/exchange_source_operator.h
+++ b/be/src/exec/operator/exchange_source_operator.h
@@ -46,6 +46,23 @@ public:
Status close(RuntimeState* state) override;
std::string debug_string(int indentation_level) const override;
+ Status reset(RuntimeState* state) {
+ if (stream_recvr) {
+ stream_recvr->close();
+ }
+ create_stream_recvr(state);
+
+ is_ready = false;
+ num_rows_skipped = 0;
+
+ const auto& queues = stream_recvr->sender_queues();
+ for (size_t i = 0; i < queues.size(); i++) {
+ deps[i]->block();
+ queues[i]->set_dependency(deps[i]);
+ }
+ return Status::OK();
+ }
+
std::vector<Dependency*> dependencies() const override {
std::vector<Dependency*> dep_vec;
std::for_each(deps.begin(), deps.end(),
@@ -81,6 +98,8 @@ public:
Status init(const TPlanNode& tnode, RuntimeState* state) override;
Status prepare(RuntimeState* state) override;
+ Status reset(RuntimeState* state) override;
+
Status get_block(RuntimeState* state, Block* block, bool* eos) override;
std::string debug_string(int indentation_level = 0) const override;
diff --git a/be/src/exec/operator/operator.cpp b/be/src/exec/operator/operator.cpp
index 16cc969276d..6f32bcb2670 100644
--- a/be/src/exec/operator/operator.cpp
+++ b/be/src/exec/operator/operator.cpp
@@ -67,6 +67,10 @@
#include "exec/operator/partitioned_aggregation_source_operator.h"
#include "exec/operator/partitioned_hash_join_probe_operator.h"
#include "exec/operator/partitioned_hash_join_sink_operator.h"
+#include "exec/operator/rec_cte_anchor_sink_operator.h"
+#include "exec/operator/rec_cte_scan_operator.h"
+#include "exec/operator/rec_cte_sink_operator.h"
+#include "exec/operator/rec_cte_source_operator.h"
#include "exec/operator/repeat_operator.h"
#include "exec/operator/result_file_sink_operator.h"
#include "exec/operator/result_sink_operator.h"
@@ -814,6 +818,8 @@ DECLARE_OPERATOR(PartitionedHashJoinSinkLocalState)
DECLARE_OPERATOR(GroupCommitBlockSinkLocalState)
DECLARE_OPERATOR(CacheSinkLocalState)
DECLARE_OPERATOR(DictSinkLocalState)
+DECLARE_OPERATOR(RecCTESinkLocalState)
+DECLARE_OPERATOR(RecCTEAnchorSinkLocalState)
#undef DECLARE_OPERATOR
@@ -847,6 +853,8 @@ DECLARE_OPERATOR(MetaScanLocalState)
DECLARE_OPERATOR(LocalExchangeSourceLocalState)
DECLARE_OPERATOR(PartitionedHashJoinProbeLocalState)
DECLARE_OPERATOR(CacheSourceLocalState)
+DECLARE_OPERATOR(RecCTESourceLocalState)
+DECLARE_OPERATOR(RecCTEScanLocalState)
#ifdef BE_TEST
DECLARE_OPERATOR(MockLocalState)
@@ -882,6 +890,7 @@ template class PipelineXSinkLocalState<SetSharedState>;
template class PipelineXSinkLocalState<LocalExchangeSharedState>;
template class PipelineXSinkLocalState<BasicSharedState>;
template class PipelineXSinkLocalState<DataQueueSharedState>;
+template class PipelineXSinkLocalState<RecCTESharedState>;
template class PipelineXLocalState<HashJoinSharedState>;
template class PipelineXLocalState<PartitionedHashJoinSharedState>;
@@ -899,6 +908,7 @@ template class PipelineXLocalState<PartitionSortNodeSharedState>;
template class PipelineXLocalState<SetSharedState>;
template class PipelineXLocalState<LocalExchangeSharedState>;
template class PipelineXLocalState<BasicSharedState>;
+template class PipelineXLocalState<RecCTESharedState>;
template class AsyncWriterSink<doris::VFileResultWriter, ResultFileSinkOperatorX>;
template class AsyncWriterSink<doris::VJdbcTableWriter, JdbcTableSinkOperatorX>;
diff --git a/be/src/exec/operator/operator.h b/be/src/exec/operator/operator.h
index e354aa26a32..8c880b24331 100644
--- a/be/src/exec/operator/operator.h
+++ b/be/src/exec/operator/operator.h
@@ -29,6 +29,7 @@
#include <vector>
#include "common/be_mock_util.h"
+#include "common/exception.h"
#include "common/logging.h"
#include "common/status.h"
#include "core/block/block.h"
@@ -185,6 +186,10 @@ public:
[[nodiscard]] virtual DataDistribution required_data_distribution(
RuntimeState* /*state*/) const;
+ virtual Status reset(RuntimeState* state) {
+ return Status::InternalError("Reset is not implemented in operator: {}", get_name());
+ }
+
protected:
OperatorPtr _child = nullptr;
@@ -583,6 +588,8 @@ public:
// For agg/sort/join sink.
virtual Status init(const TPlanNode& tnode, RuntimeState* state);
+ virtual bool reset_to_rerun(RuntimeState* state, OperatorXBase* root) const { return false; }
+
Status init(const TDataSink& tsink) override;
[[nodiscard]] virtual Status init(RuntimeState* state, ExchangeType type, const int num_buckets,
const bool use_global_hash_shuffle,
diff --git a/be/src/exec/operator/partitioned_hash_join_sink_operator.cpp b/be/src/exec/operator/partitioned_hash_join_sink_operator.cpp
index c75db33799b..dacceabb59c 100644
--- a/be/src/exec/operator/partitioned_hash_join_sink_operator.cpp
+++ b/be/src/exec/operator/partitioned_hash_join_sink_operator.cpp
@@ -478,6 +478,10 @@ Status PartitionedHashJoinSinkLocalState::_setup_internal_operator(RuntimeState*
/// Set these two values after all the work is ready.
_shared_state->_inner_shared_state = std::move(inner_shared_state);
_shared_state->_inner_runtime_state = std::move(inner_runtime_state);
+ // The inner (spill) runtime state registers its own runtime filters. Merge those IDs
+ // into the parent state so they are tracked for deregistration during recursive CTE rerun.
+ state->merge_register_runtime_filter(
+ _shared_state->_inner_runtime_state->get_deregister_runtime_filter());
return Status::OK();
}
diff --git a/be/src/exec/operator/rec_cte_anchor_sink_operator.cpp b/be/src/exec/operator/rec_cte_anchor_sink_operator.cpp
new file mode 100644
index 00000000000..6d1aeec6b77
--- /dev/null
+++ b/be/src/exec/operator/rec_cte_anchor_sink_operator.cpp
@@ -0,0 +1,56 @@
+// 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 "exec/operator/rec_cte_anchor_sink_operator.h"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+
+Status RecCTEAnchorSinkLocalState::open(RuntimeState* state) {
+ SCOPED_TIMER(exec_time_counter());
+ SCOPED_TIMER(_open_timer);
+ RETURN_IF_ERROR(Base::open(state));
+ auto& p = _parent->cast<Parent>();
+ _child_expr.resize(p._child_expr.size());
+ for (size_t i = 0; i < p._child_expr.size(); i++) {
+ RETURN_IF_ERROR(p._child_expr[i]->clone(state, _child_expr[i]));
+ }
+ return Status::OK();
+}
+
+Status RecCTEAnchorSinkOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
+ RETURN_IF_ERROR(Base::init(tnode, state));
+ DCHECK(tnode.__isset.rec_cte_node);
+ {
+ const auto& texprs = tnode.rec_cte_node.result_expr_lists[0];
+ VExprContextSPtrs ctxs;
+ RETURN_IF_ERROR(VExpr::create_expr_trees(texprs, ctxs));
+ _child_expr = ctxs;
+ }
+ _name = "REC_CTE_ANCHOR_SINK_OPERATOR";
+ return Status::OK();
+}
+
+Status RecCTEAnchorSinkOperatorX::prepare(RuntimeState* state) {
+ RETURN_IF_ERROR(Base::prepare(state));
+ RETURN_IF_ERROR(VExpr::prepare(_child_expr, state, _child->row_desc()));
+ RETURN_IF_ERROR(VExpr::check_expr_output_type(_child_expr, _child->row_desc()));
+ RETURN_IF_ERROR(VExpr::open(_child_expr, state));
+ return Status::OK();
+}
+
+} // namespace doris
diff --git a/be/src/exec/operator/rec_cte_anchor_sink_operator.h b/be/src/exec/operator/rec_cte_anchor_sink_operator.h
new file mode 100644
index 00000000000..19585ff02a8
--- /dev/null
+++ b/be/src/exec/operator/rec_cte_anchor_sink_operator.h
@@ -0,0 +1,127 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+
+#include "common/status.h"
+#include "core/block/block.h"
+#include "exec/operator/union_sink_operator.h"
+#include "exec/pipeline/rec_cte_shared_state.h"
+#include "operator.h"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+class RuntimeState;
+
+class DataQueue;
+
+class RecCTEAnchorSinkOperatorX;
+class RecCTEAnchorSinkLocalState final : public PipelineXSinkLocalState<RecCTESharedState> {
+public:
+ ENABLE_FACTORY_CREATOR(RecCTEAnchorSinkLocalState);
+ RecCTEAnchorSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state)
+ : Base(parent, state) {}
+ Status open(RuntimeState* state) override;
+
+ bool is_blockable() const override { return true; }
+
+private:
+ friend class RecCTEAnchorSinkOperatorX;
+ using Base = PipelineXSinkLocalState<RecCTESharedState>;
+ using Parent = RecCTEAnchorSinkOperatorX;
+
+ VExprContextSPtrs _child_expr;
+};
+
+class RecCTEAnchorSinkOperatorX MOCK_REMOVE(final)
+ : public DataSinkOperatorX<RecCTEAnchorSinkLocalState> {
+public:
+ using Base = DataSinkOperatorX<RecCTEAnchorSinkLocalState>;
+
+ friend class RecCTEAnchorSinkLocalState;
+ RecCTEAnchorSinkOperatorX(int sink_id, int dest_id, const TPlanNode& tnode,
+ const DescriptorTbl& descs)
+ : Base(sink_id, tnode.node_id, dest_id),
+ _row_descriptor(descs, tnode.row_tuples, tnode.nullable_tuples) {}
+
+ ~RecCTEAnchorSinkOperatorX() override = default;
+
+ Status init(const TPlanNode& tnode, RuntimeState* state) override;
+ Status prepare(RuntimeState* state) override;
+
+ bool is_serial_operator() const override { return true; }
+
+ DataDistribution required_data_distribution(RuntimeState* /*state*/) const override {
+ return {ExchangeType::NOOP};
+ }
+
+ Status terminate(RuntimeState* state) override {
+ RETURN_IF_ERROR(_notify_rec_side_ready_if_needed(state));
+ return Base::terminate(state);
+ }
+
+ Status close(RuntimeState* state) override {
+ RETURN_IF_ERROR(_notify_rec_side_ready_if_needed(state));
+ return Base::close(state);
+ }
+
+ Status sink(RuntimeState* state, Block* input_block, bool eos) override {
+ auto& local_state = get_local_state(state);
+
+ RETURN_IF_ERROR(_notify_rec_side_ready_if_needed(state));
+
+ COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)input_block->rows());
+ if (input_block->rows() != 0) {
+ Block block;
+ RETURN_IF_ERROR(materialize_block(local_state._child_expr, input_block, &block, true));
+ RETURN_IF_ERROR(local_state._shared_state->emplace_block(state, std::move(block)));
+ }
+
+ if (eos) {
+ local_state._shared_state->anchor_dep->set_ready();
+ }
+ return Status::OK();
+ }
+
+ std::shared_ptr<BasicSharedState> create_shared_state() const override {
+ std::shared_ptr<BasicSharedState> ss = std::make_shared<RecCTESharedState>();
+ ss->id = operator_id();
+ for (const auto& dest : dests_id()) {
+ ss->related_op_ids.insert(dest);
+ }
+ return ss;
+ }
+
+private:
+ Status _notify_rec_side_ready_if_needed(RuntimeState* state) {
+ if (_need_notify_rec_side_ready) {
+ _need_notify_rec_side_ready = false;
+ RETURN_IF_ERROR(get_local_state(state)._shared_state->send_data_to_targets(state, 0));
+ }
+ return Status::OK();
+ }
+
+ const RowDescriptor _row_descriptor;
+ VExprContextSPtrs _child_expr;
+
+ bool _need_notify_rec_side_ready = true;
+};
+
+#include "common/compile_check_end.h"
+} // namespace doris
\ No newline at end of file
diff --git a/be/src/exec/operator/rec_cte_scan_operator.h b/be/src/exec/operator/rec_cte_scan_operator.h
new file mode 100644
index 00000000000..470f497ef91
--- /dev/null
+++ b/be/src/exec/operator/rec_cte_scan_operator.h
@@ -0,0 +1,86 @@
+// 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.
+
+#pragma once
+
+#include "common/status.h"
+#include "exec/operator/operator.h"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+class RuntimeState;
+
+class RecCTEScanOperatorX;
+class RecCTEScanLocalState final : public PipelineXLocalState<> {
+public:
+ ENABLE_FACTORY_CREATOR(RecCTEScanLocalState);
+
+ RecCTEScanLocalState(RuntimeState* state, OperatorXBase* parent)
+ : PipelineXLocalState<>(state, parent) {
+ _scan_dependency = Dependency::create_shared(_parent->operator_id(), _parent->node_id(),
+ _parent->get_name() + "_DEPENDENCY");
+ state->get_query_ctx()->registe_cte_scan(state->fragment_instance_id(), parent->node_id(),
+ this);
+ }
+ ~RecCTEScanLocalState() override {
+ state()->get_query_ctx()->deregiste_cte_scan(state()->fragment_instance_id(),
+ parent()->node_id());
+ }
+
+ Status add_block(const PBlock& pblock) {
+ Block block;
+ size_t uncompressed_bytes;
+ int64_t decompress_time;
+ RETURN_IF_ERROR(block.deserialize(pblock, &uncompressed_bytes, &decompress_time));
+ _blocks.emplace_back(std::move(block));
+ return Status::OK();
+ }
+
+ void set_ready() { _scan_dependency->set_ready(); }
+
+ std::vector<Dependency*> dependencies() const override { return {_scan_dependency.get()}; }
+
+private:
+ friend class RecCTEScanOperatorX;
+ std::vector<Block> _blocks;
+ DependencySPtr _scan_dependency = nullptr;
+};
+
+class RecCTEScanOperatorX final : public OperatorX<RecCTEScanLocalState> {
+public:
+ RecCTEScanOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id,
+ const DescriptorTbl& descs)
+ : OperatorX<RecCTEScanLocalState>(pool, tnode, operator_id, descs) {}
+
+ Status get_block(RuntimeState* state, Block* block, bool* eos) override {
+ auto& local_state = get_local_state(state);
+
+ if (local_state._blocks.empty()) {
+ *eos = true;
+ return Status::OK();
+ }
+ *block = std::move(local_state._blocks.back());
+ RETURN_IF_ERROR(local_state.filter_block(local_state.conjuncts(), block, block->columns()));
+ local_state._blocks.pop_back();
+ return Status::OK();
+ }
+
+ bool is_source() const override { return true; }
+};
+
+#include "common/compile_check_end.h"
+} // namespace doris
\ No newline at end of file
diff --git a/be/src/exec/operator/rec_cte_sink_operator.cpp b/be/src/exec/operator/rec_cte_sink_operator.cpp
new file mode 100644
index 00000000000..92667b1f310
--- /dev/null
+++ b/be/src/exec/operator/rec_cte_sink_operator.cpp
@@ -0,0 +1,55 @@
+// 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 "exec/operator/rec_cte_sink_operator.h"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+
+Status RecCTESinkLocalState::open(RuntimeState* state) {
+ SCOPED_TIMER(exec_time_counter());
+ SCOPED_TIMER(_open_timer);
+ RETURN_IF_ERROR(Base::open(state));
+ auto& p = _parent->cast<Parent>();
+ _child_expr.resize(p._child_expr.size());
+ for (size_t i = 0; i < p._child_expr.size(); i++) {
+ RETURN_IF_ERROR(p._child_expr[i]->clone(state, _child_expr[i]));
+ }
+ return Status::OK();
+}
+
+Status RecCTESinkOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
+ RETURN_IF_ERROR(Base::init(tnode, state));
+ DCHECK(tnode.__isset.rec_cte_node);
+ {
+ const auto& texprs = tnode.rec_cte_node.result_expr_lists[1];
+ VExprContextSPtrs ctxs;
+ RETURN_IF_ERROR(VExpr::create_expr_trees(texprs, ctxs));
+ _child_expr = ctxs;
+ }
+ return Status::OK();
+}
+
+Status RecCTESinkOperatorX::prepare(RuntimeState* state) {
+ RETURN_IF_ERROR(Base::prepare(state));
+ RETURN_IF_ERROR(VExpr::prepare(_child_expr, state, _child->row_desc()));
+ RETURN_IF_ERROR(VExpr::check_expr_output_type(_child_expr, _child->row_desc()));
+ RETURN_IF_ERROR(VExpr::open(_child_expr, state));
+ return Status::OK();
+}
+
+} // namespace doris
diff --git a/be/src/exec/operator/rec_cte_sink_operator.h b/be/src/exec/operator/rec_cte_sink_operator.h
new file mode 100644
index 00000000000..a8a526854df
--- /dev/null
+++ b/be/src/exec/operator/rec_cte_sink_operator.h
@@ -0,0 +1,103 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+
+#include "common/status.h"
+#include "core/block/block.h"
+#include "exec/operator/union_sink_operator.h"
+#include "exec/pipeline/rec_cte_shared_state.h"
+#include "operator.h"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+class RuntimeState;
+
+class DataQueue;
+
+class RecCTESinkOperatorX;
+class RecCTESinkLocalState final : public PipelineXSinkLocalState<RecCTESharedState> {
+public:
+ ENABLE_FACTORY_CREATOR(RecCTESinkLocalState);
+ RecCTESinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state)
+ : Base(parent, state) {}
+ Status open(RuntimeState* state) override;
+
+private:
+ friend class RecCTESinkOperatorX;
+ using Base = PipelineXSinkLocalState<RecCTESharedState>;
+ using Parent = RecCTESinkOperatorX;
+
+ VExprContextSPtrs _child_expr;
+};
+
+class RecCTESinkOperatorX MOCK_REMOVE(final) : public DataSinkOperatorX<RecCTESinkLocalState> {
+public:
+ using Base = DataSinkOperatorX<RecCTESinkLocalState>;
+
+ friend class RecCTESinkLocalState;
+ RecCTESinkOperatorX(int sink_id, int dest_id, const TPlanNode& tnode,
+ const DescriptorTbl& descs)
+ : Base(sink_id, tnode.node_id, dest_id),
+ _row_descriptor(descs, tnode.row_tuples, tnode.nullable_tuples) {}
+
+ ~RecCTESinkOperatorX() override = default;
+
+ Status init(const TPlanNode& tnode, RuntimeState* state) override;
+ Status prepare(RuntimeState* state) override;
+
+ bool reset_to_rerun(RuntimeState* state, OperatorXBase* root) const override {
+ auto* shared_state = get_local_state(state)._shared_state;
+ if (shared_state->ready_to_return == false) {
+ THROW_IF_ERROR(root->reset(state));
+ // must set_ready after root(exchange) reset
+ // if next round executed before exchange source reset, it maybe cant find receiver and cause blocked forever
+ shared_state->source_dep->set_ready();
+ return true;
+ }
+ return false;
+ }
+
+ std::shared_ptr<BasicSharedState> create_shared_state() const override { return nullptr; }
+
+ bool is_serial_operator() const override { return true; }
+
+ DataDistribution required_data_distribution(RuntimeState* /*state*/) const override {
+ return {ExchangeType::NOOP};
+ }
+
+ Status sink(RuntimeState* state, Block* input_block, bool eos) override {
+ auto& local_state = get_local_state(state);
+
+ COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)input_block->rows());
+ if (input_block->rows() != 0) {
+ Block block;
+ RETURN_IF_ERROR(materialize_block(local_state._child_expr, input_block, &block, true));
+ RETURN_IF_ERROR(local_state._shared_state->emplace_block(state, std::move(block)));
+ }
+ return Status::OK();
+ }
+
+private:
+ const RowDescriptor _row_descriptor;
+ VExprContextSPtrs _child_expr;
+};
+
+#include "common/compile_check_end.h"
+} // namespace doris
\ No newline at end of file
diff --git a/be/src/exec/operator/rec_cte_source_operator.cpp b/be/src/exec/operator/rec_cte_source_operator.cpp
new file mode 100644
index 00000000000..20026428512
--- /dev/null
+++ b/be/src/exec/operator/rec_cte_source_operator.cpp
@@ -0,0 +1,85 @@
+// 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 "exec/operator/rec_cte_source_operator.h"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+
+RecCTESourceLocalState::RecCTESourceLocalState(RuntimeState* state, OperatorXBase* parent)
+ : Base(state, parent) {}
+
+Status RecCTESourceLocalState::open(RuntimeState* state) {
+ SCOPED_TIMER(exec_time_counter());
+ SCOPED_TIMER(_open_timer);
+ RETURN_IF_ERROR(Base::open(state));
+ return Status::OK();
+}
+
+Status RecCTESourceLocalState::init(RuntimeState* state, LocalStateInfo& info) {
+ RETURN_IF_ERROR(Base::init(state, info));
+ _shared_state->targets = _parent->cast<RecCTESourceOperatorX>()._targets;
+ _shared_state->max_recursion_depth =
+ _parent->cast<RecCTESourceOperatorX>()._max_recursion_depth;
+ _shared_state->source_dep = _dependency;
+ _anchor_dependency = Dependency::create_shared(_parent->operator_id(), _parent->node_id(),
+ _parent->get_name() + "_ANCHOR_DEPENDENCY");
+ _shared_state->anchor_dep = _anchor_dependency.get();
+
+ auto& p = _parent->cast<Parent>();
+ if (!_parent->cast<RecCTESourceOperatorX>()._is_union_all) {
+ _shared_state->agg_data = std::make_unique<DistinctDataVariants>();
+ RETURN_IF_ERROR(init_hash_method<DistinctDataVariants>(_shared_state->agg_data.get(),
+ p._hash_table_key_types, false));
+ }
+
+ _shared_state->hash_table_compute_timer =
+ ADD_TIMER(Base::custom_profile(), "HashTableComputeTime");
+ _shared_state->hash_table_emplace_timer =
+ ADD_TIMER(Base::custom_profile(), "HashTableEmplaceTime");
+ _shared_state->hash_table_input_counter =
+ ADD_COUNTER(Base::custom_profile(), "HashTableInputCount", TUnit::UNIT);
+ return Status::OK();
+}
+
+Status RecCTESourceOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
+ RETURN_IF_ERROR(Base::init(tnode, state));
+ DCHECK(tnode.__isset.rec_cte_node);
+
+ _max_recursion_depth = state->cte_max_recursion_depth();
+
+ {
+ const auto& texprs = tnode.rec_cte_node.result_expr_lists[1];
+ VExprContextSPtrs ctxs;
+ RETURN_IF_ERROR(VExpr::create_expr_trees(texprs, ctxs));
+ _hash_table_key_types = get_data_types(ctxs);
+ }
+ return Status::OK();
+}
+
+Status RecCTESourceOperatorX::prepare(RuntimeState* state) {
+ RETURN_IF_ERROR(Base::prepare(state));
+ for (auto fragment : _fragments_to_reset) {
+ if (state->fragment_id() == fragment.fragment_id) {
+ return Status::InternalError("Fragment {} contains a recursive CTE node",
+ fragment.fragment_id);
+ }
+ }
+ return Status::OK();
+}
+
+} // namespace doris
diff --git a/be/src/exec/operator/rec_cte_source_operator.h b/be/src/exec/operator/rec_cte_source_operator.h
new file mode 100644
index 00000000000..92ffe43794c
--- /dev/null
+++ b/be/src/exec/operator/rec_cte_source_operator.h
@@ -0,0 +1,330 @@
+// 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.
+#pragma once
+
+#include <gen_cpp/internal_service.pb.h>
+
+#include "common/status.h"
+#include "core/block/block.h"
+#include "exec/operator/union_sink_operator.h"
+#include "exec/pipeline/rec_cte_shared_state.h"
+#include "operator.h"
+#include "util/brpc_client_cache.h"
+#include "util/uid_util.h"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+class RuntimeState;
+
+class DataQueue;
+
+class RecCTESourceOperatorX;
+class RecCTESourceLocalState final : public PipelineXLocalState<RecCTESharedState> {
+public:
+ ENABLE_FACTORY_CREATOR(RecCTESourceLocalState);
+ using Base = PipelineXLocalState<RecCTESharedState>;
+ using Parent = RecCTESourceOperatorX;
+ RecCTESourceLocalState(RuntimeState* state, OperatorXBase* parent);
+
+ Status open(RuntimeState* state) override;
+ Status init(RuntimeState* state, LocalStateInfo& info) override;
+
+ bool is_blockable() const override { return true; }
+
+ std::vector<Dependency*> dependencies() const override {
+ return std::vector<Dependency*> {_dependency, _anchor_dependency.get()};
+ }
+
+private:
+ friend class RecCTESourceOperatorX;
+ friend class OperatorX<RecCTESourceLocalState>;
+
+ std::shared_ptr<Dependency> _anchor_dependency = nullptr;
+};
+
+// RecCTESourceOperatorX drives the recursive CTE fragment rerun lifecycle.
+//
+// It orchestrates child fragment destruction and recreation for each recursion round
+// via the rerun_fragment RPC (defined in internal_service.proto). The 4 opcodes are:
+//
+// wait_for_destroy (=1) : notify old PFC to close, async wait for tasks to finish
+// rebuild (=2) : increment stage, deregister old runtime filters, create new PFC
+// submit (=3) : submit new PFC pipeline tasks for execution
+// final_close (=4) : last round cleanup, send final report, destroy fragment
+//
+// State Transition Diagram
+// ========================
+//
+// ┌────────────────────────────────────┐
+// │ Query Start (FE) │
+// │ FE sets need_notify_close = true │
+// │ on recursive-side child fragments │
+// └──────────────┬─────────────────────┘
+// │
+// ▼
+// ┌────────────────────────────────────┐
+// │ Initial Registration (BE) │
+// │ FragmentMgr saves params in │
+// │ _rerunnable_params_map │
+// │ PFC created, prepared & submitted │
+// └──────────────┬─────────────────────┘
+// │
+// ▼
+// ┌────────────────────────────────────┐
+// │ Fragment Running │
+// │ PFC executing pipeline tasks │◄─────────────────────────────┐
+// │ _need_notify_close = true │ │
+// └──────────────┬─────────────────────┘ │
+// │ │
+// │ tasks complete, but PFC does NOT │
+// │ self-remove (blocked by _need_notify_close) │
+// ▼ │
+// ┌────────────────────────────────────┐ │
+// │ RecCTESourceOperatorX::get_block()│ │
+// │ ready_to_return? round < max? │ │
+// └───────┬─────────────────┬──────────┘ │
+// │ │ │
+// more rounds remain no more rounds │
+// │ │ │
+// ▼ │ │
+// ┌──────────────────┐ │ │
+// │_recursive_process│ │ │
+// └───────┬──────────┘ │ │
+// │ │ │
+// ▼ │ │
+// ┌────────────────────────────────────┐ │
+// │ Step 1: wait_for_destroy (=1) │ │
+// │ • collect deregister RF IDs │ │
+// │ • store brpc closure guard │ │
+// │ • notify_close() on old PFC │ │
+// │ → _need_notify_close = false │ │
+// │ → old PFC begins shutdown │ │
+// └──────────────┬─────────────────────┘ │
+// │ │
+// ▼ │
+// ┌────────────────────────────────────┐ │
+// │ Step 1.5: reset_global_rf │ │
+// │ • reset global runtime filters │ │
+// │ on merge coordinator │ │
+// └──────────────┬─────────────────────┘ │
+// │ │
+// ▼ │
+// ┌────────────────────────────────────┐ │
+// │ Step 2: rebuild (=2) │ │
+// │ • increment recursion stage │ │
+// │ • deregister old runtime filters │ │
+// │ • create NEW PFC from saved params│ │
+// │ • prepare() new PFC │ │
+// │ • insert into _pipeline_map │ │
+// └──────────────┬─────────────────────┘ │
+// │ │
+// ▼ │
+// ┌────────────────────────────────────┐ │
+// │ Step 3: submit (=3) │ │
+// │ • find new PFC in pipeline_map │ │
+// │ • call fragment_ctx->submit() │ │
+// │ • pipeline tasks start running │ │
+// └──────────────┬─────────────────────┘ │
+// │ │
+// ▼ │
+// ┌────────────────────────────────────┐ │
+// │ Step 4: send_data_to_targets │ │
+// │ • transmit_rec_cte_block() to │ │
+// │ RecursiveCteScanNode targets │ │
+// │ • new recursion round begins │ │
+// └──────────────┬─────────────────────┘ │
+// │ │
+// └──── loop back to Fragment Running ─────────────────┘
+//
+// (when no more rounds remain)
+// │
+// ▼
+// ┌────────────────────────────────────┐
+// │ final_close (=4) │
+// │ • listen_wait_close(guard, │
+// │ need_send_report = true) │
+// │ • notify_close() on PFC │
+// │ • send final status report │
+// │ • clean up completely │
+// └──────────────┬─────────────────────┘
+// │
+// ▼
+// ┌────────────────────────────────────┐
+// │ Fragment Destroyed │
+// │ rerunnable_params removed on │
+// │ query end │
+// └────────────────────────────────────┘
+//
+class RecCTESourceOperatorX : public OperatorX<RecCTESourceLocalState> {
+public:
+ using Base = OperatorX<RecCTESourceLocalState>;
+ RecCTESourceOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id,
+ const DescriptorTbl& descs)
+ : Base(pool, tnode, operator_id, descs),
+ _is_union_all(tnode.rec_cte_node.is_union_all),
+ _targets(tnode.rec_cte_node.targets),
+ _fragments_to_reset(tnode.rec_cte_node.fragments_to_reset),
+ _global_rf_ids(tnode.rec_cte_node.rec_side_runtime_filter_ids),
+ _is_used_by_other_rec_cte(tnode.rec_cte_node.is_used_by_other_rec_cte) {
+ DCHECK(tnode.__isset.rec_cte_node);
+ }
+ ~RecCTESourceOperatorX() override = default;
+
+ Status init(const TPlanNode& tnode, RuntimeState* state) override;
+ Status prepare(RuntimeState* state) override;
+
+ Status terminate(RuntimeState* state) override {
+ RETURN_IF_ERROR(_send_close(state));
+ return Base::terminate(state);
+ }
+
+ Status close(RuntimeState* state) override {
+ RETURN_IF_ERROR(_send_close(state));
+ return Base::close(state);
+ }
+
+ Status set_child(OperatorPtr child) override {
+ Base::_child = child;
+ return Status::OK();
+ }
+
+ bool is_serial_operator() const override { return true; }
+
+ DataDistribution required_data_distribution(RuntimeState* /*state*/) const override {
+ return {ExchangeType::NOOP};
+ }
+
+ Status get_block(RuntimeState* state, Block* block, bool* eos) override {
+ auto& local_state = get_local_state(state);
+ auto& ctx = local_state._shared_state;
+ ctx->update_ready_to_return();
+
+ if (!ctx->ready_to_return) {
+ if (ctx->current_round + 1 > _max_recursion_depth) {
+ return Status::Aborted("reach cte_max_recursion_depth {}", _max_recursion_depth);
+ }
+
+ ctx->source_dep->block();
+ // ctx->blocks.size() may be changed after _recursive_process
+ int current_blocks_size = int(ctx->blocks.size());
+ RETURN_IF_ERROR(_recursive_process(state, ctx->last_round_offset));
+ ctx->current_round++;
+ ctx->last_round_offset = current_blocks_size;
+ } else {
+ if (ctx->blocks.empty()) {
+ *eos = true;
+ } else {
+ block->swap(ctx->blocks.back());
+ RETURN_IF_ERROR(
+ local_state.filter_block(local_state.conjuncts(), block, block->columns()));
+ ctx->blocks.pop_back();
+ }
+ }
+ return Status::OK();
+ }
+
+ bool is_source() const override { return true; }
+
+private:
+ Status _send_close(RuntimeState* state) {
+ if (!_already_send_close && !_is_used_by_other_rec_cte) {
+ _already_send_close = true;
+ auto* round_counter = ADD_COUNTER(get_local_state(state).Base::custom_profile(),
+ "RecursiveRound", TUnit::UNIT);
+ round_counter->set(int64_t(get_local_state(state)._shared_state->current_round));
+
+ RETURN_IF_ERROR(_send_rerun_fragments(state, PRerunFragmentParams::FINAL_CLOSE));
+ }
+ return Status::OK();
+ }
+
+ Status _recursive_process(RuntimeState* state, size_t last_round_offset) const {
+ RETURN_IF_ERROR(_send_rerun_fragments(state, PRerunFragmentParams::WAIT_FOR_DESTROY));
+ RETURN_IF_ERROR(_send_reset_global_rf(state));
+ RETURN_IF_ERROR(_send_rerun_fragments(state, PRerunFragmentParams::REBUILD));
+ RETURN_IF_ERROR(_send_rerun_fragments(state, PRerunFragmentParams::SUBMIT));
+ RETURN_IF_ERROR(get_local_state(state)._shared_state->send_data_to_targets(
+ state, last_round_offset));
+ return Status::OK();
+ }
+
+ Status _send_reset_global_rf(RuntimeState* state) const {
+ TNetworkAddress addr;
+ RETURN_IF_ERROR(state->global_runtime_filter_mgr()->get_merge_addr(&addr));
+ auto stub =
+ state->get_query_ctx()->exec_env()->brpc_internal_client_cache()->get_client(addr);
+ PResetGlobalRfParams request;
+ request.mutable_query_id()->CopyFrom(UniqueId(state->query_id()).to_proto());
+ for (auto filter_id : _global_rf_ids) {
+ request.add_filter_ids(filter_id);
+ }
+
+ PResetGlobalRfResult result;
+ brpc::Controller controller;
+ controller.set_timeout_ms(
+ get_execution_rpc_timeout_ms(state->get_query_ctx()->execution_timeout()));
+ stub->reset_global_rf(&controller, &request, &result, brpc::DoNothing());
+ brpc::Join(controller.call_id());
+ return Status::create(result.status());
+ }
+
+ Status _send_rerun_fragments(RuntimeState* state, PRerunFragmentParams_Opcode stage) const {
+ for (auto fragment : _fragments_to_reset) {
+ auto stub =
+ state->get_query_ctx()->exec_env()->brpc_internal_client_cache()->get_client(
+ fragment.addr);
+
+ PRerunFragmentParams request;
+ request.mutable_query_id()->CopyFrom(UniqueId(state->query_id()).to_proto());
+ request.set_fragment_id(fragment.fragment_id);
+ request.set_stage(stage);
+
+ PRerunFragmentResult result;
+ brpc::Controller controller;
+ controller.set_timeout_ms(
+ get_execution_rpc_timeout_ms(state->get_query_ctx()->execution_timeout()));
+ stub->rerun_fragment(&controller, &request, &result, brpc::DoNothing());
+ brpc::Join(controller.call_id());
+ if (controller.Failed()) {
+ return Status::InternalError(controller.ErrorText());
+ }
+
+ auto rpc_st = Status::create(result.status());
+ if (!rpc_st.ok()) {
+ return rpc_st;
+ }
+ }
+ return Status::OK();
+ }
+
+ friend class RecCTESourceLocalState;
+ std::vector<DataTypePtr> _hash_table_key_types;
+
+ const bool _is_union_all = false;
+ std::vector<TRecCTETarget> _targets;
+ std::vector<TRecCTEResetInfo> _fragments_to_reset;
+ std::vector<int> _global_rf_ids;
+
+ int _max_recursion_depth = 0;
+
+ bool _already_send_close = false;
+
+ bool _is_used_by_other_rec_cte = false;
+};
+
+#include "common/compile_check_end.h"
+} // namespace doris
\ No newline at end of file
diff --git a/be/src/exec/operator/scan_operator.cpp b/be/src/exec/operator/scan_operator.cpp
index 3864fbf63fc..06ef4557060 100644
--- a/be/src/exec/operator/scan_operator.cpp
+++ b/be/src/exec/operator/scan_operator.cpp
@@ -192,9 +192,9 @@ Status ScanLocalState<Derived>::open(RuntimeState* state) {
auto status = _eos ? Status::OK() : _prepare_scanners();
RETURN_IF_ERROR(status);
- if (_scanner_ctx) {
+ if (auto ctx = _scanner_ctx.load()) {
DCHECK(!_eos && _num_scanners->value() > 0);
- RETURN_IF_ERROR(_scanner_ctx->init());
+ RETURN_IF_ERROR(ctx->init());
}
_opened = true;
return status;
@@ -633,9 +633,9 @@ std::string ScanLocalState<Derived>::debug_string(int indentation_level) const {
fmt::format_to(debug_string_buffer, "{}, _eos = {} , _opened = {}",
PipelineXLocalState<>::debug_string(indentation_level), _eos.load(),
_opened.load());
- if (_scanner_ctx) {
+ if (auto ctx = _scanner_ctx.load()) {
fmt::format_to(debug_string_buffer, "");
- fmt::format_to(debug_string_buffer, ", Scanner Context: {}", _scanner_ctx->debug_string());
+ fmt::format_to(debug_string_buffer, ", Scanner Context: {}", ctx->debug_string());
} else {
fmt::format_to(debug_string_buffer, "");
fmt::format_to(debug_string_buffer, ", Scanner Context: NULL");
@@ -1280,8 +1280,11 @@ Status ScanLocalState<Derived>::close(RuntimeState* state) {
SCOPED_TIMER(_close_timer);
SCOPED_TIMER(exec_time_counter());
- if (_scanner_ctx) {
- _scanner_ctx->stop_scanners(state);
+ if (auto ctx = _scanner_ctx.load()) {
+ ctx->stop_scanners(state);
+ // _scanner_ctx may be accessed in debug_string concurrently
+ // so use atomic shared ptr to avoid use after free
+ _scanner_ctx.store(nullptr);
}
std::list<std::shared_ptr<ScannerDelegate>> {}.swap(_scanners);
COUNTER_SET(_wait_for_dependency_timer, _scan_dependency->watcher_elapse_time());
@@ -1296,8 +1299,8 @@ Status ScanOperatorX<LocalStateType>::get_block(RuntimeState* state, Block* bloc
SCOPED_TIMER(local_state.exec_time_counter());
if (state->is_cancelled()) {
- if (local_state._scanner_ctx) {
- local_state._scanner_ctx->stop_scanners(state);
+ if (auto ctx = local_state._scanner_ctx.load()) {
+ ctx->stop_scanners(state);
}
return state->cancel_reason();
}
@@ -1307,13 +1310,15 @@ Status ScanOperatorX<LocalStateType>::get_block(RuntimeState* state, Block* bloc
return Status::OK();
}
- DCHECK(local_state._scanner_ctx != nullptr);
- RETURN_IF_ERROR(local_state._scanner_ctx->get_block_from_queue(state, block, eos, 0));
+ auto ctx = local_state._scanner_ctx.load();
+
+ DCHECK(ctx != nullptr);
+ RETURN_IF_ERROR(ctx->get_block_from_queue(state, block, eos, 0));
local_state.reached_limit(block, eos);
if (*eos) {
// reach limit, stop the scanners.
- local_state._scanner_ctx->stop_scanners(state);
+ ctx->stop_scanners(state);
local_state._scanner_profile->add_info_string("EOS", "True");
}
@@ -1323,16 +1328,16 @@ Status ScanOperatorX<LocalStateType>::get_block(RuntimeState* state, Block* bloc
template <typename LocalStateType>
size_t ScanOperatorX<LocalStateType>::get_reserve_mem_size(RuntimeState* state) {
auto& local_state = get_local_state(state);
- if (!local_state._opened || local_state._closed || !local_state._scanner_ctx) {
+ auto ctx = local_state._scanner_ctx.load();
+ if (!local_state._opened || local_state._closed || !ctx) {
return config::doris_scanner_row_bytes;
}
if (local_state.low_memory_mode()) {
- return local_state._scanner_ctx->low_memory_mode_scan_bytes_per_scanner() *
- local_state._scanner_ctx->low_memory_mode_scanners();
+ return ctx->low_memory_mode_scan_bytes_per_scanner() * ctx->low_memory_mode_scanners();
} else {
const auto peak_usage = local_state._memory_used_counter->value();
- const auto block_usage = local_state._scanner_ctx->block_memory_usage();
+ const auto block_usage = ctx->block_memory_usage();
if (peak_usage > 0) {
// It is only a safty check, to avoid some counter not right.
if (peak_usage > block_usage) {
diff --git a/be/src/exec/operator/scan_operator.h b/be/src/exec/operator/scan_operator.h
index 91815d8c64f..46297a71547 100644
--- a/be/src/exec/operator/scan_operator.h
+++ b/be/src/exec/operator/scan_operator.h
@@ -308,7 +308,7 @@ protected:
VExprContextSPtrs _stale_expr_ctxs;
VExprContextSPtrs _common_expr_ctxs_push_down;
- std::shared_ptr<ScannerContext> _scanner_ctx = nullptr;
+ atomic_shared_ptr<ScannerContext> _scanner_ctx = nullptr;
// Save all function predicates which may be pushed down to data source.
std::vector<FunctionFilter> _push_down_functions;
@@ -371,8 +371,8 @@ public:
void set_low_memory_mode(RuntimeState* state) override {
auto& local_state = get_local_state(state);
- if (local_state._scanner_ctx) {
- local_state._scanner_ctx->clear_free_blocks();
+ if (auto ctx = local_state._scanner_ctx.load()) {
+ ctx->clear_free_blocks();
}
}
diff --git a/be/src/exec/operator/sort_sink_operator.cpp b/be/src/exec/operator/sort_sink_operator.cpp
index 9079a283f95..d64f6a8047e 100644
--- a/be/src/exec/operator/sort_sink_operator.cpp
+++ b/be/src/exec/operator/sort_sink_operator.cpp
@@ -185,9 +185,10 @@ Status SortSinkOperatorX::merge_sort_read_for_spill(RuntimeState* state, doris::
return local_state._shared_state->sorter->merge_sort_read_for_spill(state, block, batch_size,
eos);
}
-void SortSinkOperatorX::reset(RuntimeState* state) {
+Status SortSinkOperatorX::reset(RuntimeState* state) {
auto& local_state = get_local_state(state);
local_state._shared_state->sorter->reset();
+ return Status::OK();
}
#include "common/compile_check_end.h"
} // namespace doris
diff --git a/be/src/exec/operator/sort_sink_operator.h b/be/src/exec/operator/sort_sink_operator.h
index f1b05a08c0d..d70fba0866f 100644
--- a/be/src/exec/operator/sort_sink_operator.h
+++ b/be/src/exec/operator/sort_sink_operator.h
@@ -100,7 +100,7 @@ public:
Status merge_sort_read_for_spill(RuntimeState* state, doris::Block* block, int batch_size,
bool* eos);
- void reset(RuntimeState* state);
+ Status reset(RuntimeState* state) override;
int64_t limit() const { return _limit; }
int64_t offset() const { return _offset; }
diff --git a/be/src/exec/operator/spill_sort_sink_operator.cpp b/be/src/exec/operator/spill_sort_sink_operator.cpp
index 3baec697de2..1e88f1e3850 100644
--- a/be/src/exec/operator/spill_sort_sink_operator.cpp
+++ b/be/src/exec/operator/spill_sort_sink_operator.cpp
@@ -214,9 +214,9 @@ Status SpillSortSinkLocalState::_execute_spill_sort(RuntimeState* state) {
RETURN_IF_ERROR(_spilling_writer->write_block(state, block));
block.clear_column_data();
}
+ RETURN_IF_ERROR(parent._sort_sink_operator->reset(_runtime_state.get()));
RETURN_IF_ERROR(_spilling_writer->close());
- parent._sort_sink_operator->reset(_runtime_state.get());
return Status::OK();
}
diff --git a/be/src/exec/operator/union_sink_operator.h b/be/src/exec/operator/union_sink_operator.h
index 7d0b3782b69..cec47aa126d 100644
--- a/be/src/exec/operator/union_sink_operator.h
+++ b/be/src/exec/operator/union_sink_operator.h
@@ -28,6 +28,24 @@
namespace doris {
#include "common/compile_check_begin.h"
+
+// Free function used by rec_cte sink operators to materialize a block through expressions.
+inline Status materialize_block(const VExprContextSPtrs& exprs, Block* src_block, Block* res_block,
+ bool need_clone) {
+ ColumnsWithTypeAndName columns;
+ auto rows = src_block->rows();
+ for (const auto& expr : exprs) {
+ ColumnWithTypeAndName result_data;
+ RETURN_IF_ERROR(expr->execute(src_block, result_data));
+ if (need_clone) {
+ result_data.column = result_data.column->clone_resized(rows);
+ }
+ columns.emplace_back(result_data);
+ }
+ *res_block = {columns};
+ return Status::OK();
+}
+
class RuntimeState;
class DataQueue;
diff --git a/be/src/exec/pipeline/pipeline.h b/be/src/exec/pipeline/pipeline.h
index e85c373c411..75e801ddc82 100644
--- a/be/src/exec/pipeline/pipeline.h
+++ b/be/src/exec/pipeline/pipeline.h
@@ -133,7 +133,7 @@ public:
fmt::format_to(debug_string_buffer, "\n{}", _operators[i]->debug_string(i));
}
fmt::format_to(debug_string_buffer, "\n{}",
- _sink->debug_string(cast_set<int>(_operators.size())));
+ _sink ? _sink->debug_string(cast_set<int>(_operators.size())) : "null");
return fmt::to_string(debug_string_buffer);
}
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp b/be/src/exec/pipeline/pipeline_fragment_context.cpp
index dc08bbbd989..b603b482e18 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp
@@ -87,6 +87,10 @@
#include "exec/operator/partitioned_aggregation_source_operator.h"
#include "exec/operator/partitioned_hash_join_probe_operator.h"
#include "exec/operator/partitioned_hash_join_sink_operator.h"
+#include "exec/operator/rec_cte_anchor_sink_operator.h"
+#include "exec/operator/rec_cte_scan_operator.h"
+#include "exec/operator/rec_cte_sink_operator.h"
+#include "exec/operator/rec_cte_source_operator.h"
#include "exec/operator/repeat_operator.h"
#include "exec/operator/result_file_sink_operator.h"
#include "exec/operator/result_sink_operator.h"
@@ -139,7 +143,9 @@ PipelineFragmentContext::PipelineFragmentContext(
_is_report_on_cancel(true),
_report_status_cb(std::move(report_status_cb)),
_params(request),
- _parallel_instances(_params.__isset.parallel_instances ? _params.parallel_instances : 0) {
+ _parallel_instances(_params.__isset.parallel_instances ? _params.parallel_instances : 0),
+ _need_notify_close(request.__isset.need_notify_close ? request.need_notify_close
+ : false) {
_fragment_watcher.start();
}
@@ -147,24 +153,13 @@ PipelineFragmentContext::~PipelineFragmentContext() {
LOG_INFO("PipelineFragmentContext::~PipelineFragmentContext")
.tag("query_id", print_id(_query_id))
.tag("fragment_id", _fragment_id);
- // The memory released by the query end is recorded in the query mem tracker.
- SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_query_ctx->query_mem_tracker());
- auto st = _query_ctx->exec_status();
- for (size_t i = 0; i < _tasks.size(); i++) {
- if (!_tasks[i].empty()) {
- _call_back(_tasks[i].front().first->runtime_state(), &st);
- }
+ _release_resource();
+ {
+ // The memory released by the query end is recorded in the query mem tracker.
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_query_ctx->query_mem_tracker());
+ _runtime_state.reset();
+ _query_ctx.reset();
}
- _tasks.clear();
- _dag.clear();
- _pip_id_to_pipeline.clear();
- _pipelines.clear();
- _sink.reset();
- _root_op.reset();
- _runtime_state.reset();
- _runtime_filter_mgr_map.clear();
- _op_id_to_shared_state.clear();
- _query_ctx.reset();
}
bool PipelineFragmentContext::is_timeout(timespec now) const {
@@ -174,6 +169,35 @@ bool PipelineFragmentContext::is_timeout(timespec now) const {
return _fragment_watcher.elapsed_time_seconds(now) > _timeout;
}
+// notify_close() transitions the PFC from "waiting for external close notification" to
+// "self-managed close". For recursive CTE fragments, the old PFC is kept alive until
+// the rerun_fragment(wait_for_destroy) RPC calls this to trigger shutdown.
+// Returns true if all tasks have already closed (i.e., the PFC can be safely destroyed).
+bool PipelineFragmentContext::notify_close() {
+ bool all_closed = false;
+ bool need_remove = false;
+ {
+ std::lock_guard<std::mutex> l(_task_mutex);
+ if (_closed_tasks >= _total_tasks) {
+ if (_need_notify_close) {
+ // Fragment was cancelled and waiting for notify to close.
+ // Record that we need to remove from fragment mgr, but do it
+ // after releasing _task_mutex to avoid ABBA deadlock with
+ // dump_pipeline_tasks() (which acquires _pipeline_map lock
+ // first, then _task_mutex via debug_string()).
+ need_remove = true;
+ }
+ all_closed = true;
+ }
+ // make fragment release by self after cancel
+ _need_notify_close = false;
+ }
+ if (need_remove) {
+ _exec_env->fragment_mgr()->remove_pipeline_context({_query_id, _fragment_id});
+ }
+ return all_closed;
+}
+
// Must not add lock in this method. Because it will call query ctx cancel. And
// QueryCtx cancel will call fragment ctx cancel. And Also Fragment ctx's running
// Method like exchange sink buffer will call query ctx cancel. If we add lock here
@@ -183,12 +207,8 @@ void PipelineFragmentContext::cancel(const Status reason) {
.tag("query_id", print_id(_query_id))
.tag("fragment_id", _fragment_id)
.tag("reason", reason.to_string());
- {
- std::lock_guard<std::mutex> l(_task_mutex);
- if (_closed_tasks >= _total_tasks) {
- // All tasks in this PipelineXFragmentContext already closed.
- return;
- }
+ if (notify_close()) {
+ return;
}
// Timeout is a special error code, we need print current stack to debug timeout issue.
if (reason.is<ErrorCode::TIMEOUT>()) {
@@ -258,6 +278,54 @@ PipelinePtr PipelineFragmentContext::add_pipeline(PipelinePtr parent, int idx) {
return pipeline;
}
+Status PipelineFragmentContext::_build_and_prepare_full_pipeline(ThreadPool* thread_pool) {
+ {
+ SCOPED_TIMER(_build_pipelines_timer);
+ // 2. Build pipelines with operators in this fragment.
+ auto root_pipeline = add_pipeline();
+ RETURN_IF_ERROR(_build_pipelines(_runtime_state->obj_pool(), *_query_ctx->desc_tbl,
+ &_root_op, root_pipeline));
+
+ // 3. Create sink operator
+ if (!_params.fragment.__isset.output_sink) {
+ return Status::InternalError("No output sink in this fragment!");
+ }
+ RETURN_IF_ERROR(_create_data_sink(_runtime_state->obj_pool(), _params.fragment.output_sink,
+ _params.fragment.output_exprs, _params,
+ root_pipeline->output_row_desc(), _runtime_state.get(),
+ *_desc_tbl, root_pipeline->id()));
+ RETURN_IF_ERROR(_sink->init(_params.fragment.output_sink));
+ RETURN_IF_ERROR(root_pipeline->set_sink(_sink));
+
+ for (PipelinePtr& pipeline : _pipelines) {
+ DCHECK(pipeline->sink() != nullptr) << pipeline->operators().size();
+ RETURN_IF_ERROR(pipeline->sink()->set_child(pipeline->operators().back()));
+ }
+ }
+ // 4. Build local exchanger
+ if (_runtime_state->enable_local_shuffle()) {
+ SCOPED_TIMER(_plan_local_exchanger_timer);
+ RETURN_IF_ERROR(_plan_local_exchange(_params.num_buckets,
+ _params.bucket_seq_to_instance_idx,
+ _params.shuffle_idx_to_instance_idx));
+ }
+
+ // 5. Initialize global states in pipelines.
+ for (PipelinePtr& pipeline : _pipelines) {
+ SCOPED_TIMER(_prepare_all_pipelines_timer);
+ pipeline->children().clear();
+ RETURN_IF_ERROR(pipeline->prepare(_runtime_state.get()));
+ }
+
+ {
+ SCOPED_TIMER(_build_tasks_timer);
+ // 6. Build pipeline tasks and initialize local state.
+ RETURN_IF_ERROR(_build_pipeline_tasks(thread_pool));
+ }
+
+ return Status::OK();
+}
+
Status PipelineFragmentContext::prepare(ThreadPool* thread_pool) {
if (_prepared) {
return Status::InternalError("Already prepared");
@@ -327,49 +395,7 @@ Status PipelineFragmentContext::prepare(ThreadPool* thread_pool) {
}
}
- {
- SCOPED_TIMER(_build_pipelines_timer);
- // 2. Build pipelines with operators in this fragment.
- auto root_pipeline = add_pipeline();
- RETURN_IF_ERROR(_build_pipelines(_runtime_state->obj_pool(), *_query_ctx->desc_tbl,
- &_root_op, root_pipeline));
-
- // 3. Create sink operator
- if (!_params.fragment.__isset.output_sink) {
- return Status::InternalError("No output sink in this fragment!");
- }
- RETURN_IF_ERROR(_create_data_sink(_runtime_state->obj_pool(), _params.fragment.output_sink,
- _params.fragment.output_exprs, _params,
- root_pipeline->output_row_desc(), _runtime_state.get(),
- *_desc_tbl, root_pipeline->id()));
- RETURN_IF_ERROR(_sink->init(_params.fragment.output_sink));
- RETURN_IF_ERROR(root_pipeline->set_sink(_sink));
-
- for (PipelinePtr& pipeline : _pipelines) {
- DCHECK(pipeline->sink() != nullptr) << pipeline->operators().size();
- RETURN_IF_ERROR(pipeline->sink()->set_child(pipeline->operators().back()));
- }
- }
- // 4. Build local exchanger
- if (_runtime_state->enable_local_shuffle()) {
- SCOPED_TIMER(_plan_local_exchanger_timer);
- RETURN_IF_ERROR(_plan_local_exchange(_params.num_buckets,
- _params.bucket_seq_to_instance_idx,
- _params.shuffle_idx_to_instance_idx));
- }
-
- // 5. Initialize global states in pipelines.
- for (PipelinePtr& pipeline : _pipelines) {
- SCOPED_TIMER(_prepare_all_pipelines_timer);
- pipeline->children().clear();
- RETURN_IF_ERROR(pipeline->prepare(_runtime_state.get()));
- }
-
- {
- SCOPED_TIMER(_build_tasks_timer);
- // 6. Build pipeline tasks and initialize local state.
- RETURN_IF_ERROR(_build_pipeline_tasks(thread_pool));
- }
+ RETURN_IF_ERROR(_build_and_prepare_full_pipeline(thread_pool));
_init_next_report_time();
@@ -1684,6 +1710,42 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
RETURN_IF_ERROR(cur_pipe->add_operator(op, _parallel_instances));
break;
}
+ case TPlanNodeType::REC_CTE_NODE: {
+ op = std::make_shared<RecCTESourceOperatorX>(pool, tnode, next_operator_id(), descs);
+ RETURN_IF_ERROR(cur_pipe->add_operator(op, _parallel_instances));
+
+ const auto downstream_pipeline_id = cur_pipe->id();
+ if (!_dag.contains(downstream_pipeline_id)) {
+ _dag.insert({downstream_pipeline_id, {}});
+ }
+
+ PipelinePtr anchor_side_pipe = add_pipeline(cur_pipe);
+ _dag[downstream_pipeline_id].push_back(anchor_side_pipe->id());
+
+ DataSinkOperatorPtr anchor_sink;
+ anchor_sink = std::make_shared<RecCTEAnchorSinkOperatorX>(next_sink_operator_id(),
+ op->operator_id(), tnode, descs);
+ RETURN_IF_ERROR(anchor_side_pipe->set_sink(anchor_sink));
+ RETURN_IF_ERROR(anchor_side_pipe->sink()->init(tnode, _runtime_state.get()));
+ _pipeline_parent_map.push(op->node_id(), anchor_side_pipe);
+
+ PipelinePtr rec_side_pipe = add_pipeline(cur_pipe);
+ _dag[downstream_pipeline_id].push_back(rec_side_pipe->id());
+
+ DataSinkOperatorPtr rec_sink;
+ rec_sink = std::make_shared<RecCTESinkOperatorX>(next_sink_operator_id(), op->operator_id(),
+ tnode, descs);
+ RETURN_IF_ERROR(rec_side_pipe->set_sink(rec_sink));
+ RETURN_IF_ERROR(rec_side_pipe->sink()->init(tnode, _runtime_state.get()));
+ _pipeline_parent_map.push(op->node_id(), rec_side_pipe);
+
+ break;
+ }
+ case TPlanNodeType::REC_CTE_SCAN_NODE: {
+ op = std::make_shared<RecCTEScanOperatorX>(pool, tnode, next_operator_id(), descs);
+ RETURN_IF_ERROR(cur_pipe->add_operator(op, _parallel_instances));
+ break;
+ }
default:
return Status::InternalError("Unsupported exec type in pipeline: {}",
print_plan_node_type(tnode.node_type));
@@ -1754,9 +1816,16 @@ Status PipelineFragmentContext::submit() {
}
}
if (!st.ok()) {
- std::lock_guard<std::mutex> l(_task_mutex);
- if (_closed_tasks >= _total_tasks) {
- _close_fragment_instance();
+ bool need_remove = false;
+ {
+ std::lock_guard<std::mutex> l(_task_mutex);
+ if (_closed_tasks >= _total_tasks) {
+ need_remove = _close_fragment_instance();
+ }
+ }
+ // Call remove_pipeline_context() outside _task_mutex to avoid ABBA deadlock.
+ if (need_remove) {
+ _exec_env->fragment_mgr()->remove_pipeline_context({_query_id, _fragment_id});
}
return Status::InternalError("Submit pipeline failed. err = {}, BE: {}", st.to_string(),
BackendOptions::get_localhost());
@@ -1784,13 +1853,25 @@ void PipelineFragmentContext::print_profile(const std::string& extra_info) {
}
// If all pipeline tasks binded to the fragment instance are finished, then we could
// close the fragment instance.
-void PipelineFragmentContext::_close_fragment_instance() {
+// Returns true if the caller should call remove_pipeline_context() **after** releasing
+// _task_mutex. We must not call remove_pipeline_context() here because it acquires
+// _pipeline_map's shard lock, and this function is called while _task_mutex is held.
+// Acquiring _pipeline_map while holding _task_mutex creates an ABBA deadlock with
+// dump_pipeline_tasks(), which acquires _pipeline_map first and then _task_mutex
+// (via debug_string()).
+bool PipelineFragmentContext::_close_fragment_instance() {
if (_is_fragment_instance_closed) {
- return;
+ return false;
}
Defer defer_op {[&]() { _is_fragment_instance_closed = true; }};
_fragment_level_profile->total_time_counter()->update(_fragment_watcher.elapsed_time());
- static_cast<void>(send_report(true));
+ if (!_need_notify_close) {
+ auto st = send_report(true);
+ if (!st) {
+ LOG(WARNING) << fmt::format("Failed to send report for query {}, fragment {}: {}",
+ print_id(_query_id), _fragment_id, st.to_string());
+ }
+ }
// Print profile content in info log is a tempoeray solution for stream load and external_connector.
// Since stream load does not have someting like coordinator on FE, so
// backend can not report profile to FE, ant its profile can not be shown
@@ -1823,8 +1904,9 @@ void PipelineFragmentContext::_close_fragment_instance() {
collect_realtime_load_channel_profile());
}
- // all submitted tasks done
- _exec_env->fragment_mgr()->remove_pipeline_context({_query_id, _fragment_id});
+ // Return whether the caller needs to remove from the pipeline map.
+ // The caller must do this after releasing _task_mutex.
+ return !_need_notify_close;
}
void PipelineFragmentContext::decrement_running_task(PipelineId pipeline_id) {
@@ -1837,10 +1919,17 @@ void PipelineFragmentContext::decrement_running_task(PipelineId pipeline_id) {
}
}
}
- std::lock_guard<std::mutex> l(_task_mutex);
- ++_closed_tasks;
- if (_closed_tasks >= _total_tasks) {
- _close_fragment_instance();
+ bool need_remove = false;
+ {
+ std::lock_guard<std::mutex> l(_task_mutex);
+ ++_closed_tasks;
+ if (_closed_tasks >= _total_tasks) {
+ need_remove = _close_fragment_instance();
+ }
+ }
+ // Call remove_pipeline_context() outside _task_mutex to avoid ABBA deadlock.
+ if (need_remove) {
+ _exec_env->fragment_mgr()->remove_pipeline_context({_query_id, _fragment_id});
}
}
@@ -1879,7 +1968,7 @@ Status PipelineFragmentContext::send_report(bool done) {
// Load will set _is_report_success to true because load wants to know
// the process.
if (!_is_report_success && done && exec_status.ok()) {
- return Status::NeedSendAgain("");
+ return Status::OK();
}
// If both _is_report_success and _is_report_on_cancel are false,
@@ -1889,6 +1978,10 @@ Status PipelineFragmentContext::send_report(bool done) {
// When limit is reached the fragment is also cancelled, but _is_report_on_cancel will
// be set to false, to avoid sending fault report to FE.
if (!_is_report_success && !_is_report_on_cancel) {
+ if (done) {
+ // if done is true, which means the query is finished successfully, we can safely close the fragment instance without sending report to FE, and just return OK status here.
+ return Status::OK();
+ }
return Status::NeedSendAgain("");
}
@@ -1962,10 +2055,12 @@ std::vector<PipelineTask*> PipelineFragmentContext::get_revocable_tasks() const
}
std::string PipelineFragmentContext::debug_string() {
+ std::lock_guard<std::mutex> l(_task_mutex);
fmt::memory_buffer debug_string_buffer;
fmt::format_to(debug_string_buffer,
- "PipelineFragmentContext Info: _closed_tasks={}, _total_tasks={}\n",
- _closed_tasks, _total_tasks);
+ "PipelineFragmentContext Info: _closed_tasks={}, _total_tasks={}, "
+ "need_notify_close={}, fragment_id={}, _rec_cte_stage={}\n",
+ _closed_tasks, _total_tasks, _need_notify_close, _fragment_id, _rec_cte_stage);
for (size_t j = 0; j < _tasks.size(); j++) {
fmt::format_to(debug_string_buffer, "Tasks in instance {}:\n", j);
for (size_t i = 0; i < _tasks[j].size(); i++) {
@@ -2039,5 +2134,48 @@ PipelineFragmentContext::collect_realtime_load_channel_profile() const {
_runtime_state->profile_level());
return load_channel_profile;
}
+
+// Collect runtime filter IDs registered by all tasks in this PFC.
+// Used during recursive CTE stage transitions to know which filters to deregister
+// before creating the new PFC for the next recursion round.
+// Called from rerun_fragment(wait_for_destroy) while tasks are still closing.
+// Thread safety: safe because _tasks is structurally immutable after prepare() —
+// the vector sizes do not change, and individual RuntimeState filter sets are
+// written only during open() which has completed by the time we reach rerun.
+std::set<int> PipelineFragmentContext::get_deregister_runtime_filter() const {
+ std::set<int> result;
+ for (const auto& _task : _tasks) {
+ for (const auto& task : _task) {
+ auto set = task.first->runtime_state()->get_deregister_runtime_filter();
+ result.merge(set);
+ }
+ }
+ if (_runtime_state) {
+ auto set = _runtime_state->get_deregister_runtime_filter();
+ result.merge(set);
+ }
+ return result;
+}
+
+void PipelineFragmentContext::_release_resource() {
+ std::lock_guard<std::mutex> l(_task_mutex);
+ // The memory released by the query end is recorded in the query mem tracker.
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_query_ctx->query_mem_tracker());
+ auto st = _query_ctx->exec_status();
+ for (auto& _task : _tasks) {
+ if (!_task.empty()) {
+ _call_back(_task.front().first->runtime_state(), &st);
+ }
+ }
+ _tasks.clear();
+ _dag.clear();
+ _pip_id_to_pipeline.clear();
+ _pipelines.clear();
+ _sink.reset();
+ _root_op.reset();
+ _runtime_filter_mgr_map.clear();
+ _op_id_to_shared_state.clear();
+}
+
#include "common/compile_check_end.h"
} // namespace doris
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.h b/be/src/exec/pipeline/pipeline_fragment_context.h
index 3c530da1baa..db7d168c72b 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.h
+++ b/be/src/exec/pipeline/pipeline_fragment_context.h
@@ -17,6 +17,7 @@
#pragma once
+#include <brpc/closure_guard.h>
#include <gen_cpp/Types_types.h>
#include <gen_cpp/types.pb.h>
@@ -26,6 +27,7 @@
#include <functional>
#include <memory>
#include <mutex>
+#include <set>
#include <string>
#include <vector>
@@ -89,10 +91,15 @@ public:
void cancel(const Status reason);
+ bool notify_close();
+
TUniqueId get_query_id() const { return _query_id; }
[[nodiscard]] int get_fragment_id() const { return _fragment_id; }
+ uint32_t rec_cte_stage() const { return _rec_cte_stage; }
+ void set_rec_cte_stage(uint32_t stage) { _rec_cte_stage = stage; }
+
void decrement_running_task(PipelineId pipeline_id);
Status send_report(bool);
@@ -113,6 +120,9 @@ public:
[[nodiscard]] std::vector<PipelineTask*> get_revocable_tasks() const;
void clear_finished_tasks() {
+ if (_need_notify_close) {
+ return;
+ }
for (size_t j = 0; j < _tasks.size(); j++) {
for (size_t i = 0; i < _tasks[j].size(); i++) {
_tasks[j][i].first->stop_if_finished();
@@ -123,7 +133,34 @@ public:
std::string get_load_error_url();
std::string get_first_error_msg();
+ std::set<int> get_deregister_runtime_filter() const;
+
+ // Store the brpc ClosureGuard so the RPC response is deferred until this PFC is destroyed.
+ // When need_send_report_on_destruction is true (final_close), send the report immediately
+ // and do not store the guard (let it fire on return to complete the RPC).
+ //
+ // Thread safety: This method is NOT thread-safe. It reads/writes _wait_close_guard without
+ // synchronization. Currently it is only called from rerun_fragment() which is invoked
+ // sequentially by RecCTESourceOperatorX (a serial operator) — one opcode at a time per
+ // fragment. Do NOT call this concurrently from multiple threads.
+ Status listen_wait_close(const std::shared_ptr<brpc::ClosureGuard>& guard,
+ bool need_send_report_on_destruction) {
+ if (_wait_close_guard) {
+ return Status::InternalError("Already listening wait close");
+ }
+ if (need_send_report_on_destruction) {
+ return send_report(true);
+ } else {
+ _wait_close_guard = guard;
+ }
+ return Status::OK();
+ }
+
private:
+ void _release_resource();
+
+ Status _build_and_prepare_full_pipeline(ThreadPool* thread_pool);
+
Status _build_pipelines(ObjectPool* pool, const DescriptorTbl& descs, OperatorPtr* root,
PipelinePtr cur_pipe);
Status _create_tree_helper(ObjectPool* pool, const std::vector<TPlanNode>& tnodes,
@@ -170,7 +207,11 @@ private:
Status _build_pipeline_tasks_for_instance(
int instance_idx,
const std::vector<std::shared_ptr<RuntimeProfile>>& pipeline_id_to_profile);
- void _close_fragment_instance();
+ // Close the fragment instance and return true if the caller should call
+ // remove_pipeline_context() **after** releasing _task_mutex. This avoids
+ // holding _task_mutex while acquiring _pipeline_map's shard lock, which
+ // would create an ABBA deadlock with dump_pipeline_tasks().
+ bool _close_fragment_instance();
void _init_next_report_time();
// Id of this query
@@ -206,7 +247,7 @@ private:
RuntimeProfile::Counter* _build_tasks_timer = nullptr;
std::function<void(RuntimeState*, Status*)> _call_back;
- bool _is_fragment_instance_closed = false;
+ std::atomic_bool _is_fragment_instance_closed = false;
// If this is set to false, and '_is_report_success' is false as well,
// This executor will not report status to FE on being cancelled.
@@ -325,5 +366,17 @@ private:
TPipelineFragmentParams _params;
int32_t _parallel_instances = 0;
+
+ std::atomic<bool> _need_notify_close = false;
+ // Holds the brpc ClosureGuard for async wait-close during recursive CTE rerun.
+ // When the PFC finishes closing and is destroyed, the shared_ptr destructor fires
+ // the ClosureGuard, which completes the brpc response to the RecCTESourceOperatorX.
+ // Only written by listen_wait_close() from a single rerun_fragment RPC thread.
+ std::shared_ptr<brpc::ClosureGuard> _wait_close_guard = nullptr;
+
+ // The recursion round number for recursive CTE fragments.
+ // Incremented each time the fragment is rebuilt via rerun_fragment(rebuild).
+ // Used to stamp runtime filter RPCs so stale messages from old rounds are discarded.
+ uint32_t _rec_cte_stage = 0;
};
} // namespace doris
diff --git a/be/src/exec/pipeline/pipeline_task.cpp b/be/src/exec/pipeline/pipeline_task.cpp
index e89a13d9631..3066b236f3d 100644
--- a/be/src/exec/pipeline/pipeline_task.cpp
+++ b/be/src/exec/pipeline/pipeline_task.cpp
@@ -32,6 +32,7 @@
#include "core/block/block.h"
#include "exec/operator/exchange_source_operator.h"
#include "exec/operator/operator.h"
+#include "exec/operator/rec_cte_source_operator.h"
#include "exec/operator/scan_operator.h"
#include "exec/pipeline/dependency.h"
#include "exec/pipeline/pipeline.h"
@@ -682,10 +683,6 @@ Status PipelineTask::execute(bool* done) {
}
}
- if (_eos) {
- RETURN_IF_ERROR(close(Status::OK(), false));
- }
-
DBUG_EXECUTE_IF("PipelineTask::execute.sink_eos_sleep", {
auto required_pipeline_id =
DebugPoints::instance()->get_debug_param_or_default<int32_t>(
@@ -724,6 +721,14 @@ Status PipelineTask::execute(bool* done) {
RETURN_IF_ERROR(block->check_type_and_column());
status = _sink->sink(_state, block, _eos);
+ if (_eos) {
+ if (_sink->reset_to_rerun(_state, _root)) {
+ _eos = false;
+ } else {
+ RETURN_IF_ERROR(close(Status::OK(), false));
+ }
+ }
+
if (status.is<ErrorCode::END_OF_FILE>()) {
set_wake_up_early();
return Status::OK();
@@ -905,7 +910,7 @@ Status PipelineTask::close(Status exec_status, bool close_sink) {
for (auto& op : _operators) {
auto tem = op->close(_state);
if (!tem.ok() && s.ok()) {
- s = tem;
+ s = std::move(tem);
}
}
}
@@ -1043,7 +1048,9 @@ Status PipelineTask::wake_up(Dependency* dep) {
_blocked_dep = nullptr;
auto holder = std::dynamic_pointer_cast<PipelineTask>(shared_from_this());
RETURN_IF_ERROR(_state_transition(PipelineTask::State::RUNNABLE));
- RETURN_IF_ERROR(_state->get_query_ctx()->get_pipe_exec_scheduler()->submit(holder));
+ if (auto f = _fragment_context.lock(); f) {
+ RETURN_IF_ERROR(_state->get_query_ctx()->get_pipe_exec_scheduler()->submit(holder));
+ }
return Status::OK();
}
diff --git a/be/src/exec/pipeline/rec_cte_shared_state.h b/be/src/exec/pipeline/rec_cte_shared_state.h
new file mode 100644
index 00000000000..2727fda2814
--- /dev/null
+++ b/be/src/exec/pipeline/rec_cte_shared_state.h
@@ -0,0 +1,179 @@
+// 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.
+
+#pragma once
+
+#include "dependency.h"
+#include "exec/common/distinct_agg_utils.h"
+#include "util/brpc_client_cache.h"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+
+struct RecCTESharedState : public BasicSharedState {
+ std::vector<TRecCTETarget> targets;
+ std::vector<Block> blocks;
+ IColumn::Selector distinct_row;
+ Dependency* source_dep = nullptr;
+ Dependency* anchor_dep = nullptr;
+ Arena arena;
+ RuntimeProfile::Counter* hash_table_compute_timer = nullptr;
+ RuntimeProfile::Counter* hash_table_emplace_timer = nullptr;
+ RuntimeProfile::Counter* hash_table_input_counter = nullptr;
+
+ std::unique_ptr<DistinctDataVariants> agg_data = nullptr;
+
+ int current_round = 0;
+ int last_round_offset = 0;
+ int max_recursion_depth = 0;
+ bool ready_to_return = false;
+
+ void update_ready_to_return() {
+ if (last_round_offset == blocks.size()) {
+ ready_to_return = true;
+ }
+ }
+
+ Status emplace_block(RuntimeState* state, Block&& block) {
+ if (agg_data) {
+ auto num_rows = uint32_t(block.rows());
+ ColumnRawPtrs raw_columns;
+ std::vector<ColumnPtr> columns = block.get_columns_and_convert();
+ for (auto& col : columns) {
+ raw_columns.push_back(col.get());
+ }
+
+ std::visit(Overload {[&](std::monostate& arg) -> void {
+ throw doris::Exception(ErrorCode::INTERNAL_ERROR,
+ "uninited hash table");
+ },
+ [&](auto& agg_method) -> void {
+ SCOPED_TIMER(hash_table_compute_timer);
+ using HashMethodType = std::decay_t<decltype(agg_method)>;
+ using AggState = typename HashMethodType::State;
+
+ AggState agg_state(raw_columns);
+ agg_method.init_serialized_keys(raw_columns, num_rows);
+ distinct_row.clear();
+
+ size_t row = 0;
+ auto creator = [&](const auto& ctor, auto& key, auto& origin) {
+ HashMethodType::try_presis_key(key, origin, arena);
+ ctor(key);
+ distinct_row.push_back(row);
+ };
+ auto creator_for_null_key = [&]() {
+ distinct_row.push_back(row);
+ };
+
+ SCOPED_TIMER(hash_table_emplace_timer);
+ for (; row < num_rows; ++row) {
+ agg_method.lazy_emplace(agg_state, row, creator,
+ creator_for_null_key);
+ }
+ COUNTER_UPDATE(hash_table_input_counter, num_rows);
+ }},
+ agg_data->method_variant);
+
+ if (distinct_row.size() == block.rows()) {
+ blocks.emplace_back(std::move(block));
+ } else if (!distinct_row.empty()) {
+ auto distinct_block = MutableBlock(block.clone_empty());
+ RETURN_IF_ERROR(block.append_to_block_by_selector(&distinct_block, distinct_row));
+ blocks.emplace_back(distinct_block.to_block());
+ }
+ } else {
+ blocks.emplace_back(std::move(block));
+ }
+ return Status::OK();
+ }
+
+ PTransmitRecCTEBlockParams build_basic_param(RuntimeState* state,
+ const TRecCTETarget& target) const {
+ PTransmitRecCTEBlockParams request;
+ request.set_node_id(target.node_id);
+ request.mutable_query_id()->CopyFrom(UniqueId(state->query_id()).to_proto());
+ request.mutable_fragment_instance_id()->CopyFrom(
+ UniqueId(target.fragment_instance_id).to_proto());
+ return request;
+ }
+
+ Status send_data_to_targets(RuntimeState* state, size_t round_offset) const {
+ if (targets.size() == 0) {
+ return Status::OK();
+ }
+ int send_multi_blocks_byte_size = state->query_options().exchange_multi_blocks_byte_size;
+ int block_number_per_target =
+ int(blocks.size() - round_offset + targets.size() - 1) / targets.size();
+ for (auto target : targets) {
+ auto stub =
+ state->get_query_ctx()->exec_env()->brpc_internal_client_cache()->get_client(
+ target.addr);
+ if (!stub) {
+ return Status::InternalError(fmt::format("Get rpc stub failed, host={}, port={}",
+ target.addr.hostname, target.addr.port));
+ }
+
+ // send blocks
+ int step = block_number_per_target;
+ while (round_offset < blocks.size() && step > 0) {
+ PTransmitRecCTEBlockParams request = build_basic_param(state, target);
+ auto current_bytes = 0;
+ while (round_offset < blocks.size() && step > 0 &&
+ current_bytes < send_multi_blocks_byte_size) {
+ auto* pblock = request.add_blocks();
+ size_t uncompressed_bytes = 0;
+ size_t compressed_bytes = 0;
+ int64_t compress_time;
+ RETURN_IF_ERROR(blocks[round_offset].serialize(
+ state->be_exec_version(), pblock, &uncompressed_bytes,
+ &compressed_bytes, &compress_time,
+ state->fragement_transmission_compression_type()));
+ round_offset++;
+ step--;
+ current_bytes += compressed_bytes;
+ }
+ request.set_eos(false);
+
+ PTransmitRecCTEBlockResult result;
+ brpc::Controller controller;
+ controller.set_timeout_ms(
+ get_execution_rpc_timeout_ms(state->get_query_ctx()->execution_timeout()));
+
+ stub->transmit_rec_cte_block(&controller, &request, &result, brpc::DoNothing());
+ brpc::Join(controller.call_id());
+ RETURN_IF_ERROR(Status::create(result.status()));
+ }
+
+ // send eos
+ {
+ PTransmitRecCTEBlockParams request = build_basic_param(state, target);
+ request.set_eos(true);
+
+ PTransmitRecCTEBlockResult result;
+ brpc::Controller controller;
+ stub->transmit_rec_cte_block(&controller, &request, &result, brpc::DoNothing());
+ brpc::Join(controller.call_id());
+ RETURN_IF_ERROR(Status::create(result.status()));
+ }
+ }
+ return Status::OK();
+ }
+};
+
+#include "common/compile_check_end.h"
+} // namespace doris
diff --git a/be/src/exec/runtime_filter/runtime_filter.cpp b/be/src/exec/runtime_filter/runtime_filter.cpp
index 41843427834..4a6489a5132 100644
--- a/be/src/exec/runtime_filter/runtime_filter.cpp
+++ b/be/src/exec/runtime_filter/runtime_filter.cpp
@@ -35,6 +35,7 @@ Status RuntimeFilter::_push_to_remote(RuntimeState* state, const TNetworkAddress
}
auto merge_filter_request = std::make_shared<PMergeFilterRequest>();
+ merge_filter_request->set_stage(_stage);
auto merge_filter_callback = DummyBrpcCallback<PMergeFilterResponse>::create_shared();
auto merge_filter_closure =
AutoReleaseClosure<PMergeFilterRequest, DummyBrpcCallback<PMergeFilterResponse>>::
@@ -123,8 +124,9 @@ Status RuntimeFilter::_init_with_desc(const TRuntimeFilterDesc* desc,
}
std::string RuntimeFilter::_debug_string() const {
- return fmt::format("{}, mode: {}", _wrapper ? _wrapper->debug_string() : "<null wrapper>",
- _has_remote_target ? "GLOBAL" : "LOCAL");
+ return fmt::format("{}, mode: {}, stage: {}",
+ _wrapper ? _wrapper->debug_string() : "<null wrapper>",
+ _has_remote_target ? "GLOBAL" : "LOCAL", _stage);
}
void RuntimeFilter::_check_wrapper_state(
diff --git a/be/src/exec/runtime_filter/runtime_filter.h b/be/src/exec/runtime_filter/runtime_filter.h
index 82d27cbbed2..ed4357dd772 100644
--- a/be/src/exec/runtime_filter/runtime_filter.h
+++ b/be/src/exec/runtime_filter/runtime_filter.h
@@ -44,6 +44,9 @@ public:
bool has_remote_target() const { return _has_remote_target; }
+ uint32_t stage() const { return _stage; }
+ void set_stage(uint32_t stage) { _stage = stage; }
+
template <class T>
Status assign(const T& request, butil::IOBufAsZeroCopyInputStream* data) {
std::unique_lock<std::recursive_mutex> l(_rmtx);
@@ -111,6 +114,10 @@ protected:
// _wrapper is a runtime filter function wrapper
std::shared_ptr<RuntimeFilterWrapper> _wrapper;
+ // The recursion round number for recursive CTE.
+ // Stamped onto outgoing RPC requests so stale messages from old rounds are discarded.
+ uint32_t _stage = 0;
+
// will apply to remote node
bool _has_remote_target = false;
diff --git a/be/src/exec/runtime_filter/runtime_filter_consumer.h b/be/src/exec/runtime_filter/runtime_filter_consumer.h
index fd0176904cc..ad399db05f6 100644
--- a/be/src/exec/runtime_filter/runtime_filter_consumer.h
+++ b/be/src/exec/runtime_filter/runtime_filter_consumer.h
@@ -41,11 +41,11 @@ public:
APPLIED, // The consumer will switch to this state after the expression is acquired
};
- static Status create(const QueryContext* query_ctx, const TRuntimeFilterDesc* desc, int node_id,
+ static Status create(const RuntimeState* state, const TRuntimeFilterDesc* desc, int node_id,
std::shared_ptr<RuntimeFilterConsumer>* res) {
*res = std::shared_ptr<RuntimeFilterConsumer>(
- new RuntimeFilterConsumer(query_ctx, desc, node_id));
- RETURN_IF_ERROR((*res)->_init_with_desc(desc, &query_ctx->query_options()));
+ new RuntimeFilterConsumer(state, desc, node_id));
+ RETURN_IF_ERROR((*res)->_init_with_desc(desc, &state->query_options()));
return Status::OK();
}
@@ -86,15 +86,15 @@ public:
}
private:
- RuntimeFilterConsumer(const QueryContext* query_ctx, const TRuntimeFilterDesc* desc,
- int node_id)
+ RuntimeFilterConsumer(const RuntimeState* state, const TRuntimeFilterDesc* desc, int node_id)
: RuntimeFilter(desc),
_probe_expr(desc->planId_to_target_expr.find(node_id)->second),
_registration_time(MonotonicMillis()),
_rf_state(State::NOT_READY) {
+ auto* query_ctx = state->get_query_ctx();
// If bitmap filter is not applied, it will cause the query result to be incorrect
// local rf must wait until timeout, otherwise it may lead results incorrectness, because LEFT_SEMI_DIRECT_RETURN_OPT
- bool wait_infinitely = query_ctx->runtime_filter_wait_infinitely() ||
+ bool wait_infinitely = state->runtime_filter_wait_infinitely() ||
_runtime_filter_type == RuntimeFilterType::BITMAP_FILTER ||
!has_remote_target();
_rf_wait_time_ms = wait_infinitely ? query_ctx->execution_timeout() * 1000
diff --git a/be/src/exec/runtime_filter/runtime_filter_merger.h b/be/src/exec/runtime_filter/runtime_filter_merger.h
index 4c5ae8bb182..41ab571d6a2 100644
--- a/be/src/exec/runtime_filter/runtime_filter_merger.h
+++ b/be/src/exec/runtime_filter/runtime_filter_merger.h
@@ -73,11 +73,16 @@ public:
}
void set_expected_producer_num(int num) {
- DCHECK_EQ(_received_producer_num, 0);
- DCHECK_EQ(_received_rf_size_num, 0);
+ if (_received_producer_num > 0 || _received_rf_size_num > 0) {
+ throw Exception(ErrorCode::INTERNAL_ERROR,
+ "runtime filter merger set expected producer after receive data, {}",
+ debug_string());
+ }
_expected_producer_num = num;
}
+ int get_expected_producer_num() const { return _expected_producer_num; }
+
bool add_rf_size(uint64_t size) {
_received_rf_size_num++;
if (_expected_producer_num < _received_rf_size_num) {
diff --git a/be/src/exec/runtime_filter/runtime_filter_mgr.cpp b/be/src/exec/runtime_filter/runtime_filter_mgr.cpp
index 5f1dd60f960..3615299fa38 100644
--- a/be/src/exec/runtime_filter/runtime_filter_mgr.cpp
+++ b/be/src/exec/runtime_filter/runtime_filter_mgr.cpp
@@ -25,6 +25,7 @@
#include <gen_cpp/internal_service.pb.h>
#include <gen_cpp/types.pb.h>
+#include <mutex>
#include <ostream>
#include <string>
#include <utility>
@@ -62,13 +63,13 @@ std::vector<std::shared_ptr<RuntimeFilterConsumer>> RuntimeFilterMgr::get_consum
}
Status RuntimeFilterMgr::register_consumer_filter(
- const QueryContext* query_ctx, const TRuntimeFilterDesc& desc, int node_id,
+ const RuntimeState* state, const TRuntimeFilterDesc& desc, int node_id,
std::shared_ptr<RuntimeFilterConsumer>* consumer) {
SCOPED_CONSUME_MEM_TRACKER(_tracker.get());
int32_t key = desc.filter_id;
std::lock_guard<std::mutex> l(_lock);
- RETURN_IF_ERROR(RuntimeFilterConsumer::create(query_ctx, &desc, node_id, consumer));
+ RETURN_IF_ERROR(RuntimeFilterConsumer::create(state, &desc, node_id, consumer));
_consumer_map[key].push_back(*consumer);
return Status::OK();
}
@@ -103,11 +104,21 @@ Status LocalMergeContext::register_producer(const QueryContext* query_ctx,
const TRuntimeFilterDesc* desc,
std::shared_ptr<RuntimeFilterProducer> producer) {
std::lock_guard<std::mutex> l(mtx);
+ if (producer->stage() > stage) {
+ // New recursive CTE round: discard stale merger and producers from
+ // the previous round and recreate the merger for the new round.
+ merger.reset();
+ producers.clear();
+ stage = producer->stage();
+ }
if (!merger) {
RETURN_IF_ERROR(RuntimeFilterMerger::create(query_ctx, desc, &merger));
}
producers.emplace_back(producer);
merger->set_expected_producer_num(cast_set<int>(producers.size()));
+ // Sync the local merger's stage from the producer so that outgoing merge RPCs
+ // (via _push_to_remote) carry the correct recursive CTE round number.
+ merger->set_stage(producer->stage());
return Status::OK();
}
@@ -120,10 +131,10 @@ Status RuntimeFilterMgr::get_local_merge_producer_filters(int filter_id,
std::lock_guard<std::mutex> l(_lock);
auto iter = _local_merge_map.find(filter_id);
if (iter == _local_merge_map.end()) {
- return Status::InternalError(
- "get_local_merge_producer_filters meet unknown filter: {}, role: "
- "LOCAL_MERGE_PRODUCER.",
- filter_id);
+ // Filter may have been removed during a recursive CTE stage reset.
+ // Return OK with nullptr to let the caller skip gracefully.
+ *local_merge_filters = nullptr;
+ return Status::OK();
}
*local_merge_filters = &iter->second;
DCHECK(iter->second.merger);
@@ -233,6 +244,13 @@ Status RuntimeFilterMergeControllerEntity::send_filter_size(std::shared_ptr<Quer
}
auto& cnt_val = iter->second;
std::unique_lock<std::mutex> l(iter->second.mtx);
+ // Discard stale-stage runtime filter size requests from old recursive CTE rounds.
+ // Each round increments the stage counter; only messages matching the current stage
+ // should be processed. This prevents old PFC's runtime filters from corrupting
+ // the merge state of the new round's filters.
+ if (request->stage() != iter->second.stage) {
+ return Status::OK();
+ }
cnt_val.source_addrs.push_back(request->source_addr());
Status st = Status::OK();
@@ -250,9 +268,12 @@ Status RuntimeFilterMergeControllerEntity::send_filter_size(std::shared_ptr<Quer
continue;
}
+ auto sync_request = std::make_shared<PSyncFilterSizeRequest>();
+ sync_request->set_stage(iter->second.stage);
+
auto closure = AutoReleaseClosure<PSyncFilterSizeRequest,
DummyBrpcCallback<PSyncFilterSizeResponse>>::
- create_unique(std::make_shared<PSyncFilterSizeRequest>(),
+ create_unique(sync_request,
DummyBrpcCallback<PSyncFilterSizeResponse>::create_shared(), ctx);
auto* pquery_id = closure->request_->mutable_query_id();
@@ -278,6 +299,10 @@ Status RuntimeFilterMergeControllerEntity::send_filter_size(std::shared_ptr<Quer
Status RuntimeFilterMgr::sync_filter_size(const PSyncFilterSizeRequest* request) {
LocalMergeContext* local_merge_filters = nullptr;
RETURN_IF_ERROR(get_local_merge_producer_filters(request->filter_id(), &local_merge_filters));
+ if (local_merge_filters == nullptr) {
+ // Filter was removed during a recursive CTE stage reset; discard stale request.
+ return Status::OK();
+ }
for (auto producer : local_merge_filters->producers) {
producer->set_synced_size(request->filter_size());
}
@@ -323,6 +348,14 @@ Status RuntimeFilterMergeControllerEntity::merge(std::shared_ptr<QueryContext> q
bool is_ready = false;
{
std::lock_guard<std::mutex> l(iter->second.mtx);
+ // Discard stale-stage merge requests from old recursive CTE rounds.
+ if (request->stage() != iter->second.stage) {
+ return Status::OK();
+ }
+ if (cnt_val.merger == nullptr) {
+ return Status::InternalError("Merger is null for filter id {}",
+ std::to_string(request->filter_id()));
+ }
// Skip the other broadcast join runtime filter
if (cnt_val.arrive_id.size() == 1 && cnt_val.runtime_filter_desc.is_broadcast_join) {
return Status::OK();
@@ -365,6 +398,8 @@ Status RuntimeFilterMergeControllerEntity::_send_rf_to_target(GlobalMergeContext
butil::IOBuf request_attachment;
PPublishFilterRequestV2 apply_request;
+ apply_request.set_stage(cnt_val.stage);
+
// serialize filter
void* data = nullptr;
int len = 0;
@@ -433,11 +468,44 @@ Status RuntimeFilterMergeControllerEntity::_send_rf_to_target(GlobalMergeContext
return st;
}
+// Reset merge context for the next recursive CTE round.
+// Recreates the merger to clear accumulated state, preserving expected producer count.
+// Increments the stage counter so stale merge/size RPCs from old rounds are discarded.
+Status GlobalMergeContext::reset(QueryContext* query_ctx) {
+ std::unique_lock<std::mutex> lock(mtx);
+ // Merger must exist: reset() is only called on fully initialized merge contexts.
+ DORIS_CHECK(merger);
+ int producer_size = merger->get_expected_producer_num();
+ RETURN_IF_ERROR(RuntimeFilterMerger::create(query_ctx, &runtime_filter_desc, &merger));
+ merger->set_expected_producer_num(producer_size);
+ arrive_id.clear();
+ source_addrs.clear();
+ done = false;
+ stage++;
+ // Keep the Merger's own stage in sync for consistent debug output.
+ merger->set_stage(stage);
+ return Status::OK();
+}
+
+Status RuntimeFilterMergeControllerEntity::reset_global_rf(
+ QueryContext* query_ctx, const google::protobuf::RepeatedField<int32_t>& filter_ids) {
+ for (const auto& filter_id : filter_ids) {
+ GlobalMergeContext* cnt_val;
+ {
+ std::unique_lock<std::shared_mutex> guard(_filter_map_mutex);
+ cnt_val = &_filter_map[filter_id]; // may inplace construct default object
+ }
+ RETURN_IF_ERROR(cnt_val->reset(query_ctx));
+ }
+ return Status::OK();
+}
+
std::string RuntimeFilterMergeControllerEntity::debug_string() {
std::string result = "RuntimeFilterMergeControllerEntity Info:\n";
std::shared_lock<std::shared_mutex> guard(_filter_map_mutex);
for (const auto& [filter_id, ctx] : _filter_map) {
- result += fmt::format("{}\n", ctx.merger->debug_string());
+ result += fmt::format("filter_id: {}, stage: {}, {}\n", filter_id, ctx.stage,
+ ctx.merger->debug_string());
}
return result;
}
diff --git a/be/src/exec/runtime_filter/runtime_filter_mgr.h b/be/src/exec/runtime_filter/runtime_filter_mgr.h
index c680c646610..f822e01196f 100644
--- a/be/src/exec/runtime_filter/runtime_filter_mgr.h
+++ b/be/src/exec/runtime_filter/runtime_filter_mgr.h
@@ -28,7 +28,6 @@
#include <memory>
#include <mutex>
#include <shared_mutex>
-#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>
@@ -58,6 +57,9 @@ struct LocalMergeContext {
std::mutex mtx;
std::shared_ptr<RuntimeFilterMerger> merger;
std::vector<std::shared_ptr<RuntimeFilterProducer>> producers;
+ // Tracks the recursive CTE round. When a producer from a newer round
+ // registers, the context is reset (merger recreated, old producers dropped).
+ uint32_t stage = 0;
Status register_producer(const QueryContext* query_ctx, const TRuntimeFilterDesc* desc,
std::shared_ptr<RuntimeFilterProducer> producer);
@@ -71,6 +73,12 @@ struct GlobalMergeContext {
std::unordered_set<UniqueId> arrive_id;
std::vector<PNetworkAddress> source_addrs;
std::atomic<bool> done = false;
+
+ // for represent the round number of recursive cte
+ // if lower stage rf input to higher stage, we just discard the rf
+ uint32_t stage = 0;
+
+ Status reset(QueryContext* query_ctx);
};
// owned by RuntimeState
@@ -81,7 +89,7 @@ public:
// get/set consumer
std::vector<std::shared_ptr<RuntimeFilterConsumer>> get_consume_filters(int filter_id);
- Status register_consumer_filter(const QueryContext* query_ctx, const TRuntimeFilterDesc& desc,
+ Status register_consumer_filter(const RuntimeState* state, const TRuntimeFilterDesc& desc,
int node_id,
std::shared_ptr<RuntimeFilterConsumer>* consumer_filter);
@@ -102,6 +110,17 @@ public:
std::string debug_string();
+ void remove_filter(int32_t filter_id) {
+ std::lock_guard<std::mutex> l(_lock);
+ _consumer_map.erase(filter_id);
+ // NOTE: _local_merge_map is NOT erased here. It is reset lazily in
+ // LocalMergeContext::register_producer when a producer from a newer
+ // recursive CTE round registers. Erasing eagerly here would race with
+ // multi-fragment REBUILD: a consumer-only fragment's remove_filter could
+ // delete the entry that the producer fragment just re-registered.
+ _producer_id_set.erase(filter_id);
+ }
+
private:
/**
* `_is_global = true` means this runtime filter manager menages query-level runtime filters.
@@ -153,6 +172,9 @@ public:
return _filter_map.empty();
}
+ Status reset_global_rf(QueryContext* query_ctx,
+ const google::protobuf::RepeatedField<int32_t>& filter_ids);
+
private:
Status _init_with_desc(std::shared_ptr<QueryContext> query_ctx,
const TRuntimeFilterDesc* runtime_filter_desc,
diff --git a/be/src/exec/runtime_filter/runtime_filter_producer.cpp b/be/src/exec/runtime_filter/runtime_filter_producer.cpp
index e1a1616e44e..6105014df10 100644
--- a/be/src/exec/runtime_filter/runtime_filter_producer.cpp
+++ b/be/src/exec/runtime_filter/runtime_filter_producer.cpp
@@ -57,6 +57,10 @@ Status RuntimeFilterProducer::publish(RuntimeState* state, bool build_hash_table
LocalMergeContext* context = nullptr;
RETURN_IF_ERROR(state->global_runtime_filter_mgr()->get_local_merge_producer_filters(
_wrapper->filter_id(), &context));
+ if (context == nullptr) {
+ // Filter was removed during a recursive CTE stage reset; this producer is stale.
+ return Status::OK();
+ }
std::lock_guard l(context->mtx);
RETURN_IF_ERROR(context->merger->merge_from(this));
if (context->merger->ready()) {
@@ -160,6 +164,10 @@ Status RuntimeFilterProducer::send_size(RuntimeState* state, uint64_t local_filt
LocalMergeContext* merger_context = nullptr;
RETURN_IF_ERROR(state->global_runtime_filter_mgr()->get_local_merge_producer_filters(
_wrapper->filter_id(), &merger_context));
+ if (merger_context == nullptr) {
+ // Filter was removed during a recursive CTE stage reset; this producer is stale.
+ return Status::OK();
+ }
std::lock_guard merger_lock(merger_context->mtx);
if (merger_context->merger->add_rf_size(local_filter_size)) {
if (!_has_remote_target) {
@@ -189,6 +197,8 @@ Status RuntimeFilterProducer::send_size(RuntimeState* state, uint64_t local_filt
}
auto request = std::make_shared<PSendFilterSizeRequest>();
+ request->set_stage(_stage);
+
auto callback = DummyBrpcCallback<PSendFilterSizeResponse>::create_shared();
// RuntimeFilter maybe deconstructed before the rpc finished, so that could not use
// a raw pointer in closure. Has to use the context's shared ptr.
diff --git a/be/src/exec/scan/scanner_context.h b/be/src/exec/scan/scanner_context.h
index 939635751bc..553408ebc96 100644
--- a/be/src/exec/scan/scanner_context.h
+++ b/be/src/exec/scan/scanner_context.h
@@ -63,10 +63,6 @@ public:
DorisMetrics::instance()->scanner_task_cnt->increment(1);
}
- ScanTask(std::shared_ptr<ResourceContext> resource_ctx,
- std::weak_ptr<ScannerDelegate> delegate_scanner)
- : _resource_ctx(std::move(resource_ctx)), scanner(delegate_scanner) {}
-
~ScanTask() {
SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_resource_ctx->memory_context()->mem_tracker());
cached_blocks.clear();
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index aeabb28db5f..ee352c66b4a 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -280,13 +280,11 @@ Status ConcurrentContextMap<Key, Value, ValueType>::apply_if_not_exists(
}
template <typename Key, typename Value, typename ValueType>
-void ConcurrentContextMap<Key, Value, ValueType>::erase(const Key& query_id) {
+bool ConcurrentContextMap<Key, Value, ValueType>::erase(const Key& query_id) {
auto id = get_map_id(query_id, _internal_map.size());
- {
- std::unique_lock lock(*_internal_map[id].first);
- auto& map = _internal_map[id].second;
- map.erase(query_id);
- }
+ std::unique_lock lock(*_internal_map[id].first);
+ auto& map = _internal_map[id].second;
+ return map.erase(query_id) != 0;
}
template <typename Key, typename Value, typename ValueType>
@@ -344,6 +342,10 @@ void FragmentMgr::stop() {
// destructred and remove it from _query_ctx_map_delay_delete which is destructring. it's UB.
_query_ctx_map_delay_delete.clear();
_pipeline_map.clear();
+ {
+ std::lock_guard<std::mutex> lk(_rerunnable_params_lock);
+ _rerunnable_params_map.clear();
+ }
}
std::string FragmentMgr::to_http_path(const std::string& file_name) {
@@ -676,16 +678,28 @@ Status FragmentMgr::start_query_execution(const PExecPlanFragmentStartRequest* r
}
void FragmentMgr::remove_pipeline_context(std::pair<TUniqueId, int> 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);
-
- _pipeline_map.erase(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);
+ }
}
void FragmentMgr::remove_query_context(const TUniqueId& key) {
+ // Clean up any saved rerunnable params for this query to avoid memory leaks.
+ // This covers both cancel and normal destruction paths.
+ {
+ std::lock_guard<std::mutex> lk(_rerunnable_params_lock);
+ for (auto it = _rerunnable_params_map.begin(); it != _rerunnable_params_map.end();) {
+ if (it->first.first == key) {
+ it = _rerunnable_params_map.erase(it);
+ } else {
+ ++it;
+ }
+ }
+ }
_query_ctx_map_delay_delete.erase(key);
#ifndef BE_TEST
_query_ctx_map.erase(key);
@@ -921,6 +935,19 @@ Status FragmentMgr::exec_plan_fragment(const TPipelineFragmentParams& params,
_pipeline_map.insert({params.query_id, params.fragment_id}, context);
}
+ // Save params for recursive CTE child fragments so we can recreate the PFC later.
+ // For recursive CTE, the child fragment needs to be destroyed and rebuilt between rounds,
+ // so we save the original params here and use them in rerun_fragment(rebuild).
+ if (params.__isset.need_notify_close && params.need_notify_close) {
+ std::lock_guard<std::mutex> lk(_rerunnable_params_lock);
+ _rerunnable_params_map[{params.query_id, params.fragment_id}] = {
+ .deregister_runtime_filter_ids = {},
+ .params = params,
+ .parent = parent,
+ .finish_callback = cb,
+ .query_ctx = query_ctx};
+ }
+
if (!params.__isset.need_wait_execution_trigger || !params.need_wait_execution_trigger) {
query_ctx->set_ready_to_execute_only();
}
@@ -1333,6 +1360,10 @@ Status FragmentMgr::apply_filterv2(const PPublishFilterRequestV2* request,
// 2. create the filter wrapper to replace or ignore/disable the target filters
if (!filters.empty()) {
+ // Discard stale-stage requests from old recursive CTE rounds.
+ if (filters[0]->stage() != request->stage()) {
+ return Status::OK();
+ }
RETURN_IF_ERROR(filters[0]->assign(*request, attach_data));
std::ranges::for_each(filters, [&](auto& filter) { filter->signal(filters[0].get()); });
}
@@ -1444,6 +1475,140 @@ Status FragmentMgr::get_query_statistics(const TUniqueId& query_id, TQueryStatis
print_id(query_id), query_stats);
}
+Status FragmentMgr::transmit_rec_cte_block(
+ const TUniqueId& query_id, const TUniqueId& instance_id, int node_id,
+ const google::protobuf::RepeatedPtrField<doris::PBlock>& pblocks, bool eos) {
+ if (auto q_ctx = get_query_ctx(query_id)) {
+ SCOPED_ATTACH_TASK(q_ctx.get());
+ return q_ctx->send_block_to_cte_scan(instance_id, node_id, pblocks, eos);
+ } else {
+ return Status::EndOfFile(
+ "Transmit rec cte block failed: Query context (query-id: {}) not found, maybe "
+ "finished",
+ print_id(query_id));
+ }
+}
+
+// Orchestrates the recursive CTE fragment lifecycle through 4 phases:
+//
+// wait_for_destroy: collect deregister RF IDs, store brpc closure, trigger old PFC close
+// rebuild: increment stage, deregister old RFs, create+prepare new PFC from saved params
+// submit: submit the new PFC's pipeline tasks for execution
+// final_close: async wait for close, send final report, clean up (last round only)
+//
+// The brpc ClosureGuard is stored in the PFC so the RPC response is deferred until
+// the PFC is fully destroyed. This gives the caller (RecCTESourceOperatorX) a
+// synchronization point to know when the old PFC has finished all its tasks.
+Status FragmentMgr::rerun_fragment(const std::shared_ptr<brpc::ClosureGuard>& guard,
+ const TUniqueId& query_id, int fragment_id,
+ PRerunFragmentParams_Opcode stage) {
+ if (stage == PRerunFragmentParams::WAIT_FOR_DESTROY ||
+ stage == PRerunFragmentParams::FINAL_CLOSE) {
+ auto fragment_ctx = _pipeline_map.find({query_id, fragment_id});
+ if (!fragment_ctx) {
+ return Status::NotFound("Fragment context (query-id: {}, fragment-id: {}) not found",
+ print_id(query_id), fragment_id);
+ }
+
+ if (stage == PRerunFragmentParams::WAIT_FOR_DESTROY) {
+ std::unique_lock<std::mutex> lk(_rerunnable_params_lock);
+ auto it = _rerunnable_params_map.find({query_id, fragment_id});
+ if (it == _rerunnable_params_map.end()) {
+ lk.unlock();
+ auto st = fragment_ctx->listen_wait_close(guard, true);
+ if (!st.ok()) {
+ LOG(WARNING) << fmt::format(
+ "wait_for_destroy fragment context (query-id: {}, fragment-id: "
+ "{}) failed: {}",
+ print_id(query_id), fragment_id, st.to_string());
+ }
+ return Status::NotFound(
+ "Rerunnable params (query-id: {}, fragment-id: {}) not found",
+ print_id(query_id), fragment_id);
+ }
+
+ it->second.deregister_runtime_filter_ids.merge(
+ fragment_ctx->get_deregister_runtime_filter());
+ }
+
+ auto* query_ctx = fragment_ctx->get_query_ctx();
+ SCOPED_ATTACH_TASK(query_ctx);
+ RETURN_IF_ERROR(
+ fragment_ctx->listen_wait_close(guard, stage == PRerunFragmentParams::FINAL_CLOSE));
+ fragment_ctx->notify_close();
+ return Status::OK();
+ } else if (stage == PRerunFragmentParams::REBUILD) {
+ auto q_ctx = get_query_ctx(query_id);
+ if (!q_ctx) {
+ return Status::NotFound(
+ "rerun_fragment: Query context (query-id: {}) not found, maybe finished",
+ print_id(query_id));
+ }
+ SCOPED_ATTACH_TASK(q_ctx.get());
+ RerunableFragmentInfo info;
+ {
+ std::lock_guard<std::mutex> lk(_rerunnable_params_lock);
+ auto it = _rerunnable_params_map.find({query_id, fragment_id});
+ if (it == _rerunnable_params_map.end()) {
+ return Status::NotFound("rebuild (query-id: {}, fragment-id: {}) not found",
+ print_id(query_id), fragment_id);
+ }
+ it->second.stage++;
+ // Deregister old runtime filters so new ones can be registered in the new PFC.
+ for (int32_t filter_id : it->second.deregister_runtime_filter_ids) {
+ q_ctx->runtime_filter_mgr()->remove_filter(filter_id);
+ }
+ info = it->second;
+ }
+
+ auto context = std::make_shared<PipelineFragmentContext>(
+ q_ctx->query_id(), info.params, q_ctx, _exec_env, info.finish_callback,
+ [this](const ReportStatusRequest& req, auto&& ctx) {
+ return this->trigger_pipeline_context_report(req, std::move(ctx));
+ });
+ // Propagate the recursion stage so that runtime filters created by this PFC
+ // carry the correct stage number.
+ context->set_rec_cte_stage(info.stage);
+
+ Status prepare_st = Status::OK();
+ ASSIGN_STATUS_IF_CATCH_EXCEPTION(prepare_st = context->prepare(_thread_pool.get()),
+ prepare_st);
+ if (!prepare_st.ok()) {
+ q_ctx->cancel(prepare_st, info.params.fragment_id);
+ return prepare_st;
+ }
+
+ // Insert new PFC into _pipeline_map (old one was removed)
+ _pipeline_map.insert({info.params.query_id, info.params.fragment_id}, context);
+
+ // Update QueryContext mapping (must support overwrite)
+ q_ctx->set_pipeline_context(info.params.fragment_id, context);
+ return Status::OK();
+
+ } else if (stage == PRerunFragmentParams::SUBMIT) {
+ auto fragment_ctx = _pipeline_map.find({query_id, fragment_id});
+ if (!fragment_ctx) {
+ return Status::NotFound("Fragment context (query-id: {}, fragment-id: {}) not found",
+ print_id(query_id), fragment_id);
+ }
+ return fragment_ctx->submit();
+ } else {
+ return Status::InvalidArgument("Unknown rerun fragment opcode: {}", stage);
+ }
+}
+
+Status FragmentMgr::reset_global_rf(const TUniqueId& query_id,
+ const google::protobuf::RepeatedField<int32_t>& filter_ids) {
+ if (auto q_ctx = get_query_ctx(query_id)) {
+ SCOPED_ATTACH_TASK(q_ctx.get());
+ return q_ctx->reset_global_rf(filter_ids);
+ } else {
+ return Status::NotFound(
+ "reset_fragment: Query context (query-id: {}) not found, maybe finished",
+ print_id(query_id));
+ }
+}
+
#include "common/compile_check_end.h"
} // namespace doris
diff --git a/be/src/runtime/fragment_mgr.h b/be/src/runtime/fragment_mgr.h
index 9fbb2d10035..ec30528afb1 100644
--- a/be/src/runtime/fragment_mgr.h
+++ b/be/src/runtime/fragment_mgr.h
@@ -17,6 +17,7 @@
#pragma once
+#include <brpc/closure_guard.h>
#include <gen_cpp/FrontendService_types.h>
#include <gen_cpp/QueryPlanExtra_types.h>
#include <gen_cpp/Types_types.h>
@@ -25,8 +26,10 @@
#include <cstdint>
#include <functional>
#include <iosfwd>
+#include <map>
#include <memory>
#include <mutex>
+#include <set>
#include <string>
#include <unordered_map>
#include <vector>
@@ -74,7 +77,7 @@ public:
Value find(const Key& query_id);
void insert(const Key& query_id, std::shared_ptr<ValueType>);
void clear();
- void erase(const Key& query_id);
+ bool erase(const Key& query_id);
size_t num_items() const {
size_t n = 0;
for (auto& pair : _internal_map) {
@@ -182,6 +185,18 @@ public:
std::shared_ptr<QueryContext> get_query_ctx(const TUniqueId& query_id);
+ Status transmit_rec_cte_block(const TUniqueId& query_id, const TUniqueId& instance_id,
+ int node_id,
+ const google::protobuf::RepeatedPtrField<PBlock>& pblocks,
+ bool eos);
+
+ Status rerun_fragment(const std::shared_ptr<brpc::ClosureGuard>& guard,
+ const TUniqueId& query_id, int fragment,
+ PRerunFragmentParams_Opcode stage);
+
+ Status reset_global_rf(const TUniqueId& query_id,
+ const google::protobuf::RepeatedField<int32_t>& filter_ids);
+
private:
struct BrpcItem {
TNetworkAddress network_address;
@@ -217,6 +232,26 @@ private:
PipelineFragmentContext>
_pipeline_map;
+ // Saved params and callback for rerunnable (recursive CTE) fragments.
+ // Only populated when need_notify_close == true during exec_plan_fragment.
+ // Lifecycle: created in exec_plan_fragment(), used in rerun_fragment(rebuild)
+ // to recreate PFC with fresh state, cleaned up in remove_query_context().
+ struct RerunableFragmentInfo {
+ // Runtime filter IDs registered by the old PFC, collected during wait_for_destroy.
+ // These are deregistered from the RuntimeFilterMgr before the new PFC is created.
+ std::set<int> deregister_runtime_filter_ids;
+ // Original params from FE, used to recreate the PFC each round.
+ TPipelineFragmentParams params;
+ TPipelineFragmentParamsList parent;
+ FinishCallback finish_callback;
+ // Hold query_ctx to prevent it from being destroyed while rerunnable fragments exist.
+ std::shared_ptr<QueryContext> query_ctx;
+ // Monotonically increasing stage counter, stamps runtime filter RPCs.
+ uint32_t stage = 0;
+ };
+ std::mutex _rerunnable_params_lock;
+ std::map<std::pair<TUniqueId, int>, RerunableFragmentInfo> _rerunnable_params_map;
+
// query id -> QueryContext
ConcurrentContextMap<TUniqueId, std::weak_ptr<QueryContext>, QueryContext> _query_ctx_map;
// keep query ctx do not delete immediately to make rf coordinator merge filter work well after query eos
diff --git a/be/src/runtime/query_context.cpp b/be/src/runtime/query_context.cpp
index d641bad3c61..198666346c0 100644
--- a/be/src/runtime/query_context.cpp
+++ b/be/src/runtime/query_context.cpp
@@ -32,6 +32,7 @@
#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"
@@ -387,7 +388,9 @@ std::string QueryContext::print_all_pipeline_context() {
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);
- _fragment_id_to_pipeline_ctx.insert({fragment_id, pip_ctx});
+ // 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() {
@@ -508,4 +511,46 @@ TReportExecStatusParams QueryContext::get_realtime_exec_status() {
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();
+}
+
} // namespace doris
diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h
index 932fff73bf0..c4ed6cafe85 100644
--- a/be/src/runtime/query_context.h
+++ b/be/src/runtime/query_context.h
@@ -48,6 +48,7 @@ namespace doris {
class PipelineFragmentContext;
class PipelineTask;
class Dependency;
+class RecCTEScanLocalState;
struct ReportStatusRequest {
const Status status;
@@ -166,11 +167,6 @@ public:
return _query_options.runtime_filter_wait_time_ms;
}
- bool runtime_filter_wait_infinitely() const {
- return _query_options.__isset.runtime_filter_wait_infinitely &&
- _query_options.runtime_filter_wait_infinitely;
- }
-
int be_exec_version() const {
if (!_query_options.__isset.be_exec_version) {
return 0;
@@ -297,6 +293,22 @@ public:
void set_first_error_msg(std::string error_msg);
std::string get_first_error_msg();
+ Status send_block_to_cte_scan(const TUniqueId& instance_id, int node_id,
+ const google::protobuf::RepeatedPtrField<doris::PBlock>& pblocks,
+ bool eos);
+ void registe_cte_scan(const TUniqueId& instance_id, int node_id, RecCTEScanLocalState* scan);
+ void deregiste_cte_scan(const TUniqueId& instance_id, int node_id);
+
+ std::vector<int> get_fragment_ids() {
+ std::vector<int> fragment_ids;
+ for (const auto& it : _fragment_id_to_pipeline_ctx) {
+ fragment_ids.push_back(it.first);
+ }
+ return fragment_ids;
+ }
+
+ Status reset_global_rf(const google::protobuf::RepeatedField<int32_t>& filter_ids);
+
private:
friend class QueryTaskController;
@@ -380,6 +392,10 @@ private:
// file cache context holders
std::vector<io::BlockFileCache::QueryFileCacheContextHolderPtr> _query_context_holders;
+ // instance id + node id -> cte scan
+ std::map<std::pair<TUniqueId, int>, RecCTEScanLocalState*> _cte_scan;
+ std::mutex _cte_scan_lock;
+
public:
// when fragment of pipeline is closed, it will register its profile to this map by using add_fragment_profile
void add_fragment_profile(
@@ -396,7 +412,7 @@ public:
timespec get_query_arrival_timestamp() const { return this->_query_arrival_timestamp; }
QuerySource get_query_source() const { return this->_query_source; }
- const TQueryOptions get_query_options() const { return _query_options; }
+ TQueryOptions get_query_options() const { return _query_options; }
};
} // namespace doris
diff --git a/be/src/runtime/runtime_predicate.h b/be/src/runtime/runtime_predicate.h
index 4aa52b44e0f..a56cf7efbe7 100644
--- a/be/src/runtime/runtime_predicate.h
+++ b/be/src/runtime/runtime_predicate.h
@@ -54,6 +54,7 @@ public:
void set_detected_source() {
std::unique_lock<std::shared_mutex> wlock(_rwlock);
+ _orderby_extrem = Field(PrimitiveType::TYPE_NULL);
_detected_source = true;
}
diff --git a/be/src/runtime/runtime_state.cpp b/be/src/runtime/runtime_state.cpp
index 37a2803f0b9..2be5aeeae3c 100644
--- a/be/src/runtime/runtime_state.cpp
+++ b/be/src/runtime/runtime_state.cpp
@@ -37,8 +37,11 @@
#include "common/status.h"
#include "core/value/vdatetime_value.h"
#include "exec/operator/operator.h"
+#include "exec/pipeline/pipeline_fragment_context.h"
#include "exec/pipeline/pipeline_task.h"
+#include "exec/runtime_filter/runtime_filter_consumer.h"
#include "exec/runtime_filter/runtime_filter_mgr.h"
+#include "exec/runtime_filter/runtime_filter_producer.h"
#include "io/fs/s3_file_system.h"
#include "load/load_path_mgr.h"
#include "runtime/exec_env.h"
@@ -206,10 +209,17 @@ RuntimeState::~RuntimeState() {
if (_error_log_file != nullptr && _error_log_file->is_open()) {
_error_log_file->close();
}
-
_obj_pool->clear();
}
+const std::set<int>& RuntimeState::get_deregister_runtime_filter() const {
+ return _registered_runtime_filter_ids;
+}
+
+void RuntimeState::merge_register_runtime_filter(const std::set<int>& runtime_filter_ids) {
+ _registered_runtime_filter_ids.insert(runtime_filter_ids.begin(), runtime_filter_ids.end());
+}
+
Status RuntimeState::init(const TUniqueId& fragment_instance_id, const TQueryOptions& query_options,
const TQueryGlobals& query_globals, ExecEnv* exec_env) {
_fragment_instance_id = fragment_instance_id;
@@ -490,10 +500,23 @@ RuntimeFilterMgr* RuntimeState::global_runtime_filter_mgr() {
Status RuntimeState::register_producer_runtime_filter(
const TRuntimeFilterDesc& desc, std::shared_ptr<RuntimeFilterProducer>* producer_filter) {
+ _registered_runtime_filter_ids.insert(desc.filter_id);
// Producers are created by local runtime filter mgr and shared by global runtime filter manager.
// When RF is published, consumers in both global and local RF mgr will be found.
RETURN_IF_ERROR(local_runtime_filter_mgr()->register_producer_filter(_query_ctx, desc,
producer_filter));
+ // Stamp the producer with the current recursive CTE stage so that outgoing merge RPCs
+ // carry the correct round number and stale messages from old rounds are discarded.
+ // PFC must still be alive: this runs inside a pipeline task, so the execution context
+ // cannot have expired yet.
+ // In unit-test scenarios the task execution context is never set (no PipelineFragmentContext
+ // exists), so skip the stage stamping — the default stage (0) is correct.
+ if (task_execution_context_inited()) {
+ auto pfc = std::static_pointer_cast<PipelineFragmentContext>(
+ get_task_execution_context().lock());
+ DORIS_CHECK(pfc);
+ (*producer_filter)->set_stage(pfc->rec_cte_stage());
+ }
RETURN_IF_ERROR(global_runtime_filter_mgr()->register_local_merger_producer_filter(
_query_ctx, desc, *producer_filter));
return Status::OK();
@@ -502,9 +525,23 @@ Status RuntimeState::register_producer_runtime_filter(
Status RuntimeState::register_consumer_runtime_filter(
const TRuntimeFilterDesc& desc, bool need_local_merge, int node_id,
std::shared_ptr<RuntimeFilterConsumer>* consumer_filter) {
+ _registered_runtime_filter_ids.insert(desc.filter_id);
bool need_merge = desc.has_remote_targets || need_local_merge;
RuntimeFilterMgr* mgr = need_merge ? global_runtime_filter_mgr() : local_runtime_filter_mgr();
- return mgr->register_consumer_filter(_query_ctx, desc, node_id, consumer_filter);
+ RETURN_IF_ERROR(mgr->register_consumer_filter(this, desc, node_id, consumer_filter));
+ // Stamp the consumer with the current recursive CTE stage so that incoming publish RPCs
+ // from old rounds are detected and discarded.
+ // PFC must still be alive: this runs inside a pipeline task, so the execution context
+ // cannot have expired yet.
+ // In unit-test scenarios the task execution context is never set (no PipelineFragmentContext
+ // exists), so skip the stage stamping — the default stage (0) is correct.
+ if (task_execution_context_inited()) {
+ auto pfc = std::static_pointer_cast<PipelineFragmentContext>(
+ get_task_execution_context().lock());
+ DORIS_CHECK(pfc);
+ (*consumer_filter)->set_stage(pfc->rec_cte_stage());
+ }
+ return Status::OK();
}
bool RuntimeState::is_nereids() const {
@@ -520,8 +557,7 @@ std::vector<std::shared_ptr<RuntimeProfile>> RuntimeState::build_pipeline_profil
std::size_t pipeline_size) {
std::unique_lock lc(_pipeline_profile_lock);
if (!_pipeline_id_to_profile.empty()) {
- throw Exception(ErrorCode::INTERNAL_ERROR,
- "build_pipeline_profile can only be called once.");
+ return _pipeline_id_to_profile;
}
_pipeline_id_to_profile.resize(pipeline_size);
{
diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h
index f233db496f2..9790e6c9b23 100644
--- a/be/src/runtime/runtime_state.h
+++ b/be/src/runtime/runtime_state.h
@@ -73,6 +73,7 @@ class MemTrackerLimiter;
class QueryContext;
class RuntimeFilterConsumer;
class RuntimeFilterProducer;
+class TaskExecutionContext;
// A collection of items that are part of the global state of a
// query and shared across all execution nodes of that query.
@@ -530,7 +531,7 @@ public:
_runtime_filter_mgr = runtime_filter_mgr;
}
- QueryContext* get_query_ctx() { return _query_ctx; }
+ QueryContext* get_query_ctx() const { return _query_ctx; }
[[nodiscard]] bool low_memory_mode() const;
@@ -562,6 +563,12 @@ public:
_query_options.enable_broadcast_join_force_passthrough;
}
+ int cte_max_recursion_depth() const {
+ return _query_options.__isset.cte_max_recursion_depth
+ ? _query_options.cte_max_recursion_depth
+ : 0;
+ }
+
int rpc_verbose_profile_max_instance_count() const {
return _query_options.__isset.rpc_verbose_profile_max_instance_count
? _query_options.rpc_verbose_profile_max_instance_count
@@ -638,6 +645,8 @@ public:
_task_execution_context = context;
}
+ bool task_execution_context_inited() const { return _task_execution_context_inited; }
+
std::weak_ptr<TaskExecutionContext> get_task_execution_context() {
CHECK(_task_execution_context_inited)
<< "_task_execution_context_inited == false, the ctx is not inited";
@@ -803,6 +812,15 @@ public:
_query_options.hnsw_bounded_queue, _query_options.ivf_nprobe);
}
+ bool runtime_filter_wait_infinitely() const {
+ return _query_options.__isset.runtime_filter_wait_infinitely &&
+ _query_options.runtime_filter_wait_infinitely;
+ }
+
+ const std::set<int>& get_deregister_runtime_filter() const;
+
+ void merge_register_runtime_filter(const std::set<int>& runtime_filter_ids);
+
private:
Status create_error_log_file();
@@ -935,6 +953,8 @@ private:
// used for encoding the global lazy materialize
std::shared_ptr<IdFileMap> _id_file_map = nullptr;
+
+ std::set<int> _registered_runtime_filter_ids;
};
#define RETURN_IF_CANCELLED(state) \
diff --git a/be/src/runtime/task_execution_context.cpp b/be/src/runtime/task_execution_context.cpp
new file mode 100644
index 00000000000..29e9b2d5bde
--- /dev/null
+++ b/be/src/runtime/task_execution_context.cpp
@@ -0,0 +1,32 @@
+// 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 "task_execution_context.h"
+
+#include <glog/logging.h>
+
+namespace doris {
+
+TaskExecutionContext ::TaskExecutionContext() = default;
+TaskExecutionContext ::~TaskExecutionContext() = default;
+
+HasTaskExecutionCtx::HasTaskExecutionCtx(RuntimeState* state)
+ : task_exec_ctx_(state->get_task_execution_context()) {}
+
+HasTaskExecutionCtx::~HasTaskExecutionCtx() = default;
+
+} // namespace doris
diff --git a/be/src/runtime/task_execution_context.h b/be/src/runtime/task_execution_context.h
index 7583b401f22..9b90e6ae95d 100644
--- a/be/src/runtime/task_execution_context.h
+++ b/be/src/runtime/task_execution_context.h
@@ -19,14 +19,20 @@
#include <memory>
+#include "runtime/runtime_state.h"
+
namespace doris {
-// This class act as a super class of all context like things such as
-// plan fragment executor or pipelinefragmentcontext or pipelinexfragmentcontext
+class RuntimeState;
+
+// Base class for execution contexts (e.g. PipelineFragmentContext).
+//
+// For recursive CTE, the PFC (which inherits from this class) is held by external threads
+// (scanner threads, brpc callbacks, etc.) via weak_ptr<TaskExecutionContext>.
class TaskExecutionContext : public std::enable_shared_from_this<TaskExecutionContext> {
public:
- TaskExecutionContext() = default;
- virtual ~TaskExecutionContext() = default;
+ TaskExecutionContext();
+ virtual ~TaskExecutionContext();
};
using TaskExecutionContextSPtr = std::shared_ptr<TaskExecutionContext>;
@@ -38,14 +44,9 @@ using TaskExecutionContextSPtr = std::shared_ptr<TaskExecutionContext>;
struct HasTaskExecutionCtx {
using Weak = typename TaskExecutionContextSPtr::weak_type;
- HasTaskExecutionCtx(TaskExecutionContextSPtr task_exec_ctx) : task_exec_ctx_(task_exec_ctx) {}
-
- // Init task ctx from state, the state has to own a method named get_task_execution_context()
- // like runtime state
- template <typename T>
- HasTaskExecutionCtx(T* state) : task_exec_ctx_(state->get_task_execution_context()) {}
+ HasTaskExecutionCtx(RuntimeState* state);
- virtual ~HasTaskExecutionCtx() = default;
+ virtual ~HasTaskExecutionCtx();
public:
inline TaskExecutionContextSPtr task_exec_ctx() const { return task_exec_ctx_.lock(); }
diff --git a/be/src/service/internal_service.cpp b/be/src/service/internal_service.cpp
index 58589c31aa5..d6f2b397029 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -1619,6 +1619,62 @@ void PInternalService::fold_constant_expr(google::protobuf::RpcController* contr
}
}
+void PInternalService::transmit_rec_cte_block(google::protobuf::RpcController* controller,
+ const PTransmitRecCTEBlockParams* request,
+ PTransmitRecCTEBlockResult* response,
+ google::protobuf::Closure* done) {
+ bool ret = _light_work_pool.try_offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ auto st = _exec_env->fragment_mgr()->transmit_rec_cte_block(
+ UniqueId(request->query_id()).to_thrift(),
+ UniqueId(request->fragment_instance_id()).to_thrift(), request->node_id(),
+ request->blocks(), request->eos());
+ st.to_protobuf(response->mutable_status());
+ });
+ if (!ret) {
+ offer_failed(response, done, _light_work_pool);
+ return;
+ }
+}
+
+void PInternalService::rerun_fragment(google::protobuf::RpcController* controller,
+ const PRerunFragmentParams* request,
+ PRerunFragmentResult* response,
+ google::protobuf::Closure* done) {
+ bool ret = _light_work_pool.try_offer([this, request, response, done]() {
+ // Use shared_ptr<ClosureGuard> so we can transfer ownership to the PFC.
+ // For wait_for_destroy/final_close, the guard is stored in the PFC and the RPC
+ // response is deferred until the PFC is fully destroyed. For rebuild/submit,
+ // the guard fires immediately when this lambda returns.
+ std::shared_ptr<brpc::ClosureGuard> closure_guard =
+ std::make_shared<brpc::ClosureGuard>(done);
+ auto st = _exec_env->fragment_mgr()->rerun_fragment(
+ closure_guard, UniqueId(request->query_id()).to_thrift(), request->fragment_id(),
+ request->stage());
+ st.to_protobuf(response->mutable_status());
+ });
+ if (!ret) {
+ offer_failed(response, done, _light_work_pool);
+ return;
+ }
+}
+
+void PInternalService::reset_global_rf(google::protobuf::RpcController* controller,
+ const PResetGlobalRfParams* request,
+ PResetGlobalRfResult* response,
+ google::protobuf::Closure* done) {
+ bool ret = _light_work_pool.try_offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ auto st = _exec_env->fragment_mgr()->reset_global_rf(
+ UniqueId(request->query_id()).to_thrift(), request->filter_ids());
+ st.to_protobuf(response->mutable_status());
+ });
+ if (!ret) {
+ offer_failed(response, done, _light_work_pool);
+ return;
+ }
+}
+
void PInternalService::transmit_block(google::protobuf::RpcController* controller,
const PTransmitDataParams* request,
PTransmitDataResult* response,
diff --git a/be/src/service/internal_service.h b/be/src/service/internal_service.h
index 8535cc2dc98..b1e11ec34b7 100644
--- a/be/src/service/internal_service.h
+++ b/be/src/service/internal_service.h
@@ -140,6 +140,16 @@ public:
const ::doris::PPublishFilterRequestV2* request,
::doris::PPublishFilterResponse* response,
::google::protobuf::Closure* done) override;
+ void transmit_rec_cte_block(google::protobuf::RpcController* controller,
+ const PTransmitRecCTEBlockParams* request,
+ PTransmitRecCTEBlockResult* response,
+ google::protobuf::Closure* done) override;
+ void rerun_fragment(google::protobuf::RpcController* controller,
+ const PRerunFragmentParams* request, PRerunFragmentResult* response,
+ google::protobuf::Closure* done) override;
+ void reset_global_rf(google::protobuf::RpcController* controller,
+ const PResetGlobalRfParams* request, PResetGlobalRfResult* response,
+ google::protobuf::Closure* done) override;
void transmit_block(::google::protobuf::RpcController* controller,
const ::doris::PTransmitDataParams* request,
::doris::PTransmitDataResult* response,
diff --git a/be/test/exec/runtime_filter/runtime_filter_consumer_test.cpp b/be/test/exec/runtime_filter/runtime_filter_consumer_test.cpp
index 759676abb79..a431f2adc00 100644
--- a/be/test/exec/runtime_filter/runtime_filter_consumer_test.cpp
+++ b/be/test/exec/runtime_filter/runtime_filter_consumer_test.cpp
@@ -31,7 +31,7 @@ public:
void test_signal_aquire(TRuntimeFilterDesc desc) {
std::shared_ptr<RuntimeFilterConsumer> consumer;
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
- RuntimeFilterConsumer::create(_query_ctx.get(), &desc, 0, &consumer));
+ RuntimeFilterConsumer::create(_runtime_states[0].get(), &desc, 0, &consumer));
std::shared_ptr<RuntimeFilterProducer> producer;
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
@@ -59,7 +59,7 @@ TEST_F(RuntimeFilterConsumerTest, basic) {
std::shared_ptr<RuntimeFilterConsumer> consumer;
auto desc = TRuntimeFilterDescBuilder().add_planId_to_target_expr(0).build();
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
- RuntimeFilterConsumer::create(_query_ctx.get(), &desc, 0, &consumer));
+ RuntimeFilterConsumer::create(_runtime_states[1].get(), &desc, 0, &consumer));
std::shared_ptr<RuntimeFilterConsumer> registed_consumer;
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(_runtime_states[1]->register_consumer_runtime_filter(
@@ -116,7 +116,7 @@ TEST_F(RuntimeFilterConsumerTest, timeout_aquire) {
std::shared_ptr<RuntimeFilterConsumer> consumer;
auto desc = TRuntimeFilterDescBuilder().add_planId_to_target_expr(0).build();
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
- RuntimeFilterConsumer::create(_query_ctx.get(), &desc, 0, &consumer));
+ RuntimeFilterConsumer::create(_runtime_states[1].get(), &desc, 0, &consumer));
std::shared_ptr<RuntimeFilterProducer> producer;
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
@@ -141,7 +141,7 @@ TEST_F(RuntimeFilterConsumerTest, wait_infinity) {
const_cast<TQueryOptions&>(_query_ctx->_query_options)
.__set_runtime_filter_wait_infinitely(true);
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
- RuntimeFilterConsumer::create(_query_ctx.get(), &desc, 0, &consumer));
+ RuntimeFilterConsumer::create(_runtime_states[1].get(), &desc, 0, &consumer));
std::shared_ptr<RuntimeFilterConsumer> registed_consumer;
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(_runtime_states[1]->register_consumer_runtime_filter(
@@ -152,7 +152,7 @@ TEST_F(RuntimeFilterConsumerTest, aquire_disabled) {
std::shared_ptr<RuntimeFilterConsumer> consumer;
auto desc = TRuntimeFilterDescBuilder().add_planId_to_target_expr(0).build();
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
- RuntimeFilterConsumer::create(_query_ctx.get(), &desc, 0, &consumer));
+ RuntimeFilterConsumer::create(_runtime_states[1].get(), &desc, 0, &consumer));
std::shared_ptr<RuntimeFilterProducer> producer;
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
@@ -174,7 +174,7 @@ TEST_F(RuntimeFilterConsumerTest, bitmap_filter) {
std::shared_ptr<RuntimeFilterConsumer> consumer;
{
- auto st = RuntimeFilterConsumer::create(_query_ctx.get(), &desc, 0, &consumer);
+ auto st = RuntimeFilterConsumer::create(_runtime_states[1].get(), &desc, 0, &consumer);
ASSERT_FALSE(st.ok());
}
desc.__set_src_expr(
@@ -195,13 +195,13 @@ TEST_F(RuntimeFilterConsumerTest, bitmap_filter) {
.build());
{
- auto st = RuntimeFilterConsumer::create(_query_ctx.get(), &desc, 0, &consumer);
+ auto st = RuntimeFilterConsumer::create(_runtime_states[1].get(), &desc, 0, &consumer);
ASSERT_FALSE(st.ok());
}
{
desc.__set_has_local_targets(false);
desc.__set_has_remote_targets(true);
- auto st = RuntimeFilterConsumer::create(_query_ctx.get(), &desc, 0, &consumer);
+ auto st = RuntimeFilterConsumer::create(_runtime_states[1].get(), &desc, 0, &consumer);
ASSERT_FALSE(st.ok());
desc.__set_has_local_targets(true);
desc.__set_has_remote_targets(false);
@@ -209,7 +209,7 @@ TEST_F(RuntimeFilterConsumerTest, bitmap_filter) {
desc.__set_bitmap_target_expr(TRuntimeFilterDescBuilder::get_default_expr());
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
- RuntimeFilterConsumer::create(_query_ctx.get(), &desc, 0, &consumer));
+ RuntimeFilterConsumer::create(_runtime_states[1].get(), &desc, 0, &consumer));
}
TEST_F(RuntimeFilterConsumerTest, aquire_signal_at_same_time) {
@@ -217,7 +217,7 @@ TEST_F(RuntimeFilterConsumerTest, aquire_signal_at_same_time) {
std::shared_ptr<RuntimeFilterConsumer> consumer;
auto desc = TRuntimeFilterDescBuilder().add_planId_to_target_expr(0).build();
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
- RuntimeFilterConsumer::create(_query_ctx.get(), &desc, 0, &consumer));
+ RuntimeFilterConsumer::create(_runtime_states[1].get(), &desc, 0, &consumer));
std::shared_ptr<RuntimeFilterProducer> producer;
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
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 80921628e53..d6ccc080961 100644
--- a/be/test/exec/runtime_filter/runtime_filter_mgr_test.cpp
+++ b/be/test/exec/runtime_filter/runtime_filter_mgr_test.cpp
@@ -65,7 +65,7 @@ TEST_F(RuntimeFilterMgrTest, TestRuntimeFilterMgr) {
EXPECT_TRUE(global_runtime_filter_mgr->get_consume_filters(filter_id).empty());
std::shared_ptr<RuntimeFilterConsumer> consumer_filter;
EXPECT_TRUE(global_runtime_filter_mgr
- ->register_consumer_filter(ctx.get(), desc, 0, &consumer_filter)
+ ->register_consumer_filter(&state, desc, 0, &consumer_filter)
.ok());
EXPECT_FALSE(global_runtime_filter_mgr->get_consume_filters(filter_id).empty());
}
@@ -104,9 +104,13 @@ TEST_F(RuntimeFilterMgrTest, TestRuntimeFilterMgr) {
EXPECT_NE(producer_filter, nullptr);
LocalMergeContext* local_merge_filters = nullptr;
- EXPECT_FALSE(global_runtime_filter_mgr
- ->get_local_merge_producer_filters(filter_id, &local_merge_filters)
- .ok());
+ // filter_id not yet registered: global mgr returns OK with nullptr
+ // (graceful skip for recursive CTE stage reset).
+ EXPECT_TRUE(global_runtime_filter_mgr
+ ->get_local_merge_producer_filters(filter_id, &local_merge_filters)
+ .ok());
+ EXPECT_EQ(local_merge_filters, nullptr);
+ // local mgr always returns error (not supported)
EXPECT_FALSE(local_runtime_filter_mgr
->get_local_merge_producer_filters(filter_id, &local_merge_filters)
.ok());
diff --git a/gensrc/proto/internal_service.proto b/gensrc/proto/internal_service.proto
index a5f684e6d3c..50274c17627 100644
--- a/gensrc/proto/internal_service.proto
+++ b/gensrc/proto/internal_service.proto
@@ -60,6 +60,44 @@ message PTransmitDataResult {
optional int64 receive_time = 2;
};
+message PTransmitRecCTEBlockParams {
+ repeated PBlock blocks = 1;
+ optional bool eos = 2;
+ optional PUniqueId query_id = 3;
+ optional PUniqueId fragment_instance_id = 4;
+ optional int32 node_id = 5;
+};
+
+message PTransmitRecCTEBlockResult {
+ optional PStatus status = 1;
+};
+
+
+message PRerunFragmentParams {
+ enum Opcode {
+ WAIT_FOR_DESTROY = 1; // deregister RF, destroy old PFC, async wait for tasks to close via brpc closure
+ REBUILD = 2; // rebuild next round pipeline tasks
+ SUBMIT = 3; // submit tasks to execute
+ FINAL_CLOSE = 4; // async wait for tasks to close, send report, and clean up (last round)
+ }
+ optional PUniqueId query_id = 1;
+ optional int32 fragment_id = 2;
+ optional Opcode stage = 3;
+};
+
+message PRerunFragmentResult {
+ optional PStatus status = 1;
+};
+
+message PResetGlobalRfParams {
+ optional PUniqueId query_id = 1;
+ repeated int32 filter_ids = 2;
+};
+
+message PResetGlobalRfResult {
+ optional PStatus status = 1;
+};
+
message PTabletWithPartition {
required int64 partition_id = 1;
required int64 tablet_id = 2;
@@ -576,6 +614,7 @@ message PSendFilterSizeRequest {
required PUniqueId query_id = 2;
required PNetworkAddress source_addr = 3;
required uint64 filter_size = 4;
+ optional uint32 stage = 5;
};
message PSendFilterSizeResponse {
@@ -586,6 +625,7 @@ message PSyncFilterSizeRequest {
required int32 filter_id = 1;
required PUniqueId query_id = 2;
required uint64 filter_size = 3;
+ optional uint32 stage = 5;
};
message PSyncFilterSizeResponse {
@@ -607,6 +647,7 @@ message PMergeFilterRequest {
optional bool ignored = 12;
optional uint64 local_merge_time = 13;
optional bool disabled = 14;
+ optional uint32 stage = 15;
};
message PMergeFilterResponse {
@@ -628,6 +669,7 @@ message PPublishFilterRequestV2 {
repeated int32 fragment_ids = 12; // deprecated
optional uint64 local_merge_time = 13;
optional bool disabled = 14;
+ optional uint32 stage = 15;
};
message PPublishFilterResponse {
@@ -1161,6 +1203,9 @@ service PBackendService {
rpc sync_filter_size(PSyncFilterSizeRequest) returns (PSyncFilterSizeResponse);
rpc apply_filterv2(PPublishFilterRequestV2) returns (PPublishFilterResponse);
rpc fold_constant_expr(PConstantExprRequest) returns (PConstantExprResult);
+ rpc rerun_fragment(PRerunFragmentParams) returns (PRerunFragmentResult);
+ rpc reset_global_rf(PResetGlobalRfParams) returns (PResetGlobalRfResult);
+ rpc transmit_rec_cte_block(PTransmitRecCTEBlockParams) returns (PTransmitRecCTEBlockResult);
rpc transmit_block(PTransmitDataParams) returns (PTransmitDataResult);
rpc transmit_block_by_http(PEmptyRequest) returns (PTransmitDataResult);
rpc check_rpc_channel(PCheckRPCChannelRequest) returns (PCheckRPCChannelResponse);
diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift
index 9679d549ad9..1cae84c4840 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -418,7 +418,6 @@ struct TQueryOptions {
178: optional i64 iceberg_write_target_file_size_bytes = 0;
180: optional i32 max_file_scanners_concurrency = 0;
181: optional i32 min_file_scanners_concurrency = 0;
-
182: optional i32 ivf_nprobe = 1;
179: optional bool enable_parquet_filter_by_bloom_filter = true;
@@ -470,6 +469,8 @@ struct TQueryOptions {
// session variable `spill_repartition_max_depth` in FE. Default is 8.
209: optional i32 spill_repartition_max_depth = 8
+ 184: optional i32 cte_max_recursion_depth;
+
// 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.
@@ -718,6 +719,7 @@ struct TPipelineFragmentParams {
// Used by 2.1
44: optional list<i32> topn_filter_source_node_ids
45: optional map<string, TAIResource> ai_resources
+ 46: optional bool need_notify_close
// For cloud
1000: optional bool is_mow_table;
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index f53832fc9cd..e32adb337da 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -60,7 +60,9 @@ enum TPlanNodeType {
TEST_EXTERNAL_SCAN_NODE = 31,
PARTITION_SORT_NODE = 32,
GROUP_COMMIT_SCAN_NODE = 33,
- MATERIALIZATION_NODE = 34
+ MATERIALIZATION_NODE = 34,
+ REC_CTE_NODE = 35,
+ REC_CTE_SCAN_NODE = 36
}
struct TKeyRange {
@@ -728,6 +730,29 @@ struct TFileScanNode {
2: optional string table_name
}
+struct TRecCTETarget {
+ 1: optional Types.TNetworkAddress addr
+ 2: optional Types.TUniqueId fragment_instance_id
+ 3: optional i32 node_id
+}
+
+struct TRecCTEResetInfo {
+ 1: optional Types.TNetworkAddress addr
+ 2: optional i32 fragment_id
+}
+
+struct TRecCTENode {
+ 1: optional bool is_union_all
+ 2: optional list<TRecCTETarget> targets
+ 3: optional list<TRecCTEResetInfo> fragments_to_reset
+ 4: optional list<list<Exprs.TExpr>> result_expr_lists
+ 5: optional list<i32> rec_side_runtime_filter_ids
+ 6: optional bool is_used_by_other_rec_cte
+}
+
+struct TRecCTEScanNode {
+}
+
struct TEsScanNode {
1: required Types.TTupleId tuple_id
2: optional map<string,string> properties
@@ -1476,6 +1501,7 @@ struct TPlanNode {
36: optional list<TRuntimeFilterDesc> runtime_filters
37: optional TGroupCommitScanNode group_commit_scan_node
38: optional TMaterializationNode materialization_node
+ 39: optional TRecCTENode rec_cte_node
// Use in vec exec engine
40: optional Exprs.TExpr vconjunct
@@ -1498,6 +1524,8 @@ struct TPlanNode {
50: optional list<list<Exprs.TExpr>> distribute_expr_lists
51: optional bool is_serial_operator
+ 52: optional TRecCTEScanNode rec_cte_scan_node
+
// projections is final projections, which means projecting into results and materializing them into the output block.
101: optional list<Exprs.TExpr> projections
102: optional Types.TTupleId output_tuple_id
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org