[improvement](agg) Add passthrough local exchange before streaming ag…

be/src/exec/operator/streaming_aggregation_operator.h

@@ -220,7 +220,11 @@ class StreamingAggOperatorX MOCK_REMOVE(final) : public StatefulOperatorX<Stream
             state->enable_streaming_agg_hash_join_force_passthrough()) {
             return DataDistribution(ExchangeType::PASSTHROUGH);
         }
-        if (!_needs_finalize && !state->enable_local_exchange_before_agg()) {
+        // Streaming aggregation needs pipeline decoupling but not a key repartition.
+        if (!_needs_finalize && state->enable_local_exchange_before_agg()) {
+            return DataDistribution(ExchangeType::PASSTHROUGH);
+        }
+        if (!_needs_finalize) {
             return StatefulOperatorX<StreamingAggLocalState>::required_data_distribution(state);
         }
         if (_partition_exprs.empty()) {

be/test/exec/operator/streaming_agg_operator_test.cpp

@@ -154,6 +154,15 @@ TEST_F(StreamingAggOperatorTest, test1) {
     { EXPECT_TRUE(local_state->close(state.get()).ok()); }
 }
 
+TEST_F(StreamingAggOperatorTest, require_passthrough_local_exchange_before_non_finalize_agg) {
+    state->_query_options.__set_enable_local_exchange_before_agg(true);
+    op->_needs_finalize = false;
+    op->_partition_exprs.emplace_back();
+
+    const auto distribution = op->required_data_distribution(state.get());
+    EXPECT_EQ(ExchangeType::PASSTHROUGH, distribution.distribution_type);
+}
+
 TEST_F(StreamingAggOperatorTest, test2) {
     op->_aggregate_evaluators.push_back(create_mock_agg_fn_evaluator(
             pool, MockSlotRef::create_mock_contexts(1, std::make_shared<DataTypeInt64>()), false,

fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java

@@ -219,6 +219,19 @@ public void testAdaptiveBatchSizeSessionVariables() throws Exception {
         Assert.assertEquals(1048576L, options.getPreferredBlockSizeBytes());
     }
 
+    @Test
+    public void testLocalExchangeBeforeAggDefaultsToThrift() {
+        SessionVariable var = new SessionVariable();
+        TQueryOptions options = var.toThrift();
+
+        Assert.assertTrue(var.enableLocalExchangeBeforeAgg);
+        Assert.assertTrue(options.isEnableLocalExchangeBeforeAgg());
+
+        var.enableLocalExchangeBeforeAgg = false;
+        options = var.toThrift();
+        Assert.assertFalse(options.isEnableLocalExchangeBeforeAgg());
+    }
+
     @Test
     public void testAdaptiveBatchSizeRejectsTinyNonZeroBytes() {
         SessionVariable var = new SessionVariable();