github-actions[bot] commented on code in PR #67177:
URL: https://github.com/apache/doris/pull/67177#discussion_r3868553823
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java:
##########
@@ -287,6 +316,50 @@ List<AssignedJob>
filterInstancesWhichCanReceiveDataFromRemote(
}
}
+ private boolean hasSerialReceiverPipeline(
+ PipelineDistributedPlan receiverPlan, ExchangeNode linkNode) {
+ ConnectContext connectContext = statementContext.getConnectContext();
+ if (linkNode.isSerialOperatorOnBe(connectContext)) {
+ return true;
+ }
+
+ // A non-serial Exchange does not necessarily create one receiver per
fragment
+ // instance. Build-side pipelines inherit the task count of their
parent pipeline
+ // on BE. A serial ancestor (for example, a scalar finalize
aggregation) can reduce
+ // that parent pipeline to one task per worker, and the Exchange build
pipeline
+ // inherits the same single task. Addressing the other logical
instances would
+ // send bucket rows to receivers that are never created.
+ //
+ // A LocalExchange on the path is a pipeline boundary which restores
the upstream
+ // pipeline to all fragment instances. Only serial operators after the
last such
+ // boundary can reduce the target Exchange's receiver count.
+ PlanNode receiverRoot =
receiverPlan.getFragmentJob().getFragment().getPlanRoot();
+ Optional<Boolean> serialPipeline = findSerialPipelineToExchange(
+ receiverRoot, linkNode, false, connectContext);
+ return serialPipeline.orElseThrow(() -> new IllegalStateException(
+ "Can not find receiver exchange " + linkNode.getId()
+ + " in fragment " + receiverRoot.getFragmentId()));
+ }
+
+ private Optional<Boolean> findSerialPipelineToExchange(
Review Comment:
[P2] Avoid rescanning the receiver tree for every Exchange. A wide `UNION
ALL` receiver with N Exchange leaves makes `linkPlans()` visit roughly `1 + 2 +
... + N` nodes, and bucket links call this DFS a second time. Build the
effective serial-pipeline state for all Exchange nodes in one post-rewrite
traversal per receiver fragment, then use lookups in both destination paths.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java:
##########
@@ -75,19 +78,37 @@ public class DistributePlanner {
private final boolean notNeedBackend;
private final boolean isLoadJob;
private final boolean useLoadBackendSelection;
+ private final int beExecVersion;
public DistributePlanner(StatementContext statementContext,
List<PlanFragment> fragments, boolean notNeedBackend, boolean
isLoadJob,
boolean useLoadBackendSelection) {
+ this(statementContext, fragments, notNeedBackend, isLoadJob,
useLoadBackendSelection,
+ Config.be_exec_version);
+ }
+
+ public DistributePlanner(StatementContext statementContext,
Review Comment:
[P1] Add Javadoc for this new public constructor. The current `CheckStyle`
job fails at this line with `MissingJavadocMethod`, so the head cannot pass the
mandatory FE style gate. Please document the query-level execution-version
snapshot parameter and rerun CheckStyle.
##########
regression-test/suites/nereids_p0/local_shuffle/test_local_shuffle_rqg_bugs.groovy:
##########
@@ -1663,5 +1663,86 @@ suite("test_local_shuffle_rqg_bugs") {
assertTrue(false, "Bug 26: ${t.message}")
}
+ // RQG builds 208624/208625/208649: a non-serial BUCKET_SHUFFLE Exchange
+ // addressed all pooled instances, but the scalar DISTINCT root reduced the
+ // receiver pipeline to one task per BE. Rows routed to the other logical
+ // instances had no receiver and were silently lost. Keep the original
+ // 10-bucket data and predicates so 4 BEs * 3 tasks form this mismatch.
+ try {
+ logger.info("Bug 27: scalar DISTINCT over serial BUCKET_SHUFFLE outer
joins")
+ ["bug27_t5", "bug27_t52", "bug27_t53"].each { table ->
+ sql "DROP TABLE IF EXISTS ${table}"
+ }
+ sql """CREATE TABLE bug27_t5 (
+ col_bigint BIGINT, col_v10 VARCHAR(10), col_v64
VARCHAR(64), pk INT
+ ) ENGINE=OLAP DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES ("replication_num"="1")"""
+ sql """CREATE TABLE bug27_t52 (
+ pk INT, col_bigint BIGINT, col_v10 VARCHAR(10), col_v64
VARCHAR(64)
+ ) ENGINE=OLAP DUPLICATE KEY(pk, col_bigint, col_v10)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10 PROPERTIES
("replication_num"="1")"""
+ sql """CREATE TABLE bug27_t53 (
+ pk INT, col_v10 VARCHAR(10), col_bigint BIGINT, col_v64
VARCHAR(64)
+ ) ENGINE=OLAP DUPLICATE KEY(pk, col_v10)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10 PROPERTIES
("replication_num"="1")"""
+ sql """INSERT INTO bug27_t5 VALUES
+
(-94,'had','y',0),(672609,'k','h',1),(-3766684,'a','p',2),(5070261,'on','x',3),
+
(NULL,'u','at',4),(-86,'v','c',5),(21910,'how','m',6),(-63,'that''s','go',7),
+ (-8276281,'s','a',8),(-101,'w','y',9)"""
+ sql """INSERT INTO bug27_t52 VALUES
+
(0,NULL,'g','i'),(1,-6138328,'z','do'),(2,-23217,'g','about'),(3,104,'you''re','z'),
+
(4,NULL,'oh','i'),(5,-54,'want','to'),(6,NULL,'x','c'),(7,NULL,'you''re','come'),
+ (8,3447,'really','from'),(9,-5459,'i','will')"""
+ sql """INSERT INTO bug27_t53 VALUES
+
(0,'right',NULL,'g'),(1,'on',-486256,'on'),(2,'I''ll',-1,'at'),(3,'h',29263,'don''t'),
+
(4,'a',5453,'s'),(5,'j',-119,'can''t'),(6,'one',89,'n'),(7,'s',-7227,'u'),
+ (8,'time',94,'b'),(9,'yes',1816630,'yes')"""
+
+ def vars =
"enable_local_shuffle_planner=true,enable_local_shuffle=true," +
+
"enable_bucket_shuffle_join=true,ignore_storage_data_distribution=true," +
+
"bucket_shuffle_downgrade_ratio=0.8,use_serial_exchange=false," +
+ "parallel_pipeline_task_num=3,enable_sql_cache=false"
+ def result1 = sql """SELECT /*+SET_VAR(${vars})*/ COUNT(DISTINCT t1.pk)
Review Comment:
[P2] Put these fixed expected results through the regression golden-output
path. The five new queries use `sql` plus hard-coded `assertEquals`, while the
required repository convention is `qt_`/`order_qt_` with the `.out` generated
by the regression runner. These are deterministic scalar rows, and the existing
suite output currently has no bug27 entries.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
