[opt](Nereids) remove group by key eliminate rule option (apache#32238)

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java

@@ -328,7 +328,7 @@ public class Rewriter extends AbstractBatchJobExecutor {
                     custom(RuleType.ELIMINATE_UNNECESSARY_PROJECT, EliminateUnnecessaryProject::new)
             ),
 
-            // this rule should invoke after topic "Join pull up"
+            // this rule should be invoked after topic "Join pull up"
             topic("eliminate group by keys according to fd items",
                     topDown(new EliminateGroupByKey())
             ),

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/ColumnPruning.java

@@ -18,7 +18,6 @@
 package org.apache.doris.nereids.rules.rewrite;
 
 import org.apache.doris.nereids.jobs.JobContext;
-import org.apache.doris.nereids.rules.RuleType;
 import org.apache.doris.nereids.rules.rewrite.ColumnPruning.PruneContext;
 import org.apache.doris.nereids.trees.expressions.Alias;
 import org.apache.doris.nereids.trees.expressions.Expression;
@@ -40,7 +39,6 @@
 import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
 import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
 import org.apache.doris.nereids.util.ExpressionUtils;
-import org.apache.doris.qe.ConnectContext;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -207,16 +205,8 @@ public Plan visitLogicalRepeat(LogicalRepeat<? extends Plan> repeat, PruneContex
     private Plan pruneAggregate(Aggregate agg, PruneContext context) {
         // first try to prune group by and aggregate functions
         Aggregate prunedOutputAgg = pruneOutput(agg, agg.getOutputs(), agg::pruneOutputs, context);
-        Set<Integer> enableNereidsRules = ConnectContext.get().getSessionVariable().getEnableNereidsRules();
-        Aggregate fillUpAggr;
 
-        if (!enableNereidsRules.contains(RuleType.ELIMINATE_GROUP_BY_KEY.type())) {
-            fillUpAggr = fillUpGroupByToOutput(prunedOutputAgg)
-                    .map(fullOutput -> prunedOutputAgg.withAggOutput(fullOutput))
-                    .orElse(prunedOutputAgg);
-        } else {
-            fillUpAggr = fillUpGroupByAndOutput(prunedOutputAgg);
-        }
+        Aggregate fillUpAggr = fillUpGroupByAndOutput(prunedOutputAgg);
 
         return pruneChildren(fillUpAggr);
     }
@@ -229,23 +219,6 @@ private Plan skipPruneThisAndFirstLevelChildren(Plan plan) {
         return pruneChildren(plan, requireAllOutputOfChildren);
     }
 
-    private static Optional<List<NamedExpression>> fillUpGroupByToOutput(Aggregate prunedOutputAgg) {
-        List<Expression> groupBy = prunedOutputAgg.getGroupByExpressions();
-        List<NamedExpression> output = prunedOutputAgg.getOutputExpressions();
-
-        if (output.containsAll(groupBy)) {
-            return Optional.empty();
-        }
-
-        List<NamedExpression> aggFunctions = Lists.newArrayList(output);
-        aggFunctions.removeAll(groupBy);
-
-        return Optional.of(ImmutableList.<NamedExpression>builder()
-                .addAll((List) groupBy)
-                .addAll(aggFunctions)
-                .build());
-    }
-
     private static Aggregate fillUpGroupByAndOutput(Aggregate prunedOutputAgg) {
         List<Expression> groupBy = prunedOutputAgg.getGroupByExpressions();
         List<NamedExpression> output = prunedOutputAgg.getOutputExpressions();

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/EliminateGroupByKey.java

@@ -25,7 +25,6 @@
 import org.apache.doris.nereids.trees.expressions.SlotReference;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
-import org.apache.doris.qe.ConnectContext;
 
 import com.google.common.collect.ImmutableSet;
 
@@ -42,10 +41,6 @@ public class EliminateGroupByKey extends OneRewriteRuleFactory {
     @Override
     public Rule build() {
         return logicalAggregate(logicalProject()).then(agg -> {
-            Set<Integer> enableNereidsRules = ConnectContext.get().getSessionVariable().getEnableNereidsRules();
-            if (!enableNereidsRules.contains(RuleType.ELIMINATE_GROUP_BY_KEY.type())) {
-                return null;
-            }
             LogicalPlan childPlan = agg.child();
             List<FdItem> uniqueFdItems = new ArrayList<>();
             List<FdItem> nonUniqueFdItems = new ArrayList<>();

regression-test/data/nereids_rules_p0/eliminate_gby_key/eliminate_gby_key.out

@@ -0,0 +1,164 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !1 --
+cost = 12.09
+PhysicalResultSink[358] ( outputExprs=[t2_c1#10] )
++--PhysicalDistribute[355]@7 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[352]@7 ( stats=1, projects=[c1#13 AS `t2_c1`#10] )
+      +--PhysicalHashAggregate[349]@6 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18], outputExpr=[c1#13, c3#18], partitionExpr=Optional[[c1#13, c3#18]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18]], exprIdToEquivalenceSet={13=0, 18=1} ) Order: ([])], stats=1 )
+         +--PhysicalProject[346]@5 ( stats=1, projects=[c1#13, c3#18] )
+            +--PhysicalHashJoin[343]@4 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalProject[333]@1 ( stats=1, projects=[c1#13, c3#15] )
+               |  +--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[340]@3 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[337]@3 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@2 ( stats=1 )
+
+-- !2 --
+cost = 11.09
+PhysicalResultSink[334] ( outputExprs=[t2_c2#9] )
++--PhysicalDistribute[331]@6 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[328]@6 ( stats=1, projects=[t2_c2#19 AS `t2_c2`#9] )
+      +--PhysicalHashAggregate[325]@5 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18, t2_c2#19], outputExpr=[c1#13, c3#18, t2_c2#19], partitionExpr=Optional[[c1#13, c3#18, t2_c2#19]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18, 19], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18], [19]], exprIdToEquivalenceSet={13=0, 18=1, 19=2} ) Order: ([])], stats=1 )
+         +--PhysicalProject[322]@4 ( stats=1, projects=[substring(c2#14, 1, 3) AS `t2_c2`#19, c1#13, c3#18] )
+            +--PhysicalHashJoin[319]@3 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[316]@2 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[313]@2 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@1 ( stats=1 )
+
+-- !3 --
+cost = 12.09
+PhysicalResultSink[358] ( outputExprs=[c3#11] )
++--PhysicalDistribute[355]@7 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[352]@7 ( stats=1, projects=[c3#18 AS `c3`#11] )
+      +--PhysicalHashAggregate[349]@6 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18], outputExpr=[c1#13, c3#18], partitionExpr=Optional[[c1#13, c3#18]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18]], exprIdToEquivalenceSet={13=0, 18=1} ) Order: ([])], stats=1 )
+         +--PhysicalProject[346]@5 ( stats=1, projects=[c1#13, c3#18] )
+            +--PhysicalHashJoin[343]@4 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalProject[333]@1 ( stats=1, projects=[c1#13, c3#15] )
+               |  +--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[340]@3 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[337]@3 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@2 ( stats=1 )
+
+-- !4 --
+cost = 12.09
+PhysicalResultSink[358] ( outputExprs=[cnt#12] )
++--PhysicalDistribute[355]@7 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[352]@7 ( stats=1, projects=[cnt#20 AS `cnt`#12] )
+      +--PhysicalHashAggregate[349]@6 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18], outputExpr=[c1#13, c3#18, count(*) AS `cnt`#20], partitionExpr=Optional[[c1#13, c3#18]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18]], exprIdToEquivalenceSet={13=0, 18=1} ) Order: ([])], stats=1 )
+         +--PhysicalProject[346]@5 ( stats=1, projects=[c1#13, c3#18] )
+            +--PhysicalHashJoin[343]@4 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalProject[333]@1 ( stats=1, projects=[c1#13, c3#15] )
+               |  +--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[340]@3 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[337]@3 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@2 ( stats=1 )
+
+-- !5 --
+cost = 11.09
+PhysicalResultSink[334] ( outputExprs=[t2_c2#9, t2_c1#10] )
++--PhysicalDistribute[331]@6 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[328]@6 ( stats=1, projects=[t2_c2#19 AS `t2_c2`#9, c1#13 AS `t2_c1`#10] )
+      +--PhysicalHashAggregate[325]@5 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18, t2_c2#19], outputExpr=[c1#13, c3#18, t2_c2#19], partitionExpr=Optional[[c1#13, c3#18, t2_c2#19]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18, 19], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18], [19]], exprIdToEquivalenceSet={13=0, 18=1, 19=2} ) Order: ([])], stats=1 )
+         +--PhysicalProject[322]@4 ( stats=1, projects=[substring(c2#14, 1, 3) AS `t2_c2`#19, c1#13, c3#18] )
+            +--PhysicalHashJoin[319]@3 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[316]@2 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[313]@2 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@1 ( stats=1 )
+
+-- !6 --
+cost = 12.09
+PhysicalResultSink[358] ( outputExprs=[c3#11, t2_c1#10] )
++--PhysicalDistribute[355]@7 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[352]@7 ( stats=1, projects=[c3#18 AS `c3`#11, c1#13 AS `t2_c1`#10] )
+      +--PhysicalHashAggregate[349]@6 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18], outputExpr=[c1#13, c3#18], partitionExpr=Optional[[c1#13, c3#18]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18]], exprIdToEquivalenceSet={13=0, 18=1} ) Order: ([])], stats=1 )
+         +--PhysicalProject[346]@5 ( stats=1, projects=[c1#13, c3#18] )
+            +--PhysicalHashJoin[343]@4 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalProject[333]@1 ( stats=1, projects=[c1#13, c3#15] )
+               |  +--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[340]@3 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[337]@3 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@2 ( stats=1 )
+
+-- !7 --
+cost = 11.09
+PhysicalResultSink[334] ( outputExprs=[c3#11, t2_c2#9] )
++--PhysicalDistribute[331]@6 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[328]@6 ( stats=1, projects=[c3#18 AS `c3`#11, t2_c2#19 AS `t2_c2`#9] )
+      +--PhysicalHashAggregate[325]@5 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18, t2_c2#19], outputExpr=[c1#13, c3#18, t2_c2#19], partitionExpr=Optional[[c1#13, c3#18, t2_c2#19]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18, 19], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18], [19]], exprIdToEquivalenceSet={13=0, 18=1, 19=2} ) Order: ([])], stats=1 )
+         +--PhysicalProject[322]@4 ( stats=1, projects=[substring(c2#14, 1, 3) AS `t2_c2`#19, c1#13, c3#18] )
+            +--PhysicalHashJoin[319]@3 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[316]@2 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[313]@2 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@1 ( stats=1 )
+
+-- !8 --
+cost = 12.09
+PhysicalResultSink[358] ( outputExprs=[t2_c1#10, cnt#12] )
++--PhysicalDistribute[355]@7 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[352]@7 ( stats=1, projects=[c1#13 AS `t2_c1`#10, cnt#20 AS `cnt`#12] )
+      +--PhysicalHashAggregate[349]@6 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18], outputExpr=[c1#13, c3#18, count(*) AS `cnt`#20], partitionExpr=Optional[[c1#13, c3#18]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18]], exprIdToEquivalenceSet={13=0, 18=1} ) Order: ([])], stats=1 )
+         +--PhysicalProject[346]@5 ( stats=1, projects=[c1#13, c3#18] )
+            +--PhysicalHashJoin[343]@4 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalProject[333]@1 ( stats=1, projects=[c1#13, c3#15] )
+               |  +--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[340]@3 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[337]@3 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@2 ( stats=1 )
+
+-- !9 --
+cost = 12.09
+PhysicalResultSink[358] ( outputExprs=[c3#11, cnt#12] )
++--PhysicalDistribute[355]@7 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[352]@7 ( stats=1, projects=[c3#18 AS `c3`#11, cnt#20 AS `cnt`#12] )
+      +--PhysicalHashAggregate[349]@6 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18], outputExpr=[c1#13, c3#18, count(*) AS `cnt`#20], partitionExpr=Optional[[c1#13, c3#18]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18]], exprIdToEquivalenceSet={13=0, 18=1} ) Order: ([])], stats=1 )
+         +--PhysicalProject[346]@5 ( stats=1, projects=[c1#13, c3#18] )
+            +--PhysicalHashJoin[343]@4 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalProject[333]@1 ( stats=1, projects=[c1#13, c3#15] )
+               |  +--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[340]@3 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[337]@3 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@2 ( stats=1 )
+
+-- !10 --
+cost = 12.09
+PhysicalResultSink[358] ( outputExprs=[t2_c1#10, c3#11, cnt#12] )
++--PhysicalDistribute[355]@7 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[352]@7 ( stats=1, projects=[c1#13 AS `t2_c1`#10, c3#18 AS `c3`#11, cnt#20 AS `cnt`#12] )
+      +--PhysicalHashAggregate[349]@6 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18], outputExpr=[c1#13, c3#18, count(*) AS `cnt`#20], partitionExpr=Optional[[c1#13, c3#18]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18]], exprIdToEquivalenceSet={13=0, 18=1} ) Order: ([])], stats=1 )
+         +--PhysicalProject[346]@5 ( stats=1, projects=[c1#13, c3#18] )
+            +--PhysicalHashJoin[343]@4 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalProject[333]@1 ( stats=1, projects=[c1#13, c3#15] )
+               |  +--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[340]@3 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[337]@3 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@2 ( stats=1 )
+
+-- !11 --
+cost = 11.09
+PhysicalResultSink[332] ( outputExprs=[t2_c2#9, c3#11, t2_c1#10] )
++--PhysicalDistribute[329]@6 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[326]@6 ( stats=1, projects=[t2_c2#19 AS `t2_c2`#9, c3#18 AS `c3`#11, c1#13 AS `t2_c1`#10] )
+      +--PhysicalHashAggregate[323]@5 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18, t2_c2#19], outputExpr=[c1#13, c3#18, t2_c2#19], partitionExpr=Optional[[c1#13, c3#18, t2_c2#19]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18, 19], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18], [19]], exprIdToEquivalenceSet={13=0, 18=1, 19=2} ) Order: ([])], stats=1 )
+         +--PhysicalProject[320]@4 ( stats=1, projects=[substring(c2#14, 1, 3) AS `t2_c2`#19, c1#13, c3#18] )
+            +--PhysicalHashJoin[317]@3 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[314]@2 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[311]@2 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@1 ( stats=1 )
+
+-- !12 --
+cost = 11.09
+PhysicalResultSink[330] ( outputExprs=[t2_c2#9, c3#11, t2_c1#10, cnt#12] )
++--PhysicalDistribute[327]@6 ( stats=1, distributionSpec=DistributionSpecGather )
+   +--PhysicalProject[324]@6 ( stats=1, projects=[t2_c2#19 AS `t2_c2`#9, c3#18 AS `c3`#11, c1#13 AS `t2_c1`#10, cnt#20 AS `cnt`#12] )
+      +--PhysicalHashAggregate[321]@5 ( aggPhase=LOCAL, aggMode=INPUT_TO_RESULT, maybeUseStreaming=false, groupByExpr=[c1#13, c3#18, t2_c2#19], outputExpr=[c1#13, c3#18, t2_c2#19, count(*) AS `cnt`#20], partitionExpr=Optional[[c1#13, c3#18, t2_c2#19]], requireProperties=[DistributionSpecHash ( orderedShuffledColumns=[13, 18, 19], shuffleType=REQUIRE, tableId=-1, selectedIndexId=-1, partitionIds=[], equivalenceExprIds=[[13], [18], [19]], exprIdToEquivalenceSet={13=0, 18=1, 19=2} ) Order: ([])], stats=1 )
+         +--PhysicalProject[318]@4 ( stats=1, projects=[substring(c2#14, 1, 3) AS `t2_c2`#19, c1#13, c3#18] )
+            +--PhysicalHashJoin[315]@3 ( type=INNER_JOIN, stats=1, hashCondition=[(c3#15 = c2#17)], otherCondition=[], markCondition=[] )
+               |--PhysicalOlapScan[t2]@0 ( stats=1 )
+               +--PhysicalDistribute[312]@2 ( stats=1, distributionSpec=DistributionSpecReplicated )
+                  +--PhysicalProject[309]@2 ( stats=1, projects=[c2#17, c3#18] )
+                     +--PhysicalOlapScan[t1]@1 ( stats=1 )
+