[feature](partition prune) support remove predicates that are always true after partition prune in list partition (apache#57169)

Problem Summary:

After partition pruning on a list-partitioned table, the partitioning
key predicates used for pruning are often redundant within the remaining
partitions, leading to unnecessary column scan and computation.

Given a table:

CREATE TABLE sales (
    id INT,
    region VARCHAR(10)
) PARTITION BY LIST (region) (
    PARTITION p_north VALUES IN ('bj', 'tj'),
    PARTITION p_south VALUES IN ('gz', 'sz')
);

query:
SELECT * FROM sales WHERE region IN ('bj', 'tj');

Plan Before:

-> TableReader(Partition p_north)
    -> TableScan(Filter: region in ('bj', 'tj'))

Plan After:

-> TableReader(Partition p_north)
    -> TableScan() // Filter is removed

Author: feiniaofeiafei

fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java

@@ -274,6 +274,7 @@ public enum TableFrom {
     private final Set<List<String>> materializationRewrittenSuccessSet = new HashSet<>();
 
     private boolean isInsert = false;
+    private boolean skipPrunePredicate = false;
 
     public StatementContext() {
         this(ConnectContext.get(), null, 0);
@@ -994,4 +995,11 @@ public void setIsInsert(boolean isInsert) {
     public boolean isInsert() {
         return isInsert;
     }
+
+    public boolean isSkipPrunePredicate() {
+        return skipPrunePredicate;
+    }
+    public void setSkipPrunePredicate(boolean skipPrunePredicate) {
+        this.skipPrunePredicate = skipPrunePredicate;
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadPlanInfoCollector.java

@@ -49,6 +49,7 @@
 import org.apache.doris.nereids.rules.analysis.ExpressionAnalyzer;
 import org.apache.doris.nereids.rules.expression.rules.ConvertAggStateCast;
 import org.apache.doris.nereids.rules.expression.rules.PartitionPruner;
+import org.apache.doris.nereids.rules.expression.rules.PartitionPruner.PartitionTableType;
 import org.apache.doris.nereids.rules.expression.rules.SortedPartitionRanges;
 import org.apache.doris.nereids.trees.expressions.Alias;
 import org.apache.doris.nereids.trees.expressions.Cast;
@@ -499,7 +500,7 @@ private List<Long> getAllPartitionIds() throws DdlException, AnalysisException {
                 List<Long> prunedPartitions = PartitionPruner.prune(
                         partitionSlots, filterPredicate, idToPartitions,
                         CascadesContext.initContext(new StatementContext(), logicalPlan, PhysicalProperties.ANY),
-                        PartitionPruner.PartitionTableType.OLAP, sortedPartitionRanges);
+                        PartitionTableType.OLAP, sortedPartitionRanges).first;
                 return prunedPartitions;
             } else {
                 return null;

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/PartitionPruner.java

@@ -20,6 +20,7 @@
 import org.apache.doris.catalog.ListPartitionItem;
 import org.apache.doris.catalog.PartitionItem;
 import org.apache.doris.catalog.RangePartitionItem;
+import org.apache.doris.common.Pair;
 import org.apache.doris.common.profile.SummaryProfile;
 import org.apache.doris.nereids.CascadesContext;
 import org.apache.doris.nereids.rules.expression.ExpressionRewriteContext;
@@ -35,7 +36,11 @@
 import org.apache.doris.nereids.trees.expressions.literal.DateTimeLiteral;
 import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
 import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
 import org.apache.doris.nereids.types.DateTimeType;
+import org.apache.doris.nereids.util.ExpressionUtils;
 import org.apache.doris.nereids.util.Utils;
 
 import com.google.common.collect.ImmutableList;
@@ -46,6 +51,7 @@
 import com.google.common.collect.RangeSet;
 import com.google.common.collect.Sets;
 
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -110,17 +116,21 @@ public Expression visitComparisonPredicate(ComparisonPredicate cp, Void context)
     }
 
     /** prune */
-    public <K extends Comparable<K>> List<K> prune() {
+    public <K extends Comparable<K>> Pair<List<K>, Boolean> prune() {
         Builder<K> scanPartitionIdents = ImmutableList.builder();
+        boolean canPredicatePruned = true;
         for (OnePartitionEvaluator partition : partitions) {
-            if (!canBePrunedOut(partitionPredicate, partition)) {
+            Pair<Boolean, Boolean> res = canBePrunedOut(partitionPredicate, partition);
+            if (!res.first) {
+                canPredicatePruned = canPredicatePruned && res.second;
                 scanPartitionIdents.add((K) partition.getPartitionIdent());
             }
         }
-        return scanPartitionIdents.build();
+        return Pair.of(scanPartitionIdents.build(), canPredicatePruned);
     }
 
-    public static <K extends Comparable<K>> List<K> prune(List<Slot> partitionSlots, Expression partitionPredicate,
+    public static <K extends Comparable<K>> Pair<List<K>, Optional<Expression>> prune(List<Slot> partitionSlots,
+            Expression partitionPredicate,
             Map<K, PartitionItem> idToPartitions, CascadesContext cascadesContext,
             PartitionTableType partitionTableType) {
         return prune(partitionSlots, partitionPredicate, idToPartitions,
@@ -130,7 +140,8 @@ public static <K extends Comparable<K>> List<K> prune(List<Slot> partitionSlots,
     /**
      * prune partition with `idToPartitions` as parameter.
      */
-    public static <K extends Comparable<K>> List<K> prune(List<Slot> partitionSlots, Expression partitionPredicate,
+    public static <K extends Comparable<K>> Pair<List<K>, Optional<Expression>> prune(List<Slot> partitionSlots,
+            Expression partitionPredicate,
             Map<K, PartitionItem> idToPartitions, CascadesContext cascadesContext,
             PartitionTableType partitionTableType, Optional<SortedPartitionRanges<K>> sortedPartitionRanges) {
         long startAt = System.currentTimeMillis();
@@ -146,40 +157,53 @@ public static <K extends Comparable<K>> List<K> prune(List<Slot> partitionSlots,
         }
     }
 
-    private static <K extends Comparable<K>> List<K> pruneInternal(List<Slot> partitionSlots,
+    private static <K extends Comparable<K>> Pair<List<K>, Optional<Expression>> pruneInternal(
+            List<Slot> partitionSlots,
             Expression partitionPredicate,
             Map<K, PartitionItem> idToPartitions, CascadesContext cascadesContext,
             PartitionTableType partitionTableType, Optional<SortedPartitionRanges<K>> sortedPartitionRanges) {
         partitionPredicate = PartitionPruneExpressionExtractor.extract(
                 partitionPredicate, ImmutableSet.copyOf(partitionSlots), cascadesContext);
+        Expression originalPartitionPredicate = partitionPredicate;
         partitionPredicate = PredicateRewriteForPartitionPrune.rewrite(partitionPredicate, cascadesContext);
-
         int expandThreshold = cascadesContext.getAndCacheSessionVariable(
                 "partitionPruningExpandThreshold",
                 10, sessionVariable -> sessionVariable.partitionPruningExpandThreshold);
 
         partitionPredicate = OrToIn.EXTRACT_MODE_INSTANCE.rewriteTree(
                 partitionPredicate, new ExpressionRewriteContext(cascadesContext));
         if (BooleanLiteral.TRUE.equals(partitionPredicate)) {
-            return Utils.fastToImmutableList(idToPartitions.keySet());
+            // The partition column predicate is always true and can be deleted, the partition cannot be pruned
+            return Pair.of(Utils.fastToImmutableList(idToPartitions.keySet()), Optional.of(originalPartitionPredicate));
         } else if (BooleanLiteral.FALSE.equals(partitionPredicate) || partitionPredicate.isNullLiteral()) {
-            return ImmutableList.of();
+            // The partition column predicate is always false, and all partitions can be pruned.
+            return Pair.of(ImmutableList.of(), Optional.empty());
         }
 
         if (sortedPartitionRanges.isPresent()) {
             RangeSet<MultiColumnBound> predicateRanges = partitionPredicate.accept(
                     new PartitionPredicateToRange(partitionSlots), null);
             if (predicateRanges != null) {
-                return binarySearchFiltering(
+                Pair<List<K>, Boolean> res = binarySearchFiltering(
                         sortedPartitionRanges.get(), partitionSlots, partitionPredicate, cascadesContext,
                         expandThreshold, predicateRanges
                 );
+                if (res.second) {
+                    return Pair.of(res.first, Optional.of(originalPartitionPredicate));
+                } else {
+                    return Pair.of(res.first, Optional.empty());
+                }
             }
         }
 
-        return sequentialFiltering(
+        Pair<List<K>, Boolean> res = sequentialFiltering(
                 idToPartitions, partitionSlots, partitionPredicate, cascadesContext, expandThreshold
         );
+        if (res.second) {
+            return Pair.of(res.first, Optional.of(originalPartitionPredicate));
+        } else {
+            return Pair.of(res.first, Optional.empty());
+        }
     }
 
     /**
@@ -198,14 +222,15 @@ public static <K> OnePartitionEvaluator<K> toPartitionEvaluator(K id, PartitionI
         }
     }
 
-    private static <K extends Comparable<K>> List<K> binarySearchFiltering(
+    private static <K extends Comparable<K>> Pair<List<K>, Boolean> binarySearchFiltering(
             SortedPartitionRanges<K> sortedPartitionRanges, List<Slot> partitionSlots,
             Expression partitionPredicate, CascadesContext cascadesContext, int expandThreshold,
             RangeSet<MultiColumnBound> predicateRanges) {
         List<PartitionItemAndRange<K>> sortedPartitions = sortedPartitionRanges.sortedPartitions;
 
         Set<K> selectedIdSets = Sets.newTreeSet();
         int leftIndex = 0;
+        boolean canPredicatePruned = true;
         for (Range<MultiColumnBound> predicateRange : predicateRanges.asRanges()) {
             int rightIndex = sortedPartitions.size();
             if (leftIndex >= rightIndex) {
@@ -246,8 +271,10 @@ private static <K extends Comparable<K>> List<K> binarySearchFiltering(
 
                 OnePartitionEvaluator<K> partitionEvaluator = toPartitionEvaluator(
                         partitionId, partition.partitionItem, partitionSlots, cascadesContext, expandThreshold);
-                if (!canBePrunedOut(partitionPredicate, partitionEvaluator)) {
+                Pair<Boolean, Boolean> res = canBePrunedOut(partitionPredicate, partitionEvaluator);
+                if (!res.first) {
                     selectedIdSets.add(partitionId);
+                    canPredicatePruned = canPredicatePruned && res.second;
                 }
             }
         }
@@ -256,15 +283,17 @@ private static <K extends Comparable<K>> List<K> binarySearchFiltering(
             K partitionId = defaultPartition.id;
             OnePartitionEvaluator<K> partitionEvaluator = toPartitionEvaluator(
                     partitionId, defaultPartition.partitionItem, partitionSlots, cascadesContext, expandThreshold);
-            if (!canBePrunedOut(partitionPredicate, partitionEvaluator)) {
+            Pair<Boolean, Boolean> res = canBePrunedOut(partitionPredicate, partitionEvaluator);
+            if (!res.first) {
                 selectedIdSets.add(partitionId);
+                canPredicatePruned = canPredicatePruned && res.second;
             }
         }
 
-        return Utils.fastToImmutableList(selectedIdSets);
+        return Pair.of(Utils.fastToImmutableList(selectedIdSets), canPredicatePruned);
     }
 
-    private static <K extends Comparable<K>> List<K> sequentialFiltering(
+    private static <K extends Comparable<K>> Pair<List<K>, Boolean> sequentialFiltering(
             Map<K, PartitionItem> idToPartitions, List<Slot> partitionSlots,
             Expression partitionPredicate, CascadesContext cascadesContext, int expandThreshold) {
         List<OnePartitionEvaluator<?>> evaluators = Lists.newArrayListWithCapacity(idToPartitions.size());
@@ -278,18 +307,70 @@ private static <K extends Comparable<K>> List<K> sequentialFiltering(
     }
 
     /**
-     * return true if partition is not qualified. that is, can be pruned out.
+     * return Pair
+     *     pair.first is true if partition can be pruned
+     *     pair.second is true if partitionPredicate is always true in this partition
      */
-    private static <K> boolean canBePrunedOut(Expression partitionPredicate, OnePartitionEvaluator<K> evaluator) {
+    private static <K> Pair<Boolean, Boolean> canBePrunedOut(Expression partitionPredicate,
+            OnePartitionEvaluator<K> evaluator) {
         List<Map<Slot, PartitionSlotInput>> onePartitionInputs = evaluator.getOnePartitionInputs();
-        for (Map<Slot, PartitionSlotInput> currentInputs : onePartitionInputs) {
-            // evaluate whether there's possible for this partition to accept this predicate
-            Expression result = evaluator.evaluateWithDefaultPartition(partitionPredicate, currentInputs);
-            if (!result.equals(BooleanLiteral.FALSE) && !(result instanceof NullLiteral)) {
-                return false;
+        if (evaluator instanceof OneListPartitionEvaluator) {
+            // if a table has default partition, the predicate should not be pruned,
+            // because evaluateWithDefaultPartition always return true in default partition
+            // e.g. PARTITION BY LIST(k1) (
+            //     PARTITION p1 VALUES IN ("1","2","3","4"),
+            //     PARTITION p2 VALUES IN ("5","6","7","8"),
+            //     PARTITION p3 )  p3 is default partition
+            boolean notDefaultPartition = !evaluator.isDefaultPartition();
+            Pair<Boolean, Boolean> res = Pair.of(notDefaultPartition, notDefaultPartition);
+            for (Map<Slot, PartitionSlotInput> currentInputs : onePartitionInputs) {
+                // evaluate whether there's possible for this partition to accept this predicate
+                Expression result = evaluator.evaluateWithDefaultPartition(partitionPredicate, currentInputs);
+                if (result.equals(BooleanLiteral.FALSE) || (result instanceof NullLiteral)) {
+                    // Indicates that there is a partition value that does not satisfy the predicate
+                    res.second = false;
+                } else if (result.equals(BooleanLiteral.TRUE)) {
+                    // Indicates that there is a partition value that satisfies the predicate
+                    res.first = false;
+                } else {
+                    // Indicates that this partition value may or may not satisfy the predicate
+                    res.second = false;
+                    res.first = false;
+                }
+                if (!res.first && !res.second) {
+                    break;
+                }
             }
+            return res;
+        } else {
+            // only prune partition predicates in list partition, therefore set pair.second always be false,
+            // meaning not to prune partition predicates in range partition
+            for (Map<Slot, PartitionSlotInput> currentInputs : onePartitionInputs) {
+                Expression result = evaluator.evaluateWithDefaultPartition(partitionPredicate, currentInputs);
+                if (!result.equals(BooleanLiteral.FALSE) && !(result instanceof NullLiteral)) {
+                    return Pair.of(false, false);
+                }
+            }
+            // only have false result: Can be pruned out. have other exprs: CanNot be pruned out
+            return Pair.of(true, false);
+        }
+    }
+
+    /** remove predicates that are always true*/
+    public static Plan prunePredicate(boolean skipPrunePredicate, Optional<Expression> prunedPredicates,
+            LogicalFilter<? extends Plan> filter, LogicalRelation scan) {
+        if (!skipPrunePredicate && prunedPredicates.isPresent()) {
+            Set<Expression> conjuncts = new LinkedHashSet<>(filter.getConjuncts());
+            Expression deletedPredicate = prunedPredicates.get();
+            Set<Expression> deletedPredicateSet = ExpressionUtils.extractConjunctionToSet(deletedPredicate);
+            conjuncts.removeAll(deletedPredicateSet);
+            if (conjuncts.isEmpty()) {
+                return scan;
+            } else {
+                return filter.withConjunctsAndChild(conjuncts, scan);
+            }
+        } else {
+            return filter.withChildren(ImmutableList.of(scan));
         }
-        // only have false result: Can be pruned out. have other exprs: CanNot be pruned out
-        return true;
     }
 }

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

@@ -20,6 +20,7 @@
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.PartitionItem;
 import org.apache.doris.catalog.SupportBinarySearchFilteringPartitions;
+import org.apache.doris.common.Pair;
 import org.apache.doris.common.cache.NereidsSortedPartitionsCacheManager;
 import org.apache.doris.datasource.ExternalTable;
 import org.apache.doris.nereids.CascadesContext;
@@ -28,6 +29,7 @@
 import org.apache.doris.nereids.rules.expression.rules.PartitionPruner;
 import org.apache.doris.nereids.rules.expression.rules.PartitionPruner.PartitionTableType;
 import org.apache.doris.nereids.rules.expression.rules.SortedPartitionRanges;
+import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions;
@@ -100,10 +102,10 @@ private SelectedPartitions pruneExternalPartitions(ExternalTable externalTable,
             sortedPartitionRanges = (Optional) partitionsCacheManager.get(
                             (SupportBinarySearchFilteringPartitions) externalTable, scan);
         }
-
-        List<String> prunedPartitions = new ArrayList<>(PartitionPruner.prune(
+        Pair<List<String>, Optional<Expression>> res = PartitionPruner.prune(
                 partitionSlots, filter.getPredicate(), nameToPartitionItem, ctx,
-                PartitionTableType.EXTERNAL, sortedPartitionRanges));
+                PartitionTableType.EXTERNAL, sortedPartitionRanges);
+        List<String> prunedPartitions = new ArrayList<>(res.first);
 
         for (String name : prunedPartitions) {
             selectedPartitionItems.put(name, nameToPartitionItem.get(name));