update some fix, todo and comment

BiteTheDDDDt · BiteTheDDDDt · commit 76d7ef8ef13f · 2025-12-22T00:17:38.000+08:00
fix fix fix Revert "fix" This reverts commit 41bba83. fix update some todo and comment update fix update fix test tets tets remove nullable_tuples update format update update fix update fix fix fix Revert "fix" This reverts commit ce86242. Reapply "fix" This reverts commit 5c58d3d. fix update fix
diff --git a/be/src/pipeline/exec/exchange_sink_buffer.cpp b/be/src/pipeline/exec/exchange_sink_buffer.cpp
@@ -92,7 +92,7 @@ namespace pipeline {
 ExchangeSinkBuffer::ExchangeSinkBuffer(PUniqueId query_id, PlanNodeId dest_node_id,
                                        PlanNodeId node_id, RuntimeState* state,
                                        const std::vector<InstanceLoId>& sender_ins_ids)
-        : HasTaskExecutionCtx(state),
+        : HasTaskExecutionCtx(state, false),
           _queue_capacity(0),
           _is_failed(false),
           _query_id(std::move(query_id)),
diff --git a/be/src/pipeline/exec/exchange_sink_buffer.h b/be/src/pipeline/exec/exchange_sink_buffer.h
@@ -269,7 +269,7 @@ class ExchangeSinkBuffer : public HasTaskExecutionCtx {
                        RuntimeState* state, const std::vector<InstanceLoId>& sender_ins_ids);
 #ifdef BE_TEST
     ExchangeSinkBuffer(RuntimeState* state, int64_t sinknum)
-            : HasTaskExecutionCtx(state), _state(state), _exchange_sink_num(sinknum) {};
+            : HasTaskExecutionCtx(state, false), _state(state), _exchange_sink_num(sinknum) {};
 #endif
 
     ~ExchangeSinkBuffer() override = default;
diff --git a/be/src/pipeline/exec/exchange_source_operator.cpp b/be/src/pipeline/exec/exchange_source_operator.cpp
@@ -50,7 +50,7 @@ ExchangeLocalState::~ExchangeLocalState() {
 std::string ExchangeLocalState::debug_string(int indentation_level) const {
     fmt::memory_buffer debug_string_buffer;
     fmt::format_to(debug_string_buffer, "{}, recvr: ({})", Base::debug_string(indentation_level),
-                   stream_recvr->debug_string());
+                   stream_recvr ? stream_recvr->debug_string() : "null");
     return fmt::to_string(debug_string_buffer);
 }
 
@@ -210,6 +210,7 @@ Status ExchangeLocalState::close(RuntimeState* state) {
     }
     if (stream_recvr != nullptr) {
         stream_recvr->close();
+        stream_recvr.reset();
     }
     if (_parent->cast<ExchangeSourceOperatorX>()._is_merging) {
         vsort_exec_exprs.close(state);
diff --git a/be/src/pipeline/exec/rec_cte_anchor_sink_operator.h b/be/src/pipeline/exec/rec_cte_anchor_sink_operator.h
@@ -58,8 +58,7 @@ class RecCTEAnchorSinkOperatorX MOCK_REMOVE(final)
     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) {}
+            : Base(sink_id, tnode.node_id, dest_id), _row_descriptor(descs, tnode.row_tuples) {}
 
     ~RecCTEAnchorSinkOperatorX() override = default;
 
@@ -72,13 +71,20 @@ class RecCTEAnchorSinkOperatorX MOCK_REMOVE(final)
         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, vectorized::Block* input_block, bool eos) override {
         auto& local_state = get_local_state(state);
 
-        if (_need_notify_rec_side_ready) {
-            RETURN_IF_ERROR(get_local_state(state)._shared_state->send_data_to_targets(state, 0));
-            _need_notify_rec_side_ready = false;
-        }
+        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) {
@@ -103,6 +109,14 @@ class RecCTEAnchorSinkOperatorX MOCK_REMOVE(final)
     }
 
 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;
     vectorized::VExprContextSPtrs _child_expr;
 
diff --git a/be/src/pipeline/exec/rec_cte_sink_operator.h b/be/src/pipeline/exec/rec_cte_sink_operator.h
@@ -55,8 +55,7 @@ class RecCTESinkOperatorX MOCK_REMOVE(final) : public DataSinkOperatorX<RecCTESi
     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) {}
