upd

Mryange · Mryange · commit 61cbca23264c · 2026-04-01T20:19:15.000+08:00
diff --git a/be/src/exec/operator/distinct_streaming_aggregation_operator.cpp b/be/src/exec/operator/distinct_streaming_aggregation_operator.cpp
@@ -153,18 +153,14 @@ Status DistinctStreamingAggLocalState::_distinct_pre_agg_with_serialized_key(
 
     size_t key_size = _probe_expr_ctxs.size();
     ColumnRawPtrs key_columns(key_size);
-    std::vector<int> result_idxs(key_size);
+    Columns key_column_ptrs(key_size);
     {
         SCOPED_TIMER(_expr_timer);
         for (size_t i = 0; i < key_size; ++i) {
-            int result_column_id = -1;
-            RETURN_IF_ERROR(_probe_expr_ctxs[i]->execute(in_block, &result_column_id));
-            in_block->get_by_position(result_column_id).column =
-                    in_block->get_by_position(result_column_id)
-                            .column->convert_to_full_column_if_const();
-            key_columns[i] = in_block->get_by_position(result_column_id).column.get();
+            RETURN_IF_ERROR(_probe_expr_ctxs[i]->execute(in_block, key_column_ptrs[i]));
+            key_column_ptrs[i] = key_column_ptrs[i]->convert_to_full_column_if_const();
+            key_columns[i] = key_column_ptrs[i].get();
             key_columns[i]->assume_mutable()->replace_float_special_values();
-            result_idxs[i] = result_column_id;
         }
     }
 
@@ -198,12 +194,9 @@ Status DistinctStreamingAggLocalState::_distinct_pre_agg_with_serialized_key(
         }
         DCHECK_EQ(out_block->columns(), key_size);
         if (_stop_emplace_flag && _distinct_row.empty()) {
-            // If _stop_emplace_flag is true and _distinct_row is also empty, it means it is in streaming mode, outputting what is input
-            // swap the column directly, to solve Check failed: d.column->use_count() == 1 (2 vs. 1)
+            // Streaming mode: move key columns directly into output block
             for (int i = 0; i < key_size; ++i) {
-                auto output_column = out_block->get_by_position(i).column;
-                out_block->replace_by_position(i, key_columns[i]->assume_mutable());
-                in_block->replace_by_position(result_idxs[i], output_column);
+                out_block->replace_by_position(i, std::move(key_column_ptrs[i]));
             }
         } else {
             DCHECK_EQ(_cache_block.rows(), 0);
@@ -231,7 +224,7 @@ Status DistinctStreamingAggLocalState::_distinct_pre_agg_with_serialized_key(
         ColumnsWithTypeAndName columns_with_schema;
         for (int i = 0; i < key_size; ++i) {
             if (_stop_emplace_flag) {
-                columns_with_schema.emplace_back(key_columns[i]->assume_mutable(),
+                columns_with_schema.emplace_back(std::move(key_column_ptrs[i]),
                                                  _probe_expr_ctxs[i]->root()->data_type(),
                                                  _probe_expr_ctxs[i]->root()->expr_name());
             } else {
diff --git a/be/src/exec/operator/hashjoin_probe_operator.cpp b/be/src/exec/operator/hashjoin_probe_operator.cpp
@@ -135,10 +135,9 @@ Status HashJoinProbeLocalState::close(RuntimeState* state) {
     return JoinProbeLocalState<HashJoinSharedState, HashJoinProbeLocalState>::close(state);
 }
 
-bool HashJoinProbeLocalState::_need_probe_null_map(Block& block,
-                                                   const std::vector<int>& res_col_ids) {
+bool HashJoinProbeLocalState::_need_probe_null_map(const std::vector<ColumnPtr>& res_columns) {
     for (size_t i = 0; i < _probe_expr_ctxs.size(); ++i) {
-        const auto* column = block.get_by_position(res_col_ids[i]).column.get();
+        const auto* column = res_columns[i].get();
         if (column->is_nullable() &&
             !_parent->cast<HashJoinProbeOperatorX>()._serialize_null_into_key[i]) {
             return true;
@@ -329,7 +328,7 @@ std::string HashJoinProbeLocalState::debug_string(int indentation_level) const {
 }
 
 Status HashJoinProbeLocalState::_extract_join_column(Block& block,
-                                                     const std::vector<int>& res_col_ids) {
+                                                     const std::vector<ColumnPtr>& res_columns) {
     if (empty_right_table_shortcut()) {
         return Status::OK();
     }
@@ -338,7 +337,7 @@ Status HashJoinProbeLocalState::_extract_join_column(Block& block,
 
     if (!_has_set_need_null_map_for_probe) {
         _has_set_need_null_map_for_probe = true;
-        _need_null_map_for_probe = _need_probe_null_map(block, res_col_ids);
+        _need_null_map_for_probe = _need_probe_null_map(res_columns);
     }
     if (_need_null_map_for_probe) {
         if (!_null_map_column) {
@@ -349,11 +348,10 @@ Status HashJoinProbeLocalState::_extract_join_column(Block& block,
 
     auto& shared_state = *_shared_state;
     for (size_t i = 0; i < shared_state.build_exprs_size; ++i) {
-        const auto* column = block.get_by_position(res_col_ids[i]).column.get();
+        const auto* column = res_columns[i].get();
         if (!column->is_nullable() &&
             _parent->cast<HashJoinProbeOperatorX>()._serialize_null_into_key[i]) {
-            _key_columns_holder.emplace_back(
-                    make_nullable(block.get_by_position(res_col_ids[i]).column));
+            _key_columns_holder.emplace_back(make_nullable(res_columns[i]));
             _probe_columns[i] = _key_columns_holder.back().get();
         } else if (const auto* nullable = check_and_get_column<ColumnNullable>(*column);
                    nullable &&
@@ -363,8 +361,11 @@ Status HashJoinProbeLocalState::_extract_join_column(Block& block,
             const auto& col_nullmap = nullable->get_null_map_data();
             DCHECK(_null_map_column);
             VectorizedUtils::update_null_map(_null_map_column->get_data(), col_nullmap);
+            // Hold the column to keep nested column pointer alive
+            _key_columns_holder.emplace_back(res_columns[i]);
             _probe_columns[i] = &col_nested;
         } else {
+            _key_columns_holder.emplace_back(res_columns[i]);
             _probe_columns[i] = column;
         }
     }
@@ -411,19 +412,16 @@ bool HashJoinProbeOperatorX::need_more_input_data(RuntimeState* state) const {
 
 Status HashJoinProbeOperatorX::_do_evaluate(Block& block, VExprContextSPtrs& exprs,
                                             RuntimeProfile::Counter& expr_call_timer,
-                                            std::vector<int>& res_col_ids) const {
+                                            std::vector<ColumnPtr>& res_columns) const {
     for (size_t i = 0; i < exprs.size(); ++i) {
-        int result_col_id = -1;
-        // execute build column
+        // execute probe column
         {
             SCOPED_TIMER(&expr_call_timer);
-            RETURN_IF_ERROR(exprs[i]->execute(&block, &result_col_id));
+            RETURN_IF_ERROR(exprs[i]->execute(&block, res_columns[i]));
         }
 
         // TODO: opt the column is const
-        block.get_by_position(result_col_id).column =
-                block.get_by_position(result_col_id).column->convert_to_full_column_if_const();
-        res_col_ids[i] = result_col_id;
+        res_columns[i] = res_columns[i]->convert_to_full_column_if_const();
     }
     return Status::OK();
 }
@@ -438,15 +436,17 @@ Status HashJoinProbeOperatorX::push(RuntimeState* state, Block* input_block, boo
 
     if (rows > 0) {
         COUNTER_UPDATE(local_state._probe_rows_counter, rows);
-        std::vector<int> res_col_ids(local_state._probe_expr_ctxs.size());
+        std::vector<ColumnPtr> res_columns(local_state._probe_expr_ctxs.size());
         RETURN_IF_ERROR(_do_evaluate(*input_block, local_state._probe_expr_ctxs,
-                                     *local_state._probe_expr_call_timer, res_col_ids));
+                                     *local_state._probe_expr_call_timer, res_columns));
         if (_join_op == TJoinOp::RIGHT_OUTER_JOIN || _join_op == TJoinOp::FULL_OUTER_JOIN) {
             local_state._probe_column_convert_to_null =
                     local_state._convert_block_to_null(*input_block);
         }
 
-        RETURN_IF_ERROR(local_state._extract_join_column(*input_block, res_col_ids));
+        RETURN_IF_ERROR(local_state._extract_join_column(*input_block, res_columns));
+        // Release ColumnPtrs to restore reference counts on block columns
+        res_columns.clear();
 
         local_state._estimate_memory_usage += (input_block->allocated_bytes() - origin_size);
 
diff --git a/be/src/exec/operator/hashjoin_probe_operator.h b/be/src/exec/operator/hashjoin_probe_operator.h
@@ -69,9 +69,9 @@ class HashJoinProbeLocalState MOCK_REMOVE(final)
 
 private:
     void _prepare_probe_block();
-    bool _need_probe_null_map(Block& block, const std::vector<int>& res_col_ids);
+    bool _need_probe_null_map(const std::vector<ColumnPtr>& res_columns);
     std::vector<uint16_t> _convert_block_to_null(Block& block);
-    Status _extract_join_column(Block& block, const std::vector<int>& res_col_ids);
+    Status _extract_join_column(Block& block, const std::vector<ColumnPtr>& res_columns);
     friend class HashJoinProbeOperatorX;
     template <int JoinOpType>
     friend struct ProcessHashTableProbe;
@@ -182,7 +182,7 @@ class HashJoinProbeOperatorX MOCK_REMOVE(final)
 private:
     Status _do_evaluate(Block& block, VExprContextSPtrs& exprs,
                         RuntimeProfile::Counter& expr_call_timer,
-                        std::vector<int>& res_col_ids) const;
+                        std::vector<ColumnPtr>& res_columns) const;
     friend class HashJoinProbeLocalState;
     template <int JoinOpType>
     friend struct ProcessHashTableProbe;
diff --git a/be/src/exec/operator/set_probe_sink_operator.cpp b/be/src/exec/operator/set_probe_sink_operator.cpp
@@ -142,15 +142,13 @@ Status SetProbeSinkOperatorX<is_intersect>::_extract_probe_column(
     auto& build_not_ignore_null = local_state._shared_state->build_not_ignore_null;
 
     auto& child_exprs = local_state._child_exprs;
+    local_state._probe_column_holders.resize(child_exprs.size());
     for (size_t i = 0; i < child_exprs.size(); ++i) {
-        int result_col_id = -1;
-        RETURN_IF_ERROR(child_exprs[i]->execute(&block, &result_col_id));
+        ColumnPtr result_column;
+        RETURN_IF_ERROR(child_exprs[i]->execute(&block, result_column));
+        result_column = result_column->convert_to_full_column_if_const();
 
-        block.get_by_position(result_col_id).column =
-                block.get_by_position(result_col_id).column->convert_to_full_column_if_const();
-        const auto* column = block.get_by_position(result_col_id).column.get();
-
-        if (const auto* nullable = check_and_get_column<ColumnNullable>(*column)) {
+        if (const auto* nullable = check_and_get_column<ColumnNullable>(*result_column)) {
             if (!build_not_ignore_null[i]) {
                 return Status::InternalError(
                         "SET operator expects a nullable : {} column in column {}, but the "
@@ -160,16 +158,13 @@ Status SetProbeSinkOperatorX<is_intersect>::_extract_probe_column(
                         nullable->get_nested_column_ptr()->is_nullable());
             }
             raw_ptrs[i] = nullable;
+            local_state._probe_column_holders[i] = std::move(result_column);
         } else {
             if (build_not_ignore_null[i]) {
-                auto column_ptr = make_nullable(block.get_by_position(result_col_id).column, false);
-                local_state._probe_column_inserted_id.emplace_back(block.columns());
-                block.insert(
-                        {column_ptr, make_nullable(block.get_by_position(result_col_id).type), ""});
-                column = column_ptr.get();
+                result_column = make_nullable(result_column, false);
             }
-
-            raw_ptrs[i] = column;
+            raw_ptrs[i] = result_column.get();
+            local_state._probe_column_holders[i] = std::move(result_column);
         }
     }
     return Status::OK();
diff --git a/be/src/exec/operator/set_probe_sink_operator.h b/be/src/exec/operator/set_probe_sink_operator.h
@@ -54,8 +54,8 @@ class SetProbeSinkLocalState final : public PipelineXSinkLocalState<SetSharedSta
 
     int64_t _estimate_memory_usage = 0;
 
-    //record insert column id during probe
-    std::vector<uint16_t> _probe_column_inserted_id;
+    // Holds ColumnPtr references to keep _probe_columns raw pointers valid
+    std::vector<ColumnPtr> _probe_column_holders;
     ColumnRawPtrs _probe_columns;
     // every child has its result expr list
     VExprContextSPtrs _child_exprs;
diff --git a/be/src/exec/scan/scanner.cpp b/be/src/exec/scan/scanner.cpp
@@ -189,13 +189,15 @@ Status Scanner::_do_projections(Block* origin_block, Block* output_block) {
     }
     Block input_block = *origin_block;
 
-    std::vector<int> result_column_ids;
     for (auto& projections : _intermediate_projections) {
-        result_column_ids.resize(projections.size());
+        ColumnsWithTypeAndName columns_with_schema;
+        columns_with_schema.reserve(projections.size());
         for (int i = 0; i < projections.size(); i++) {
-            RETURN_IF_ERROR(projections[i]->execute(&input_block, &result_column_ids[i]));
+            ColumnWithTypeAndName result_data;
+            RETURN_IF_ERROR(projections[i]->execute(&input_block, result_data));
+            columns_with_schema.emplace_back(std::move(result_data));
         }
-        input_block.shuffle_columns(result_column_ids);
+        input_block = Block(std::move(columns_with_schema));
     }
 
     DCHECK_EQ(rows, input_block.rows());
diff --git a/be/src/exprs/vectorized_agg_fn.cpp b/be/src/exprs/vectorized_agg_fn.cpp
@@ -31,7 +31,6 @@
 #include "common/object_pool.h"
 #include "core/block/block.h"
 #include "core/block/column_with_type_and_name.h"
-#include "core/block/materialize_block.h"
 #include "core/data_type/data_type_agg_state.h"
 #include "core/data_type/data_type_factory.hpp"
 #include "exec/common/util.hpp"
@@ -282,28 +281,35 @@ void AggFnEvaluator::destroy(AggregateDataPtr place) {
 
 Status AggFnEvaluator::execute_single_add(Block* block, AggregateDataPtr place, Arena& arena) {
     RETURN_IF_ERROR(_calc_argument_columns(block));
-    _function->add_batch_single_place(block->rows(), place, _agg_columns.data(), arena);
+    _function->add_batch_single_place(block->rows(), place, _agg_raw_input_columns.data(), arena);
+    _reset_input_columns();
     return Status::OK();
 }
 
 Status AggFnEvaluator::execute_batch_add(Block* block, size_t offset, AggregateDataPtr* places,
                                          Arena& arena, bool agg_many) {
     RETURN_IF_ERROR(_calc_argument_columns(block));
-    _function->add_batch(block->rows(), places, offset, _agg_columns.data(), arena, agg_many);
+    _function->add_batch(block->rows(), places, offset, _agg_raw_input_columns.data(), arena,
+                         agg_many);
+    _reset_input_columns();
     return Status::OK();
 }
 
 Status AggFnEvaluator::execute_batch_add_selected(Block* block, size_t offset,
                                                   AggregateDataPtr* places, Arena& arena) {
     RETURN_IF_ERROR(_calc_argument_columns(block));
-    _function->add_batch_selected(block->rows(), places, offset, _agg_columns.data(), arena);
+    _function->add_batch_selected(block->rows(), places, offset, _agg_raw_input_columns.data(),
+                                  arena);
+    _reset_input_columns();
     return Status::OK();
 }
 
 Status AggFnEvaluator::streaming_agg_serialize_to_column(Block* block, MutableColumnPtr& dst,
                                                          const size_t num_rows, Arena& arena) {
     RETURN_IF_ERROR(_calc_argument_columns(block));
-    _function->streaming_agg_serialize_to_column(_agg_columns.data(), dst, num_rows, arena);
+    _function->streaming_agg_serialize_to_column(_agg_raw_input_columns.data(), dst, num_rows,
+                                                 arena);
+    _reset_input_columns();
     return Status::OK();
 }
 
@@ -340,23 +346,28 @@ std::string AggFnEvaluator::debug_string() const {
     return out.str();
 }
 
-Status AggFnEvaluator::_calc_argument_columns(Block* block) {
+Status AggFnEvaluator::_calc_argument_columns(const Block* block) {
     SCOPED_TIMER(_expr_timer);
-    _agg_columns.resize(_input_exprs_ctxs.size());
-    std::vector<int> column_ids(_input_exprs_ctxs.size());
+    _agg_input_columns.resize(_input_exprs_ctxs.size(), nullptr);
+    _agg_raw_input_columns.resize(_input_exprs_ctxs.size(), nullptr);
     for (int i = 0; i < _input_exprs_ctxs.size(); ++i) {
-        int column_id = -1;
-        RETURN_IF_ERROR(_input_exprs_ctxs[i]->execute(block, &column_id));
-        column_ids[i] = column_id;
-    }
-    materialize_block_inplace(*block, column_ids.data(),
-                              column_ids.data() + _input_exprs_ctxs.size());
-    for (int i = 0; i < _input_exprs_ctxs.size(); ++i) {
-        _agg_columns[i] = block->get_by_position(column_ids[i]).column.get();
+        RETURN_IF_ERROR(_input_exprs_ctxs[i]->execute(block, _agg_input_columns[i]));
+        _agg_input_columns[i] = _agg_input_columns[i]->convert_to_full_column_if_const();
+        _agg_raw_input_columns[i] = _agg_input_columns[i].get();
     }
     return Status::OK();
 }
 
+void AggFnEvaluator::_reset_input_columns() {
+    SCOPED_TIMER(_expr_timer);
+    for (auto& col : _agg_input_columns) {
+        col = nullptr;
+    }
+    for (auto& col : _agg_raw_input_columns) {
+        col = nullptr;
+    }
+}
+
 AggFnEvaluator* AggFnEvaluator::clone(RuntimeState* state, ObjectPool* pool) {
     return pool->add(AggFnEvaluator::create_unique(*this, state).release());
 }
@@ -374,7 +385,8 @@ AggFnEvaluator::AggFnEvaluator(AggFnEvaluator& evaluator, RuntimeState* state)
           _data_type(evaluator._data_type),
           _function(evaluator._function),
           _expr_name(evaluator._expr_name),
-          _agg_columns(evaluator._agg_columns) {
+          _agg_input_columns(evaluator._agg_input_columns),
+          _agg_raw_input_columns(evaluator._agg_raw_input_columns) {
     if (evaluator._fn.binary_type == TFunctionBinaryType::JAVA_UDF) {
         DataTypes tmp_argument_types;
         tmp_argument_types.reserve(evaluator._input_exprs_ctxs.size());
diff --git a/be/src/exprs/vectorized_agg_fn.h b/be/src/exprs/vectorized_agg_fn.h
@@ -127,7 +127,9 @@ class AggFnEvaluator {
               _without_key(without_key),
               _is_window_function(is_window_function) {};
 #endif
-    Status _calc_argument_columns(Block* block);
+    Status _calc_argument_columns(const Block* block);
+
+    void _reset_input_columns();
 
     DataTypes _argument_types_with_sort;
     DataTypes _real_argument_types;
@@ -149,7 +151,9 @@ class AggFnEvaluator {
 
     std::string _expr_name;
 
-    std::vector<const IColumn*> _agg_columns;
+    Columns _agg_input_columns;
+
+    std::vector<const IColumn*> _agg_raw_input_columns;
 };
 
 #include "common/compile_check_end.h"

Original file line number	Diff line number	Diff line change
`@@ -189,13 +189,15 @@ Status Scanner::_do_projections(Block* origin_block, Block* output_block) {`
`189`	`189`	`}`
`190`	`190`	`Block input_block = *origin_block;`
`191`	`191`
`192`		`- std::vector<int> result_column_ids;`
`193`	`192`	`for (auto& projections : _intermediate_projections) {`
`194`		`- result_column_ids.resize(projections.size());`
	`193`	`+ ColumnsWithTypeAndName columns_with_schema;`
	`194`	`+ columns_with_schema.reserve(projections.size());`
`195`	`195`	`for (int i = 0; i < projections.size(); i++) {`
`196`		`- RETURN_IF_ERROR(projections[i]->execute(&input_block, &result_column_ids[i]));`
	`196`	`+ ColumnWithTypeAndName result_data;`
	`197`	`+ RETURN_IF_ERROR(projections[i]->execute(&input_block, result_data));`
	`198`	`+ columns_with_schema.emplace_back(std::move(result_data));`
`197`	`199`	`}`
`198`		`- input_block.shuffle_columns(result_column_ids);`
	`200`	`+ input_block = Block(std::move(columns_with_schema));`
`199`	`201`	`}`
`200`	`202`
`201`	`203`	`DCHECK_EQ(rows, input_block.rows());`