[refactor](merger) Simplify sort merger (#47689)

1. Delete unused variable in `VSortExecExprs`.
2. De-couple sort operator and local exchange operator.
3. Use local exchange 's profile to collect sort merger's metrics
instead of sort operator's.

Author: Gabriel39

be/src/pipeline/exec/exchange_source_operator.cpp

@@ -151,7 +151,7 @@ Status ExchangeSourceOperatorX::get_block(RuntimeState* state, vectorized::Block
     if (_is_merging && !local_state.is_ready) {
         SCOPED_TIMER(local_state.create_merger_timer);
         RETURN_IF_ERROR(local_state.stream_recvr->create_merger(
-                local_state.vsort_exec_exprs.lhs_ordering_expr_ctxs(), _is_asc_order, _nulls_first,
+                local_state.vsort_exec_exprs.ordering_expr_ctxs(), _is_asc_order, _nulls_first,
                 state->batch_size(), _limit, _offset));
         local_state.is_ready = true;
         return Status::OK();

be/src/pipeline/exec/sort_source_operator.cpp

@@ -67,18 +67,5 @@ const vectorized::SortDescription& SortSourceOperatorX::get_sort_description(
     return local_state._shared_state->sorter->get_sort_description();
 }
 
-Status SortSourceOperatorX::build_merger(RuntimeState* state,
-                                         std::unique_ptr<vectorized::VSortedRunMerger>& merger,
-                                         RuntimeProfile* profile) {
-    // now only use in LocalMergeSortExchanger::get_block
-    vectorized::VSortExecExprs vsort_exec_exprs;
-    // clone vsort_exec_exprs in LocalMergeSortExchanger
-    RETURN_IF_ERROR(_vsort_exec_exprs.clone(state, vsort_exec_exprs));
-    merger = std::make_unique<vectorized::VSortedRunMerger>(
-            vsort_exec_exprs.lhs_ordering_expr_ctxs(), _is_asc_order, _nulls_first,
-            state->batch_size(), _limit, _offset, profile);
-    return Status::OK();
-}
-
 #include "common/compile_check_end.h"
 } // namespace doris::pipeline

be/src/pipeline/exec/sort_source_operator.h

@@ -56,11 +56,10 @@ class SortSourceOperatorX final : public OperatorX<SortLocalState> {
     bool use_local_merge() const { return _merge_by_exchange; }
     const vectorized::SortDescription& get_sort_description(RuntimeState* state) const;
 
-    Status build_merger(RuntimeState* state, std::unique_ptr<vectorized::VSortedRunMerger>& merger,
-                        RuntimeProfile* profile);
-
 private:
+    friend class PipelineFragmentContext;
     friend class SortLocalState;
+
     const bool _merge_by_exchange;
     std::vector<bool> _is_asc_order;
     std::vector<bool> _nulls_first;

be/src/pipeline/local_exchange/local_exchanger.cpp

@@ -20,7 +20,6 @@
 #include "common/cast_set.h"
 #include "common/status.h"
 #include "pipeline/exec/sort_sink_operator.h"
-#include "pipeline/exec/sort_source_operator.h"
 #include "pipeline/local_exchange/local_exchange_sink_operator.h"
 #include "pipeline/local_exchange/local_exchange_source_operator.h"
 #include "vec/runtime/partitioner.h"
@@ -410,7 +409,14 @@ void LocalMergeSortExchanger::finalize() {
 
 Status LocalMergeSortExchanger::build_merger(RuntimeState* state,
                                              LocalExchangeSourceLocalState* local_state) {
-    RETURN_IF_ERROR(_sort_source->build_merger(state, _merger, local_state->profile()));
+    vectorized::VExprContextSPtrs ordering_expr_ctxs;
+    ordering_expr_ctxs.resize(_merge_info.ordering_expr_ctxs.size());
+    for (size_t i = 0; i < ordering_expr_ctxs.size(); i++) {
+        RETURN_IF_ERROR(_merge_info.ordering_expr_ctxs[i]->clone(state, ordering_expr_ctxs[i]));
+    }
+    _merger = std::make_unique<vectorized::VSortedRunMerger>(
+            ordering_expr_ctxs, _merge_info.is_asc_order, _merge_info.nulls_first,
+            state->batch_size(), _merge_info.limit, _merge_info.offset, local_state->profile());
     std::vector<vectorized::BlockSupplier> child_block_suppliers;
     for (int channel_id = 0; channel_id < _num_partitions; channel_id++) {
         vectorized::BlockSupplier block_supplier = [&, local_state, id = channel_id](

be/src/pipeline/local_exchange/local_exchanger.h

@@ -29,7 +29,6 @@ namespace pipeline {
 class LocalExchangeSourceLocalState;
 class LocalExchangeSinkLocalState;
 class BlockWrapper;
-class SortSourceOperatorX;
 
 struct Profile {
     RuntimeProfile::Counter* compute_hash_value_timer = nullptr;
@@ -335,11 +334,18 @@ class PassToOneExchanger final : public Exchanger<BlockWrapperSPtr> {
 
 class LocalMergeSortExchanger final : public Exchanger<BlockWrapperSPtr> {
 public:
+    struct MergeInfo {
+        const std::vector<bool>& is_asc_order;
+        const std::vector<bool>& nulls_first;
+        const int64_t limit;
+        const int64_t offset;
+        const vectorized::VExprContextSPtrs& ordering_expr_ctxs;
+    };
     ENABLE_FACTORY_CREATOR(LocalMergeSortExchanger);
-    LocalMergeSortExchanger(std::shared_ptr<SortSourceOperatorX> sort_source,
-                            int running_sink_operators, int num_partitions, int free_block_limit)
+    LocalMergeSortExchanger(MergeInfo&& merge_info, int running_sink_operators, int num_partitions,
+                            int free_block_limit)
             : Exchanger<BlockWrapperSPtr>(running_sink_operators, num_partitions, free_block_limit),
-              _sort_source(std::move(sort_source)) {}
+              _merge_info(std::move(merge_info)) {}
     ~LocalMergeSortExchanger() override = default;
     Status sink(RuntimeState* state, vectorized::Block* in_block, bool eos, Profile&& profile,
                 SinkInfo&& sink_info) override;
@@ -355,7 +361,7 @@ class LocalMergeSortExchanger final : public Exchanger<BlockWrapperSPtr> {
 
 private:
     std::unique_ptr<vectorized::VSortedRunMerger> _merger;
-    std::shared_ptr<SortSourceOperatorX> _sort_source;
+    MergeInfo _merge_info;
     std::vector<std::atomic_int64_t> _queues_mem_usege;
 };
 

be/src/pipeline/pipeline_fragment_context.cpp

@@ -818,7 +818,10 @@ Status PipelineFragmentContext::_add_local_exchange_impl(
                     child_op->get_name());
         }
         shared_state->exchanger = LocalMergeSortExchanger::create_unique(
-                sort_source, cur_pipe->num_tasks(), _num_instances,
+                LocalMergeSortExchanger::MergeInfo {
+                        sort_source->_is_asc_order, sort_source->_nulls_first, sort_source->_limit,
+                        sort_source->_offset, sort_source->_vsort_exec_exprs.ordering_expr_ctxs()},
+                cur_pipe->num_tasks(), _num_instances,
                 _runtime_state->query_options().__isset.local_exchange_free_blocks_limit
                         ? cast_set<int>(
                                   _runtime_state->query_options().local_exchange_free_blocks_limit)

be/src/vec/common/sort/heap_sorter.cpp

@@ -195,9 +195,9 @@ void HeapSorter::_do_filter(HeapSortCursorBlockView& block_view, size_t num_rows
 }
 
 Status HeapSorter::_prepare_sort_descs(Block* block) {
-    _sort_description.resize(_vsort_exec_exprs.lhs_ordering_expr_ctxs().size());
+    _sort_description.resize(_vsort_exec_exprs.ordering_expr_ctxs().size());
     for (int i = 0; i < _sort_description.size(); i++) {
-        const auto& ordering_expr = _vsort_exec_exprs.lhs_ordering_expr_ctxs()[i];
+        const auto& ordering_expr = _vsort_exec_exprs.ordering_expr_ctxs()[i];
         RETURN_IF_ERROR(ordering_expr->execute(block, &_sort_description[i].column_number));
 
         _sort_description[i].direction = _is_asc_order[i] ? 1 : -1;

be/src/vec/common/sort/sorter.cpp

@@ -180,10 +180,10 @@ Status Sorter::partial_sort(Block& src_block, Block& dest_block) {
         dest_block.swap(new_block);
     }
 
-    _sort_description.resize(_vsort_exec_exprs.lhs_ordering_expr_ctxs().size());
+    _sort_description.resize(_vsort_exec_exprs.ordering_expr_ctxs().size());
     Block* result_block = _materialize_sort_exprs ? &dest_block : &src_block;
     for (int i = 0; i < _sort_description.size(); i++) {
-        const auto& ordering_expr = _vsort_exec_exprs.lhs_ordering_expr_ctxs()[i];
+        const auto& ordering_expr = _vsort_exec_exprs.ordering_expr_ctxs()[i];
         RETURN_IF_ERROR(ordering_expr->execute(result_block, &_sort_description[i].column_number));
 
         _sort_description[i].direction = _is_asc_order[i] ? 1 : -1;

be/src/vec/common/sort/vsort_exec_exprs.cpp

@@ -48,7 +48,7 @@ Status VSortExecExprs::init(const TSortInfo& sort_info, ObjectPool* pool) {
 
 Status VSortExecExprs::init(const std::vector<TExpr>& ordering_exprs,
                             const std::vector<TExpr>* sort_tuple_slot_exprs, ObjectPool* pool) {
-    RETURN_IF_ERROR(VExpr::create_expr_trees(ordering_exprs, _lhs_ordering_expr_ctxs));
+    RETURN_IF_ERROR(VExpr::create_expr_trees(ordering_exprs, _ordering_expr_ctxs));
     if (sort_tuple_slot_exprs != NULL) {
         _materialize_tuple = true;
         RETURN_IF_ERROR(
@@ -59,44 +59,29 @@ Status VSortExecExprs::init(const std::vector<TExpr>& ordering_exprs,
     return Status::OK();
 }
 
-Status VSortExecExprs::init(const VExprContextSPtrs& lhs_ordering_expr_ctxs,
-                            const VExprContextSPtrs& rhs_ordering_expr_ctxs) {
-    _lhs_ordering_expr_ctxs = lhs_ordering_expr_ctxs;
-    _rhs_ordering_expr_ctxs = rhs_ordering_expr_ctxs;
-    return Status::OK();
-}
-
 Status VSortExecExprs::prepare(RuntimeState* state, const RowDescriptor& child_row_desc,
                                const RowDescriptor& output_row_desc) {
     if (_materialize_tuple) {
         RETURN_IF_ERROR(VExpr::prepare(_sort_tuple_slot_expr_ctxs, state, child_row_desc));
     }
-    RETURN_IF_ERROR(VExpr::prepare(_lhs_ordering_expr_ctxs, state, output_row_desc));
+    RETURN_IF_ERROR(VExpr::prepare(_ordering_expr_ctxs, state, output_row_desc));
     return Status::OK();
 }
 
 Status VSortExecExprs::open(RuntimeState* state) {
     if (_materialize_tuple) {
         RETURN_IF_ERROR(VExpr::open(_sort_tuple_slot_expr_ctxs, state));
     }
-    RETURN_IF_ERROR(VExpr::open(_lhs_ordering_expr_ctxs, state));
-    RETURN_IF_ERROR(
-            VExpr::clone_if_not_exists(_lhs_ordering_expr_ctxs, state, _rhs_ordering_expr_ctxs));
+    RETURN_IF_ERROR(VExpr::open(_ordering_expr_ctxs, state));
     return Status::OK();
 }
 
 void VSortExecExprs::close(RuntimeState* state) {}
 
 Status VSortExecExprs::clone(RuntimeState* state, VSortExecExprs& new_exprs) {
-    new_exprs._lhs_ordering_expr_ctxs.resize(_lhs_ordering_expr_ctxs.size());
-    new_exprs._rhs_ordering_expr_ctxs.resize(_rhs_ordering_expr_ctxs.size());
-    for (size_t i = 0; i < _lhs_ordering_expr_ctxs.size(); i++) {
-        RETURN_IF_ERROR(
-                _lhs_ordering_expr_ctxs[i]->clone(state, new_exprs._lhs_ordering_expr_ctxs[i]));
-    }
-    for (size_t i = 0; i < _rhs_ordering_expr_ctxs.size(); i++) {
-        RETURN_IF_ERROR(
-                _rhs_ordering_expr_ctxs[i]->clone(state, new_exprs._rhs_ordering_expr_ctxs[i]));
+    new_exprs._ordering_expr_ctxs.resize(_ordering_expr_ctxs.size());
+    for (size_t i = 0; i < _ordering_expr_ctxs.size(); i++) {
+        RETURN_IF_ERROR(_ordering_expr_ctxs[i]->clone(state, new_exprs._ordering_expr_ctxs[i]));
     }
     new_exprs._sort_tuple_slot_expr_ctxs.resize(_sort_tuple_slot_expr_ctxs.size());
     for (size_t i = 0; i < _sort_tuple_slot_expr_ctxs.size(); i++) {

be/src/vec/common/sort/vsort_exec_exprs.h

@@ -58,10 +58,7 @@ class VSortExecExprs {
     }
 
     // Can only be used after calling prepare()
-    const VExprContextSPtrs& lhs_ordering_expr_ctxs() const { return _lhs_ordering_expr_ctxs; }
-
-    // Can only be used after calling open()
-    const VExprContextSPtrs& rhs_ordering_expr_ctxs() const { return _rhs_ordering_expr_ctxs; }
+    const VExprContextSPtrs& ordering_expr_ctxs() const { return _ordering_expr_ctxs; }
 
     bool need_materialize_tuple() const { return _materialize_tuple; }
 
@@ -73,8 +70,7 @@ class VSortExecExprs {
 
 private:
     // Create two VExprContexts for evaluating over the TupleRows.
-    VExprContextSPtrs _lhs_ordering_expr_ctxs;
-    VExprContextSPtrs _rhs_ordering_expr_ctxs;
+    VExprContextSPtrs _ordering_expr_ctxs;
 
     // If true, the tuples to be sorted are materialized by
     // _sort_tuple_slot_exprs before the actual sort is performed.
@@ -85,16 +81,10 @@ class VSortExecExprs {
     // _materialize_tuple is true.
     VExprContextSPtrs _sort_tuple_slot_expr_ctxs;
 
-    // for some reason, _sort_tuple_slot_expr_ctxs is not-null but _lhs_ordering_expr_ctxs is nullable
+    // for some reason, _sort_tuple_slot_expr_ctxs is not-null but _ordering_expr_ctxs is nullable
     // this flag list would be used to convert column to nullable.
     std::vector<bool> _need_convert_to_nullable_flags;
 
-    // Initialize directly from already-created VExprContexts. Callers should manually call
-    // Prepare(), Open(), and Close() on input VExprContexts (instead of calling the
-    // analogous functions in this class). Used for testing.
-    Status init(const VExprContextSPtrs& lhs_ordering_expr_ctxs,
-                const VExprContextSPtrs& rhs_ordering_expr_ctxs);
-
     // Initialize the ordering and (optionally) materialization expressions from the thrift
     // TExprs into the specified pool. sort_tuple_slot_exprs is NULL if the tuple is not
     // materialized.