database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[branch-2.1](pick) Pick 2 PRs to branch-2.1 (#39604)

Browse Source 
## Proposed changes

pick #39480 #39589

<!--Describe your changes.-->
This commit is contained in:
Gabriel
2024-08-20 17:10:30 +08:00
committed by GitHub
parent 5fcd6e6270
commit e302882e52
2 changed files with 21 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
View File

@ -2107,10 +2107,11 @@ public class Coordinator implements CoordInterface {
if ((isColocateFragment(fragment, fragment.getPlanRoot())
&& fragmentIdToSeqToAddressMap.containsKey(fragment.getFragmentId())
&& fragmentIdToSeqToAddressMap.get(fragment.getFragmentId()).size() > 0)) {
computeColocateJoinInstanceParam(fragment.getFragmentId(), parallelExecInstanceNum, params);
computeColocateJoinInstanceParam(fragment.getFragmentId(), parallelExecInstanceNum, params,
fragment.hasNullAwareLeftAntiJoin());
} else if (bucketShuffleJoinController.isBucketShuffleJoin(fragment.getFragmentId().asInt())) {
bucketShuffleJoinController.computeInstanceParam(fragment.getFragmentId(),
parallelExecInstanceNum, params);
parallelExecInstanceNum, params, fragment.hasNullAwareLeftAntiJoin());
} else {
// case A
for (Entry<TNetworkAddress, Map<Integer, List<TScanRangeParams>>> entry : fragmentExecParamsMap.get(
@ -2135,7 +2136,8 @@ public class Coordinator implements CoordInterface {
int expectedInstanceNum = Math.min(parallelExecInstanceNum,
leftMostNode.getNumInstances());
boolean forceToLocalShuffle = context != null
&& context.getSessionVariable().isForceToLocalShuffle();
&& context.getSessionVariable().isForceToLocalShuffle()
&& !fragment.hasNullAwareLeftAntiJoin() && useNereids;
boolean ignoreStorageDataDistribution = forceToLocalShuffle || (scanNodes.stream()
.allMatch(scanNode -> scanNode.ignoreStorageDataDistribution(context,
addressToBackendID.size())) && useNereids);
@ -2304,9 +2306,9 @@ public class Coordinator implements CoordInterface {
}
private void computeColocateJoinInstanceParam(PlanFragmentId fragmentId,
int parallelExecInstanceNum, FragmentExecParams params) {
int parallelExecInstanceNum, FragmentExecParams params, boolean hasNullAwareLeftAntiJoin) {
assignScanRanges(fragmentId, parallelExecInstanceNum, params, fragmentIdTobucketSeqToScanRangeMap,
fragmentIdToSeqToAddressMap, fragmentIdToScanNodeIds);
fragmentIdToSeqToAddressMap, fragmentIdToScanNodeIds, hasNullAwareLeftAntiJoin);
}
private Map<TNetworkAddress, Long> getReplicaNumPerHostForOlapTable() {
@ -3049,16 +3051,16 @@ public class Coordinator implements CoordInterface {
}
private void computeInstanceParam(PlanFragmentId fragmentId,
int parallelExecInstanceNum, FragmentExecParams params) {
int parallelExecInstanceNum, FragmentExecParams params, boolean hasNullAwareLeftAntiJoin) {
assignScanRanges(fragmentId, parallelExecInstanceNum, params, fragmentIdBucketSeqToScanRangeMap,
fragmentIdToSeqToAddressMap, fragmentIdToScanNodeIds);
fragmentIdToSeqToAddressMap, fragmentIdToScanNodeIds, hasNullAwareLeftAntiJoin);
}
}
private void assignScanRanges(PlanFragmentId fragmentId, int parallelExecInstanceNum, FragmentExecParams params,
Map<PlanFragmentId, BucketSeqToScanRange> fragmentIdBucketSeqToScanRangeMap,
Map<PlanFragmentId, Map<Integer, TNetworkAddress>> curFragmentIdToSeqToAddressMap,
Map<PlanFragmentId, Set<Integer>> fragmentIdToScanNodeIds) {
Map<PlanFragmentId, Set<Integer>> fragmentIdToScanNodeIds, boolean hasNullAwareLeftAntiJoin) {
Map<Integer, TNetworkAddress> bucketSeqToAddress = curFragmentIdToSeqToAddressMap.get(fragmentId);
BucketSeqToScanRange bucketSeqToScanRange = fragmentIdBucketSeqToScanRangeMap.get(fragmentId);
Set<Integer> scanNodeIds = fragmentIdToScanNodeIds.get(fragmentId);
@ -3092,7 +3094,7 @@ public class Coordinator implements CoordInterface {
* 2. Use Nereids planner.
*/
boolean forceToLocalShuffle = context != null
&& context.getSessionVariable().isForceToLocalShuffle();
&& context.getSessionVariable().isForceToLocalShuffle() && !hasNullAwareLeftAntiJoin && useNereids;
boolean ignoreStorageDataDistribution = forceToLocalShuffle || (scanNodes.stream()
.allMatch(node -> node.ignoreStorageDataDistribution(context,
addressToBackendID.size()))