support count pushdown. tpcds 14/ tpch13

ds14 增加了agg push，执行时间 4.7 -> 4.8
h13 增加了 agg push，应该让p6 恢复到 p4 的成绩，提高一些

当任一 group key的ndv 接近 行数（0.9 倍）时，不下推agg

DORIS-24367 case-when 不能下推join 补null的一侧

pick unnest-subquery-cte

ut-tmp

adjust rt

update-shape

fix eager_agg.groovy, runtime_filter_mode=OFF;

fmt

14/67 因为rebase后增加了 repeat 拆分, 形状变化

支持  min(if), max(if), 增加了context.isValid检查,避免无效下推

doris-24240: rewriteRoot 检查nullable失败则不做eagerAgg

column pruning 不产生不合法 的setOp

24207-2: orExpansion union 字段没对齐

DORIS-24239 context.groupKeys 不能为空

DORIS-24206: fix EliminateGroupByKeyByUniform bug:没有替换alias的exprId

DORIS-24205
1. union 的孩子不能部分改写
2. agg 输入字段和 group key 有交集,则不下推

LogicalProject 构造projectMap时不能有unbound

DORIS-23842 没有aggFunc时 下推包含所有group key
的分支，而不是大分支. ds37/38/82/87 受到影响。select distinct A from T1 join T2 on ... group by A`

aliasMap 使用HashMap，不用IdentityMap

DORIS-24149

DORIS-24151

doris-24150 rt case

1. exprId 的等值判断, 2.update rt.

DORIS-24150

update shape

remove unused code

1. sum-if 不考虑穿过bigJoin， 2. 支持union

q5 两个sum（0）错误去重了

sum-if 基本款 (还没有支持union), 43 有提升

simple sum-if no union

检查context的字段 是project的输出.拒绝 sum(A) 下推 proj(x, x+y as A) 且x 不是group key

derive deep false

throw exception for eager agg when FeDebug

1. remove finalGroupKeys, 2. project 下推后改写projects

push agg on join

group key only slotreference

do not support avg/count

mode=1 时 即使没有经过big join 也要 强制 下推

shape with/without pkfk based on tpc_preview

Author: englefly

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

@@ -17,8 +17,6 @@
 
 package org.apache.doris.nereids.rules.rewrite;
 
-import org.apache.doris.common.util.DebugUtil;
-import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.jobs.JobContext;
 import org.apache.doris.nereids.properties.OrderKey;
 import org.apache.doris.nereids.trees.expressions.Alias;
@@ -52,15 +50,14 @@
 import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
 import org.apache.doris.nereids.util.ExpressionUtils;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.LinkedHashMultimap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Multimap;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -74,25 +71,10 @@
  * So, we need add a rule to adjust all expression's nullable attribute after rewrite.
  */
 public class AdjustNullable extends DefaultPlanRewriter<Map<ExprId, Slot>> implements CustomRewriter {
-
-    private static final Logger LOG = LogManager.getLogger(AdjustNullable.class);
-
     private final boolean isAnalyzedPhase;
 
-    /**
-     * When check is true, if we find a slot that is non-nullable in the plan,
-     * but we infer it should be nullable from the plan's subtree, and fe_debug is true,
-     * then throw an exception.
-     */
-    private final boolean check;
-
-    public AdjustNullable(boolean isAnalyzedPhase, boolean check) {
-        this.isAnalyzedPhase = isAnalyzedPhase;
-        this.check = check;
-    }
-
     public AdjustNullable(boolean isAnalyzedPhase) {
-        this(isAnalyzedPhase, !isAnalyzedPhase);
+        this.isAnalyzedPhase = isAnalyzedPhase;
     }
 
     @Override
@@ -460,7 +442,7 @@ private <T extends Expression> Optional<T> updateExpression(Optional<T> input,
     private <T extends Expression> Optional<T> updateExpression(T input,
             Map<ExprId, Slot> replaceMap, boolean debugCheck) {
         AtomicBoolean changed = new AtomicBoolean(false);
-        Expression replaced = doUpdateExpression(changed, input, replaceMap, check && debugCheck);
+        Expression replaced = doUpdateExpression(changed, input, replaceMap, !isAnalyzedPhase && debugCheck);
         return changed.get() ? Optional.of((T) replaced) : Optional.empty();
     }
 
@@ -497,14 +479,9 @@ private static Expression doUpdateExpression(AtomicBoolean changed, Expression i
                 // repeat may check fail.
                 if (!slotReference.nullable() && newSlotReference.nullable()
                         && check && ConnectContext.get() != null) {
-                    if (ConnectContext.get().getSessionVariable().feDebug) {
-                        throw new AnalysisException("AdjustNullable convert slot " + slotReference
-                                + " from not-nullable to nullable. You can disable check by set fe_debug = false.");
-                    } else {
-                        LOG.warn("adjust nullable convert slot '" + slotReference
-                                + "' from not-nullable to nullable for query "
-                                + DebugUtil.printId(ConnectContext.get().queryId()));
-                    }
+                    SessionVariable.throwAnalysisExceptionWhenFeDebug("AdjustNullable convert slot "
+                            + slotReference
+                            + " from not-nullable to nullable. You can disable check by set fe_debug = false.");
                 }
                 return newSlotReference;
             } else {

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

@@ -59,7 +59,6 @@
 import org.apache.doris.qe.ConnectContext;
 
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableList.Builder;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import org.roaringbitmap.RoaringBitmap;
@@ -69,7 +68,6 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Function;
-import java.util.stream.IntStream;
 
 /**
  * ColumnPruning.
@@ -221,28 +219,21 @@ public Plan visitLogicalUnion(LogicalUnion union, PruneContext context) {
         }
         LogicalUnion prunedOutputUnion = pruneUnionOutput(union, context);
         // start prune children of union
-        List<Slot> originOutput = union.getOutput();
-        Set<Slot> prunedOutput = prunedOutputUnion.getOutputSet();
-        List<Integer> prunedOutputIndexes = IntStream.range(0, originOutput.size())
-                .filter(index -> prunedOutput.contains(originOutput.get(index)))
-                .boxed()
-                .collect(ImmutableList.toImmutableList());
-
         ImmutableList.Builder<Plan> prunedChildren = ImmutableList.builder();
         ImmutableList.Builder<List<SlotReference>> prunedChildrenOutputs = ImmutableList.builder();
         for (int i = 0; i < prunedOutputUnion.arity(); i++) {
             List<SlotReference> regularChildOutputs = prunedOutputUnion.getRegularChildOutput(i);
 
             RoaringBitmap prunedChildOutputExprIds = new RoaringBitmap();
-            Builder<SlotReference> prunedChildOutputBuilder
-                    = ImmutableList.builderWithExpectedSize(regularChildOutputs.size());
-            for (Integer index : prunedOutputIndexes) {
-                SlotReference slot = regularChildOutputs.get(index);
-                prunedChildOutputBuilder.add(slot);
-                prunedChildOutputExprIds.add(slot.getExprId().asInt());
-            }
-
-            List<SlotReference> prunedChildOutput = prunedChildOutputBuilder.build();
+            //Builder<SlotReference> prunedChildOutputBuilder
+            //        = ImmutableList.builderWithExpectedSize(regularChildOutputs.size());
+            //for (Integer index : prunedOutputIndexes) {
+            //    SlotReference slot = regularChildOutputs.get(index);
+            //    prunedChildOutputBuilder.add(slot);
+            //    prunedChildOutputExprIds.add(slot.getExprId().asInt());
+            //}
+            regularChildOutputs.forEach(col -> prunedChildOutputExprIds.add(col.getExprId().asInt()));
+            List<SlotReference> prunedChildOutput = regularChildOutputs; //prunedChildOutputBuilder.build();
             Plan prunedChild = doPruneChild(
                     prunedOutputUnion, prunedOutputUnion.child(i), prunedChildOutputExprIds,
                     prunedChildOutput, true
@@ -420,15 +411,15 @@ private LogicalUnion pruneUnionOutput(LogicalUnion union, PruneContext context)
                 extractColumnIndex.add(i);
             }
         }
-
         ImmutableList.Builder<List<NamedExpression>> prunedConstantExprsList
                 = ImmutableList.builderWithExpectedSize(constantExprsList.size());
+        List<List<SlotReference>> prunedRegularChildrenOutputs =
+                Lists.newArrayListWithCapacity(regularChildrenOutputs.size());
         if (prunedOutputs.isEmpty()) {
             // process prune all columns
             NamedExpression originSlot = originOutput.get(0);
             prunedOutputs = ImmutableList.of(new SlotReference(originSlot.getExprId(), originSlot.getName(),
                     TinyIntType.INSTANCE, false, originSlot.getQualifier()));
-            regularChildrenOutputs = Lists.newArrayListWithCapacity(regularChildrenOutputs.size());
             children = Lists.newArrayListWithCapacity(children.size());
             for (int i = 0; i < union.getArity(); i++) {
                 Plan child = union.child(i);
@@ -442,28 +433,36 @@ private LogicalUnion pruneUnionOutput(LogicalUnion union, PruneContext context)
                 } else {
                     project = new LogicalProject<>(newProjectOutput, child);
                 }
-                regularChildrenOutputs.add((List) project.getOutput());
+                prunedRegularChildrenOutputs.add((List) project.getOutput());
                 children.add(project);
             }
             for (int i = 0; i < constantExprsList.size(); i++) {
                 prunedConstantExprsList.add(ImmutableList.of(new Alias(new TinyIntLiteral((byte) 1))));
             }
         } else {
-            int len = extractColumnIndex.size();
+            int prunedOutputSize = extractColumnIndex.size();
             for (List<NamedExpression> row : constantExprsList) {
-                ImmutableList.Builder<NamedExpression> newRow = ImmutableList.builderWithExpectedSize(len);
+                ImmutableList.Builder<NamedExpression> newRow = ImmutableList.builderWithExpectedSize(prunedOutputSize);
                 for (int idx : extractColumnIndex) {
                     newRow.add(row.get(idx));
                 }
                 prunedConstantExprsList.add(newRow.build());
             }
+
+            for (int childIdx = 0; childIdx < union.getRegularChildrenOutputs().size(); childIdx++) {
+                List<SlotReference> regular = Lists.newArrayListWithExpectedSize(prunedOutputSize);
+                for (int colIdx : extractColumnIndex) {
+                    regular.add(regularChildrenOutputs.get(childIdx).get(colIdx));
+                }
+                prunedRegularChildrenOutputs.add(regular);
+            }
         }
 
         if (prunedOutputs.equals(originOutput) && !context.requiredSlotsIds.isEmpty()) {
             return union;
         } else {
             return union.withNewOutputsChildrenAndConstExprsList(prunedOutputs, children,
-                    regularChildrenOutputs, prunedConstantExprsList.build());
+                    prunedRegularChildrenOutputs, prunedConstantExprsList.build());
         }
     }
 

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

@@ -176,10 +176,13 @@ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, O
         }
         //4. union all joins and put producers to context
         List<List<SlotReference>> childrenOutputs = joins.stream()
-                .map(j -> j.getOutput().stream()
+                .map(j -> j.getOutput().stream() //.map(j -> j.getOutput().stream().distinct()
                         .map(SlotReference.class::cast)
                         .collect(ImmutableList.toImmutableList()))
                 .collect(ImmutableList.toImmutableList());
+        //LogicalUnion union = new LogicalUnion(Qualifier.ALL,
+        //        new ArrayList<>(join.getOutput().stream().distinct().collect(Collectors.toList())),
+        //        childrenOutputs, ImmutableList.of(), false, joins);
         LogicalUnion union = new LogicalUnion(Qualifier.ALL, new ArrayList<>(join.getOutput()),
                 childrenOutputs, ImmutableList.of(), false, joins);
         ctx.cteProducerList.add(leftProducer);
@@ -319,8 +322,26 @@ private List<Plan> expandInnerJoin(CascadesContext ctx, Pair<List<Expression>,
 
             LogicalCTEConsumer left = new LogicalCTEConsumer(ctx.getStatementContext().getNextRelationId(),
                     leftProducer.getCteId(), "", leftProducer);
+            List<NamedExpression> leftOutput = new ArrayList<>();
+            for (Slot producerOutputSlot : leftProducer.getOutput()) {
+                for (Slot consumerSlot : left.getProducerToConsumerOutputMap().get(producerOutputSlot)) {
+                    if (!leftOutput.contains(consumerSlot)) {
+                        leftOutput.add(consumerSlot);
+                        break;
+                    }
+                }
+            }
             LogicalCTEConsumer right = new LogicalCTEConsumer(ctx.getStatementContext().getNextRelationId(),
                     rightProducer.getCteId(), "", rightProducer);
+            List<NamedExpression> rightOutput = new ArrayList<>();
+            for (Slot producerOutputSlot : rightProducer.getOutput()) {
+                for (Slot consumerSlot : right.getProducerToConsumerOutputMap().get(producerOutputSlot)) {
+                    if (!rightOutput.contains(consumerSlot)) {
+                        rightOutput.add(consumerSlot);
+                        break;
+                    }
+                }
+            }
             ctx.putCTEIdToConsumer(left);
             ctx.putCTEIdToConsumer(right);
 
@@ -335,7 +356,10 @@ private List<Plan> expandInnerJoin(CascadesContext ctx, Pair<List<Expression>,
 
             LogicalJoin<? extends Plan, ? extends Plan> newJoin = new LogicalJoin<>(
                     JoinType.INNER_JOIN, hashCond, otherCond, join.getDistributeHint(),
-                    join.getMarkJoinSlotReference(), left, right, null);
+                    join.getMarkJoinSlotReference(),
+                    new LogicalProject<>(leftOutput, left),
+                    new LogicalProject<>(rightOutput, right),
+                    null);
             if (newJoin.getHashJoinConjuncts().stream()
                     .anyMatch(equalTo -> equalTo.children().stream().anyMatch(e -> !(e instanceof Slot)))) {
                 Plan plan = PushDownExpressionsInHashCondition.pushDownHashExpression(newJoin);