This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c79dc9ce0b7 [fix](fe) Forward four local exchange session variables to
the master (#70197)
c79dc9ce0b7 is described below
commit c79dc9ce0b7539a36360d38d284a962c08364dc3
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Fri Nov 6 01:15:21 2026 +0000
[fix](fe) Forward four local exchange session variables to the master
(#70197)
### What problem does this PR solve?
Issue Number: None
Related PR: #69273, #70177
Problem Summary:
**In short.** Found while bringing branch-4.2 #66216 to the FE local
shuffle planner (#70177): four session variables that decide the local
exchanges of a plan are not forwarded to the master. #69273 made the
planner and executor variables travel with a statement that a follower
or an observer forwards to the master,
`enable_local_exchange_before_agg` among them, but not
`enable_local_exchange_before_streaming_agg`,
`enable_streaming_agg_hash_join_force_passthrough`,
`enable_distinct_streaming_agg_force_passthrough` and
`force_to_local_shuffle`. A session-level `SET` of one of them is
dropped on that path, and the master plans the statement with its own
global value. This PR forwards the four as well.
**Background.** A statement that has to run on the master FE (`INSERT
INTO ... SELECT`, `CREATE TABLE ... AS SELECT`, ...) and arrives at a
follower or an observer is forwarded: `FEOpExecutor` sends
`SessionVariable#getForwardVariables()`, and `ConnectProcessor` on the
master applies them with `setForwardedSessionVariables()` to a session
that starts from the master's global variables. Only variables whose
`@VarAttr` sets `needForward` (or one of the `affectQueryResult*`
attributes) are sent.
**The problem, and what it cost.** These four do not set it:
| Variable | Default | What it decides |
|---|---|---|
| `enable_local_exchange_before_streaming_agg` (#66222) | false |
whether a streaming pre-aggregation asks for a local hash exchange on
its grouping keys: BE `StreamingAggOperatorX`, and the FE planner with
#70177 |
| `enable_streaming_agg_hash_join_force_passthrough` | true | whether a
streaming pre-aggregation right above a hash join asks for a passthrough
local exchange: the FE planner (`AggregationNode`) and BE
`StreamingAggOperatorX` |
| `enable_distinct_streaming_agg_force_passthrough` | true | the same
for a distinct streaming pre-aggregation: the FE planner and BE
`DistinctStreamingAggOperatorX` |
| `force_to_local_shuffle` | false | makes every scan serial
(`ScanNode#isSerialNode`), so the FE planner pools the scan and adds a
local shuffle above it |
So a session that sets one of them gets that plan only for the
statements its own FE runs. A forwarded statement, such as an `INSERT
INTO ... SELECT` or a CTAS issued through a follower or an observer, is
planned with the master's global value, and `EXPLAIN` on the follower
shows a plan that the master does not run. Query results stay the same,
since these variables only decide how rows move between the pipeline
tasks of a BE; what differs is the plan, and with it the speed of the
statement.
**How this PR fixes it.** Set `needForward = true` on the four, as
#69273 does for the others, and add them to the list in
`SessionVariablesTest#testForwardPlanAndExecutionKnobs`.
**Results.**
| Statement forwarded from a follower after `SET <variable> =
<non-default value>` | Before | After |
|---|---|---|
| Value the master plans it with | the master's global value | the
session's value |
branch-4.1 and branch-4.2 have the same gap, and #69273 is not on them
either.
---
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java | 8 ++++----
.../src/test/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 a622f153161..afee10603f2 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 class SessionVariable implements Serializable,
Writable {
@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 class SessionVariable implements Serializable,
Writable {
@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 a880f6aaeab..ef6514ceb04 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 class SessionVariablesTest extends
TestWithFeService {
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,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
