[refactor] Remove free blocks and block reuse

Author: mrhhsg

be/src/pipeline/exec/data_queue.cpp

@@ -32,8 +32,6 @@ namespace pipeline {
 DataQueue::DataQueue(int child_count)
         : _queue_blocks_lock(child_count),
           _queue_blocks(child_count),
-          _free_blocks_lock(child_count),
-          _free_blocks(child_count),
           _child_count(child_count),
           _is_finished(child_count),
           _is_canceled(child_count),
@@ -42,7 +40,6 @@ DataQueue::DataQueue(int child_count)
           _flag_queue_idx(0) {
     for (int i = 0; i < child_count; ++i) {
         _queue_blocks_lock[i].reset(new std::mutex());
-        _free_blocks_lock[i].reset(new std::mutex());
         _is_finished[i] = false;
         _is_canceled[i] = false;
         _cur_bytes_in_queue[i] = 0;
@@ -52,36 +49,6 @@ DataQueue::DataQueue(int child_count)
     _sink_dependencies.resize(child_count, nullptr);
 }
 
-std::unique_ptr<vectorized::Block> DataQueue::get_free_block(int child_idx) {
-    {
-        INJECT_MOCK_SLEEP(std::lock_guard<std::mutex> l(*_free_blocks_lock[child_idx]));
-        if (!_free_blocks[child_idx].empty()) {
-            auto block = std::move(_free_blocks[child_idx].front());
-            _free_blocks[child_idx].pop_front();
-            return block;
-        }
-    }
-
-    return vectorized::Block::create_unique();
-}
-
-void DataQueue::push_free_block(std::unique_ptr<vectorized::Block> block, int child_idx) {
-    DCHECK(block->rows() == 0);
-
-    if (!_is_low_memory_mode) {
-        INJECT_MOCK_SLEEP(std::lock_guard<std::mutex> l(*_free_blocks_lock[child_idx]));
-        _free_blocks[child_idx].emplace_back(std::move(block));
-    }
-}
-
-void DataQueue::clear_free_blocks() {
-    for (size_t child_idx = 0; child_idx < _free_blocks.size(); ++child_idx) {
-        std::lock_guard<std::mutex> l(*_free_blocks_lock[child_idx]);
-        std::deque<std::unique_ptr<vectorized::Block>> tmp_queue;
-        _free_blocks[child_idx].swap(tmp_queue);
-    }
-}
-
 void DataQueue::terminate() {
     for (int i = 0; i < _queue_blocks.size(); i++) {
         set_finish(i);
@@ -93,7 +60,6 @@ void DataQueue::terminate() {
             _sink_dependencies[i]->set_always_ready();
         }
     }
-    clear_free_blocks();
 }
 
 //check which queue have data, and save the idx in _flag_queue_idx,

be/src/pipeline/exec/data_queue.h

@@ -42,12 +42,8 @@ class DataQueue {
 
     Status push_block(std::unique_ptr<vectorized::Block> block, int child_idx = 0);
 
-    std::unique_ptr<vectorized::Block> get_free_block(int child_idx = 0);
-
     void push_free_block(std::unique_ptr<vectorized::Block> output_block, int child_idx = 0);
 
-    void clear_free_blocks();
-
     void set_finish(int child_idx = 0);
     void set_canceled(int child_idx = 0); // should set before finish
     bool is_finish(int child_idx = 0);
@@ -76,7 +72,6 @@ class DataQueue {
     void set_low_memory_mode() {
         _is_low_memory_mode = true;
         _max_blocks_in_sub_queue = 1;
-        clear_free_blocks();
     }
 
     void terminate();
@@ -85,9 +80,6 @@ class DataQueue {
     std::vector<std::unique_ptr<std::mutex>> _queue_blocks_lock;
     std::vector<std::deque<std::unique_ptr<vectorized::Block>>> _queue_blocks;
 
-    std::vector<std::unique_ptr<std::mutex>> _free_blocks_lock;
-    std::vector<std::deque<std::unique_ptr<vectorized::Block>>> _free_blocks;
-
     //how many deque will be init, always will be one
     int _child_count = 0;
     std::vector<std::atomic_bool> _is_finished;

be/src/pipeline/exec/operator.cpp

@@ -17,6 +17,8 @@
 
 #include "operator.h"
 
+#include <glog/logging.h>
+
 #include "common/status.h"
 #include "pipeline/dependency.h"
 #include "pipeline/exec/aggregation_sink_operator.h"
@@ -84,6 +86,8 @@
 #include "util/debug_util.h"
 #include "util/runtime_profile.h"
 #include "util/string_util.h"
+#include "vec/columns/column_nullable.h"
+#include "vec/core/block.h"
 #include "vec/exprs/vexpr.h"
 #include "vec/exprs/vexpr_context.h"
 #include "vec/utils/util.hpp"
@@ -322,46 +326,31 @@ Status OperatorXBase::do_projections(RuntimeState* state, vectorized::Block* ori
     }
 
     DCHECK_EQ(rows, input_block.rows());
-    auto insert_column_datas = [&](auto& to, vectorized::ColumnPtr& from, size_t rows) {
-        if (to->is_nullable() && !from->is_nullable()) {
-            if (_keep_origin || !from->is_exclusive()) {
-                auto& null_column = reinterpret_cast<vectorized::ColumnNullable&>(*to);
-                null_column.get_nested_column().insert_range_from(*from, 0, rows);
-                null_column.get_null_map_column().get_data().resize_fill(rows, 0);
-                bytes_usage += null_column.allocated_bytes();
-            } else {
-                to = make_nullable(from, false)->assume_mutable();
-            }
-        } else {
-            if (_keep_origin || !from->is_exclusive()) {
-                to->insert_range_from(*from, 0, rows);
-                bytes_usage += from->allocated_bytes();
-            } else {
-                to = from->assume_mutable();
-            }
-        }
-    };
-
-    using namespace vectorized;
-    vectorized::MutableBlock mutable_block =
-            vectorized::VectorizedUtils::build_mutable_mem_reuse_block(output_block,
-                                                                       *_output_row_descriptor);
     if (rows != 0) {
+        using namespace vectorized;
+        MutableBlock mutable_block =
+                VectorizedUtils::build_mutable_block(output_block, *_output_row_descriptor);
         auto& mutable_columns = mutable_block.mutable_columns();
-        const size_t origin_columns_count = input_block.columns();
         DCHECK_EQ(mutable_columns.size(), local_state->_projections.size()) << debug_string();
         for (int i = 0; i < mutable_columns.size(); ++i) {
-            auto result_column_id = -1;
             ColumnPtr column_ptr;
             RETURN_IF_ERROR(local_state->_projections[i]->execute(&input_block, column_ptr));
             column_ptr = column_ptr->convert_to_full_column_if_const();
-            if (result_column_id >= origin_columns_count) {
-                bytes_usage += column_ptr->allocated_bytes();
+
+            if (mutable_columns[i]->is_nullable() && !column_ptr->is_nullable()) {
+                const auto size = column_ptr->size();
+                mutable_columns[i] = ColumnNullable::create(column_ptr->assume_mutable(),
+                                                            ColumnUInt8::create(size, 0))
+                                             ->assume_mutable();
+            } else {
+                mutable_columns[i] = column_ptr->assume_mutable();
             }
-            insert_column_datas(mutable_columns[i], column_ptr, rows);
         }
-        DCHECK(mutable_block.rows() == rows);
+        DCHECK_EQ(mutable_block.rows(), rows);
+        auto empty_columns = origin_block->clone_empty_columns();
+        origin_block->set_columns(std::move(empty_columns));
         output_block->set_columns(std::move(mutable_columns));
+        DCHECK_EQ(output_block->rows(), rows);
     }
 
     local_state->_estimate_memory_usage += bytes_usage;

be/src/pipeline/exec/repeat_operator.cpp

@@ -114,8 +114,7 @@ Status RepeatLocalState::get_repeated_block(vectorized::Block* input_block, int
     size_t input_column_size = input_block->columns();
     size_t output_column_size = p._output_slots.size();
     DCHECK_LT(input_column_size, output_column_size);
-    auto m_block = vectorized::VectorizedUtils::build_mutable_mem_reuse_block(output_block,
-                                                                              p._output_slots);
+    auto m_block = vectorized::VectorizedUtils::build_mutable_block(output_block, p._output_slots);
     auto& output_columns = m_block.mutable_columns();
     /* Fill all slots according to child, for example:select tc1,tc2,sum(tc3) from t1 group by grouping sets((tc1),(tc2));
      * insert into t1 values(1,2,1),(1,3,1),(2,1,1),(3,1,1);
@@ -238,8 +237,8 @@ Status RepeatOperatorX::pull(doris::RuntimeState* state, vectorized::Block* outp
                 _repeat_id_idx = 0;
             }
         } else if (local_state._expr_ctxs.empty()) {
-            auto m_block = vectorized::VectorizedUtils::build_mutable_mem_reuse_block(
-                    output_block, _output_slots);
+            auto m_block =
+                    vectorized::VectorizedUtils::build_mutable_block(output_block, _output_slots);
             auto rows = _child_block.rows();
             auto& columns = m_block.mutable_columns();
 

be/src/pipeline/exec/scan_operator.h

@@ -361,13 +361,7 @@ class ScanOperatorX : public OperatorX<LocalStateType> {
         return {ExchangeType::BUCKET_HASH_SHUFFLE};
     }
 
-    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();
-        }
-    }
+    void set_low_memory_mode(RuntimeState* state) override {}
 
     using OperatorX<LocalStateType>::node_id;
     using OperatorX<LocalStateType>::operator_id;

be/src/pipeline/exec/table_function_operator.cpp

@@ -155,8 +155,8 @@ bool TableFunctionLocalState::_is_inner_and_empty() {
 Status TableFunctionLocalState::get_expanded_block(RuntimeState* state,
                                                    vectorized::Block* output_block, bool* eos) {
     auto& p = _parent->cast<TableFunctionOperatorX>();
-    vectorized::MutableBlock m_block = vectorized::VectorizedUtils::build_mutable_mem_reuse_block(
-            output_block, p._output_slots);
+    vectorized::MutableBlock m_block =
+            vectorized::VectorizedUtils::build_mutable_block(output_block, p._output_slots);
     vectorized::MutableColumns& columns = m_block.mutable_columns();
 
     for (int i = 0; i < p._fn_num; i++) {

be/src/pipeline/exec/union_sink_operator.cpp

@@ -102,8 +102,7 @@ Status UnionSinkOperatorX::sink(RuntimeState* state, vectorized::Block* in_block
     SCOPED_TIMER(local_state.exec_time_counter());
     COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)in_block->rows());
     if (local_state._output_block == nullptr) {
-        local_state._output_block =
-                local_state._shared_state->data_queue.get_free_block(_cur_child_id);
+        local_state._output_block = vectorized::Block::create_unique();
     }
     if (_cur_child_id < _get_first_materialized_child_idx()) { //pass_through
         if (in_block->rows() > 0) {

be/src/pipeline/exec/union_sink_operator.h

@@ -149,9 +149,8 @@ class UnionSinkOperatorX MOCK_REMOVE(final) : public DataSinkOperatorX<UnionSink
         DCHECK_LT(child_id, _child_size);
         DCHECK(!is_child_passthrough(child_id));
         if (input_block->rows() > 0) {
-            vectorized::MutableBlock mblock =
-                    vectorized::VectorizedUtils::build_mutable_mem_reuse_block(output_block,
-                                                                               row_descriptor());
+            vectorized::MutableBlock mblock = vectorized::VectorizedUtils::build_mutable_block(
+                    output_block, row_descriptor());
             vectorized::Block res;
             RETURN_IF_ERROR(materialize_block(state, input_block, child_id, &res));
             RETURN_IF_ERROR(mblock.merge(res));

be/src/pipeline/exec/union_source_operator.cpp

@@ -137,8 +137,6 @@ Status UnionSourceOperatorX::get_block(RuntimeState* state, vectorized::Block* b
             return Status::OK();
         }
         block->swap(*output_block);
-        output_block->clear_column_data(row_descriptor().num_materialized_slots());
-        local_state._shared_state->data_queue.push_free_block(std::move(output_block), child_idx);
     }
     local_state.reached_limit(block, eos);
     return Status::OK();
@@ -153,7 +151,7 @@ Status UnionSourceOperatorX::get_next_const(RuntimeState* state, vectorized::Blo
 
     auto& _const_expr_list_idx = local_state._const_expr_list_idx;
     vectorized::MutableBlock mblock =
-            vectorized::VectorizedUtils::build_mutable_mem_reuse_block(block, row_descriptor());
+            vectorized::VectorizedUtils::build_mutable_block(block, row_descriptor());
 
     vectorized::ColumnsWithTypeAndName tmp_block_columns;
     for (; _const_expr_list_idx < _const_expr_lists.size() && mblock.rows() < state->batch_size();

be/src/pipeline/local_exchange/local_exchanger.cpp

@@ -165,7 +165,7 @@ Status ShuffleExchanger::get_block(RuntimeState* state, vectorized::Block* block
     if (_dequeue_data(source_info.local_state, partitioned_block, eos, block,
                       source_info.channel_id)) {
         SCOPED_TIMER(profile.copy_data_timer);
-        mutable_block = vectorized::VectorizedUtils::build_mutable_mem_reuse_block(
+        mutable_block = vectorized::VectorizedUtils::build_mutable_block(
                 block, partitioned_block.first->_data_block);
         RETURN_IF_ERROR(get_data());
     }
@@ -198,9 +198,6 @@ Status ShuffleExchanger::_split_rows(RuntimeState* state, const uint32_t* __rest
 
     vectorized::Block data_block;
     std::shared_ptr<BlockWrapper> new_block_wrapper;
-    if (!_free_blocks.try_dequeue(data_block)) {
-        data_block = block->clone_empty();
-    }
     data_block.swap(*block);
     new_block_wrapper =
             BlockWrapper::create_shared(std::move(data_block), local_state->_shared_state, -1);
@@ -264,9 +261,6 @@ Status ShuffleExchanger::_split_rows(RuntimeState* state, const uint32_t* __rest
 
     vectorized::Block data_block;
     std::shared_ptr<BlockWrapper> new_block_wrapper;
-    if (!_free_blocks.try_dequeue(data_block)) {
-        data_block = block->clone_empty();
-    }
     data_block.swap(*block);
     new_block_wrapper = BlockWrapper::create_shared(std::move(data_block), nullptr, -1);
     if (new_block_wrapper->_data_block.empty()) {
@@ -289,9 +283,6 @@ Status PassthroughExchanger::sink(RuntimeState* state, vectorized::Block* in_blo
         return Status::OK();
     }
     vectorized::Block new_block;
-    if (!_free_blocks.try_dequeue(new_block)) {
-        new_block = {in_block->clone_empty()};
-    }
     new_block.swap(*in_block);
     auto channel_id = ((*sink_info.channel_id)++) % _num_partitions;
     BlockWrapperSPtr wrapper = BlockWrapper::create_shared(
@@ -341,9 +332,6 @@ Status PassToOneExchanger::sink(RuntimeState* state, vectorized::Block* in_block
         return Status::OK();
     }
     vectorized::Block new_block;
-    if (!_free_blocks.try_dequeue(new_block)) {
-        new_block = {in_block->clone_empty()};
-    }
     new_block.swap(*in_block);
 
     BlockWrapperSPtr wrapper = BlockWrapper::create_shared(
@@ -370,10 +358,6 @@ Status PassToOneExchanger::get_block(RuntimeState* state, vectorized::Block* blo
 
 void ExchangerBase::finalize() {
     DCHECK(_running_source_operators == 0);
-    vectorized::Block block;
-    while (_free_blocks.try_dequeue(block)) {
-        // do nothing
-    }
 }
 
 Status BroadcastExchanger::sink(RuntimeState* state, vectorized::Block* in_block, bool eos,
@@ -382,9 +366,6 @@ Status BroadcastExchanger::sink(RuntimeState* state, vectorized::Block* in_block
         return Status::OK();
     }
     vectorized::Block new_block;
-    if (!_free_blocks.try_dequeue(new_block)) {
-        new_block = {in_block->clone_empty()};
-    }
     new_block.swap(*in_block);
     auto wrapper = BlockWrapper::create_shared(
             std::move(new_block),
@@ -415,9 +396,8 @@ Status BroadcastExchanger::get_block(RuntimeState* state, vectorized::Block* blo
     if (_dequeue_data(source_info.local_state, partitioned_block, eos, block,
                       source_info.channel_id)) {
         SCOPED_TIMER(profile.copy_data_timer);
-        vectorized::MutableBlock mutable_block =
-                vectorized::VectorizedUtils::build_mutable_mem_reuse_block(
-                        block, partitioned_block.first->_data_block);
+        vectorized::MutableBlock mutable_block = vectorized::VectorizedUtils::build_mutable_block(
+                block, partitioned_block.first->_data_block);
         auto block_wrapper = partitioned_block.first;
         RETURN_IF_ERROR(mutable_block.add_rows(&block_wrapper->_data_block,
                                                partitioned_block.second.offset_start,
@@ -431,9 +411,6 @@ Status AdaptivePassthroughExchanger::_passthrough_sink(RuntimeState* state,
                                                        vectorized::Block* in_block,
                                                        SinkInfo&& sink_info) {
     vectorized::Block new_block;
-    if (!_free_blocks.try_dequeue(new_block)) {
-        new_block = {in_block->clone_empty()};
-    }
     new_block.swap(*in_block);
     auto channel_id = ((*sink_info.channel_id)++) % _num_partitions;
     _enqueue_data_and_set_ready(