Repository: asterixdb Updated Branches: refs/heads/master ed4693812 -> dc7e68a02
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/dc7e68a0/asterixdb/asterix-app/src/test/resources/optimizerts/results/tpcds/query-ASTERIXDB-1581.plan ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/test/resources/optimizerts/results/tpcds/query-ASTERIXDB-1581.plan b/asterixdb/asterix-app/src/test/resources/optimizerts/results/tpcds/query-ASTERIXDB-1581.plan index 14b31eb..f914055 100644 --- a/asterixdb/asterix-app/src/test/resources/optimizerts/results/tpcds/query-ASTERIXDB-1581.plan +++ b/asterixdb/asterix-app/src/test/resources/optimizerts/results/tpcds/query-ASTERIXDB-1581.plan @@ -26,155 +26,146 @@ -- NESTED_TUPLE_SOURCE |LOCAL| } -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- STABLE_SORT [$$92(ASC)] |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| + -- HYBRID_HASH_JOIN [$$92][$$93] |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- HYBRID_HASH_JOIN [$$92][$$93] |PARTITIONED| - -- HASH_PARTITION_EXCHANGE [$$92] |PARTITIONED| - -- PRE_CLUSTERED_GROUP_BY[$$32] |PARTITIONED| - { - -- AGGREGATE |LOCAL| + -- PRE_CLUSTERED_GROUP_BY[$$32] |PARTITIONED| + { + -- AGGREGATE |LOCAL| + -- AGGREGATE |LOCAL| + -- ASSIGN |LOCAL| -- AGGREGATE |LOCAL| - -- ASSIGN |LOCAL| - -- AGGREGATE |LOCAL| - -- STREAM_SELECT |LOCAL| - -- NESTED_TUPLE_SOURCE |LOCAL| - } + -- STREAM_SELECT |LOCAL| + -- NESTED_TUPLE_SOURCE |LOCAL| + } + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- STABLE_SORT [$$32(ASC)] |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- STABLE_SORT [$$32(ASC)] |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- HYBRID_HASH_JOIN [$$32][$$85] |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- ASSIGN |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- REPLICATE |PARTITIONED| - -- HASH_PARTITION_EXCHANGE [$$94] |PARTITIONED| - -- STREAM_PROJECT |UNPARTITIONED| - -- ASSIGN |UNPARTITIONED| - -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| - -- REPLICATE |UNPARTITIONED| - -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| - -- AGGREGATE |UNPARTITIONED| - -- AGGREGATE |UNPARTITIONED| - -- RANDOM_MERGE_EXCHANGE |PARTITIONED| - -- AGGREGATE |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- STREAM_SELECT |PARTITIONED| - -- ASSIGN |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- DATASOURCE_SCAN |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- EMPTY_TUPLE_SOURCE |PARTITIONED| - -- HASH_PARTITION_EXCHANGE [$$85] |PARTITIONED| - -- ASSIGN |PARTITIONED| + -- HYBRID_HASH_JOIN [$$32][$$85] |PARTITIONED| + -- HASH_PARTITION_EXCHANGE [$$32] |PARTITIONED| + -- STREAM_PROJECT |UNPARTITIONED| + -- ASSIGN |UNPARTITIONED| + -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| + -- REPLICATE |UNPARTITIONED| + -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| + -- AGGREGATE |UNPARTITIONED| + -- AGGREGATE |UNPARTITIONED| + -- RANDOM_MERGE_EXCHANGE |PARTITIONED| + -- AGGREGATE |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- STREAM_SELECT |PARTITIONED| + -- ASSIGN |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- DATASOURCE_SCAN |PARTITIONED| + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- EMPTY_TUPLE_SOURCE |PARTITIONED| + -- HASH_PARTITION_EXCHANGE [$$85] |PARTITIONED| + -- ASSIGN |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- STREAM_SELECT |PARTITIONED| -- STREAM_PROJECT |PARTITIONED| - -- STREAM_SELECT |PARTITIONED| + -- ASSIGN |PARTITIONED| -- STREAM_PROJECT |PARTITIONED| -- ASSIGN |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- ASSIGN |PARTITIONED| + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- REPLICATE |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- REPLICATE |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- DATASOURCE_SCAN |PARTITIONED| - -- BROADCAST_EXCHANGE |PARTITIONED| - -- STREAM_SELECT |UNPARTITIONED| + -- DATASOURCE_SCAN |PARTITIONED| + -- BROADCAST_EXCHANGE |PARTITIONED| + -- STREAM_SELECT |UNPARTITIONED| + -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| + -- REPLICATE |UNPARTITIONED| -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| - -- REPLICATE |UNPARTITIONED| - -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| - -- AGGREGATE |UNPARTITIONED| - -- AGGREGATE |UNPARTITIONED| - -- RANDOM_MERGE_EXCHANGE |PARTITIONED| - -- AGGREGATE |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- STREAM_SELECT |PARTITIONED| - -- ASSIGN |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| + -- AGGREGATE |UNPARTITIONED| + -- AGGREGATE |UNPARTITIONED| + -- RANDOM_MERGE_EXCHANGE |PARTITIONED| + -- AGGREGATE |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- STREAM_SELECT |PARTITIONED| + -- ASSIGN |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- DATASOURCE_SCAN |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- DATASOURCE_SCAN |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- EMPTY_TUPLE_SOURCE |PARTITIONED| - -- HASH_PARTITION_EXCHANGE [$$93] |PARTITIONED| - -- ASSIGN |PARTITIONED| + -- EMPTY_TUPLE_SOURCE |PARTITIONED| + -- HASH_PARTITION_EXCHANGE [$$93] |PARTITIONED| + -- ASSIGN |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- STREAM_SELECT |PARTITIONED| -- STREAM_PROJECT |PARTITIONED| - -- STREAM_SELECT |PARTITIONED| + -- ASSIGN |PARTITIONED| -- STREAM_PROJECT |PARTITIONED| - -- ASSIGN |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- DATASOURCE_SCAN |PARTITIONED| - -- BROADCAST_EXCHANGE |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- STREAM_SELECT |PARTITIONED| - -- ASSIGN |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- PRE_CLUSTERED_GROUP_BY[$$94] |PARTITIONED| - { + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- DATASOURCE_SCAN |PARTITIONED| + -- BROADCAST_EXCHANGE |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- STREAM_SELECT |PARTITIONED| + -- ASSIGN |PARTITIONED| + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- PRE_CLUSTERED_GROUP_BY[$$94] |PARTITIONED| + { + -- AGGREGATE |LOCAL| + -- AGGREGATE |LOCAL| -- AGGREGATE |LOCAL| - -- AGGREGATE |LOCAL| - -- AGGREGATE |LOCAL| - -- STREAM_SELECT |LOCAL| - -- NESTED_TUPLE_SOURCE |LOCAL| - } + -- STREAM_SELECT |LOCAL| + -- NESTED_TUPLE_SOURCE |LOCAL| + } + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- STABLE_SORT [$$94(ASC)] |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- STABLE_SORT [$$94(ASC)] |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- HYBRID_HASH_JOIN [$$94][$$95] |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- REPLICATE |PARTITIONED| - -- HASH_PARTITION_EXCHANGE [$$94] |PARTITIONED| - -- STREAM_PROJECT |UNPARTITIONED| - -- ASSIGN |UNPARTITIONED| - -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| - -- REPLICATE |UNPARTITIONED| - -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| - -- AGGREGATE |UNPARTITIONED| - -- AGGREGATE |UNPARTITIONED| - -- RANDOM_MERGE_EXCHANGE |PARTITIONED| - -- AGGREGATE |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- STREAM_SELECT |PARTITIONED| - -- ASSIGN |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- DATASOURCE_SCAN |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- EMPTY_TUPLE_SOURCE |PARTITIONED| - -- HASH_PARTITION_EXCHANGE [$$95] |PARTITIONED| - -- ASSIGN |PARTITIONED| + -- HYBRID_HASH_JOIN [$$94][$$95] |PARTITIONED| + -- HASH_PARTITION_EXCHANGE [$$94] |PARTITIONED| + -- STREAM_PROJECT |UNPARTITIONED| + -- ASSIGN |UNPARTITIONED| + -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| + -- REPLICATE |UNPARTITIONED| + -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| + -- AGGREGATE |UNPARTITIONED| + -- AGGREGATE |UNPARTITIONED| + -- RANDOM_MERGE_EXCHANGE |PARTITIONED| + -- AGGREGATE |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- STREAM_SELECT |PARTITIONED| + -- ASSIGN |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- DATASOURCE_SCAN |PARTITIONED| + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- EMPTY_TUPLE_SOURCE |PARTITIONED| + -- HASH_PARTITION_EXCHANGE [$$95] |PARTITIONED| + -- ASSIGN |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- STREAM_SELECT |PARTITIONED| -- STREAM_PROJECT |PARTITIONED| - -- STREAM_SELECT |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- ASSIGN |PARTITIONED| + -- ASSIGN |PARTITIONED| + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- REPLICATE |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- REPLICATE |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- DATASOURCE_SCAN |PARTITIONED| - -- BROADCAST_EXCHANGE |PARTITIONED| - -- STREAM_SELECT |UNPARTITIONED| + -- DATASOURCE_SCAN |PARTITIONED| + -- BROADCAST_EXCHANGE |PARTITIONED| + -- STREAM_SELECT |UNPARTITIONED| + -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| + -- REPLICATE |UNPARTITIONED| -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| - -- REPLICATE |UNPARTITIONED| - -- ONE_TO_ONE_EXCHANGE |UNPARTITIONED| - -- AGGREGATE |UNPARTITIONED| - -- AGGREGATE |UNPARTITIONED| - -- RANDOM_MERGE_EXCHANGE |PARTITIONED| - -- AGGREGATE |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| - -- STREAM_SELECT |PARTITIONED| - -- ASSIGN |PARTITIONED| - -- STREAM_PROJECT |PARTITIONED| + -- AGGREGATE |UNPARTITIONED| + -- AGGREGATE |UNPARTITIONED| + -- RANDOM_MERGE_EXCHANGE |PARTITIONED| + -- AGGREGATE |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- STREAM_SELECT |PARTITIONED| + -- ASSIGN |PARTITIONED| + -- STREAM_PROJECT |PARTITIONED| + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + -- DATASOURCE_SCAN |PARTITIONED| -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- DATASOURCE_SCAN |PARTITIONED| - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - -- EMPTY_TUPLE_SOURCE |PARTITIONED| + -- EMPTY_TUPLE_SOURCE |PARTITIONED| http://git-wip-us.apache.org/repos/asf/asterixdb/blob/dc7e68a0/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AbstractPreclusteredGroupByPOperator.java ---------------------------------------------------------------------- diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AbstractPreclusteredGroupByPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AbstractPreclusteredGroupByPOperator.java index ab68b68..75970ac 100644 --- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AbstractPreclusteredGroupByPOperator.java +++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AbstractPreclusteredGroupByPOperator.java @@ -22,6 +22,7 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; import org.apache.commons.lang3.mutable.Mutable; @@ -86,6 +87,10 @@ public abstract class AbstractPreclusteredGroupByPOperator extends AbstractPhysi ILogicalOperator op2 = gby.getInputs().get(0).getValue(); IPhysicalPropertiesVector childProp = op2.getDeliveredPhysicalProperties(); IPartitioningProperty pp = childProp.getPartitioningProperty(); + Map<LogicalVariable, LogicalVariable> ppSubstMap = computePartitioningPropertySubstitutionMap(gby, pp); + if (ppSubstMap != null) { + pp.substituteColumnVars(ppSubstMap); + } List<ILocalStructuralProperty> childLocals = childProp.getLocalProperties(); if (childLocals == null) { deliveredProperties = new StructuralPropertiesVector(pp, propsLocal); @@ -100,6 +105,38 @@ public abstract class AbstractPreclusteredGroupByPOperator extends AbstractPhysi deliveredProperties = new StructuralPropertiesVector(pp, propsLocal); } + // If we have "gby var1 as var3, var2 as var4" + // and input is partitioned on (var1,var2) then output is partitioned on (var3,var4) + private Map<LogicalVariable, LogicalVariable> computePartitioningPropertySubstitutionMap(GroupByOperator gbyOp, + IPartitioningProperty childpp) { + Set<LogicalVariable> childPartitioningColumns = new HashSet<>(); + childpp.getColumns(childPartitioningColumns); + + List<Pair<LogicalVariable, Mutable<ILogicalExpression>>> groupByList = gbyOp.getGroupByList(); + if (groupByList.size() != childPartitioningColumns.size()) { + return null; + } + + Map<LogicalVariable, LogicalVariable> substMap = null; + for (Pair<LogicalVariable, Mutable<ILogicalExpression>> ve : groupByList) { + ILogicalExpression expr = ve.second.getValue(); + if (expr.getExpressionTag() != LogicalExpressionTag.VARIABLE) { + return null; + } + VariableReferenceExpression varRefExpr = (VariableReferenceExpression) expr; + LogicalVariable var = varRefExpr.getVariableReference(); + if (!childPartitioningColumns.remove(var)) { + return null; + } + if (substMap == null) { + substMap = new HashMap<>(); + } + substMap.put(var, ve.first); + } + + return childPartitioningColumns.isEmpty() ? substMap : null; + } + @Override public PhysicalRequirements getRequiredPropertiesForChildren(ILogicalOperator op, IPhysicalPropertiesVector reqdByParent, IOptimizationContext context) {
