Remove unnecessary DynamicFilters and fix UT for removing dynamic filters

fix predicates for join

Add session config for optimize the generation for dynamic filters

Remove unused session variable

presto-main/src/main/java/io/prestosql/SystemSessionProperties.java

@@ -144,6 +144,7 @@ public final class SystemSessionProperties
     public static final String ENABLE_CROSS_REGION_DYNAMIC_FILTER = "cross-region-dynamic-filter-enabled";
     public static final String ENABLE_HEURISTICINDEX_FILTER = "heuristicindex_filter_enabled";
     public static final String PUSH_TABLE_THROUGH_SUBQUERY = "push_table_through_subquery";
+    public static final String OPTIMIZE_DYNAMIC_FILTER_GENERATION = "optimize_dynamic_filter_generation";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
@@ -640,6 +641,11 @@ public final class SystemSessionProperties
                         "Expected FPP for BloomFilter which is used in dynamic filtering",
                         featuresConfig.getDynamicFilteringBloomFilterFpp(),
                         false),
+                booleanProperty(
+                        OPTIMIZE_DYNAMIC_FILTER_GENERATION,
+                        "Generate dynamic filters based on the selectivity",
+                        true,
+                        false),
                 booleanProperty(
                         ENABLE_EXECUTION_PLAN_CACHE,
                         "Enable execution plan caching",
@@ -1143,6 +1149,11 @@ public final class SystemSessionProperties
         return session.getSystemProperty(DYNAMIC_FILTERING_BLOOM_FILTER_FPP, Double.class);
     }
 
+    public static boolean isOptimizeDynamicFilterGeneration(Session session)
+    {
+        return session.getSystemProperty(OPTIMIZE_DYNAMIC_FILTER_GENERATION, Boolean.class);
+    }
+
     public static boolean isExecutionPlanCacheEnabled(Session session)
     {
         return session.getSystemProperty(ENABLE_EXECUTION_PLAN_CACHE, Boolean.class);

presto-main/src/main/java/io/prestosql/sql/planner/PlanOptimizers.java

@@ -286,7 +286,6 @@ public class PlanOptimizers
                         .addAll(new CanonicalizeExpressions(metadata, typeAnalyzer).rules())
                         .build());
 
-        PlanOptimizer predicatePushDown = new StatsRecordingPlanOptimizer(optimizerStats, new PredicatePushDown(metadata, typeAnalyzer, true));
         builder.add(
                 // Clean up all the sugar in expressions, e.g. AtTimeZone, must be run before all the other optimizers
                 new IterativeOptimizer(
@@ -412,7 +411,7 @@ public class PlanOptimizers
                 new CheckSubqueryNodesAreRewritten(),
                 new StatsRecordingPlanOptimizer(
                         optimizerStats,
-                        new PredicatePushDown(metadata, typeAnalyzer, false)),
+                        new PredicatePushDown(metadata, typeAnalyzer, false, false)),
                 new PruneUnreferencedOutputs(), // Prune unreferenced outputs to make the sub-query simple
                 inlineProjections,              // Remove redundant projects to make the sub-query simple
                 new SubQueryPushDown(metadata), // SubQueryPushDown is introduced in Hetu. It must run before AddExchanges
@@ -476,7 +475,7 @@ public class PlanOptimizers
                         statsCalculator,
                         estimatedExchangesCostCalculator,
                         ImmutableSet.of(new EliminateCrossJoins())), // This can pull up Filter and Project nodes from between Joins, so we need to push them down again
-                predicatePushDown,
+                new StatsRecordingPlanOptimizer(optimizerStats, new PredicatePushDown(metadata, typeAnalyzer, true, false)),
                 simplifyOptimizer, // Should be always run after PredicatePushDown
                 new IterativeOptimizer(
                         ruleStats,
@@ -578,7 +577,7 @@ public class PlanOptimizers
                         costCalculator,
                         ImmutableSet.of(new RemoveEmptyDelete()))); // Run RemoveEmptyDelete after table scan is removed by PickTableLayout/AddExchanges
 
-        builder.add(predicatePushDown); // Run predicate push down one more time in case we can leverage new information from layouts' effective predicate
+        builder.add(new StatsRecordingPlanOptimizer(optimizerStats, new PredicatePushDown(metadata, typeAnalyzer, true, true))); // Run predicate push down one more time in case we can leverage new information from layouts' effective predicate
         builder.add(new RemoveUnsupportedDynamicFilters(metadata, statsCalculator));
         builder.add(simplifyOptimizer); // Should be always run after PredicatePushDown
         builder.add(projectionPushDown);

presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/RemoveUnsupportedDynamicFilters.java

@@ -35,6 +35,7 @@ import io.prestosql.sql.planner.plan.JoinNode;
 import io.prestosql.sql.planner.plan.PlanNode;
 import io.prestosql.sql.planner.plan.PlanVisitor;
 import io.prestosql.sql.planner.plan.ProjectNode;
+import io.prestosql.sql.planner.plan.SimplePlanRewriter;
 import io.prestosql.sql.planner.plan.TableScanNode;
 import io.prestosql.sql.tree.Expression;
 
@@ -48,6 +49,7 @@ import java.util.stream.Collectors;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static io.prestosql.SystemSessionProperties.getDynamicFilteringMaxSize;
+import static io.prestosql.SystemSessionProperties.isOptimizeDynamicFilterGeneration;
 import static io.prestosql.sql.DynamicFilters.extractDynamicFilters;
 import static io.prestosql.sql.DynamicFilters.getDescriptor;
 import static io.prestosql.sql.ExpressionUtils.combineConjuncts;
@@ -65,12 +67,13 @@ import static java.util.stream.Collectors.toList;
 public class RemoveUnsupportedDynamicFilters
         implements PlanOptimizer
 {
-    private static final double DEFAULT_SELECTIVITY_THRESHOLD = 0.5D;
+    private static final double DEFAULT_GENERATE_SELECTIVITY_THRESHOLD = 0.5D;
+    private static final double DEFAULT_REMOVE_SELECTIVITY_THRESHOLD = 0.01D;
 
     private final Metadata metadata;
     private final StatsCalculator statsCalculator;
-    private Session session;
     private StatsProvider statsProvider;
+    private final Set<String> removedDynamicFilterIds = new HashSet<>();
 
     public RemoveUnsupportedDynamicFilters(Metadata metadata, StatsCalculator statsCalculator)
     {
@@ -81,15 +84,25 @@ public class RemoveUnsupportedDynamicFilters
     @Override
     public PlanNode optimize(PlanNode plan, Session session, TypeProvider types, SymbolAllocator symbolAllocator, PlanNodeIdAllocator idAllocator, WarningCollector warningCollector)
     {
-        this.session = session;
         this.statsProvider = new CachingStatsProvider(statsCalculator, session, symbolAllocator.getTypes());
-        PlanWithConsumedDynamicFilters result = plan.accept(new RemoveUnsupportedDynamicFilters.Rewriter(), ImmutableSet.of());
+        PlanWithConsumedDynamicFilters result = plan.accept(new RemoveUnsupportedDynamicFilters.Rewriter(session, metadata, removedDynamicFilterIds), ImmutableSet.of());
-        return result.getNode();
+        return SimplePlanRewriter.rewriteWith(new RemoveFilterVisitor(removedDynamicFilterIds), result.getNode(), null);
     }
 
     private class Rewriter
             extends PlanVisitor<PlanWithConsumedDynamicFilters, Set<String>>
     {
+        private final Metadata metadata;
+        private final Session session;
+        private final Set<String> removedDynamicFilterIds;
+
+        public Rewriter(Session session, Metadata metadata, Set<String> removedDynamicFilterIds)
+        {
+            this.session = session;
+            this.metadata = metadata;
+            this.removedDynamicFilterIds = removedDynamicFilterIds;
+        }
+
         @Override
         protected PlanWithConsumedDynamicFilters visitPlan(PlanNode node, Set<String> allowedDynamicFilterIds)
         {
@@ -115,7 +128,7 @@ public class RemoveUnsupportedDynamicFilters
         public PlanWithConsumedDynamicFilters visitJoin(JoinNode node, Set<String> allowedDynamicFilterIds)
         {
             ImmutableSet.Builder<String> builder = ImmutableSet.<String>builder().addAll(allowedDynamicFilterIds);
-            if (!hasHighSelectivity(node.getRight())) {
+            if (!isOptimizeDynamicFilterGeneration(session) || (isOptimizeDynamicFilterGeneration(session) && !hasHighSelectivity(node.getRight()))) {
                 builder.addAll(node.getDynamicFilters().keySet());
             }
             ImmutableSet<String> allowedDynamicFilterIdsProbeSide = builder.build();
@@ -164,8 +177,16 @@ public class RemoveUnsupportedDynamicFilters
             PlanNode source = result.getNode();
             Expression modified;
             if (source instanceof TableScanNode) {
+                // Keep only small table
+                DynamicFilters.ExtractResult extractResult = extractDynamicFilters(original);
+                if (isOptimizeDynamicFilterGeneration(session) && !highSelectivity(node)) {
+                    modified = removeDynamicFilters(original, ImmutableSet.of(), consumedDynamicFilterIds);
+                    extractResult.getDynamicConjuncts().forEach(descriptor -> removedDynamicFilterIds.add(descriptor.getId()));
+                }
                 // Keep only allowed dynamic filters
-                modified = removeDynamicFilters(original, allowedDynamicFilterIds, consumedDynamicFilterIds);
+                else {
+                    modified = removeDynamicFilters(original, allowedDynamicFilterIds, consumedDynamicFilterIds);
+                }
             }
             else {
                 modified = removeAllDynamicFilters(original);
@@ -197,19 +218,21 @@ public class RemoveUnsupportedDynamicFilters
             // Only handle the case that build side of JoinNode is0
             // TableScanNode or FilterNode above TableScanNode
             // as the estimates will be more accurate
+            Optional<Expression> predicates = Optional.empty();
             if (node instanceof TableScanNode) {
                 buildSideTableScanNode = Optional.of(node);
+                predicates = ((TableScanNode) buildSideTableScanNode.get()).getPredicate();
             }
 
             if (node instanceof FilterNode) {
                 PlanNode sourceNode = ((FilterNode) node).getSource();
                 if (sourceNode instanceof TableScanNode) {
                     buildSideTableScanNode = Optional.of(sourceNode);
+                    predicates = Optional.of(((FilterNode) node).getPredicate());
                 }
             }
 
             if (buildSideTableScanNode.isPresent()) {
-                Optional<Expression> predicates = ((TableScanNode) buildSideTableScanNode.get()).getPredicate();
                 // If there is dynamic filters applied on the build side,
                 // the selectivity cannot be easily calculated,
                 // thus we assume it's not high selectivity
@@ -236,13 +259,33 @@ public class RemoveUnsupportedDynamicFilters
 
                     // If selectivity too low, no need to create Dynamic Filter
                     double selectivity = filteredStats.getOutputRowCount() / totalRowCount.getValue();
-                    return selectivity > DEFAULT_SELECTIVITY_THRESHOLD;
+                    return selectivity > DEFAULT_GENERATE_SELECTIVITY_THRESHOLD;
                 }
             }
 
             return false;
         }
 
+        private boolean highSelectivity(FilterNode node)
+        {
+            Estimate totalRowCount = metadata.getTableStatistics(session, ((TableScanNode) node.getSource()).getTable(), Constraint.alwaysTrue()).getRowCount();
+            PlanNodeStatsEstimate filteredStats = statsProvider.getStats(node);
+
+            if (!filteredStats.isOutputRowCountUnknown() && !totalRowCount.isUnknown()) {
+                // If filtered row count is too big, no need to create Dynamic Filter
+                if (filteredStats.getOutputRowCount() > getDynamicFilteringMaxSize(session)) {
+                    return true;
+                }
+
+                // If selectivity too low, no need to create Dynamic Filter
+                double selectivity = filteredStats.getOutputRowCount() / totalRowCount.getValue();
+                return selectivity > DEFAULT_REMOVE_SELECTIVITY_THRESHOLD;
+            }
+            else {
+                return true;
+            }
+        }
+
         private Expression removeDynamicFilters(Expression expression, Set<String> allowedDynamicFilterIds, ImmutableSet.Builder<String> consumedDynamicFilterIds)
         {
             return combineConjuncts(extractConjuncts(expression)
@@ -290,4 +333,63 @@ public class RemoveUnsupportedDynamicFilters
             return consumedDynamicFilterIds;
         }
     }
+
+    private static class RemoveFilterVisitor
+            extends SimplePlanRewriter<Void>
+    {
+        private final Set<String> removedDynamicFilterIds;
+
+        public RemoveFilterVisitor(Set<String> removedDynamicFilterIds)
+        {
+            this.removedDynamicFilterIds = removedDynamicFilterIds;
+        }
+
+        @Override
+        public PlanNode visitFilter(FilterNode node, RewriteContext<Void> context)
+        {
+            PlanNode source = context.rewrite(node.getSource());
+            Expression original = node.getPredicate();
+            Expression modified;
+            if (source instanceof TableScanNode) {
+                modified = combineConjuncts(extractConjuncts(original)
+                        .stream()
+                        .filter(conjunct ->
+                                getDescriptor(conjunct)
+                                        .map(descriptor -> !removedDynamicFilterIds.contains(descriptor.getId())).orElse(true))
+                        .collect(toImmutableList()));
+            }
+            else {
+                modified = original;
+            }
+            return new FilterNode(node.getId(), source, modified);
+        }
+
+        @Override
+        public PlanNode visitJoin(JoinNode node, RewriteContext<Void> context)
+        {
+            PlanNode leftSource = context.rewrite(node.getLeft());
+            PlanNode rightSource = context.rewrite(node.getRight());
+            Map<String, Symbol> dynamicFilters = node.getDynamicFilters().entrySet().stream()
+                    .filter(entry -> !removedDynamicFilterIds.contains(entry.getKey()))
+                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+            if (leftSource != node.getLeft() ||
+                    rightSource != node.getRight() ||
+                    !dynamicFilters.equals(node.getDynamicFilters())) {
+                return new JoinNode(
+                        node.getId(),
+                        node.getType(),
+                        leftSource,
+                        rightSource,
+                        node.getCriteria(),
+                        node.getOutputSymbols(),
+                        node.getFilter(),
+                        node.getLeftHashSymbol(),
+                        node.getRightHashSymbol(),
+                        node.getDistributionType(),
+                        node.isSpillable(),
+                        dynamicFilters);
+            }
+            return node;
+        }
+    }
 }

presto-main/src/main/java/io/prestosql/sql/planner/optimizations/PredicatePushDown.java

@@ -105,13 +105,15 @@ public class PredicatePushDown
     private final LiteralEncoder literalEncoder;
     private final TypeAnalyzer typeAnalyzer;
     private final boolean useTableProperties;
+    private final boolean dynamicFiltering;
 
-    public PredicatePushDown(Metadata metadata, TypeAnalyzer typeAnalyzer, boolean useTableProperties)
+    public PredicatePushDown(Metadata metadata, TypeAnalyzer typeAnalyzer, boolean useTableProperties, boolean dynamicFiltering)
     {
         this.metadata = requireNonNull(metadata, "metadata is null");
         this.literalEncoder = new LiteralEncoder(metadata);
         this.typeAnalyzer = requireNonNull(typeAnalyzer, "typeAnalyzer is null");
         this.useTableProperties = useTableProperties;
+        this.dynamicFiltering = dynamicFiltering;
     }
 
     @Override
@@ -127,7 +129,7 @@ public class PredicatePushDown
                 metadata,
                 useTableProperties && isPredicatePushdownUseTableProperties(session));
         return SimplePlanRewriter.rewriteWith(
-                new Rewriter(symbolAllocator, idAllocator, metadata, literalEncoder, effectivePredicateExtractor, typeAnalyzer, session, types),
+                new Rewriter(symbolAllocator, idAllocator, metadata, literalEncoder, effectivePredicateExtractor, typeAnalyzer, session, types, dynamicFiltering),
                 plan,
                 TRUE_LITERAL);
     }
@@ -144,6 +146,7 @@ public class PredicatePushDown
         private final Session session;
         private final TypeProvider types;
         private final ExpressionEquivalence expressionEquivalence;
+        private final boolean dynamicFiltering;
 
         private Rewriter(
                 SymbolAllocator symbolAllocator,
@@ -153,7 +156,8 @@ public class PredicatePushDown
                 EffectivePredicateExtractor effectivePredicateExtractor,
                 TypeAnalyzer typeAnalyzer,
                 Session session,
-                TypeProvider types)
+                TypeProvider types,
+                boolean dynamicFiltering)
         {
             this.symbolAllocator = requireNonNull(symbolAllocator, "symbolAllocator is null");
             this.idAllocator = requireNonNull(idAllocator, "idAllocator is null");
@@ -164,6 +168,7 @@ public class PredicatePushDown
             this.session = requireNonNull(session, "session is null");
             this.types = requireNonNull(types, "types is null");
             this.expressionEquivalence = new ExpressionEquivalence(metadata, typeAnalyzer);
+            this.dynamicFiltering = dynamicFiltering;
         }
 
         @Override
@@ -568,7 +573,7 @@ public class PredicatePushDown
         {
             Map<String, Symbol> dynamicFilters = ImmutableMap.of();
             List<Expression> predicates = ImmutableList.of();
-            if ((node.getType() == INNER || node.getType() == RIGHT) && isEnableDynamicFiltering(session)) {
+            if ((node.getType() == INNER || node.getType() == RIGHT) && isEnableDynamicFiltering(session) && dynamicFiltering) {
                 // New equiJoinClauses could potentially not contain symbols used in current dynamic filters.
                 // Since we use PredicatePushdown to push dynamic filters themselves,
                 // instead of separate ApplyDynamicFilters rule we derive dynamic filters within PredicatePushdown itself.

presto-main/src/test/java/io/prestosql/sql/planner/TestDynamicFilter.java

@@ -16,6 +16,7 @@ package io.prestosql.sql.planner;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import io.prestosql.Session;
+import io.prestosql.sql.analyzer.FeaturesConfig;
 import io.prestosql.sql.planner.assertions.BasePlanTest;
 import io.prestosql.sql.planner.plan.EnforceSingleRowNode;
 import io.prestosql.sql.planner.plan.FilterNode;
@@ -26,6 +27,8 @@ import java.util.Optional;
 
 import static io.prestosql.SystemSessionProperties.DYNAMIC_FILTERING_MAX_SIZE;
 import static io.prestosql.SystemSessionProperties.ENABLE_DYNAMIC_FILTERING;
+import static io.prestosql.SystemSessionProperties.JOIN_DISTRIBUTION_TYPE;
+import static io.prestosql.SystemSessionProperties.JOIN_REORDERING_STRATEGY;
 import static io.prestosql.sql.planner.assertions.PlanMatchPattern.anyNot;
 import static io.prestosql.sql.planner.assertions.PlanMatchPattern.anyTree;
 import static io.prestosql.sql.planner.assertions.PlanMatchPattern.equiJoinClause;
@@ -45,7 +48,10 @@ public class TestDynamicFilter
     TestDynamicFilter()
     {
         // in order to test testUncorrelatedSubqueries with Dynamic Filtering, enable it
-        super(ImmutableMap.of(ENABLE_DYNAMIC_FILTERING, "true"));
+        super(ImmutableMap.of(
+                ENABLE_DYNAMIC_FILTERING, "true",
+                JOIN_REORDERING_STRATEGY, FeaturesConfig.JoinReorderingStrategy.NONE.name(),
+                JOIN_DISTRIBUTION_TYPE, FeaturesConfig.JoinDistributionType.BROADCAST.name()));
     }
 
     @Test
@@ -258,8 +264,8 @@ public class TestDynamicFilter
                 anyTree(
                         join(INNER, ImmutableList.of(equiJoinClause("LINEITEM_OK", "PART_PK")),
                                 join(INNER, ImmutableList.of(equiJoinClause("LINEITEM_OK", "ORDERS_OK")),
-                                        project(
+                                        project(node(FilterNode.class,
-                                                tableScan("lineitem", ImmutableMap.of("LINEITEM_OK", "orderkey"))),
+                                                tableScan("lineitem", ImmutableMap.of("LINEITEM_OK", "orderkey")))),
                                         exchange(project(
                                                 tableScan("orders", ImmutableMap.of("ORDERS_OK", "orderkey"))))),
                                 exchange(

presto-main/src/test/java/io/prestosql/sql/planner/optimizations/TestReorderWindows.java

@@ -323,7 +323,7 @@ public class TestReorderWindows
     {
         List<PlanOptimizer> optimizers = ImmutableList.of(
                 new UnaliasSymbolReferences(),
-                new PredicatePushDown(getQueryRunner().getMetadata(), new TypeAnalyzer(getQueryRunner().getSqlParser(), getQueryRunner().getMetadata()), false), new IterativeOptimizer(
+                new PredicatePushDown(getQueryRunner().getMetadata(), new TypeAnalyzer(getQueryRunner().getSqlParser(), getQueryRunner().getMetadata()), false, false), new IterativeOptimizer(
                         new RuleStatsRecorder(),
                         getQueryRunner().getStatsCalculator(),
                         getQueryRunner().getEstimatedExchangesCostCalculator(),

presto-tests/src/main/java/io/prestosql/tests/util/PrePushDownPlanGenerator.java

@@ -173,7 +173,7 @@ public class PrePushDownPlanGenerator
                 estimatedExchangesCostCalculator,
                 new SimplifyExpressions(metadata, typeAnalyzer).rules());
 
-        PlanOptimizer predicatePushDown = new StatsRecordingPlanOptimizer(optimizerStats, new PredicatePushDown(metadata, typeAnalyzer, false));
+        PlanOptimizer predicatePushDown = new StatsRecordingPlanOptimizer(optimizerStats, new PredicatePushDown(metadata, typeAnalyzer, false, false));
 
         builder.add(
                 // Clean up all the sugar in expressions, e.g. AtTimeZone, must be run before all the other optimizers