From 097b15b767528cefff8a7b09e971f5f4a80e50ea Mon Sep 17 00:00:00 2001 From: morningman Date: Wed, 4 Nov 2026 16:10:20 +0800 Subject: [PATCH] [fix](fe) Forward four local exchange session variables to the master ### What problem does this PR solve? Issue Number: None Related PR: #69273, #70177 Problem Summary: A statement that has to run on the master FE (INSERT INTO ... SELECT, CTAS, ...) and arrives at a follower or an observer is forwarded with the session variables that `SessionVariable#getForwardVariables()` sends, and the master applies them to a session that starts from its own global variables. Only variables whose `@VarAttr` sets `needForward` are sent. #69273 flagged the planner and executor variables, `enable_local_exchange_before_agg` among them, but not four that also decide the local exchanges of a plan: `enable_local_exchange_before_streaming_agg` (a local hash exchange before a streaming pre-aggregation, BE `StreamingAggOperatorX` and the FE planner with #70177), `enable_streaming_agg_hash_join_force_passthrough` and `enable_distinct_streaming_agg_force_passthrough` (a passthrough local exchange below a streaming pre-aggregation, FE `AggregationNode` and the BE operators), and `force_to_local_shuffle` (every scan serial, `ScanNode#isSerialNode`). A session-level SET of one of them was dropped when the statement was forwarded, so the master planned it with its global value and EXPLAIN on the follower showed a plan the master did not run. Results do not change, only the plan and its speed. The four now set `needForward = true`, as #69273 does for the others. ### Release note The session variables enable_local_exchange_before_streaming_agg, enable_streaming_agg_hash_join_force_passthrough, enable_distinct_streaming_agg_force_passthrough and force_to_local_shuffle now apply to statements that a follower or observer FE forwards to the master. ### Check List (For Author) - Test: Unit Test - `SessionVariablesTest#testForwardPlanAndExecutionKnobs` lists the four; with `SessionVariable.java` reverted it fails. The test classes that call `getForwardVariables()` or `setForwardedSessionVariables()` pass, and fe-core Checkstyle reports nothing. Not run on a cluster with two FEs. - Behavior changed: Yes. A forwarded statement uses the session's value of these four variables instead of the master's global value. - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 --- .../main/java/org/apache/doris/qe/SessionVariable.java | 8 ++++---- .../java/org/apache/doris/qe/SessionVariablesTest.java | 7 +++++-- 2 files changed, 9 insertions(+), 6 deletions(-) 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 a622f153161f6f..afee10603f2561 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 @@ -1451,16 +1451,16 @@ public void checkQuerySlotCount(String slotCnt) { @VarAttrDef.VarAttr(name = ENABLE_DISTINCT_STREAMING_AGGREGATION, needForward = true, fuzzy = true) public boolean enableDistinctStreamingAggregation = true; - @VarAttrDef.VarAttr(name = ENABLE_STREAMING_AGG_HASH_JOIN_FORCE_PASSTHROUGH, fuzzy = true) + @VarAttrDef.VarAttr(name = ENABLE_STREAMING_AGG_HASH_JOIN_FORCE_PASSTHROUGH, needForward = true, fuzzy = true) public boolean enableStreamingAggHashJoinForcePassthrough = true; @VarAttrDef.VarAttr(name = ENABLE_LOCAL_EXCHANGE_BEFORE_AGG, needForward = true, fuzzy = true) public boolean enableLocalExchangeBeforeAgg = true; - @VarAttrDef.VarAttr(name = ENABLE_LOCAL_EXCHANGE_BEFORE_STREAMING_AGG, fuzzy = true) + @VarAttrDef.VarAttr(name = ENABLE_LOCAL_EXCHANGE_BEFORE_STREAMING_AGG, needForward = true, fuzzy = true) public boolean enableLocalExchangeBeforeStreamingAgg = false; - @VarAttrDef.VarAttr(name = ENABLE_DISTINCT_STREAMING_AGG_FORCE_PASSTHROUGH, fuzzy = true) + @VarAttrDef.VarAttr(name = ENABLE_DISTINCT_STREAMING_AGG_FORCE_PASSTHROUGH, needForward = true, fuzzy = true) public boolean enableDistinctStreamingAggForcePassthrough = true; @VarAttrDef.VarAttr(name = ENABLE_BROADCAST_JOIN_FORCE_PASSTHROUGH, needForward = true, fuzzy = true) @@ -1777,7 +1777,7 @@ public boolean isEnableDynamicPartitionDropProtection() { @VarAttrDef.VarAttr( name = FORCE_TO_LOCAL_SHUFFLE, fuzzy = false, varType = VariableAnnotation.EXPERIMENTAL, - description = "Whether to force to local shuffle on pipelineX engine.") + description = "Whether to force to local shuffle on pipelineX engine.", needForward = true) private boolean forceToLocalShuffle = false; @VarAttrDef.VarAttr( diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java index a880f6aaeab04f..ef6514ceb04bb8 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java @@ -564,12 +564,15 @@ public void testForwardPlanAndExecutionKnobs() { SessionVariable.ENABLE_BUCKET_SHUFFLE_JOIN, SessionVariable.ENABLE_COST_BASED_JOIN_REORDER, SessionVariable.ENABLE_CTE_MATERIALIZE, SessionVariable.ENABLE_DECOUPLED_RUNTIME_FILTER, SessionVariable.ENABLE_DISTINCT_STREAMING_AGGREGATION, + SessionVariable.ENABLE_DISTINCT_STREAMING_AGG_FORCE_PASSTHROUGH, SessionVariable.ENABLE_DPHYP_OPTIMIZER, SessionVariable.ENABLE_LOCAL_EXCHANGE, - SessionVariable.ENABLE_LOCAL_EXCHANGE_BEFORE_AGG, SessionVariable.ENABLE_PARALLEL_OUTFILE, + SessionVariable.ENABLE_LOCAL_EXCHANGE_BEFORE_AGG, + SessionVariable.ENABLE_LOCAL_EXCHANGE_BEFORE_STREAMING_AGG, SessionVariable.ENABLE_PARALLEL_OUTFILE, SessionVariable.ENABLE_PARTITION_TOPN, SessionVariable.ENABLE_PUSH_DOWN_NO_GROUP_AGG, SessionVariable.ENABLE_SHARED_EXCHANGE_SINK_BUFFER, SessionVariable.ENABLE_SQL_CACHE, + SessionVariable.ENABLE_STREAMING_AGG_HASH_JOIN_FORCE_PASSTHROUGH, SessionVariable.EXCHANGE_MULTI_BLOCKS_BYTE_SIZE, - SessionVariable.EXPAND_RUNTIME_FILTER_BY_INNER_JION, + SessionVariable.EXPAND_RUNTIME_FILTER_BY_INNER_JION, SessionVariable.FORCE_TO_LOCAL_SHUFFLE, SessionVariable.GLOBAL_PARTITION_TOPN_THRESHOLD, SessionVariable.IGNORE_RUNTIME_FILTER_IDS, SessionVariable.INLINE_CTE_REFERENCED_THRESHOLD, SessionVariable.LOCAL_EXCHANGE_FREE_BLOCKS_LIMIT,