+            : Base(sink_id, tnode.node_id, dest_id), _row_descriptor(descs, tnode.row_tuples) {}
 
     ~RecCTESinkOperatorX() override = default;
 
diff --git a/be/src/pipeline/exec/rec_cte_source_operator.cpp b/be/src/pipeline/exec/rec_cte_source_operator.cpp
@@ -41,14 +41,10 @@ Status RecCTESourceLocalState::init(RuntimeState* state, LocalStateInfo& info) {
     _shared_state->anchor_dep = _anchor_dependency.get();
 
     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]));
-    }
     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(),
-                                                               get_data_types(_child_expr), false));
+                                                               p._hash_table_key_types, false));
     }
 
     _shared_state->hash_table_compute_timer =
@@ -70,16 +66,19 @@ Status RecCTESourceOperatorX::init(const TPlanNode& tnode, RuntimeState* state)
         const auto& texprs = tnode.rec_cte_node.result_expr_lists[1];
         vectorized::VExprContextSPtrs ctxs;
         RETURN_IF_ERROR(vectorized::VExpr::create_expr_trees(texprs, ctxs));
-        _child_expr = ctxs;
+        _hash_table_key_types = get_data_types(ctxs);
     }
     return Status::OK();
 }
 
 Status RecCTESourceOperatorX::prepare(RuntimeState* state) {
     RETURN_IF_ERROR(Base::prepare(state));
-    RETURN_IF_ERROR(vectorized::VExpr::prepare(_child_expr, state, _child->row_desc()));
-    RETURN_IF_ERROR(vectorized::VExpr::check_expr_output_type(_child_expr, _child->row_desc()));
-    RETURN_IF_ERROR(vectorized::VExpr::open(_child_expr, 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();
 }
 
diff --git a/be/src/pipeline/exec/rec_cte_source_operator.h b/be/src/pipeline/exec/rec_cte_source_operator.h
@@ -58,8 +58,6 @@ class RecCTESourceLocalState final : public PipelineXLocalState<RecCTESharedStat
     friend class RecCTESourceOperatorX;
     friend class OperatorX<RecCTESourceLocalState>;
 
-    vectorized::VExprContextSPtrs _child_expr;
-
     std::shared_ptr<Dependency> _anchor_dependency = nullptr;
 };
 
@@ -135,13 +133,13 @@ class RecCTESourceOperatorX : public OperatorX<RecCTESourceLocalState> {
 
 private:
     Status _send_close(RuntimeState* state) {
-        if (!_aready_send_close && !_is_used_by_other_rec_cte) {
-            RETURN_IF_ERROR(_send_rerun_fragments(state, PRerunFragmentParams::close));
-            _aready_send_close = true;
-
+        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::close));
         }
         return Status::OK();
     }
@@ -179,10 +177,6 @@ class RecCTESourceOperatorX : public OperatorX<RecCTESourceLocalState> {
 
     Status _send_rerun_fragments(RuntimeState* state, PRerunFragmentParams_Opcode stage) const {
         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);
-            }
             auto stub =
                     state->get_query_ctx()->exec_env()->brpc_internal_client_cache()->get_client(
                             fragment.addr);
@@ -202,14 +196,16 @@ class RecCTESourceOperatorX : public OperatorX<RecCTESourceLocalState> {
                 return Status::InternalError(controller.ErrorText());
             }
 
-            RETURN_IF_ERROR(Status::create(result.status()));
+            auto rpc_st = Status::create(result.status());
+            if (!rpc_st.ok()) {
+                return rpc_st;
+            }
         }
         return Status::OK();
     }
 
     friend class RecCTESourceLocalState;
-
-    vectorized::VExprContextSPtrs _child_expr;
+    std::vector<vectorized::DataTypePtr> _hash_table_key_types;
 
     const bool _is_union_all = false;
     std::vector<TRecCTETarget> _targets;
@@ -218,7 +214,7 @@ class RecCTESourceOperatorX : public OperatorX<RecCTESourceLocalState> {
 
     int _max_recursion_depth = 0;
 
-    bool _aready_send_close = false;
+    bool _already_send_close = false;
 
     bool _is_used_by_other_rec_cte = false;
 };
