From 8cbcfa30be55be4e370e3bad6121eeb85ee020f7 Mon Sep 17 00:00:00 2001
From: Mryange <2319153948@qq.com>
Date: Mon, 23 Sep 2024 19:29:17 +0800
Subject: [PATCH] enablePassToOneExchange

---
 be/src/pipeline/exec/exchange_sink_operator.cpp            | 5 +++--
 be/src/pipeline/exec/exchange_sink_operator.h              | 1 +
 be/src/runtime/runtime_state.h                             | 5 +++++
 .../src/main/java/org/apache/doris/qe/SessionVariable.java | 7 +++++++
 gensrc/thrift/PaloInternalService.thrift                   | 1 +
 5 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/be/src/pipeline/exec/exchange_sink_operator.cpp b/be/src/pipeline/exec/exchange_sink_operator.cpp
index f34da3f476bcdd..d73085aef811db 100644
--- a/be/src/pipeline/exec/exchange_sink_operator.cpp
+++ b/be/src/pipeline/exec/exchange_sink_operator.cpp
@@ -306,7 +306,8 @@ ExchangeSinkOperatorX::ExchangeSinkOperatorX(
           _tablet_sink_txn_id(sink.tablet_sink_txn_id),
           _t_tablet_sink_exprs(&sink.tablet_sink_exprs),
           _enable_local_merge_sort(state->enable_local_merge_sort()),
-          _is_multi_cast(is_multi_cast) {
+          _is_multi_cast(is_multi_cast),
+          _enable_pass_to_one_exchange(state->enable_pass_to_one_exchange()) {
     DCHECK_GT(destinations.size(), 0);
     DCHECK(sink.output_partition.type == TPartitionType::UNPARTITIONED ||
            sink.output_partition.type == TPartitionType::HASH_PARTITIONED ||
@@ -683,7 +684,7 @@ DataDistribution ExchangeSinkOperatorX::required_data_distribution() const {
             sort_source && sort_source->use_local_merge()) {
             // Sort the data local
             return ExchangeType::LOCAL_MERGE_SORT;
-        } else if (!_is_multi_cast) {
+        } else if (!_is_multi_cast && _enable_pass_to_one_exchange) {
             return ExchangeType::PASS_TO_ONE_EXCHANGE;
         }
     }
diff --git a/be/src/pipeline/exec/exchange_sink_operator.h b/be/src/pipeline/exec/exchange_sink_operator.h
index 9469db19f4d18b..2752a7ab3851c6 100644
--- a/be/src/pipeline/exec/exchange_sink_operator.h
+++ b/be/src/pipeline/exec/exchange_sink_operator.h
@@ -283,6 +283,7 @@ class ExchangeSinkOperatorX final : public DataSinkOperatorX<ExchangeSinkLocalSt
     int _writer_count = 1;
     const bool _enable_local_merge_sort;
     const bool _is_multi_cast;
+    const bool _enable_pass_to_one_exchange;
 
     int _close_sender_number = 1;
 };
diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h
index 90cf1bc34bdeca..035d0403016080 100644
--- a/be/src/runtime/runtime_state.h
+++ b/be/src/runtime/runtime_state.h
@@ -587,6 +587,11 @@ class RuntimeState {
                _query_options.enable_local_merge_sort;
     }
 
+    bool enable_pass_to_one_exchange() const {
+        return _query_options.__isset.enable_pass_to_one_exchange &&
+               _query_options.enable_pass_to_one_exchange;
+    }
+
     int64_t min_revocable_mem() const {
         if (_query_options.__isset.min_revocable_mem) {
             return std::max(_query_options.min_revocable_mem, (int64_t)1);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index e4fb6153e3359c..b478c70c321830 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -278,6 +278,8 @@ public class SessionVariable implements Serializable, Writable {
 
     public static final String ENABLE_LOCAL_MERGE_SORT = "enable_local_merge_sort";
 
+    public static final String ENABLE_PASS_TO_ONE_EXCHANGE = "enable_pass_to_one_exchange";
+
     public static final String ENABLE_AGG_STATE = "enable_agg_state";
 
     public static final String ENABLE_RPC_OPT_FOR_PIPELINE = "enable_rpc_opt_for_pipeline";
@@ -1068,6 +1070,9 @@ public class SessionVariable implements Serializable, Writable {
     @VariableMgr.VarAttr(name = ENABLE_LOCAL_MERGE_SORT)
     private boolean enableLocalMergeSort = true;
 
+    @VariableMgr.VarAttr(name = ENABLE_PASS_TO_ONE_EXCHANGE)
+    private boolean enablePassToOneExchange = false;
+
     @VariableMgr.VarAttr(name = ENABLE_AGG_STATE, fuzzy = false, varType = VariableAnnotation.EXPERIMENTAL,
             needForward = true)
     public boolean enableAggState = false;
@@ -2215,6 +2220,7 @@ public void initFuzzyModeVariables() {
         this.parallelPrepareThreshold = random.nextInt(32) + 1;
         this.enableCommonExprPushdown = random.nextBoolean();
         this.enableLocalExchange = random.nextBoolean();
+        this.enablePassToOneExchange = random.nextBoolean();
         // This will cause be dead loop, disable it first
         // this.disableJoinReorder = random.nextBoolean();
         this.enableCommonExpPushDownForInvertedIndex = random.nextBoolean();
@@ -3774,6 +3780,7 @@ public TQueryOptions toThrift() {
         tResult.setDataQueueMaxBlocks(dataQueueMaxBlocks);
 
         tResult.setEnableLocalMergeSort(enableLocalMergeSort);
+        tResult.setEnablePassToOneExchange(enablePassToOneExchange);
         tResult.setEnableParallelResultSink(enableParallelResultSink);
         tResult.setEnableShortCircuitQueryAccessColumnStore(enableShortCircuitQueryAcessColumnStore);
         tResult.setReadCsvEmptyLineAsNull(readCsvEmptyLineAsNull);
diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift
index 48f41e8e0ab9f9..d0f4f3798484ca 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -348,6 +348,7 @@ struct TQueryOptions {
   // 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.
   // In read path, read from file cache or remote storage when execute query.
+  135: optional bool enable_pass_to_one_exchange = true;
   1000: optional bool disable_file_cache = false
 }