github-actions[bot] commented on code in PR #66295: URL: https://github.com/apache/doris/pull/66295#discussion_r3688822419
########## fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlacementAuditTest.java: ########## @@ -0,0 +1,254 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.qe; + +import org.apache.doris.nereids.NereidsPlanner; +import org.apache.doris.planner.ExchangeNode; +import org.apache.doris.planner.HashJoinNode; +import org.apache.doris.planner.LocalExchangeNode; +import org.apache.doris.planner.LocalExchangeNode.LocalExchangeType; +import org.apache.doris.planner.OlapScanNode; +import org.apache.doris.planner.PlanFragment; +import org.apache.doris.planner.PlanNode; +import org.apache.doris.planner.SetOperationNode; +import org.apache.doris.thrift.TPartitionType; +import org.apache.doris.utframe.TestWithFeService; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * Audit: for every multi-input operator that depends on hash placement, all of its input + * branches must end up on the SAME placement function. + * + * <p>Motivation: {@code RequireHash} means "I need hash-partitioned input"; it is satisfied by + * GLOBAL_EXECUTION_HASH_SHUFFLE, LOCAL_EXECUTION_HASH_SHUFFLE and BUCKET_HASH_SHUFFLE alike. + * It does NOT mean "all my branches must use the same hash function". A multi-input operator + * that hands the generic require to every branch can therefore end up with one branch keeping + * a storage-bucket placement while another is re-partitioned by execution hash — the same key + * then sits in two different pipeline tasks. This audit walks the finished plan and flags that + * mix instead of relying on a reviewer noticing it. + */ +public class LocalExchangePlacementAuditTest extends TestWithFeService { + @Override + protected int backendNum() { + return 3; + } + + @Override + protected void runBeforeAll() throws Exception { + createDatabase("test"); + useDatabase("test"); + // b1/b2 share bucket count so the optimizer may bucket-shuffle one onto the other; + // b3 has a different bucket count so it must be re-shuffled. + createTable("CREATE TABLE test.b1 (k INT, k2 INT, v INT) DISTRIBUTED BY HASH(k) BUCKETS 6 " + + "PROPERTIES ('replication_num'='1')"); + createTable("CREATE TABLE test.b2 (k INT, k2 INT, v INT) DISTRIBUTED BY HASH(k) BUCKETS 6 " + + "PROPERTIES ('replication_num'='1')"); + createTable("CREATE TABLE test.b3 (k INT, k2 INT, v INT) DISTRIBUTED BY HASH(k) BUCKETS 7 " + + "PROPERTIES ('replication_num'='1')"); + } + + /** The placement a branch actually lands on, as observed from the finished plan tree. */ + private static LocalExchangeType effectivePlacement(PlanNode node, ConnectContext ctx) { + if (node instanceof LocalExchangeNode) { + LocalExchangeType type = ((LocalExchangeNode) node).getExchangeType(); + // A PASSTHROUGH/BROADCAST/... wrapper does not decide hash placement; look through it. + return type.isHashShuffle() ? type : effectivePlacement(node.getChild(0), ctx); + } + if (node instanceof ExchangeNode) { + TPartitionType partitionType = ((ExchangeNode) node).getPartitionType(); + if (partitionType == TPartitionType.HASH_PARTITIONED) { + return LocalExchangeType.GLOBAL_EXECUTION_HASH_SHUFFLE; + } + if (partitionType == TPartitionType.BUCKET_SHFFULE_HASH_PARTITIONED) { + return LocalExchangeType.BUCKET_HASH_SHUFFLE; + } + return LocalExchangeType.NOOP; + } + if (node instanceof OlapScanNode) { + // Mirrors OlapScanNode.enforceAndDeriveLocalExchange: a pooling (serial) scan claims + // nothing, a non-pooling bucket scan claims its storage bucket distribution. + boolean pooling = node.getFragment() != null && node.getFragment().useSerialSource(ctx); + return pooling ? LocalExchangeType.NOOP : LocalExchangeType.BUCKET_HASH_SHUFFLE; + } + // Pass-through operators report their child's placement upward. + if (node.getChildren().size() == 1) { + return effectivePlacement(node.getChild(0), ctx); + } + return LocalExchangeType.NOOP; + } + + /** Returns "" when the plan is consistent, otherwise a description of the mixed placement. */ + private static String findMixedPlacement(List<PlanFragment> fragments, ConnectContext ctx) { + StringBuilder problems = new StringBuilder(); + for (PlanFragment fragment : fragments) { + walk(fragment.getPlanRoot(), problems, ctx); + } + return problems.toString(); + } + + private static void walk(PlanNode node, StringBuilder problems, ConnectContext ctx) { + boolean placementSensitive = node instanceof SetOperationNode || node instanceof HashJoinNode; + if (placementSensitive && node.getChildren().size() > 1) { + Map<LocalExchangeType, List<Integer>> byPlacement = new LinkedHashMap<>(); + for (int i = 0; i < node.getChildren().size(); i++) { + LocalExchangeType placement = effectivePlacement(node.getChild(i), ctx); + if (placement.isHashShuffle()) { Review Comment: [P2] Make missing and nested placements visible to this audit `effectivePlacement()` returns `NOOP` for every multi-input child, and this filter then discards every `NOOP`/non-hash branch. A bucket/partitioned `HashJoinNode`, `IntersectNode`, or `ExceptNode` with one pooling branch left unshuffled and one BUCKET/GLOBAL sibling therefore leaves a one-entry map and passes; an outer operator likewise cannot see a nested set operation's output placement. Those are precisely missing-realignment regressions this matrix is meant to catch. Please make the check distribution-mode aware and require every correctness-sensitive branch to resolve to a compatible placement, with focused failing cases for `NOOP + BUCKET` and a nested multi-input child. ########## fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlannerTest.java: ########## @@ -531,6 +531,39 @@ public void testAnalyticPlanContainsPassthroughAndLocalHashShuffle() throws Exce olapScan("t1"))))))))); } + @Test + public void testBucketShuffleUnionUnderWindowStaysBucketAligned() throws Exception { + // Regression: UNION ALL feeding a window function returned duplicate row_number()=1 for + // the same partition key. The union is bucket-shuffled (t1 keeps its storage buckets, + // t2 arrives through a bucket-shuffle exchange onto them), and the analytic sort below + // the AnalyticEval requires hash-partitioned input. With the generic hash requirement + // the t2 branch satisfied it and kept its bucket placement while the t1 branch — serial + // under the pooling scan, so it claims no distribution — was re-partitioned by + // LOCAL_EXECUTION_HASH_SHUFFLE. The two placements put one partition key in two + // different pipeline tasks and the analytic numbered each task from 1. + // + // AnalyticEval ← Sort ← Union ← LE(BUCKET_HASH) ← LE(PT) ← scan(t1) + // ← Exchange (bucket shuffle) ← scan(t2) + setupLocalShuffleSession(sv -> { + sv.setForceToLocalShuffle(true); + sv.setBucketShuffleDowngradeRatio(0); Review Comment: [P2] Restore the bucket-shuffle downgrade setting after this test `TestWithFeService` is `PER_CLASS`, so every method here shares this `ConnectContext`. `setupLocalShuffleSession()` resets the other local-shuffle knobs but not `bucketShuffleDowngradeRatio`; this test changes the default from `0.8` to `0` and leaves it there, so whichever planner tests run afterward can choose different bucket-shuffle plans than they do in isolation. Please reset this variable in the common setup before applying tweaks, or snapshot and restore it in this test. ########## fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlacementAuditTest.java: ########## @@ -0,0 +1,254 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.qe; + +import org.apache.doris.nereids.NereidsPlanner; +import org.apache.doris.planner.ExchangeNode; +import org.apache.doris.planner.HashJoinNode; +import org.apache.doris.planner.LocalExchangeNode; +import org.apache.doris.planner.LocalExchangeNode.LocalExchangeType; +import org.apache.doris.planner.OlapScanNode; +import org.apache.doris.planner.PlanFragment; +import org.apache.doris.planner.PlanNode; +import org.apache.doris.planner.SetOperationNode; +import org.apache.doris.thrift.TPartitionType; +import org.apache.doris.utframe.TestWithFeService; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * Audit: for every multi-input operator that depends on hash placement, all of its input + * branches must end up on the SAME placement function. + * + * <p>Motivation: {@code RequireHash} means "I need hash-partitioned input"; it is satisfied by + * GLOBAL_EXECUTION_HASH_SHUFFLE, LOCAL_EXECUTION_HASH_SHUFFLE and BUCKET_HASH_SHUFFLE alike. + * It does NOT mean "all my branches must use the same hash function". A multi-input operator + * that hands the generic require to every branch can therefore end up with one branch keeping + * a storage-bucket placement while another is re-partitioned by execution hash — the same key + * then sits in two different pipeline tasks. This audit walks the finished plan and flags that + * mix instead of relying on a reviewer noticing it. + */ +public class LocalExchangePlacementAuditTest extends TestWithFeService { + @Override + protected int backendNum() { + return 3; + } + + @Override + protected void runBeforeAll() throws Exception { + createDatabase("test"); + useDatabase("test"); + // b1/b2 share bucket count so the optimizer may bucket-shuffle one onto the other; + // b3 has a different bucket count so it must be re-shuffled. + createTable("CREATE TABLE test.b1 (k INT, k2 INT, v INT) DISTRIBUTED BY HASH(k) BUCKETS 6 " + + "PROPERTIES ('replication_num'='1')"); + createTable("CREATE TABLE test.b2 (k INT, k2 INT, v INT) DISTRIBUTED BY HASH(k) BUCKETS 6 " + + "PROPERTIES ('replication_num'='1')"); + createTable("CREATE TABLE test.b3 (k INT, k2 INT, v INT) DISTRIBUTED BY HASH(k) BUCKETS 7 " + + "PROPERTIES ('replication_num'='1')"); + } + + /** The placement a branch actually lands on, as observed from the finished plan tree. */ + private static LocalExchangeType effectivePlacement(PlanNode node, ConnectContext ctx) { + if (node instanceof LocalExchangeNode) { + LocalExchangeType type = ((LocalExchangeNode) node).getExchangeType(); + // A PASSTHROUGH/BROADCAST/... wrapper does not decide hash placement; look through it. + return type.isHashShuffle() ? type : effectivePlacement(node.getChild(0), ctx); + } + if (node instanceof ExchangeNode) { + TPartitionType partitionType = ((ExchangeNode) node).getPartitionType(); + if (partitionType == TPartitionType.HASH_PARTITIONED) { + return LocalExchangeType.GLOBAL_EXECUTION_HASH_SHUFFLE; + } + if (partitionType == TPartitionType.BUCKET_SHFFULE_HASH_PARTITIONED) { + return LocalExchangeType.BUCKET_HASH_SHUFFLE; + } + return LocalExchangeType.NOOP; + } + if (node instanceof OlapScanNode) { + // Mirrors OlapScanNode.enforceAndDeriveLocalExchange: a pooling (serial) scan claims + // nothing, a non-pooling bucket scan claims its storage bucket distribution. + boolean pooling = node.getFragment() != null && node.getFragment().useSerialSource(ctx); + return pooling ? LocalExchangeType.NOOP : LocalExchangeType.BUCKET_HASH_SHUFFLE; + } + // Pass-through operators report their child's placement upward. + if (node.getChildren().size() == 1) { + return effectivePlacement(node.getChild(0), ctx); + } + return LocalExchangeType.NOOP; + } + + /** Returns "" when the plan is consistent, otherwise a description of the mixed placement. */ + private static String findMixedPlacement(List<PlanFragment> fragments, ConnectContext ctx) { + StringBuilder problems = new StringBuilder(); + for (PlanFragment fragment : fragments) { + walk(fragment.getPlanRoot(), problems, ctx); + } + return problems.toString(); + } + + private static void walk(PlanNode node, StringBuilder problems, ConnectContext ctx) { + boolean placementSensitive = node instanceof SetOperationNode || node instanceof HashJoinNode; + if (placementSensitive && node.getChildren().size() > 1) { + Map<LocalExchangeType, List<Integer>> byPlacement = new LinkedHashMap<>(); + for (int i = 0; i < node.getChildren().size(); i++) { + LocalExchangeType placement = effectivePlacement(node.getChild(i), ctx); + if (placement.isHashShuffle()) { + byPlacement.computeIfAbsent(placement, k -> new ArrayList<>()).add(i); + } + } + if (byPlacement.size() > 1) { + problems.append(node.getClass().getSimpleName()) + .append('(').append(node.getId().asInt()).append(") mixes ") + .append(byPlacement).append('\n'); + } + } + for (PlanNode child : node.getChildren()) { + walk(child, problems, ctx); + } + } + + private String audit(String label, String sql, boolean pooling, boolean bucketUpgrade) + throws Exception { + SessionVariable sv = connectContext.getSessionVariable(); + sv.setEnableLocalShufflePlanner(true); + sv.setEnableLocalShuffle(true); + sv.setEnableNereidsDistributePlanner(true); + // ignore_storage_data_distribution is the real pooling switch: useSerialSource() gates on + // it. force_to_local_shuffle only forces ScanNode.isSerialNode(), which a small table + // already satisfies via `scanRangeNum < parallelExecInstanceNum * numScanBackends`, so + // flipping it alone would not give a non-pooling arm at all. + sv.setIgnoreStorageDataDistribution(pooling); + sv.setForceToLocalShuffle(pooling); + sv.setPipelineTaskNum(bucketUpgrade ? "16" : "8"); + sv.setBucketShuffleDowngradeRatio(0); + // ratio <= 1 disables the upgrade entirely; 1.01 makes it fire whenever instances + // slightly exceed buckets-with-data, which is what a bucket join above a mis-claiming + // child needs in order to be fooled by that claim. + sv.setLocalShuffleBucketUpgradeRatio(bucketUpgrade ? 1.01 : 1.5); Review Comment: [P2] Actually disable the upgrade in the false matrix arm `shouldUpgradeBucketParallelism()` treats every ratio above 1 as enabled; it disables the optimization only for `ratio <= 1`. Here the six/seven buckets are spread over three backends while the test requests eight instances per worker, so `1.5` can still satisfy `instances > bucketsPerWorker * ratio` and both boolean arms can exercise the upgraded branch. Please use `0` or `1` for the false arm (or assert the computed eligibility) so this matrix really covers both production paths. -- 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]
