Nitin-Kashyap commented on code in PR #27784:
URL: https://github.com/apache/doris/pull/27784#discussion_r1414265683
##########
fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java:
##########
@@ -626,17 +639,83 @@ private boolean canBucketShuffleJoin(HashJoinNode node,
PlanFragment leftChildFr
leftRoot = leftRoot.getChild(0);
}
if (leftRoot instanceof OlapScanNode) {
- return canBucketShuffleJoin(node, leftRoot, rhsHashExprs);
+ return canBucketShuffleJoin(node, (OlapScanNode) leftRoot,
rhsHashExprs);
+ } else if (leftRoot instanceof HiveScanNode) {
+ return canBucketShuffleJoin(node, (HiveScanNode) leftRoot,
rhsHashExprs, hashType);
}
}
return false;
}
+ private boolean canBucketShuffleJoin(HashJoinNode node, HiveScanNode
leftScanNode,
+ List<Expr> rhsJoinExprs,
Ref<THashType> hashType) {
+ HMSExternalTable leftTable = leftScanNode.getHiveTable();
+
+ DistributionInfo leftDistribution =
leftTable.getDefaultDistributionInfo();
+ if (leftDistribution == null || !(leftDistribution instanceof
HiveExternalDistributionInfo)) {
+ return false;
+ }
+
+ HiveExternalDistributionInfo hiveDistributionInfo =
(HiveExternalDistributionInfo) leftDistribution;
+
+ List<Column> leftDistributeColumns =
hiveDistributionInfo.getDistributionColumns();
+ List<String> leftDistributeColumnNames = leftDistributeColumns.stream()
+ .map(col -> leftTable.getName() + "." +
col.getName().toLowerCase()).collect(Collectors.toList());
+
+ List<String> leftJoinColumnNames = new ArrayList<>();
+ List<Expr> rightExprs = new ArrayList<>();
+ List<BinaryPredicate> eqJoinConjuncts = node.getEqJoinConjuncts();
+
+ for (BinaryPredicate eqJoinPredicate : eqJoinConjuncts) {
+ Expr lhsJoinExpr = eqJoinPredicate.getChild(0);
+ Expr rhsJoinExpr = eqJoinPredicate.getChild(1);
+ if (lhsJoinExpr.unwrapSlotRef() == null ||
rhsJoinExpr.unwrapSlotRef() == null) {
+ continue;
+ }
+
+ SlotRef leftSlot =
node.getChild(0).findSrcSlotRef(lhsJoinExpr.unwrapSlotRef());
+ if (leftSlot.getTable() instanceof HMSExternalTable
+ &&
leftScanNode.desc.getSlots().contains(leftSlot.getDesc())) {
+ // table name in SlotRef is not the really name. `select *
from test as t`
+ // table name in SlotRef is `t`, but here we need is `test`.
+ leftJoinColumnNames.add(leftSlot.getTable().getName() + "."
+ + leftSlot.getColumnName().toLowerCase());
+ rightExprs.add(rhsJoinExpr);
+ }
+ }
+
+ //2 the join columns should contains all left table distribute columns
to enable bucket shuffle join
+ for (int i = 0; i < leftDistributeColumnNames.size(); i++) {
+ String distributeColumnName = leftDistributeColumnNames.get(i);
+ boolean findRhsExprs = false;
+ // check the join column name is same as distribute column name and
+ // check the rhs join expr type is same as distribute column
+ for (int j = 0; j < leftJoinColumnNames.size(); j++) {
+ if (leftJoinColumnNames.get(j).equals(distributeColumnName)) {
+ // varchar and string type don't need to check the length
property
+ if ((rightExprs.get(j).getType().isVarcharOrStringType()
+ &&
leftDistributeColumns.get(i).getType().isVarcharOrStringType())
+ ||
(rightExprs.get(j).getType().equals(leftDistributeColumns.get(i).getType()))) {
+ rhsJoinExprs.add(rightExprs.get(j));
+ findRhsExprs = true;
+ break;
+ }
+ }
+ }
+
+ if (!findRhsExprs) {
+ return false;
+ }
+ }
+
+ hashType.value = THashType.SPARK_MURMUR32;
Review Comment:
added check to create Hash distribution only in case of Spark generated
tables; for Hive/Hudi it will return as RandomDistribution
--
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: commits-unsubscr...@doris.apache.org
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
