You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by GitBox <gi...@apache.org> on 2022/12/05 10:29:49 UTC
[GitHub] [doris] 924060929 opened a new pull request, #14827: (Refactor) (Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
924060929 opened a new pull request, #14827:
URL: https://github.com/apache/doris/pull/14827
# Proposed changes
Issue Number: close #xxx
## Problem summary
Describe your changes.
## Checklist(Required)
1. Does it affect the original behavior:
- [ ] Yes
- [ ] No
- [ ] I don't know
2. Has unit tests been added:
- [ ] Yes
- [ ] No
- [ ] No Need
3. Has document been added or modified:
- [ ] Yes
- [ ] No
- [ ] No Need
4. Does it need to update dependencies:
- [ ] Yes
- [ ] No
5. Are there any changes that cannot be rolled back:
- [ ] Yes (If Yes, please explain WHY)
- [ ] No
## Further comments
If this is a relatively large or complex change, kick off the discussion at [dev@doris.apache.org](mailto:dev@doris.apache.org) by explaining why you chose the solution you did and what alternatives you considered, etc...
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] hello-stephen commented on pull request #14827: (Refactor) (Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
hello-stephen commented on PR #14827:
URL: https://github.com/apache/doris/pull/14827#issuecomment-1337491743
TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 35.46 seconds
load time: 451 seconds
storage size: 17123356271 Bytes
https://doris-community-test-1308700295.cos.ap-hongkong.myqcloud.com/tmp/20221205143926_clickbench_pr_58109.html
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] 924060929 commented on a diff in pull request #14827: [Refactor](Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
924060929 commented on code in PR #14827:
URL: https://github.com/apache/doris/pull/14827#discussion_r1050647059
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/NormalizeAggregate.java:
##########
@@ -56,100 +52,86 @@
* After rule:
* Project(k1#1, Alias(SR#9)#4, Alias(k1#1 + 1)#5, Alias(SR#10))#6, Alias(SR#11))#7, Alias(SR#10 + 1)#8)
* +-- Aggregate(keys:[k1#1, SR#9], outputs:[k1#1, SR#9, Alias(SUM(v1#3))#10, Alias(SUM(v1#3 + 1))#11])
- * +-- Project(k1#1, Alias(K2#2 + 1)#9, v1#3)
+ * +-- Project(k1#1, Alias(K2#2 + 1)#9, v1#3)
* <p>
* More example could get from UT {NormalizeAggregateTest}
*/
-public class NormalizeAggregate extends OneRewriteRuleFactory {
+public class NormalizeAggregate extends OneRewriteRuleFactory implements NormalizeToSlot {
@Override
public Rule build() {
return logicalAggregate().whenNot(LogicalAggregate::isNormalized).then(aggregate -> {
- // substitution map used to substitute expression in aggregate's output to use it as top projections
- Map<Expression, Expression> substitutionMap = Maps.newHashMap();
- List<Expression> keys = aggregate.getGroupByExpressions();
- List<NamedExpression> newOutputs = Lists.newArrayList();
-
- // keys
- Map<Boolean, List<Expression>> partitionedKeys = keys.stream()
- .collect(Collectors.groupingBy(SlotReference.class::isInstance));
- List<Expression> newKeys = Lists.newArrayList();
- List<NamedExpression> bottomProjections = Lists.newArrayList();
- if (partitionedKeys.containsKey(false)) {
- // process non-SlotReference keys
- newKeys.addAll(partitionedKeys.get(false).stream()
- .map(e -> new Alias(e, e.toSql()))
- .peek(a -> substitutionMap.put(a.child(), a.toSlot()))
- .peek(bottomProjections::add)
- .map(Alias::toSlot)
- .collect(Collectors.toList()));
- }
- if (partitionedKeys.containsKey(true)) {
- // process SlotReference keys
- partitionedKeys.get(true).stream()
- .map(SlotReference.class::cast)
- .peek(s -> substitutionMap.put(s, s))
- .peek(bottomProjections::add)
- .forEach(newKeys::add);
- }
- // add all necessary key to output
- substitutionMap.entrySet().stream()
- .filter(kv -> aggregate.getOutputExpressions().stream()
- .anyMatch(e -> e.anyMatch(kv.getKey()::equals)))
- .map(Entry::getValue)
- .map(NamedExpression.class::cast)
- .forEach(newOutputs::add);
-
- // if we generate bottom, we need to generate to project too.
- // output
- List<NamedExpression> outputs = aggregate.getOutputExpressions();
- Map<Boolean, List<NamedExpression>> partitionedOutputs = outputs.stream()
- .collect(Collectors.groupingBy(e -> e.anyMatch(AggregateFunction.class::isInstance)));
-
- boolean needBottomProjects = partitionedKeys.containsKey(false);
- if (partitionedOutputs.containsKey(true)) {
- // process expressions that contain aggregate function
- Set<AggregateFunction> aggregateFunctions = partitionedOutputs.get(true).stream()
- .flatMap(e -> e.<Set<AggregateFunction>>collect(AggregateFunction.class::isInstance).stream())
- .collect(Collectors.toSet());
-
- // replace all non-slot expression in aggregate functions children.
- for (AggregateFunction aggregateFunction : aggregateFunctions) {
- List<Expression> newChildren = Lists.newArrayList();
- for (Expression child : aggregateFunction.getArguments()) {
- if (child instanceof SlotReference || child instanceof Literal) {
- newChildren.add(child);
- if (child instanceof SlotReference) {
- bottomProjections.add((SlotReference) child);
- }
- } else {
- needBottomProjects = true;
- Alias alias = new Alias(child, child.toSql());
- bottomProjections.add(alias);
- newChildren.add(alias.toSlot());
- }
- }
- AggregateFunction newFunction = (AggregateFunction) aggregateFunction.withChildren(newChildren);
- Alias alias = new Alias(newFunction, newFunction.toSql());
- newOutputs.add(alias);
- substitutionMap.put(aggregateFunction, alias.toSlot());
- }
- }
-
- // assemble
- LogicalPlan root = aggregate.child();
- if (needBottomProjects) {
- root = new LogicalProject<>(bottomProjections, root);
- }
- root = new LogicalAggregate<>(newKeys, newOutputs, aggregate.isDisassembled(),
- true, aggregate.isFinalPhase(), aggregate.getAggPhase(),
- aggregate.getSourceRepeat(), root);
- List<NamedExpression> projections = outputs.stream()
- .map(e -> ExpressionUtils.replace(e, substitutionMap))
- .map(NamedExpression.class::cast)
- .collect(Collectors.toList());
- root = new LogicalProject<>(projections, root);
-
- return root;
+ // push expression to bottom project
+ Set<Alias> existsAliases = ExpressionUtils.collect(
Review Comment:
the result is
```
LogicalAggregate(groupBy=(slot#1, slot#2, slot#3), output=[slot#1, slot#2, slot#3, sum(slot#4)])
|
LogicalProject(projects=[(a + 1)#1, (a + 2)#2, (a + 3)#3, b#4])
```
and
```
LogicalAggregate(groupBy=(slot#0), output=[slot#0, sum(slot#1), sum(slot#2), sum(slot#3)])
|
LogicalProject(projects=[a#0, (b + 1)#1, (b + 2)#2, (b + 3)#3])
```
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] 924060929 commented on a diff in pull request #14827: [Refactor](Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
924060929 commented on code in PR #14827:
URL: https://github.com/apache/doris/pull/14827#discussion_r1050636798
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java:
##########
@@ -314,6 +320,11 @@ public CascadesContext getCascadesContext() {
return cascadesContext;
}
+ public static PhysicalProperties buildInitRequireProperties(Plan initPlan) {
+ boolean isQuery = !(initPlan instanceof Command) || (initPlan instanceof ExplainCommand);
Review Comment:
currently, explain command only for explain query, so we should require gather for it.
when the explain command support other sqls, like `insert into`, we should change it.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] 924060929 commented on a diff in pull request #14827: [Refactor](Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
924060929 commented on code in PR #14827:
URL: https://github.com/apache/doris/pull/14827#discussion_r1050637301
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/memo/Memo.java:
##########
@@ -667,7 +668,8 @@ public String toString() {
builder.append(group).append("\n");
builder.append(" stats=").append(group.getStatistics()).append("\n");
StatsDeriveResult stats = group.getStatistics();
- if (stats != null && group.getLogicalExpressions().get(0).getPlan() instanceof LogicalOlapScan) {
+ if (stats != null && !group.getLogicalExpressions().isEmpty()
Review Comment:
bingo
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] morrySnow commented on a diff in pull request #14827: [Refactor](Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
morrySnow commented on code in PR #14827:
URL: https://github.com/apache/doris/pull/14827#discussion_r1045570822
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/AggregateInfo.java:
##########
@@ -391,6 +394,10 @@ public ArrayList<FunctionCallExpr> getMaterializedAggregateExprs() {
return result;
}
+ public List<String> getMaterializedAggregateExprLabels() {
+ return Lists.newArrayList(materializedSlotLabels);
+ }
Review Comment:
why not put it into AggregateInfoBase?
##########
fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java:
##########
@@ -1216,4 +1219,19 @@ public String getReasonOfPreAggregation() {
public String getSelectedIndexName() {
return olapTable.getIndexNameById(selectedIndexId);
}
+
+ public void finalizeForNerieds() {
+ computeNumNodes();
+ computeStatsForNerieds();
Review Comment:
could we compute them in Nereids itself?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -1055,6 +1112,18 @@ public PlanFragment visitPhysicalLimit(PhysicalLimit<? extends Plan> physicalLim
public PlanFragment visitPhysicalDistribute(PhysicalDistribute<? extends Plan> distribute,
PlanTranslatorContext context) {
PlanFragment childFragment = distribute.child().accept(this, context);
+
+ if (childFragment.getPlanRoot() instanceof AggregationNode
+ && distribute.child() instanceof PhysicalHashAggregate
+ && context.getFirstAggregateInFragment(childFragment) == distribute.child()) {
+ PhysicalHashAggregate<Plan> hashAggregate = (PhysicalHashAggregate) distribute.child();
+ if (hashAggregate.getAggPhase() == AggPhase.LOCAL
+ && hashAggregate.getAggMode() == AggMode.INPUT_TO_BUFFER) {
Review Comment:
ditto
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java:
##########
@@ -183,22 +147,32 @@ private void addRequestPropertyToChildren(PhysicalProperties... physicalProperti
requestPropertyToChildren.add(Lists.newArrayList(physicalProperties));
}
- private List<ExprId> extractFromDistinctFunction(List<NamedExpression> outputExpression) {
+ private void addRequestPropertyToChildren(List<PhysicalProperties> physicalProperties) {
+ requestPropertyToChildren.add(physicalProperties);
+ }
+
+ private List<ExprId> extractExprIdFromDistinctFunction(List<NamedExpression> outputExpression) {
+ Set<AggregateFunction> distinctAggregateFunctions = ExpressionUtils.collect(outputExpression, expr ->
+ expr instanceof AggregateFunction && ((AggregateFunction) expr).isDistinct()
+ );
List<ExprId> exprIds = Lists.newArrayList();
- for (NamedExpression originOutputExpr : outputExpression) {
- Set<AggregateFunction> aggregateFunctions
- = originOutputExpr.collect(AggregateFunction.class::isInstance);
- for (AggregateFunction aggregateFunction : aggregateFunctions) {
- if (aggregateFunction.isDistinct()) {
- for (Expression expr : aggregateFunction.children()) {
- Preconditions.checkState(expr instanceof SlotReference, "normalize aggregate failed to"
- + " normalize aggregate function " + aggregateFunction.toSql());
- exprIds.add(((SlotReference) expr).getExprId());
- }
- }
+ for (AggregateFunction aggregateFunction : distinctAggregateFunctions) {
+ for (Expression expr : aggregateFunction.children()) {
+ Preconditions.checkState(expr instanceof SlotReference, "normalize aggregate failed to"
+ + " normalize aggregate function " + aggregateFunction.toSql());
+ exprIds.add(((SlotReference) expr).getExprId());
}
}
return exprIds;
}
+
+ private void addRequestHashDistribution(List<Expression> hashColumns, ShuffleType shuffleType) {
Review Comment:
not use?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/PhysicalProperties.java:
##########
@@ -99,6 +120,15 @@ public int hashCode() {
@Override
public String toString() {
+ if (this.equals(ANY)) {
+ return "ANY";
+ }
+ if (this.equals(REPLICATED)) {
+ return "REPLICATED";
+ }
+ if (this.equals(GATHER)) {
+ return "GATHER";
+ }
Review Comment:
distributed and order is two part of Phyiscal properties, so cannot just return distributed string
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/memo/Memo.java:
##########
@@ -667,7 +668,8 @@ public String toString() {
builder.append(group).append("\n");
builder.append(" stats=").append(group.getStatistics()).append("\n");
StatsDeriveResult stats = group.getStatistics();
- if (stats != null && group.getLogicalExpressions().get(0).getPlan() instanceof LogicalOlapScan) {
+ if (stats != null && !group.getLogicalExpressions().isEmpty()
Review Comment:
this is because, some group has no logical expression?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java:
##########
@@ -218,13 +222,27 @@ public PhysicalProperties visitPhysicalNestedLoopJoin(
@Override
public PhysicalProperties visitPhysicalOlapScan(PhysicalOlapScan olapScan, PlanContext context) {
- if (olapScan.getDistributionSpec() instanceof DistributionSpecHash) {
+ if (!olapScan.getTable().isColocateTable() && olapScan.getScanTabletNum() == 1) {
+ return PhysicalProperties.GATHER;
Review Comment:
add a todo, let's find a better way to handle both tablet num == 1 and colocate table together in future
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java:
##########
@@ -314,6 +320,11 @@ public CascadesContext getCascadesContext() {
return cascadesContext;
}
+ public static PhysicalProperties buildInitRequireProperties(Plan initPlan) {
+ boolean isQuery = !(initPlan instanceof Command) || (initPlan instanceof ExplainCommand);
Review Comment:
why add ` (initPlan instanceof ExplainCommand)`
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -1233,6 +1240,16 @@ public void setEnableNereidsPlanner(boolean enableNereidsPlanner) {
this.enableNereidsPlanner = enableNereidsPlanner;
}
+ public Set<String> getDisableNereidsRules() {
+ return Arrays.stream(disableNereidsRules.split(",[\\s]*"))
+ .map(rule -> rule.toUpperCase(Locale.ROOT))
+ .collect(ImmutableSet.toImmutableSet());
Review Comment:
nit: could do split and check legality when user do SET, see more in `org.apache.doris.qe.VariableMgr.VarAttr#checker`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/algebra/OlapScan.java:
##########
@@ -0,0 +1,58 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.plans.algebra;
+
+import org.apache.doris.catalog.OlapTable;
+
+import java.util.List;
+
+/** OlapScan */
+public interface OlapScan extends Scan {
+ OlapTable getTable();
+
+ long getSelectedIndexId();
+
+ List<Long> getSelectedPartitionIds();
+
+ List<Long> getSelectedTabletIds();
+
+ /** getScanTabletNum */
+ default int getScanTabletNum() {
Review Comment:
after prune, select 0 tablet is possible
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/FunctionBuilder.java:
##########
@@ -57,8 +59,12 @@ public boolean canApply(List<? extends Object> arguments) {
}
for (int i = 0; i < arguments.size(); i++) {
Class constructorArgumentType = getConstructorArgumentType(i);
- if (!constructorArgumentType.isInstance(arguments.get(i))) {
- return false;
+ Object argument = arguments.get(i);
+ if (!constructorArgumentType.isInstance(argument)) {
+ Optional<Class> primitiveType = ReflectionUtils.getPrimitiveType(argument.getClass());
Review Comment:
this mean, currently, we could only build function with primitive type function?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java:
##########
@@ -72,12 +70,7 @@ public Double visit(Plan plan, Void context) {
}
@Override
- public Double visitPhysicalAggregate(PhysicalAggregate<? extends Plan> agg, Void context) {
- if (agg.isFinalPhase()
- && agg.getAggPhase() == AggPhase.LOCAL
- && children.get(0).getPlan() instanceof PhysicalDistribute) {
- return -1.0;
- }
+ public Double visitPhysicalHashAggregate(PhysicalHashAggregate<? extends Plan> agg, Void context) {
return 0.0;
}
Review Comment:
just remove this function
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/cascades/DeriveStatsJob.java:
##########
@@ -45,31 +47,42 @@ public class DeriveStatsJob extends Job {
* @param context context of current job
*/
public DeriveStatsJob(GroupExpression groupExpression, JobContext context) {
- super(JobType.DERIVE_STATS, context);
- this.groupExpression = groupExpression;
- this.deriveChildren = false;
+ this(groupExpression, false, context);
}
- /**
- * Copy constructor for DeriveStatsJob.
- *
- * @param other DeriveStatsJob copied from
- */
- public DeriveStatsJob(DeriveStatsJob other) {
- super(JobType.DERIVE_STATS, other.context);
- this.groupExpression = other.groupExpression;
- this.deriveChildren = other.deriveChildren;
+ private DeriveStatsJob(GroupExpression groupExpression, boolean deriveChildren, JobContext context) {
+ super(JobType.DERIVE_STATS, context);
+ this.groupExpression = groupExpression;
+ this.deriveChildren = deriveChildren;
}
@Override
public void execute() {
countJobExecutionTimesOfGroupExpressions(groupExpression);
- if (!deriveChildren) {
- deriveChildren = true;
- pushJob(new DeriveStatsJob(this));
- for (Group child : groupExpression.children()) {
- if (!child.getLogicalExpressions().isEmpty()) {
- pushJob(new DeriveStatsJob(child.getLogicalExpressions().get(0), context));
+ if (groupExpression.isStatDerived()) {
+ return;
+ }
+ if (!deriveChildren && groupExpression.arity() > 0) {
+ pushJob(new DeriveStatsJob(groupExpression, true, context));
+
+ List<Group> children = groupExpression.children();
+ for (int i = children.size() - 1; i >= 0; i--) {
+ Group childGroup = children.get(i);
+
+ List<GroupExpression> logicalExpressions = childGroup.getLogicalExpressions();
+ for (int j = logicalExpressions.size() - 1; j >= 0; j--) {
Review Comment:
need add some comment to explain, why we use all group expression to do stats derive
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java:
##########
@@ -183,22 +147,32 @@ private void addRequestPropertyToChildren(PhysicalProperties... physicalProperti
requestPropertyToChildren.add(Lists.newArrayList(physicalProperties));
}
- private List<ExprId> extractFromDistinctFunction(List<NamedExpression> outputExpression) {
+ private void addRequestPropertyToChildren(List<PhysicalProperties> physicalProperties) {
+ requestPropertyToChildren.add(physicalProperties);
+ }
+
+ private List<ExprId> extractExprIdFromDistinctFunction(List<NamedExpression> outputExpression) {
Review Comment:
not use?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalAggregate.java:
##########
@@ -82,8 +67,8 @@ public LogicalAggregate(
List<Expression> groupByExpressions,
List<NamedExpression> outputExpressions,
CHILD_TYPE child) {
- this(groupByExpressions, outputExpressions, Optional.empty(), false,
- false, true, AggPhase.LOCAL, Optional.empty(), child);
+ this(groupByExpressions, outputExpressions,
+ false, Optional.empty(), child);
Review Comment:
nit: could in one line
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -1160,10 +1230,11 @@ private PlanFragment exchangeToMergeFragment(PlanFragment inputFragment, PlanTra
Preconditions.checkState(inputFragment.isPartitioned());
// exchange node clones the behavior of its input, aside from the conjuncts
- ExchangeNode mergePlan =
- new ExchangeNode(context.nextPlanNodeId(), inputFragment.getPlanRoot(), false);
+ ExchangeNode mergePlan = new ExchangeNode(context.nextPlanNodeId(),
+ inputFragment.getPlanRoot(), false);
+ DataPartition dataPartition = DataPartition.UNPARTITIONED;
Review Comment:
why add this local variable?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -1055,6 +1112,18 @@ public PlanFragment visitPhysicalLimit(PhysicalLimit<? extends Plan> physicalLim
public PlanFragment visitPhysicalDistribute(PhysicalDistribute<? extends Plan> distribute,
PlanTranslatorContext context) {
PlanFragment childFragment = distribute.child().accept(this, context);
+
+ if (childFragment.getPlanRoot() instanceof AggregationNode
+ && distribute.child() instanceof PhysicalHashAggregate
Review Comment:
why need double check?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/DistinctToGroupBy.java:
##########
@@ -0,0 +1,32 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.rewrite.OneRewriteRuleFactory;
+
+/** AggGroupByToDistinct */
+public class DistinctToGroupBy extends OneRewriteRuleFactory {
Review Comment:
what's this?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSlotReference.java:
##########
@@ -480,7 +478,7 @@ private boolean handleNamePartsTwoOrThree(Slot boundSlot, List<String> nameParts
}
/** BoundStar is used to wrap list of slots for temporary. */
- private class BoundStar extends NamedExpression implements PropagateNullable {
+ public static class BoundStar extends NamedExpression implements PropagateNullable {
Review Comment:
move out from BindSlotReference class?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/NormalizeAggregate.java:
##########
@@ -56,100 +52,86 @@
* After rule:
* Project(k1#1, Alias(SR#9)#4, Alias(k1#1 + 1)#5, Alias(SR#10))#6, Alias(SR#11))#7, Alias(SR#10 + 1)#8)
* +-- Aggregate(keys:[k1#1, SR#9], outputs:[k1#1, SR#9, Alias(SUM(v1#3))#10, Alias(SUM(v1#3 + 1))#11])
- * +-- Project(k1#1, Alias(K2#2 + 1)#9, v1#3)
+ * +-- Project(k1#1, Alias(K2#2 + 1)#9, v1#3)
* <p>
* More example could get from UT {NormalizeAggregateTest}
*/
-public class NormalizeAggregate extends OneRewriteRuleFactory {
+public class NormalizeAggregate extends OneRewriteRuleFactory implements NormalizeToSlot {
@Override
public Rule build() {
return logicalAggregate().whenNot(LogicalAggregate::isNormalized).then(aggregate -> {
- // substitution map used to substitute expression in aggregate's output to use it as top projections
- Map<Expression, Expression> substitutionMap = Maps.newHashMap();
- List<Expression> keys = aggregate.getGroupByExpressions();
- List<NamedExpression> newOutputs = Lists.newArrayList();
-
- // keys
- Map<Boolean, List<Expression>> partitionedKeys = keys.stream()
- .collect(Collectors.groupingBy(SlotReference.class::isInstance));
- List<Expression> newKeys = Lists.newArrayList();
- List<NamedExpression> bottomProjections = Lists.newArrayList();
- if (partitionedKeys.containsKey(false)) {
- // process non-SlotReference keys
- newKeys.addAll(partitionedKeys.get(false).stream()
- .map(e -> new Alias(e, e.toSql()))
- .peek(a -> substitutionMap.put(a.child(), a.toSlot()))
- .peek(bottomProjections::add)
- .map(Alias::toSlot)
- .collect(Collectors.toList()));
- }
- if (partitionedKeys.containsKey(true)) {
- // process SlotReference keys
- partitionedKeys.get(true).stream()
- .map(SlotReference.class::cast)
- .peek(s -> substitutionMap.put(s, s))
- .peek(bottomProjections::add)
- .forEach(newKeys::add);
- }
- // add all necessary key to output
- substitutionMap.entrySet().stream()
- .filter(kv -> aggregate.getOutputExpressions().stream()
- .anyMatch(e -> e.anyMatch(kv.getKey()::equals)))
- .map(Entry::getValue)
- .map(NamedExpression.class::cast)
- .forEach(newOutputs::add);
-
- // if we generate bottom, we need to generate to project too.
- // output
- List<NamedExpression> outputs = aggregate.getOutputExpressions();
- Map<Boolean, List<NamedExpression>> partitionedOutputs = outputs.stream()
- .collect(Collectors.groupingBy(e -> e.anyMatch(AggregateFunction.class::isInstance)));
-
- boolean needBottomProjects = partitionedKeys.containsKey(false);
- if (partitionedOutputs.containsKey(true)) {
- // process expressions that contain aggregate function
- Set<AggregateFunction> aggregateFunctions = partitionedOutputs.get(true).stream()
- .flatMap(e -> e.<Set<AggregateFunction>>collect(AggregateFunction.class::isInstance).stream())
- .collect(Collectors.toSet());
-
- // replace all non-slot expression in aggregate functions children.
- for (AggregateFunction aggregateFunction : aggregateFunctions) {
- List<Expression> newChildren = Lists.newArrayList();
- for (Expression child : aggregateFunction.getArguments()) {
- if (child instanceof SlotReference || child instanceof Literal) {
- newChildren.add(child);
- if (child instanceof SlotReference) {
- bottomProjections.add((SlotReference) child);
- }
- } else {
- needBottomProjects = true;
- Alias alias = new Alias(child, child.toSql());
- bottomProjections.add(alias);
- newChildren.add(alias.toSlot());
- }
- }
- AggregateFunction newFunction = (AggregateFunction) aggregateFunction.withChildren(newChildren);
- Alias alias = new Alias(newFunction, newFunction.toSql());
- newOutputs.add(alias);
- substitutionMap.put(aggregateFunction, alias.toSlot());
- }
- }
-
- // assemble
- LogicalPlan root = aggregate.child();
- if (needBottomProjects) {
- root = new LogicalProject<>(bottomProjections, root);
- }
- root = new LogicalAggregate<>(newKeys, newOutputs, aggregate.isDisassembled(),
- true, aggregate.isFinalPhase(), aggregate.getAggPhase(),
- aggregate.getSourceRepeat(), root);
- List<NamedExpression> projections = outputs.stream()
- .map(e -> ExpressionUtils.replace(e, substitutionMap))
- .map(NamedExpression.class::cast)
- .collect(Collectors.toList());
- root = new LogicalProject<>(projections, root);
-
- return root;
+ // push expression to bottom project
+ Set<Alias> existsAliases = ExpressionUtils.collect(
Review Comment:
does it solve the problem of explosion of data? SQL like:
```sql
SELECT a + 1, a + 2, a + 3, SUM(b) FROM t GROUP BY a + 1, a + 2, a + 3;
SELECT a, SUM(b + 1), SUM(b + 2), SUM(b + 3) FROM t GROUP BY a;
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/NormalizeToSlot.java:
##########
@@ -0,0 +1,133 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+import javax.annotation.Nullable;
+
+/** NormalizeToSlot */
+public interface NormalizeToSlot {
+
+ /** NormalizeSlotContext */
+ class NormalizeToSlotContext {
+ private final Map<Expression, NormalizeToSlotTriplet> normalizeToSlotMap;
+
+ public NormalizeToSlotContext(Map<Expression, NormalizeToSlotTriplet> normalizeToSlotMap) {
+ this.normalizeToSlotMap = normalizeToSlotMap;
+ }
+
+ /** buildContext */
+ public static NormalizeToSlotContext buildContext(
+ Set<Alias> existsAliases, Set<? extends Expression> sourceExpressions) {
+ Map<Expression, NormalizeToSlotTriplet> normalizeToSlotMap = Maps.newLinkedHashMap();
+
+ Map<Expression, Alias> existsAliasMap = Maps.newLinkedHashMap();
+ for (Alias existsAlias : existsAliases) {
+ existsAliasMap.put(existsAlias.child(), existsAlias);
+ }
+
+ for (Expression expression : sourceExpressions) {
+ if (normalizeToSlotMap.containsKey(expression)) {
+ continue;
+ }
+ NormalizeToSlotTriplet normalizeToSlotTriplet =
+ NormalizeToSlotTriplet.toTriplet(expression, existsAliasMap.get(expression));
+ normalizeToSlotMap.put(expression, normalizeToSlotTriplet);
+ }
+ return new NormalizeToSlotContext(normalizeToSlotMap);
+ }
+
+ /** normalizeToUseSlotRef, no custom normalize */
+ public <E extends Expression> List<E> normalizeToUseSlotRef(List<E> expressions) {
+ return normalizeToUseSlotRef(expressions, (context, expr) -> expr);
+ }
+
+ /** normalizeToUseSlotRef */
+ public <E extends Expression> List<E> normalizeToUseSlotRef(List<E> expressions,
+ BiFunction<NormalizeToSlotContext, Expression, Expression> customNormalize) {
+ return expressions.stream()
+ .map(expr -> (E) expr.rewriteDownShortCircuit(child -> {
+ Expression newChild = customNormalize.apply(this, child);
+ if (newChild != null && newChild != child) {
+ return newChild;
+ }
+ NormalizeToSlotTriplet normalizeToSlotTriplet = normalizeToSlotMap.get(child);
+ return normalizeToSlotTriplet == null ? child : normalizeToSlotTriplet.remainExpr;
+ })).collect(ImmutableList.toImmutableList());
+ }
+
+ /**
+ * generate bottom projections with groupByExpressions.
+ * eg:
+ * groupByExpressions: k1#0, k2#1 + 1;
+ * bottom: k1#0, (k2#1 + 1) AS (k2 + 1)#2;
+ */
+ public Set<NamedExpression> pushDownToNamedExpression(Collection<? extends Expression> needToPushExpressions) {
+ return needToPushExpressions.stream()
+ .map(expr -> {
+ NormalizeToSlotTriplet normalizeToSlotTriplet = normalizeToSlotMap.get(expr);
+ return normalizeToSlotTriplet == null
+ ? (NamedExpression) expr
+ : normalizeToSlotTriplet.pushedExpr;
+ }).collect(ImmutableSet.toImmutableSet());
+ }
+ }
+
+ /** NormalizeToSlotTriplet */
+ class NormalizeToSlotTriplet {
+ public final Expression originExpr;
+ public final Slot remainExpr;
+ public final NamedExpression pushedExpr;
Review Comment:
nit: add a example to explain them
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rewrite/rules/FoldConstantRuleOnFE.java:
##########
@@ -59,143 +61,148 @@ public class FoldConstantRuleOnFE extends AbstractExpressionRewriteRule {
@Override
public Expression rewrite(Expression expr, ExpressionRewriteContext ctx) {
- return process(expr, ctx);
- }
+ if (expr instanceof AggregateFunction && ((AggregateFunction) expr).isDistinct()) {
+ return expr;
+ } else if (expr instanceof AggregateExpression && ((AggregateExpression) expr).getFunction().isDistinct()) {
+ return expr;
Review Comment:
why their child could not be fold?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/AggregateStrategies.java:
##########
@@ -0,0 +1,1234 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.annotation.DependsRules;
+import org.apache.doris.nereids.pattern.PatternDescriptor;
+import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.properties.RequireProperties;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.expression.rewrite.ExpressionRewriteContext;
+import org.apache.doris.nereids.rules.expression.rewrite.rules.FoldConstantRuleOnFE;
+import org.apache.doris.nereids.rules.expression.rewrite.rules.TypeCoercion;
+import org.apache.doris.nereids.rules.implementation.ImplementationRuleFactory;
+import org.apache.doris.nereids.rules.implementation.LogicalOlapScanToPhysicalOlapScan;
+import org.apache.doris.nereids.rules.rewrite.logical.NormalizeAggregate;
+import org.apache.doris.nereids.trees.expressions.AggregateExpression;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.IsNull;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.MultiDistinctCount;
+import org.apache.doris.nereids.trees.expressions.functions.agg.MultiDistinctSum;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.AggMode;
+import org.apache.doris.nereids.trees.plans.AggPhase;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.algebra.Project;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate.PushDownAggOp;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+
+/** AggregateStrategies */
+@DependsRules({
+ NormalizeAggregate.class,
+ FoldConstantRuleOnFE.class
+})
+public class AggregateStrategies implements ImplementationRuleFactory {
+
+ @Override
+ public List<Rule> buildRules() {
+ PatternDescriptor<LogicalAggregate<GroupPlan>> basePattern = logicalAggregate()
+ .when(LogicalAggregate::isNormalized);
+
+ return ImmutableList.of(
+ RuleType.STORAGE_LAYER_AGGREGATE_WITHOUT_PROJECT.build(
+ logicalAggregate(
+ logicalOlapScan()
+ )
+ .when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
+ .thenApply(ctx -> storageLayerAggregate(ctx.root, null, ctx.root.child(), ctx.cascadesContext))
+ ),
+ RuleType.STORAGE_LAYER_AGGREGATE_WITH_PROJECT.build(
+ logicalAggregate(
+ logicalProject(
+ logicalOlapScan()
+ )
+ )
+ .when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
+ .thenApply(ctx -> {
+ LogicalAggregate<LogicalProject<LogicalOlapScan>> agg = ctx.root;
+ LogicalProject<LogicalOlapScan> project = agg.child();
+ LogicalOlapScan olapScan = project.child();
+ return storageLayerAggregate(agg, project, olapScan, ctx.cascadesContext);
+ })
+ ),
+ RuleType.ONE_PHASE_AGGREGATE_WITHOUT_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 0)
+ .thenApplyMulti(ctx -> onePhaseAggregateWithoutDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITHOUT_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 0)
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithoutDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITH_COUNT_DISTINCT_MULTI.build(
+ basePattern
+ .when(this::containsCountDistinctMultiExpr)
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithCountDistinctMulti(ctx.root, ctx.connectContext))
+ ),
+ RuleType.THREE_PHASE_AGGREGATE_WITH_COUNT_DISTINCT_MULTI.build(
+ basePattern
+ .when(this::containsCountDistinctMultiExpr)
+ .thenApplyMulti(ctx -> threePhaseAggregateWithCountDistinctMulti(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITH_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1)
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.ONE_PHASE_AGGREGATE_SINGLE_DISTINCT_TO_MULTI.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1 && enableSingleDistinctColumnOpt())
+ .thenApplyMulti(ctx -> onePhaseAggregateWithMultiDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_SINGLE_DISTINCT_TO_MULTI.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1 && enableSingleDistinctColumnOpt())
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithMultiDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.THREE_PHASE_AGGREGATE_WITH_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1)
+ .thenApplyMulti(ctx -> threePhaseAggregateWithDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITH_MULTI_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() > 1 && !containsCountDistinctMultiExpr(agg))
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithMultiDistinct(ctx.root, ctx.connectContext))
+ )
+ );
+ }
+
+ /**
+ * sql: select count(*) from tbl
+ *
+ * before:
+ *
+ * LogicalAggregate(groupBy=[], output=[count(*)])
+ * |
+ * LogicalOlapScan(table=tbl)
+ *
+ * after:
+ *
+ * LogicalAggregate(groupBy=[], output=[count(*)])
+ * |
+ * PhysicalStorageLayerAggregate(pushAggOp=COUNT, table=PhysicalOlapScan(table=tbl))
+ *
+ */
+ private LogicalAggregate<? extends Plan> storageLayerAggregate(
+ LogicalAggregate<? extends Plan> aggregate,
+ @Nullable LogicalProject<? extends Plan> project,
+ LogicalOlapScan olapScan, CascadesContext cascadesContext) {
+ final LogicalAggregate<? extends Plan> canNotPush = aggregate;
+
+ KeysType keysType = olapScan.getTable().getKeysType();
+ if (keysType != KeysType.AGG_KEYS && keysType != KeysType.DUP_KEYS) {
+ return canNotPush;
+ }
+
+ List<Expression> groupByExpressions = aggregate.getGroupByExpressions();
+ if (!groupByExpressions.isEmpty() || !aggregate.getDistinctArguments().isEmpty()) {
+ return canNotPush;
+ }
+
+ Set<AggregateFunction> aggregateFunctions = aggregate.getAggregateFunctions();
+ Set<Class<? extends AggregateFunction>> functionClasses = aggregateFunctions
+ .stream()
+ .map(AggregateFunction::getClass)
+ .collect(Collectors.toSet());
+
+ Map<Class, PushDownAggOp> supportedAgg = PushDownAggOp.supportedFunctions();
+ if (!supportedAgg.keySet().containsAll(functionClasses)) {
+ return canNotPush;
+ }
+ if (functionClasses.contains(Count.class) && keysType != KeysType.DUP_KEYS) {
+ return canNotPush;
+ }
+ if (aggregateFunctions.stream().anyMatch(fun -> fun.arity() > 1)) {
+ return canNotPush;
+ }
+
+ // we already normalize the arguments to slotReference
+ List<Expression> argumentsOfAggregateFunction = aggregateFunctions.stream()
+ .flatMap(aggregateFunction -> aggregateFunction.getArguments().stream())
+ .collect(ImmutableList.toImmutableList());
+
+ if (project != null) {
+ argumentsOfAggregateFunction = Project.findProject(
+ (List<SlotReference>) (List) argumentsOfAggregateFunction, project.getProjects())
+ .stream()
+ .map(p -> p instanceof Alias ? p.child(0) : p)
+ .collect(ImmutableList.toImmutableList());
+ }
+
+ boolean onlyContainsSlotOrNumericCastSlot = argumentsOfAggregateFunction
+ .stream()
+ .allMatch(argument -> {
+ if (argument instanceof SlotReference) {
+ return true;
+ }
+ if (argument instanceof Cast) {
+ return argument.child(0) instanceof SlotReference
+ && argument.getDataType().isNumericType()
+ && argument.child(0).getDataType().isNumericType();
+ }
+ return false;
+ });
+ if (!onlyContainsSlotOrNumericCastSlot) {
+ return canNotPush;
+ }
+
+ Set<PushDownAggOp> pushDownAggOps = functionClasses.stream()
+ .map(supportedAgg::get)
+ .collect(Collectors.toSet());
+
+ PushDownAggOp mergeOp = pushDownAggOps.size() == 1
+ ? pushDownAggOps.iterator().next()
+ : PushDownAggOp.MIX;
+
+ Set<SlotReference> aggUsedSlots =
+ ExpressionUtils.collect(argumentsOfAggregateFunction, SlotReference.class::isInstance);
+
+ List<SlotReference> usedSlotInTable = (List<SlotReference>) (List) Project.findProject(aggUsedSlots,
+ (List<NamedExpression>) (List) olapScan.getOutput());
+
+ for (SlotReference slot : usedSlotInTable) {
+ Column column = slot.getColumn().get();
+ if (keysType == KeysType.AGG_KEYS && !column.isKey()) {
+ return canNotPush;
+ }
+ // The zone map max length of CharFamily is 512, do not
+ // over the length: https://github.com/apache/doris/pull/6293
+ if (mergeOp == PushDownAggOp.MIN_MAX || mergeOp == PushDownAggOp.MIX) {
+ PrimitiveType colType = column.getType().getPrimitiveType();
+ if (colType.isArrayType() || colType.isComplexType() || colType == PrimitiveType.STRING) {
+ return canNotPush;
+ }
+ if (colType.isCharFamily() && mergeOp != PushDownAggOp.COUNT && column.getType().getLength() > 512) {
+ return canNotPush;
+ }
+ }
+ if (mergeOp == PushDownAggOp.COUNT || mergeOp == PushDownAggOp.MIX) {
+ // NULL value behavior in `count` function is zero, so
+ // we should not use row_count to speed up query. the col
+ // must be not null
+ if (column.isAllowNull()) {
+ return canNotPush;
+ }
+ }
+ }
+
+ PhysicalOlapScan physicalOlapScan = (PhysicalOlapScan) new LogicalOlapScanToPhysicalOlapScan()
+ .build()
+ .transform(olapScan, cascadesContext)
+ .get(0);
+
+ return aggregate.withChildren(ImmutableList.of(
+ new PhysicalStorageLayerAggregate(physicalOlapScan, mergeOp)
+ ));
+ }
+
+ /**
+ * sql: select count(*) from tbl group by id
+ *
+ * before:
+ *
+ * LogicalAggregate(groupBy=[id], output=[count(*)])
+ * |
+ * LogicalOlapScan(table=tbl)
+ *
+ * after:
+ *
+ * single node aggregate:
+ *
+ * PhysicalHashAggregate(groupBy=[id], output=[count(*)])
+ * |
+ * PhysicalDistribute(distributionSpec=GATHER)
+ * |
+ * LogicalOlapScan(table=tbl)
+ *
+ * distribute node aggregate:
+ *
+ * PhysicalHashAggregate(groupBy=[id], output=[count(*)])
+ * |
+ * LogicalOlapScan(table=tbl, **already distribute by id**)
+ *
+ */
+ private List<PhysicalHashAggregate<Plan>> onePhaseAggregateWithoutDistinct(
+ LogicalAggregate<? extends Plan> logicalAgg, ConnectContext connectContext) {
+ RequireProperties requireGather = RequireProperties.of(PhysicalProperties.GATHER);
Review Comment:
gather require could be a static member of RequireProperties
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] 924060929 commented on a diff in pull request #14827: [Refactor](Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
924060929 commented on code in PR #14827:
URL: https://github.com/apache/doris/pull/14827#discussion_r1050639594
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rewrite/rules/FoldConstantRuleOnFE.java:
##########
@@ -59,143 +61,148 @@ public class FoldConstantRuleOnFE extends AbstractExpressionRewriteRule {
@Override
public Expression rewrite(Expression expr, ExpressionRewriteContext ctx) {
- return process(expr, ctx);
- }
+ if (expr instanceof AggregateFunction && ((AggregateFunction) expr).isDistinct()) {
+ return expr;
+ } else if (expr instanceof AggregateExpression && ((AggregateExpression) expr).getFunction().isDistinct()) {
+ return expr;
Review Comment:
maybe erase some distinct columns, that will be some bug
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #14827: [Refactor](Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #14827:
URL: https://github.com/apache/doris/pull/14827#issuecomment-1356802549
PR approved by anyone and no changes requested.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] 924060929 commented on a diff in pull request #14827: [Refactor](Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
924060929 commented on code in PR #14827:
URL: https://github.com/apache/doris/pull/14827#discussion_r1050649218
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/AggregateStrategies.java:
##########
@@ -0,0 +1,1234 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.annotation.DependsRules;
+import org.apache.doris.nereids.pattern.PatternDescriptor;
+import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.properties.RequireProperties;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.expression.rewrite.ExpressionRewriteContext;
+import org.apache.doris.nereids.rules.expression.rewrite.rules.FoldConstantRuleOnFE;
+import org.apache.doris.nereids.rules.expression.rewrite.rules.TypeCoercion;
+import org.apache.doris.nereids.rules.implementation.ImplementationRuleFactory;
+import org.apache.doris.nereids.rules.implementation.LogicalOlapScanToPhysicalOlapScan;
+import org.apache.doris.nereids.rules.rewrite.logical.NormalizeAggregate;
+import org.apache.doris.nereids.trees.expressions.AggregateExpression;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.IsNull;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.MultiDistinctCount;
+import org.apache.doris.nereids.trees.expressions.functions.agg.MultiDistinctSum;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.AggMode;
+import org.apache.doris.nereids.trees.plans.AggPhase;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.algebra.Project;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate.PushDownAggOp;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+
+/** AggregateStrategies */
+@DependsRules({
+ NormalizeAggregate.class,
+ FoldConstantRuleOnFE.class
+})
+public class AggregateStrategies implements ImplementationRuleFactory {
+
+ @Override
+ public List<Rule> buildRules() {
+ PatternDescriptor<LogicalAggregate<GroupPlan>> basePattern = logicalAggregate()
+ .when(LogicalAggregate::isNormalized);
+
+ return ImmutableList.of(
+ RuleType.STORAGE_LAYER_AGGREGATE_WITHOUT_PROJECT.build(
+ logicalAggregate(
+ logicalOlapScan()
+ )
+ .when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
+ .thenApply(ctx -> storageLayerAggregate(ctx.root, null, ctx.root.child(), ctx.cascadesContext))
+ ),
+ RuleType.STORAGE_LAYER_AGGREGATE_WITH_PROJECT.build(
+ logicalAggregate(
+ logicalProject(
+ logicalOlapScan()
+ )
+ )
+ .when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
+ .thenApply(ctx -> {
+ LogicalAggregate<LogicalProject<LogicalOlapScan>> agg = ctx.root;
+ LogicalProject<LogicalOlapScan> project = agg.child();
+ LogicalOlapScan olapScan = project.child();
+ return storageLayerAggregate(agg, project, olapScan, ctx.cascadesContext);
+ })
+ ),
+ RuleType.ONE_PHASE_AGGREGATE_WITHOUT_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 0)
+ .thenApplyMulti(ctx -> onePhaseAggregateWithoutDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITHOUT_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 0)
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithoutDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITH_COUNT_DISTINCT_MULTI.build(
+ basePattern
+ .when(this::containsCountDistinctMultiExpr)
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithCountDistinctMulti(ctx.root, ctx.connectContext))
+ ),
+ RuleType.THREE_PHASE_AGGREGATE_WITH_COUNT_DISTINCT_MULTI.build(
+ basePattern
+ .when(this::containsCountDistinctMultiExpr)
+ .thenApplyMulti(ctx -> threePhaseAggregateWithCountDistinctMulti(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITH_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1)
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.ONE_PHASE_AGGREGATE_SINGLE_DISTINCT_TO_MULTI.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1 && enableSingleDistinctColumnOpt())
+ .thenApplyMulti(ctx -> onePhaseAggregateWithMultiDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_SINGLE_DISTINCT_TO_MULTI.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1 && enableSingleDistinctColumnOpt())
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithMultiDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.THREE_PHASE_AGGREGATE_WITH_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1)
+ .thenApplyMulti(ctx -> threePhaseAggregateWithDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITH_MULTI_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() > 1 && !containsCountDistinctMultiExpr(agg))
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithMultiDistinct(ctx.root, ctx.connectContext))
+ )
+ );
+ }
+
+ /**
+ * sql: select count(*) from tbl
+ *
+ * before:
+ *
+ * LogicalAggregate(groupBy=[], output=[count(*)])
+ * |
+ * LogicalOlapScan(table=tbl)
+ *
+ * after:
+ *
+ * LogicalAggregate(groupBy=[], output=[count(*)])
+ * |
+ * PhysicalStorageLayerAggregate(pushAggOp=COUNT, table=PhysicalOlapScan(table=tbl))
+ *
+ */
+ private LogicalAggregate<? extends Plan> storageLayerAggregate(
+ LogicalAggregate<? extends Plan> aggregate,
+ @Nullable LogicalProject<? extends Plan> project,
+ LogicalOlapScan olapScan, CascadesContext cascadesContext) {
+ final LogicalAggregate<? extends Plan> canNotPush = aggregate;
+
+ KeysType keysType = olapScan.getTable().getKeysType();
+ if (keysType != KeysType.AGG_KEYS && keysType != KeysType.DUP_KEYS) {
+ return canNotPush;
+ }
+
+ List<Expression> groupByExpressions = aggregate.getGroupByExpressions();
+ if (!groupByExpressions.isEmpty() || !aggregate.getDistinctArguments().isEmpty()) {
+ return canNotPush;
+ }
+
+ Set<AggregateFunction> aggregateFunctions = aggregate.getAggregateFunctions();
+ Set<Class<? extends AggregateFunction>> functionClasses = aggregateFunctions
+ .stream()
+ .map(AggregateFunction::getClass)
+ .collect(Collectors.toSet());
+
+ Map<Class, PushDownAggOp> supportedAgg = PushDownAggOp.supportedFunctions();
+ if (!supportedAgg.keySet().containsAll(functionClasses)) {
+ return canNotPush;
+ }
+ if (functionClasses.contains(Count.class) && keysType != KeysType.DUP_KEYS) {
+ return canNotPush;
+ }
+ if (aggregateFunctions.stream().anyMatch(fun -> fun.arity() > 1)) {
+ return canNotPush;
+ }
+
+ // we already normalize the arguments to slotReference
+ List<Expression> argumentsOfAggregateFunction = aggregateFunctions.stream()
+ .flatMap(aggregateFunction -> aggregateFunction.getArguments().stream())
+ .collect(ImmutableList.toImmutableList());
+
+ if (project != null) {
+ argumentsOfAggregateFunction = Project.findProject(
+ (List<SlotReference>) (List) argumentsOfAggregateFunction, project.getProjects())
+ .stream()
+ .map(p -> p instanceof Alias ? p.child(0) : p)
+ .collect(ImmutableList.toImmutableList());
+ }
+
+ boolean onlyContainsSlotOrNumericCastSlot = argumentsOfAggregateFunction
+ .stream()
+ .allMatch(argument -> {
+ if (argument instanceof SlotReference) {
+ return true;
+ }
+ if (argument instanceof Cast) {
+ return argument.child(0) instanceof SlotReference
+ && argument.getDataType().isNumericType()
+ && argument.child(0).getDataType().isNumericType();
+ }
+ return false;
+ });
+ if (!onlyContainsSlotOrNumericCastSlot) {
+ return canNotPush;
+ }
+
+ Set<PushDownAggOp> pushDownAggOps = functionClasses.stream()
+ .map(supportedAgg::get)
+ .collect(Collectors.toSet());
+
+ PushDownAggOp mergeOp = pushDownAggOps.size() == 1
+ ? pushDownAggOps.iterator().next()
+ : PushDownAggOp.MIX;
+
+ Set<SlotReference> aggUsedSlots =
+ ExpressionUtils.collect(argumentsOfAggregateFunction, SlotReference.class::isInstance);
+
+ List<SlotReference> usedSlotInTable = (List<SlotReference>) (List) Project.findProject(aggUsedSlots,
+ (List<NamedExpression>) (List) olapScan.getOutput());
+
+ for (SlotReference slot : usedSlotInTable) {
+ Column column = slot.getColumn().get();
+ if (keysType == KeysType.AGG_KEYS && !column.isKey()) {
+ return canNotPush;
+ }
+ // The zone map max length of CharFamily is 512, do not
+ // over the length: https://github.com/apache/doris/pull/6293
+ if (mergeOp == PushDownAggOp.MIN_MAX || mergeOp == PushDownAggOp.MIX) {
+ PrimitiveType colType = column.getType().getPrimitiveType();
+ if (colType.isArrayType() || colType.isComplexType() || colType == PrimitiveType.STRING) {
+ return canNotPush;
+ }
+ if (colType.isCharFamily() && mergeOp != PushDownAggOp.COUNT && column.getType().getLength() > 512) {
+ return canNotPush;
+ }
+ }
+ if (mergeOp == PushDownAggOp.COUNT || mergeOp == PushDownAggOp.MIX) {
+ // NULL value behavior in `count` function is zero, so
+ // we should not use row_count to speed up query. the col
+ // must be not null
+ if (column.isAllowNull()) {
+ return canNotPush;
+ }
+ }
+ }
+
+ PhysicalOlapScan physicalOlapScan = (PhysicalOlapScan) new LogicalOlapScanToPhysicalOlapScan()
+ .build()
+ .transform(olapScan, cascadesContext)
+ .get(0);
+
+ return aggregate.withChildren(ImmutableList.of(
+ new PhysicalStorageLayerAggregate(physicalOlapScan, mergeOp)
+ ));
+ }
+
+ /**
+ * sql: select count(*) from tbl group by id
+ *
+ * before:
+ *
+ * LogicalAggregate(groupBy=[id], output=[count(*)])
+ * |
+ * LogicalOlapScan(table=tbl)
+ *
+ * after:
+ *
+ * single node aggregate:
+ *
+ * PhysicalHashAggregate(groupBy=[id], output=[count(*)])
+ * |
+ * PhysicalDistribute(distributionSpec=GATHER)
+ * |
+ * LogicalOlapScan(table=tbl)
+ *
+ * distribute node aggregate:
+ *
+ * PhysicalHashAggregate(groupBy=[id], output=[count(*)])
+ * |
+ * LogicalOlapScan(table=tbl, **already distribute by id**)
+ *
+ */
+ private List<PhysicalHashAggregate<Plan>> onePhaseAggregateWithoutDistinct(
+ LogicalAggregate<? extends Plan> logicalAgg, ConnectContext connectContext) {
+ RequireProperties requireGather = RequireProperties.of(PhysicalProperties.GATHER);
Review Comment:
RequireProperties has a `List<PhysicalProperties> properties` match to every require physical properties of children.
so `RequireProperties.GATHER` only use in the one child and not very general
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] 924060929 commented on a diff in pull request #14827: [Refactor](Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
924060929 commented on code in PR #14827:
URL: https://github.com/apache/doris/pull/14827#discussion_r1050649218
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/AggregateStrategies.java:
##########
@@ -0,0 +1,1234 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.annotation.DependsRules;
+import org.apache.doris.nereids.pattern.PatternDescriptor;
+import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.properties.RequireProperties;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.expression.rewrite.ExpressionRewriteContext;
+import org.apache.doris.nereids.rules.expression.rewrite.rules.FoldConstantRuleOnFE;
+import org.apache.doris.nereids.rules.expression.rewrite.rules.TypeCoercion;
+import org.apache.doris.nereids.rules.implementation.ImplementationRuleFactory;
+import org.apache.doris.nereids.rules.implementation.LogicalOlapScanToPhysicalOlapScan;
+import org.apache.doris.nereids.rules.rewrite.logical.NormalizeAggregate;
+import org.apache.doris.nereids.trees.expressions.AggregateExpression;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.IsNull;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.MultiDistinctCount;
+import org.apache.doris.nereids.trees.expressions.functions.agg.MultiDistinctSum;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.AggMode;
+import org.apache.doris.nereids.trees.plans.AggPhase;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.algebra.Project;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate.PushDownAggOp;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+
+/** AggregateStrategies */
+@DependsRules({
+ NormalizeAggregate.class,
+ FoldConstantRuleOnFE.class
+})
+public class AggregateStrategies implements ImplementationRuleFactory {
+
+ @Override
+ public List<Rule> buildRules() {
+ PatternDescriptor<LogicalAggregate<GroupPlan>> basePattern = logicalAggregate()
+ .when(LogicalAggregate::isNormalized);
+
+ return ImmutableList.of(
+ RuleType.STORAGE_LAYER_AGGREGATE_WITHOUT_PROJECT.build(
+ logicalAggregate(
+ logicalOlapScan()
+ )
+ .when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
+ .thenApply(ctx -> storageLayerAggregate(ctx.root, null, ctx.root.child(), ctx.cascadesContext))
+ ),
+ RuleType.STORAGE_LAYER_AGGREGATE_WITH_PROJECT.build(
+ logicalAggregate(
+ logicalProject(
+ logicalOlapScan()
+ )
+ )
+ .when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
+ .thenApply(ctx -> {
+ LogicalAggregate<LogicalProject<LogicalOlapScan>> agg = ctx.root;
+ LogicalProject<LogicalOlapScan> project = agg.child();
+ LogicalOlapScan olapScan = project.child();
+ return storageLayerAggregate(agg, project, olapScan, ctx.cascadesContext);
+ })
+ ),
+ RuleType.ONE_PHASE_AGGREGATE_WITHOUT_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 0)
+ .thenApplyMulti(ctx -> onePhaseAggregateWithoutDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITHOUT_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 0)
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithoutDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITH_COUNT_DISTINCT_MULTI.build(
+ basePattern
+ .when(this::containsCountDistinctMultiExpr)
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithCountDistinctMulti(ctx.root, ctx.connectContext))
+ ),
+ RuleType.THREE_PHASE_AGGREGATE_WITH_COUNT_DISTINCT_MULTI.build(
+ basePattern
+ .when(this::containsCountDistinctMultiExpr)
+ .thenApplyMulti(ctx -> threePhaseAggregateWithCountDistinctMulti(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITH_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1)
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.ONE_PHASE_AGGREGATE_SINGLE_DISTINCT_TO_MULTI.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1 && enableSingleDistinctColumnOpt())
+ .thenApplyMulti(ctx -> onePhaseAggregateWithMultiDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_SINGLE_DISTINCT_TO_MULTI.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1 && enableSingleDistinctColumnOpt())
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithMultiDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.THREE_PHASE_AGGREGATE_WITH_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() == 1)
+ .thenApplyMulti(ctx -> threePhaseAggregateWithDistinct(ctx.root, ctx.connectContext))
+ ),
+ RuleType.TWO_PHASE_AGGREGATE_WITH_MULTI_DISTINCT.build(
+ basePattern
+ .when(agg -> agg.getDistinctArguments().size() > 1 && !containsCountDistinctMultiExpr(agg))
+ .thenApplyMulti(ctx -> twoPhaseAggregateWithMultiDistinct(ctx.root, ctx.connectContext))
+ )
+ );
+ }
+
+ /**
+ * sql: select count(*) from tbl
+ *
+ * before:
+ *
+ * LogicalAggregate(groupBy=[], output=[count(*)])
+ * |
+ * LogicalOlapScan(table=tbl)
+ *
+ * after:
+ *
+ * LogicalAggregate(groupBy=[], output=[count(*)])
+ * |
+ * PhysicalStorageLayerAggregate(pushAggOp=COUNT, table=PhysicalOlapScan(table=tbl))
+ *
+ */
+ private LogicalAggregate<? extends Plan> storageLayerAggregate(
+ LogicalAggregate<? extends Plan> aggregate,
+ @Nullable LogicalProject<? extends Plan> project,
+ LogicalOlapScan olapScan, CascadesContext cascadesContext) {
+ final LogicalAggregate<? extends Plan> canNotPush = aggregate;
+
+ KeysType keysType = olapScan.getTable().getKeysType();
+ if (keysType != KeysType.AGG_KEYS && keysType != KeysType.DUP_KEYS) {
+ return canNotPush;
+ }
+
+ List<Expression> groupByExpressions = aggregate.getGroupByExpressions();
+ if (!groupByExpressions.isEmpty() || !aggregate.getDistinctArguments().isEmpty()) {
+ return canNotPush;
+ }
+
+ Set<AggregateFunction> aggregateFunctions = aggregate.getAggregateFunctions();
+ Set<Class<? extends AggregateFunction>> functionClasses = aggregateFunctions
+ .stream()
+ .map(AggregateFunction::getClass)
+ .collect(Collectors.toSet());
+
+ Map<Class, PushDownAggOp> supportedAgg = PushDownAggOp.supportedFunctions();
+ if (!supportedAgg.keySet().containsAll(functionClasses)) {
+ return canNotPush;
+ }
+ if (functionClasses.contains(Count.class) && keysType != KeysType.DUP_KEYS) {
+ return canNotPush;
+ }
+ if (aggregateFunctions.stream().anyMatch(fun -> fun.arity() > 1)) {
+ return canNotPush;
+ }
+
+ // we already normalize the arguments to slotReference
+ List<Expression> argumentsOfAggregateFunction = aggregateFunctions.stream()
+ .flatMap(aggregateFunction -> aggregateFunction.getArguments().stream())
+ .collect(ImmutableList.toImmutableList());
+
+ if (project != null) {
+ argumentsOfAggregateFunction = Project.findProject(
+ (List<SlotReference>) (List) argumentsOfAggregateFunction, project.getProjects())
+ .stream()
+ .map(p -> p instanceof Alias ? p.child(0) : p)
+ .collect(ImmutableList.toImmutableList());
+ }
+
+ boolean onlyContainsSlotOrNumericCastSlot = argumentsOfAggregateFunction
+ .stream()
+ .allMatch(argument -> {
+ if (argument instanceof SlotReference) {
+ return true;
+ }
+ if (argument instanceof Cast) {
+ return argument.child(0) instanceof SlotReference
+ && argument.getDataType().isNumericType()
+ && argument.child(0).getDataType().isNumericType();
+ }
+ return false;
+ });
+ if (!onlyContainsSlotOrNumericCastSlot) {
+ return canNotPush;
+ }
+
+ Set<PushDownAggOp> pushDownAggOps = functionClasses.stream()
+ .map(supportedAgg::get)
+ .collect(Collectors.toSet());
+
+ PushDownAggOp mergeOp = pushDownAggOps.size() == 1
+ ? pushDownAggOps.iterator().next()
+ : PushDownAggOp.MIX;
+
+ Set<SlotReference> aggUsedSlots =
+ ExpressionUtils.collect(argumentsOfAggregateFunction, SlotReference.class::isInstance);
+
+ List<SlotReference> usedSlotInTable = (List<SlotReference>) (List) Project.findProject(aggUsedSlots,
+ (List<NamedExpression>) (List) olapScan.getOutput());
+
+ for (SlotReference slot : usedSlotInTable) {
+ Column column = slot.getColumn().get();
+ if (keysType == KeysType.AGG_KEYS && !column.isKey()) {
+ return canNotPush;
+ }
+ // The zone map max length of CharFamily is 512, do not
+ // over the length: https://github.com/apache/doris/pull/6293
+ if (mergeOp == PushDownAggOp.MIN_MAX || mergeOp == PushDownAggOp.MIX) {
+ PrimitiveType colType = column.getType().getPrimitiveType();
+ if (colType.isArrayType() || colType.isComplexType() || colType == PrimitiveType.STRING) {
+ return canNotPush;
+ }
+ if (colType.isCharFamily() && mergeOp != PushDownAggOp.COUNT && column.getType().getLength() > 512) {
+ return canNotPush;
+ }
+ }
+ if (mergeOp == PushDownAggOp.COUNT || mergeOp == PushDownAggOp.MIX) {
+ // NULL value behavior in `count` function is zero, so
+ // we should not use row_count to speed up query. the col
+ // must be not null
+ if (column.isAllowNull()) {
+ return canNotPush;
+ }
+ }
+ }
+
+ PhysicalOlapScan physicalOlapScan = (PhysicalOlapScan) new LogicalOlapScanToPhysicalOlapScan()
+ .build()
+ .transform(olapScan, cascadesContext)
+ .get(0);
+
+ return aggregate.withChildren(ImmutableList.of(
+ new PhysicalStorageLayerAggregate(physicalOlapScan, mergeOp)
+ ));
+ }
+
+ /**
+ * sql: select count(*) from tbl group by id
+ *
+ * before:
+ *
+ * LogicalAggregate(groupBy=[id], output=[count(*)])
+ * |
+ * LogicalOlapScan(table=tbl)
+ *
+ * after:
+ *
+ * single node aggregate:
+ *
+ * PhysicalHashAggregate(groupBy=[id], output=[count(*)])
+ * |
+ * PhysicalDistribute(distributionSpec=GATHER)
+ * |
+ * LogicalOlapScan(table=tbl)
+ *
+ * distribute node aggregate:
+ *
+ * PhysicalHashAggregate(groupBy=[id], output=[count(*)])
+ * |
+ * LogicalOlapScan(table=tbl, **already distribute by id**)
+ *
+ */
+ private List<PhysicalHashAggregate<Plan>> onePhaseAggregateWithoutDistinct(
+ LogicalAggregate<? extends Plan> logicalAgg, ConnectContext connectContext) {
+ RequireProperties requireGather = RequireProperties.of(PhysicalProperties.GATHER);
Review Comment:
RequireProperties has a `List<PhysicalProperties> properties` match to every require physical properties of children.
so `RequireProperties.GATHER` is only used for one child and not very general
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #14827: [Refactor](Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #14827:
URL: https://github.com/apache/doris/pull/14827#issuecomment-1356802536
PR approved by at least one committer and no changes requested.
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] 924060929 commented on a diff in pull request #14827: [Refactor](Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
924060929 commented on code in PR #14827:
URL: https://github.com/apache/doris/pull/14827#discussion_r1050638766
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/DistinctToGroupBy.java:
##########
@@ -0,0 +1,32 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.rewrite.OneRewriteRuleFactory;
+
+/** AggGroupByToDistinct */
+public class DistinctToGroupBy extends OneRewriteRuleFactory {
Review Comment:
deleted
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] morrySnow merged pull request #14827: [Refactor](Nerieds) Refactor aggregate function/plan/rules and support related cbo rules
Posted by GitBox <gi...@apache.org>.
morrySnow merged PR #14827:
URL: https://github.com/apache/doris/pull/14827
--
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-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org