diff --git a/be/src/pipeline/exec/scan_operator.cpp b/be/src/pipeline/exec/scan_operator.cpp
@@ -1366,6 +1366,7 @@ Status ScanLocalState<Derived>::close(RuntimeState* state) {
     SCOPED_TIMER(exec_time_counter());
     if (_scanner_ctx) {
         _scanner_ctx->stop_scanners(state);
+        _scanner_ctx.reset();
     }
     std::list<std::shared_ptr<vectorized::ScannerDelegate>> {}.swap(_scanners);
     COUNTER_SET(_wait_for_dependency_timer, _scan_dependency->watcher_elapse_time());
diff --git a/be/src/pipeline/pipeline_fragment_context.cpp b/be/src/pipeline/pipeline_fragment_context.cpp
@@ -320,7 +320,7 @@ Status PipelineFragmentContext::prepare(ThreadPool* thread_pool) {
 
         auto* fragment_context = this;
 
-        if (!_need_notify_close && _params.query_options.__isset.is_report_success) {
+        if (_params.query_options.__isset.is_report_success) {
             fragment_context->set_is_report_success(_params.query_options.is_report_success);
         }
 
@@ -1793,10 +1793,16 @@ void PipelineFragmentContext::_close_fragment_instance() {
     }
     Defer defer_op {[&]() {
         _is_fragment_instance_closed = true;
-        _close_cv.notify_all();
+        _notify_cv.notify_all();
     }};
     _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
@@ -1971,8 +1977,11 @@ 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:\nneed_notify_close: {}\n",
-                   _need_notify_close);
+    fmt::format_to(debug_string_buffer,
+                   "PipelineFragmentContext Info: _closed_tasks={}, _total_tasks={}, "
+                   "need_notify_close={}, has_task_execution_ctx_ref_count={}\n",
+                   _closed_tasks, _total_tasks, _need_notify_close,
+                   _has_task_execution_ctx_ref_count);
     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++) {
@@ -2057,10 +2066,17 @@ Status PipelineFragmentContext::wait_close(bool close) {
 
     {
         std::unique_lock<std::mutex> lock(_task_mutex);
-        _close_cv.wait(lock, [this] { return _is_fragment_instance_closed.load(); });
+        _notify_cv.wait(lock, [this] {
+            return _is_fragment_instance_closed.load() && !_has_task_execution_ctx_ref_count;
+        });
     }
 
     if (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());
+        }
         _exec_env->fragment_mgr()->remove_pipeline_context({_query_id, _fragment_id});
     }
     return Status::OK();
diff --git a/be/src/pipeline/pipeline_fragment_context.h b/be/src/pipeline/pipeline_fragment_context.h
@@ -195,7 +195,6 @@ class PipelineFragmentContext : public TaskExecutionContext {
     Pipelines _pipelines;
     PipelineId _next_pipeline_id = 0;
     std::mutex _task_mutex;
-    std::condition_variable _close_cv;
     int _closed_tasks = 0;
     // After prepared, `_total_tasks` is equal to the size of `_tasks`.
     // When submit fail, `_total_tasks` is equal to the number of tasks submitted.
diff --git a/be/src/pipeline/pipeline_task.cpp b/be/src/pipeline/pipeline_task.cpp
@@ -32,6 +32,7 @@
 #include "pipeline/dependency.h"
 #include "pipeline/exec/exchange_source_operator.h"
 #include "pipeline/exec/operator.h"
+#include "pipeline/exec/rec_cte_source_operator.h"
 #include "pipeline/exec/scan_operator.h"
 #include "pipeline/pipeline.h"
 #include "pipeline/pipeline_fragment_context.h"
@@ -553,6 +554,10 @@ Status PipelineTask::execute(bool* done) {
                 }
             }
 
+            if (_eos && !_sink->need_rerun(_state)) {
+                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>(
@@ -602,8 +607,6 @@ Status PipelineTask::execute(bool* done) {
                                 "Only ExchangeSourceOperatorX can be rerun, real is {}",
                                 _root->get_name());
                     }
-                } else {
-                    RETURN_IF_ERROR(close(Status::OK(), false));
                 }
             }
 
