Split AddExchanges.Rewriter#visitJoin into smaller methods
@@ -710,17 +710,23 @@ public class AddExchanges
List<Symbol> rightSymbols = node.getCriteria().stream()
.map(JoinNode.EquiJoinClause::getRight)
.collect(toImmutableList());
- JoinNode.Type type = node.getType();
-
- PlanWithProperties left;
- PlanWithProperties right;
JoinNode.DistributionType distributionType = node.getDistributionType().orElseThrow(() -> new IllegalArgumentException("distributionType not yet set"));
if (distributionType == JoinNode.DistributionType.PARTITIONED) {
+ return planPartitionedJoin(node, leftSymbols, rightSymbols);
+ }
+ else {
+ return planReplicatedJoin(node);
+ }
+ }
+
+ private PlanWithProperties planPartitionedJoin(JoinNode node, List<Symbol> leftSymbols, List<Symbol> rightSymbols)
+ {
SetMultimap<Symbol, Symbol> rightToLeft = createMapping(rightSymbols, leftSymbols);
SetMultimap<Symbol, Symbol> leftToRight = createMapping(leftSymbols, rightSymbols);
- left = node.getLeft().accept(this, PreferredProperties.partitioned(ImmutableSet.copyOf(leftSymbols)));
+ PlanWithProperties left = node.getLeft().accept(this, PreferredProperties.partitioned(ImmutableSet.copyOf(leftSymbols)));
+ PlanWithProperties right;
if (left.getProperties().isNodePartitionedOn(leftSymbols) && !left.getProperties().isSingleNode()) {
Partitioning rightPartitioning = left.getProperties().translate(createTranslator(leftToRight)).getNodePartitioning().get();
@@ -759,11 +765,15 @@ public class AddExchanges
partitionedExchange(idAllocator.getNextId(), REMOTE, right.getNode(), new PartitioningScheme(rightPartitioning, right.getNode().getOutputSymbols())),
right.getProperties());
}
+
+ return buildJoin(node, left, right, JoinNode.DistributionType.PARTITIONED);
}
- else {
+
+ private PlanWithProperties planReplicatedJoin(JoinNode node)
+ {
// Broadcast Join
- left = node.getLeft().accept(this, PreferredProperties.any());
- right = node.getRight().accept(this, PreferredProperties.any());
+ PlanWithProperties left = node.getLeft().accept(this, PreferredProperties.any());
+ PlanWithProperties right = node.getRight().accept(this, PreferredProperties.any());
if (left.getProperties().isSingleNode()) {
if (!right.getProperties().isSingleNode() ||
@@ -778,20 +788,24 @@ public class AddExchanges
replicatedExchange(idAllocator.getNextId(), REMOTE, right.getNode()),
right.getProperties());
}
+
+ return buildJoin(node, left, right, JoinNode.DistributionType.REPLICATED);
}
+ private PlanWithProperties buildJoin(JoinNode node, PlanWithProperties newLeft, PlanWithProperties newRight, JoinNode.DistributionType newDistributionType)
+ {
JoinNode result = new JoinNode(node.getId(),
- type,
- left.getNode(),
- right.getNode(),
+ node.getType(),
+ newLeft.getNode(),
+ newRight.getNode(),
node.getCriteria(),
node.getOutputSymbols(),
node.getFilter(),
node.getLeftHashSymbol(),
node.getRightHashSymbol(),
- node.getDistributionType());
+ Optional.of(newDistributionType));
- return new PlanWithProperties(result, deriveProperties(result, ImmutableList.of(left.getProperties(), right.getProperties())));
+ return new PlanWithProperties(result, deriveProperties(result, ImmutableList.of(newLeft.getProperties(), newRight.getProperties())));
}
@Override
