diff --git a/be/src/exec/operator/streaming_aggregation_operator.h b/be/src/exec/operator/streaming_aggregation_operator.h
index 5ed6d1481d5fcc..440df9eedece4f 100644
--- a/be/src/exec/operator/streaming_aggregation_operator.h
+++ b/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()) {
diff --git a/be/test/exec/operator/streaming_agg_operator_test.cpp b/be/test/exec/operator/streaming_agg_operator_test.cpp
index d56e88ec5a04a9..9c8654f939e75c 100644
--- a/be/test/exec/operator/streaming_agg_operator_test.cpp
+++ b/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,
diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java
index 2cc9d43b31167b..d4f3faf3137012 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/VariableMgrTest.java
+++ b/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();