@@ -739,7 +742,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);
             }
         }
     }
diff --git a/be/src/pipeline/rec_cte_shared_state.h b/be/src/pipeline/rec_cte_shared_state.h
@@ -114,6 +114,9 @@ struct RecCTESharedState : public BasicSharedState {
     }
 
     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();
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
@@ -1437,18 +1437,16 @@ Status FragmentMgr::rerun_fragment(const TUniqueId& query_id, int fragment,
 
         if (stage == PRerunFragmentParams::wait) {
             return fragment_ctx->wait_close(false);
-        }
-        if (stage == PRerunFragmentParams::release) {
+        } else if (stage == PRerunFragmentParams::release) {
             return fragment_ctx->set_to_rerun();
-        }
-        if (stage == PRerunFragmentParams::rebuild) {
+        } else if (stage == PRerunFragmentParams::rebuild) {
             return fragment_ctx->rebuild(_thread_pool.get());
-        }
-        if (stage == PRerunFragmentParams::submit) {
+        } else if (stage == PRerunFragmentParams::submit) {
             return fragment_ctx->submit();
-        }
-        if (stage == PRerunFragmentParams::close) {
+        } else if (stage == PRerunFragmentParams::close) {
             return fragment_ctx->wait_close(true);
+        } else {
+            return Status::InvalidArgument("Unknown rerun fragment opcode: {}", stage);
         }
     } else {
         return Status::NotFound(
diff --git 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.
diff --git a/be/src/runtime/task_execution_context.cpp b/be/src/runtime/task_execution_context.cpp
diff --git a/be/src/runtime/task_execution_context.h b/be/src/runtime/task_execution_context.h
diff --git a/be/src/runtime_filter/runtime_filter_mgr.cpp b/be/src/runtime_filter/runtime_filter_mgr.cpp
diff --git a/be/src/runtime_filter/runtime_filter_mgr.h b/be/src/runtime_filter/runtime_filter_mgr.h
diff --git a/be/src/vec/exec/scan/scanner_context.cpp b/be/src/vec/exec/scan/scanner_context.cpp
diff --git a/be/src/vec/runtime/vdata_stream_recvr.cpp b/be/src/vec/runtime/vdata_stream_recvr.cpp
diff --git a/be/src/vec/sink/vdata_stream_sender.cpp b/be/src/vec/sink/vdata_stream_sender.cpp

Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,7 @@ ExchangeLocalState::~ExchangeLocalState() {`
`50`	`50`	`std::string ExchangeLocalState::debug_string(int indentation_level) const {`
`51`	`51`	`fmt::memory_buffer debug_string_buffer;`
`52`	`52`	`fmt::format_to(debug_string_buffer, "{}, recvr: ({})", Base::debug_string(indentation_level),`
`53`		`- stream_recvr->debug_string());`
	`53`	`+ stream_recvr ? stream_recvr->debug_string() : "null");`
`54`	`54`	`return fmt::to_string(debug_string_buffer);`
`55`	`55`	`}`
`56`	`56`
`@@ -210,6 +210,7 @@ Status ExchangeLocalState::close(RuntimeState* state) {`
`210`	`210`	`}`
`211`	`211`	`if (stream_recvr != nullptr) {`
`212`	`212`	`stream_recvr->close();`
	`213`	`+ stream_recvr.reset();`
`213`	`214`	`}`
`214`	`215`	`if (_parent->cast<ExchangeSourceOperatorX>()._is_merging) {`
`215`	`216`	`vsort_exec_exprs.close(state);`
Original file line number	Diff line number	Diff line change
`@@ -1366,6 +1366,7 @@ Status ScanLocalState<Derived>::close(RuntimeState* state) {`
`1366`	`1366`	`SCOPED_TIMER(exec_time_counter());`
`1367`	`1367`	`if (_scanner_ctx) {`
`1368`	`1368`	`_scanner_ctx->stop_scanners(state);`
	`1369`	`+ _scanner_ctx.reset();`
`1369`	`1370`	`}`
`1370`	`1371`	`std::list<std::shared_ptr<vectorized::ScannerDelegate>> {}.swap(_scanners);`
`1371`	`1372`	`COUNTER_SET(_wait_for_dependency_timer, _scan_dependency->watcher_elapse_time());`