Force repartition when aggregation node has mixed grouping sets
@@ -240,9 +240,12 @@ public class AddExchanges
Set<VariableReferenceExpression> partitioningRequirement = ImmutableSet.copyOf(node.getGroupingKeys());
boolean preferSingleNode = hasSingleNodeExecutionPreference(node, metadata.getFunctionManager());
+ boolean hasMixedGroupingSets = node.hasEmptyGroupingSet() && node.hasNonEmptyGroupingSet();
PreferredProperties preferredProperties = preferSingleNode ? PreferredProperties.undistributed() : PreferredProperties.any();
- if (!node.getGroupingKeys().isEmpty()) {
+ // If aggregation has a mixed of non-global and global grouping set, an repartition exchange is any way needed to eliminate duplicate default outputs
+ // from partial aggregations (enforced in `ValidateAggregationWithDefaultValues.java`). Therefore, we don't have preference on what the child will return.
+ if (!node.getGroupingKeys().isEmpty() && !hasMixedGroupingSets) {
preferredProperties = PreferredProperties.partitionedWithLocal(partitioningRequirement, grouped(node.getGroupingKeys()))
.mergeWithParent(parentPreferredProperties);
}
@@ -259,7 +262,8 @@ public class AddExchanges
gatheringExchange(idAllocator.getNextId(), REMOTE_STREAMING, child.getNode()),
child.getProperties());
}
- else if (!child.getProperties().isStreamPartitionedOn(partitioningRequirement) && !child.getProperties().isNodePartitionedOn(partitioningRequirement)) {
+ else if (hasMixedGroupingSets
+ || !child.getProperties().isStreamPartitionedOn(partitioningRequirement) && !child.getProperties().isNodePartitionedOn(partitioningRequirement)) {
child = withDerivedProperties(
partitionedExchange(
idAllocator.getNextId(),
