EmmyMiao87 commented on a change in pull request #5459:
URL: https://github.com/apache/incubator-doris/pull/5459#discussion_r587974095
##########
File path:
fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java
##########
@@ -529,21 +533,40 @@ private boolean canBucketShuffleJoin(HashJoinNode node,
PlanFragment leftChildFr
}
PlanNode leftRoot = leftChildFragment.getPlanRoot();
- // leftRoot should be OlapScanNode
+ // 1.leftRoot be OlapScanNode
if (leftRoot instanceof OlapScanNode) {
return canBucketShuffleJoin(node, leftRoot, rhsHashExprs);
}
+ // 2.leftRoot be colocate hashjoin node or bucket shuffle join node
+ PlanNode rightRoot = rightChildFragment.getPlanRoot();
+ if (leftRoot instanceof HashJoinNode) {
+ while (leftRoot instanceof HashJoinNode) {
+ if (((HashJoinNode)leftRoot).isColocate() ||
((HashJoinNode)leftRoot).isBucketShuffle()) {
+ leftRoot = leftRoot.getChild(0);
+ } else {
+ return false;
+ }
+ }
+ if (leftRoot instanceof OlapScanNode) {
+ return canBucketShuffleJoin(node, leftRoot, rhsHashExprs);
+ }
+ }
+
return false;
}
//the join expr must contian left table distribute column
private boolean canBucketShuffleJoin(HashJoinNode node, PlanNode leftRoot,
List<Expr> rhsJoinExprs) {
OlapScanNode leftScanNode = ((OlapScanNode) leftRoot);
+ OlapTable leftTable = leftScanNode.getOlapTable();
- //1 the left table must be only one partition
+ //1 the left table has more than one partition or left table is not a
stable colocate table
if (leftScanNode.getSelectedPartitionIds().size() != 1) {
Review comment:
Why is it related to the number of partitions here?
##########
File path: fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
##########
@@ -1240,11 +1244,16 @@ private void computeScanRangeAssignment() throws
Exception {
scanNodeIds.add(scanNode.getId().asInt());
FragmentScanRangeAssignment assignment =
fragmentExecParamsMap.get(scanNode.getFragmentId()).scanRangeAssignment;
- if (isColocateJoin(scanNode.getFragment().getPlanRoot())) {
+ boolean fragmentContainsColocateJoin =
isColocateJoin(scanNode.getFragment().getPlanRoot());
+ boolean fragmentContainsBucketShuffleJoin =
bucketShuffleJoinController.isBucketShuffleJoin(scanNode.getFragmentId().asInt(),
scanNode.getFragment().getPlanRoot());
+
+ if (fragmentContainsColocateJoin) {
computeScanRangeAssignmentByColocate((OlapScanNode) scanNode);
- } else if
(bucketShuffleJoinController.isBucketShuffleJoin(scanNode.getFragmentId().asInt(),
scanNode.getFragment().getPlanRoot())) {
+ }
+ if (fragmentContainsBucketShuffleJoin) {
Review comment:
```suggestion
else if (fragmentContainsBucketShuffleJoin) {
```
##########
File path:
fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java
##########
@@ -529,21 +533,40 @@ private boolean canBucketShuffleJoin(HashJoinNode node,
PlanFragment leftChildFr
}
PlanNode leftRoot = leftChildFragment.getPlanRoot();
- // leftRoot should be OlapScanNode
+ // 1.leftRoot be OlapScanNode
if (leftRoot instanceof OlapScanNode) {
return canBucketShuffleJoin(node, leftRoot, rhsHashExprs);
}
+ // 2.leftRoot be colocate hashjoin node or bucket shuffle join node
+ PlanNode rightRoot = rightChildFragment.getPlanRoot();
+ if (leftRoot instanceof HashJoinNode) {
+ while (leftRoot instanceof HashJoinNode) {
+ if (((HashJoinNode)leftRoot).isColocate() ||
((HashJoinNode)leftRoot).isBucketShuffle()) {
Review comment:
how about broadcast？
##########
File path: fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
##########
@@ -1180,32 +1180,36 @@ private void
computeColocateJoinInstanceParam(PlanFragmentId fragmentId, int par
}
}
+ // set bucket for scanrange
Review comment:
It is recommended to enrich the comments and give a practical example to
illustrate what is stored in this nested type.
##########
File path: fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
##########
@@ -1240,11 +1244,16 @@ private void computeScanRangeAssignment() throws
Exception {
scanNodeIds.add(scanNode.getId().asInt());
FragmentScanRangeAssignment assignment =
fragmentExecParamsMap.get(scanNode.getFragmentId()).scanRangeAssignment;
- if (isColocateJoin(scanNode.getFragment().getPlanRoot())) {
+ boolean fragmentContainsColocateJoin =
isColocateJoin(scanNode.getFragment().getPlanRoot());
+ boolean fragmentContainsBucketShuffleJoin =
bucketShuffleJoinController.isBucketShuffleJoin(scanNode.getFragmentId().asInt(),
scanNode.getFragment().getPlanRoot());
+
+ if (fragmentContainsColocateJoin) {
computeScanRangeAssignmentByColocate((OlapScanNode) scanNode);
- } else if
(bucketShuffleJoinController.isBucketShuffleJoin(scanNode.getFragmentId().asInt(),
scanNode.getFragment().getPlanRoot())) {
+ }
+ if (fragmentContainsBucketShuffleJoin) {
bucketShuffleJoinController.computeScanRangeAssignmentByBucket((OlapScanNode)
scanNode, idToBackend, addressToBackendID);
- } else {
+ }
+ if (!(fragmentContainsColocateJoin |
fragmentContainsBucketShuffleJoin)) {
Review comment:
```suggestion
else {
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
