From e035d99daa71e26481a1e52e2f266b5e225fd0bd Mon Sep 17 00:00:00 2001 From: lizheng920625 Date: Fri, 21 Aug 2020 18:17:18 +0800 Subject: [PATCH] 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 --- .../io/prestosql/SystemSessionProperties.java | 11 ++ .../prestosql/sql/planner/PlanOptimizers.java | 7 +- .../rule/RemoveUnsupportedDynamicFilters.java | 120 ++++++++++++++++-- .../optimizations/PredicatePushDown.java | 13 +- .../sql/planner/TestDynamicFilter.java | 12 +- .../optimizations/TestReorderWindows.java | 2 +- .../tests/util/PrePushDownPlanGenerator.java | 2 +- 7 files changed, 145 insertions(+), 22 deletions(-) diff --git a/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java b/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java index 6f213a2ee..feddf5179 100644 --- a/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java +++ b/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> 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); diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/PlanOptimizers.java b/presto-main/src/main/java/io/prestosql/sql/planner/PlanOptimizers.java index 001edb615..ec9428594 100755 --- a/presto-main/src/main/java/io/prestosql/sql/planner/PlanOptimizers.java +++ b/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); diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/RemoveUnsupportedDynamicFilters.java b/presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/RemoveUnsupportedDynamicFilters.java index 062f3d802..48ef2d819 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/RemoveUnsupportedDynamicFilters.java +++ b/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 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()); - return result.getNode(); + PlanWithConsumedDynamicFilters result = plan.accept(new RemoveUnsupportedDynamicFilters.Rewriter(session, metadata, removedDynamicFilterIds), ImmutableSet.of()); + return SimplePlanRewriter.rewriteWith(new RemoveFilterVisitor(removedDynamicFilterIds), result.getNode(), null); } private class Rewriter extends PlanVisitor> { + private final Metadata metadata; + private final Session session; + private final Set removedDynamicFilterIds; + + public Rewriter(Session session, Metadata metadata, Set removedDynamicFilterIds) + { + this.session = session; + this.metadata = metadata; + this.removedDynamicFilterIds = removedDynamicFilterIds; + } + @Override protected PlanWithConsumedDynamicFilters visitPlan(PlanNode node, Set allowedDynamicFilterIds) { @@ -115,7 +128,7 @@ public class RemoveUnsupportedDynamicFilters public PlanWithConsumedDynamicFilters visitJoin(JoinNode node, Set allowedDynamicFilterIds) { ImmutableSet.Builder builder = ImmutableSet.builder().addAll(allowedDynamicFilterIds); - if (!hasHighSelectivity(node.getRight())) { + if (!isOptimizeDynamicFilterGeneration(session) || (isOptimizeDynamicFilterGeneration(session) && !hasHighSelectivity(node.getRight()))) { builder.addAll(node.getDynamicFilters().keySet()); } ImmutableSet 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 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 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 allowedDynamicFilterIds, ImmutableSet.Builder consumedDynamicFilterIds) { return combineConjuncts(extractConjuncts(expression) @@ -290,4 +333,63 @@ public class RemoveUnsupportedDynamicFilters return consumedDynamicFilterIds; } } + + private static class RemoveFilterVisitor + extends SimplePlanRewriter + { + private final Set removedDynamicFilterIds; + + public RemoveFilterVisitor(Set removedDynamicFilterIds) + { + this.removedDynamicFilterIds = removedDynamicFilterIds; + } + + @Override + public PlanNode visitFilter(FilterNode node, RewriteContext 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 context) + { + PlanNode leftSource = context.rewrite(node.getLeft()); + PlanNode rightSource = context.rewrite(node.getRight()); + Map 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; + } + } } diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/PredicatePushDown.java b/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/PredicatePushDown.java index 078a031ad..6ba65e88a 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/PredicatePushDown.java +++ b/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 dynamicFilters = ImmutableMap.of(); List 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. diff --git a/presto-main/src/test/java/io/prestosql/sql/planner/TestDynamicFilter.java b/presto-main/src/test/java/io/prestosql/sql/planner/TestDynamicFilter.java index c903b8ab1..9a5ec7339 100644 --- a/presto-main/src/test/java/io/prestosql/sql/planner/TestDynamicFilter.java +++ b/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( - tableScan("lineitem", ImmutableMap.of("LINEITEM_OK", "orderkey"))), + project(node(FilterNode.class, + tableScan("lineitem", ImmutableMap.of("LINEITEM_OK", "orderkey")))), exchange(project( tableScan("orders", ImmutableMap.of("ORDERS_OK", "orderkey"))))), exchange( diff --git a/presto-main/src/test/java/io/prestosql/sql/planner/optimizations/TestReorderWindows.java b/presto-main/src/test/java/io/prestosql/sql/planner/optimizations/TestReorderWindows.java index 1f2bf805d..4d68e3de1 100644 --- a/presto-main/src/test/java/io/prestosql/sql/planner/optimizations/TestReorderWindows.java +++ b/presto-main/src/test/java/io/prestosql/sql/planner/optimizations/TestReorderWindows.java @@ -323,7 +323,7 @@ public class TestReorderWindows { List 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(), diff --git a/presto-tests/src/main/java/io/prestosql/tests/util/PrePushDownPlanGenerator.java b/presto-tests/src/main/java/io/prestosql/tests/util/PrePushDownPlanGenerator.java index ff479c46e..571a55fad 100644 --- a/presto-tests/src/main/java/io/prestosql/tests/util/PrePushDownPlanGenerator.java +++ b/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