From b5609b1a020567dc4183ca06da8477b9dfcaba31 Mon Sep 17 00:00:00 2001 From: Sreeni Viswanadha Date: Wed, 26 Aug 2026 06:36:27 -0700 Subject: [PATCH] refactor(planner): Check determinism inside the plan copy utility (#28397) --- .../presto/sql/planner/PlannerUtils.java | 169 ++++++------------ .../planner/optimizations/JoinPrefilter.java | 41 +++-- .../optimizations/OptimizeTopNUsingRowId.java | 34 ++-- .../optimizations/PayloadJoinOptimizer.java | 14 +- .../PrefilterForLimitingAggregation.java | 12 +- 5 files changed, 111 insertions(+), 159 deletions(-) diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlannerUtils.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlannerUtils.java index d6c65cd552d5c..23efc11e73130 100644 --- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlannerUtils.java +++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlannerUtils.java @@ -330,19 +330,29 @@ public static AggregationNode.Aggregation createArrayAggregation(FunctionAndType return new AggregationNode.Aggregation(call, Optional.empty(), Optional.empty(), false, Optional.empty()); } - private static PlanNode cloneFilterNode(FilterNode filterNode, Session session, Metadata metadata, PlanNodeIdAllocator planNodeIdAllocator, List variablesToKeep, Map varMap, PlanNodeIdAllocator idAllocator) + private static Optional copyFilterNode(FilterNode filterNode, PlanNodeIdAllocator planNodeIdAllocator, List variablesToKeep, Map varMap, DeterminismEvaluator determinismEvaluator) { - PlanNode newSource = clonePlanNode(filterNode.getSource(), session, metadata, planNodeIdAllocator, variablesToKeep, varMap); - return new FilterNode( - filterNode.getSourceLocation(), - idAllocator.getNextId(), - newSource, - RowExpressionVariableInliner.inlineVariables(varMap, filterNode.getPredicate())); + if (!determinismEvaluator.isDeterministic(filterNode.getPredicate())) { + return Optional.empty(); + } + return copyDeterministicScanNodes(filterNode.getSource(), planNodeIdAllocator, variablesToKeep, varMap, determinismEvaluator) + .map(newSource -> new FilterNode( + filterNode.getSourceLocation(), + planNodeIdAllocator.getNextId(), + newSource, + RowExpressionVariableInliner.inlineVariables(varMap, filterNode.getPredicate()))); } - private static PlanNode cloneProjectNode(ProjectNode projectNode, Session session, Metadata metadata, PlanNodeIdAllocator planNodeIdAllocator, List fieldsToKeep, Map varMap, PlanNodeIdAllocator idAllocator) + private static Optional copyProjectNode(ProjectNode projectNode, PlanNodeIdAllocator planNodeIdAllocator, List fieldsToKeep, Map varMap, DeterminismEvaluator determinismEvaluator) { - PlanNode newSource = clonePlanNode(projectNode.getSource(), session, metadata, planNodeIdAllocator, fieldsToKeep, varMap); + if (!projectNode.getAssignments().getExpressions().stream().allMatch(determinismEvaluator::isDeterministic)) { + return Optional.empty(); + } + + Optional newSource = copyDeterministicScanNodes(projectNode.getSource(), planNodeIdAllocator, fieldsToKeep, varMap, determinismEvaluator); + if (!newSource.isPresent()) { + return Optional.empty(); + } Assignments.Builder newAssignments = Assignments.builder(); @@ -354,13 +364,13 @@ private static PlanNode cloneProjectNode(ProjectNode projectNode, Session sessio newAssignments.put(varMap.getOrDefault(var, var), RowExpressionVariableInliner.inlineVariables(varMap, entry.getValue())); } - return new ProjectNode( - idAllocator.getNextId(), - newSource, - newAssignments.build()); + return Optional.of(new ProjectNode( + planNodeIdAllocator.getNextId(), + newSource.get(), + newAssignments.build())); } - private static TableScanNode cloneTableScan(TableScanNode scanNode, Session session, Metadata metadata, PlanNodeIdAllocator planNodeIdAllocator, List fieldsToKeep, Map varMap) + private static TableScanNode copyTableScan(TableScanNode scanNode, PlanNodeIdAllocator planNodeIdAllocator, Map varMap) { Map assignments = scanNode.getAssignments(); @@ -396,30 +406,44 @@ private static TableScanNode cloneTableScan(TableScanNode scanNode, Session sess scanNode.getEnforcedConstraint(), scanNode.getCteMaterializationInfo()); } - public static PlanNode clonePlanNode(PlanNode planNode, Session session, Metadata metadata, PlanNodeIdAllocator planNodeIdAllocator, List fieldsToKeep, Map varMap) + /** + * Copies a scan/filter/project subtree (or a UNION ALL of such subtrees), giving every copied node a + * freshly allocated plan node id and renaming the variables listed in {@code varMap}. Sharing one + * subtree between two parents instead of copying it leaves the plan with duplicated plan node ids, + * which the plan checker rejects. + *

+ * Returns {@link Optional#empty()} when the subtree must not be duplicated: it contains a node type + * this method cannot rebuild, or a non-deterministic expression, which would make the copy produce + * different rows than the original (e.g. a rand() filter). Callers are expected to abandon their + * rewrite in that case, so they do not need to check determinism themselves. + */ + public static Optional copyDeterministicScanNodes(PlanNode planNode, Metadata metadata, PlanNodeIdAllocator planNodeIdAllocator, List fieldsToKeep, Map varMap) + { + return copyDeterministicScanNodes(planNode, planNodeIdAllocator, fieldsToKeep, varMap, new RowExpressionDeterminismEvaluator(metadata.getFunctionAndTypeManager())); + } + + private static Optional copyDeterministicScanNodes(PlanNode planNode, PlanNodeIdAllocator planNodeIdAllocator, List fieldsToKeep, Map varMap, DeterminismEvaluator determinismEvaluator) { if (planNode instanceof TableScanNode) { - TableScanNode scanNode = (TableScanNode) planNode; - return cloneTableScan(scanNode, session, metadata, planNodeIdAllocator, fieldsToKeep, varMap); + return Optional.of(copyTableScan((TableScanNode) planNode, planNodeIdAllocator, varMap)); } else if (planNode instanceof FilterNode) { - return cloneFilterNode((FilterNode) planNode, session, metadata, planNodeIdAllocator, fieldsToKeep, varMap, planNodeIdAllocator); + return copyFilterNode((FilterNode) planNode, planNodeIdAllocator, fieldsToKeep, varMap, determinismEvaluator); } else if (planNode instanceof ProjectNode) { - return cloneProjectNode((ProjectNode) planNode, session, metadata, planNodeIdAllocator, fieldsToKeep, varMap, planNodeIdAllocator); + return copyProjectNode((ProjectNode) planNode, planNodeIdAllocator, fieldsToKeep, varMap, determinismEvaluator); } else if (planNode instanceof UnionNode) { - return cloneUnionNode((UnionNode) planNode, session, metadata, planNodeIdAllocator, fieldsToKeep, varMap); + return copyUnionNode((UnionNode) planNode, planNodeIdAllocator, fieldsToKeep, varMap, determinismEvaluator); } - checkState(false, "Currently cannot clone: " + planNode.getClass().getName() + " nodes."); - return null; + return Optional.empty(); } - private static PlanNode cloneUnionNode(UnionNode unionNode, Session session, Metadata metadata, PlanNodeIdAllocator idAllocator, List fieldsToKeep, Map varMap) + private static Optional copyUnionNode(UnionNode unionNode, PlanNodeIdAllocator idAllocator, List fieldsToKeep, Map varMap, DeterminismEvaluator determinismEvaluator) { int numSources = unionNode.getSources().size(); - List clonedSources = new ArrayList<>(); + List copiedSources = new ArrayList<>(); List> legVarMaps = new ArrayList<>(); for (int i = 0; i < numSources; i++) { @@ -431,8 +455,11 @@ private static PlanNode cloneUnionNode(UnionNode unionNode, Session session, Met .filter(v -> v != null) .collect(toImmutableList()); - PlanNode clonedLeg = clonePlanNode(unionNode.getSources().get(i), session, metadata, idAllocator, legFieldsToKeep, legVarMap); - clonedSources.add(clonedLeg); + Optional copiedLeg = copyDeterministicScanNodes(unionNode.getSources().get(i), idAllocator, legFieldsToKeep, legVarMap, determinismEvaluator); + if (!copiedLeg.isPresent()) { + return Optional.empty(); + } + copiedSources.add(copiedLeg.get()); legVarMaps.add(legVarMap); } @@ -447,18 +474,18 @@ private static PlanNode cloneUnionNode(UnionNode unionNode, Session session, Met List originalInputs = unionNode.getVariableMapping().get(outputVar); for (int i = 0; i < numSources; i++) { VariableReferenceExpression originalInput = originalInputs.get(i); - VariableReferenceExpression clonedInput = legVarMaps.get(i).getOrDefault(originalInput, originalInput); - newOutputToInputs.put(newOutputVar, clonedInput); + VariableReferenceExpression copiedInput = legVarMaps.get(i).getOrDefault(originalInput, originalInput); + newOutputToInputs.put(newOutputVar, copiedInput); } } ListMultimap multimap = newOutputToInputs.build(); - return new UnionNode( + return Optional.of(new UnionNode( unionNode.getSourceLocation(), idAllocator.getNextId(), - clonedSources, + copiedSources, newOutputVars.build(), - fromListMultimap(multimap)); + fromListMultimap(multimap))); } public static String getPlanString(PlanNode planNode, Session session, TypeProvider types, Metadata metadata, boolean isVerboseOptimizerInfoEnabled) @@ -607,53 +634,6 @@ public static boolean isScanFilterProjectOrUnion(PlanNode node) node instanceof UnionNode && ((UnionNode) node).getSources().stream().allMatch(PlannerUtils::isScanFilterProjectOrUnion); } - /** - * Returns true if the scan-filter-project plan subtree contains only deterministic - * expressions in all filters and projections. This check is critical for optimizations - * that clone the subtree (e.g., JoinPrefilter), because cloning a subtree with - * non-deterministic expressions (like rand()) produces different results from each - * clone, leading to incorrect query results. - */ - public static boolean isDeterministicScanFilterProject(PlanNode node, FunctionAndTypeManager functionAndTypeManager) - { - DeterminismEvaluator determinismEvaluator = new RowExpressionDeterminismEvaluator(functionAndTypeManager); - return isDeterministicPlanSubtree(node, determinismEvaluator, false); - } - - /** - * Like {@link #isDeterministicScanFilterProject}, but additionally accepts a UnionNode - * whose every source is itself a deterministic scan/filter/project (or another such Union) - * subtree. Pair with {@link #isScanFilterProjectOrUnion} when the caller specifically - * supports cloning UNION ALL probe sides. - */ - public static boolean isDeterministicScanFilterProjectOrUnion(PlanNode node, FunctionAndTypeManager functionAndTypeManager) - { - DeterminismEvaluator determinismEvaluator = new RowExpressionDeterminismEvaluator(functionAndTypeManager); - return isDeterministicPlanSubtree(node, determinismEvaluator, true); - } - - private static boolean isDeterministicPlanSubtree(PlanNode node, DeterminismEvaluator determinismEvaluator, boolean allowUnion) - { - if (node instanceof TableScanNode) { - return true; - } - else if (node instanceof FilterNode) { - FilterNode filterNode = (FilterNode) node; - return determinismEvaluator.isDeterministic(filterNode.getPredicate()) - && isDeterministicPlanSubtree(filterNode.getSource(), determinismEvaluator, allowUnion); - } - else if (node instanceof ProjectNode) { - ProjectNode projectNode = (ProjectNode) node; - return projectNode.getAssignments().getExpressions().stream().allMatch(determinismEvaluator::isDeterministic) - && isDeterministicPlanSubtree(projectNode.getSource(), determinismEvaluator, allowUnion); - } - else if (allowUnion && node instanceof UnionNode) { - return ((UnionNode) node).getSources().stream() - .allMatch(source -> isDeterministicPlanSubtree(source, determinismEvaluator, allowUnion)); - } - return false; - } - public static CallExpression equalityPredicate(FunctionResolution functionResolution, RowExpression leftExpr, RowExpression rightExpr) { return new CallExpression(EQUAL.name(), @@ -959,37 +939,4 @@ public static Optional addPassThroughVariable( } return Optional.empty(); } - - /** - * Returns true if the plan subtree (expected to be a Filter/Project chain - * above a TableScanNode) contains any non-deterministic expressions. - * Useful for optimizations that clone the subtree — non-deterministic - * expressions would produce different values in the clone vs the original. - */ - public static boolean containsNonDeterministicExpression(PlanNode node, FunctionAndTypeManager functionAndTypeManager) - { - DeterminismEvaluator determinismEvaluator = new RowExpressionDeterminismEvaluator(functionAndTypeManager); - PlanNode current = node; - while (current != null) { - if (current instanceof ProjectNode) { - for (RowExpression expression : ((ProjectNode) current).getAssignments().getExpressions()) { - if (!determinismEvaluator.isDeterministic(expression)) { - return true; - } - } - current = ((ProjectNode) current).getSource(); - } - else if (current instanceof FilterNode) { - if (!determinismEvaluator.isDeterministic(((FilterNode) current).getPredicate())) { - return true; - } - current = ((FilterNode) current).getSource(); - } - else { - // TableScanNode or other leaf - break; - } - } - return false; - } } diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/JoinPrefilter.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/JoinPrefilter.java index 08296e9aa69a5..b95e115db327b 100644 --- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/JoinPrefilter.java +++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/JoinPrefilter.java @@ -51,10 +51,8 @@ import static com.facebook.presto.spi.plan.JoinType.INNER; import static com.facebook.presto.spi.plan.JoinType.LEFT; import static com.facebook.presto.sql.planner.PlannerUtils.addProjections; -import static com.facebook.presto.sql.planner.PlannerUtils.clonePlanNode; +import static com.facebook.presto.sql.planner.PlannerUtils.copyDeterministicScanNodes; import static com.facebook.presto.sql.planner.PlannerUtils.getVariableHash; -import static com.facebook.presto.sql.planner.PlannerUtils.isDeterministicScanFilterProject; -import static com.facebook.presto.sql.planner.PlannerUtils.isDeterministicScanFilterProjectOrUnion; import static com.facebook.presto.sql.planner.PlannerUtils.isScanFilterProject; import static com.facebook.presto.sql.planner.PlannerUtils.isScanFilterProjectOrUnion; import static com.facebook.presto.sql.planner.PlannerUtils.projectExpressions; @@ -126,16 +124,16 @@ public PlanOptimizerResult optimize(PlanNode plan, Session session, TypeProvider private static Optional findCloneableSource( PlanNode node, Set joinKeyVars, - FunctionAndTypeManager functionAndTypeManager, boolean complexEnabled) { // Base case: scan/filter/project, plus UNION ALL of such when complex mode is enabled. - // The UnionNode-aware helpers are intentionally separate from the plain - // isScanFilterProject / isDeterministicScanFilterProject so that other optimizers - // calling those don't silently gain UNION ALL handling without being audited. + // isScanFilterProjectOrUnion is intentionally separate from the plain isScanFilterProject so that + // other optimizers calling the latter don't silently gain UNION ALL handling without being audited. + // Determinism is not checked here: copyDeterministicScanNodes refuses to copy a subtree that + // contains a non-deterministic expression, and the rewrite is abandoned when it does. if (complexEnabled - ? isScanFilterProjectOrUnion(node) && isDeterministicScanFilterProjectOrUnion(node, functionAndTypeManager) - : isScanFilterProject(node) && isDeterministicScanFilterProject(node, functionAndTypeManager)) { + ? isScanFilterProjectOrUnion(node) + : isScanFilterProject(node)) { return Optional.of(node); } @@ -169,13 +167,11 @@ private static Optional findCloneableSource( } if (leftOutputs.containsAll(resolvedKeys.get()) - && isScanFilterProject(crossJoin.getLeft()) - && isDeterministicScanFilterProject(crossJoin.getLeft(), functionAndTypeManager)) { + && isScanFilterProject(crossJoin.getLeft())) { return Optional.of(crossJoin.getLeft()); } if (rightOutputs.containsAll(resolvedKeys.get()) - && isScanFilterProject(crossJoin.getRight()) - && isDeterministicScanFilterProject(crossJoin.getRight(), functionAndTypeManager)) { + && isScanFilterProject(crossJoin.getRight())) { return Optional.of(crossJoin.getRight()); } } @@ -190,8 +186,7 @@ && isDeterministicScanFilterProject(crossJoin.getRight(), functionAndTypeManager Set replicateVars = ImmutableSet.copyOf(unnest.getReplicateVariables()); if (replicateVars.containsAll(resolvedKeys.get()) - && isScanFilterProject(unnest.getSource()) - && isDeterministicScanFilterProject(unnest.getSource(), functionAndTypeManager)) { + && isScanFilterProject(unnest.getSource())) { return Optional.of(unnest.getSource()); } } @@ -209,8 +204,7 @@ && isDeterministicScanFilterProject(unnest.getSource(), functionAndTypeManager)) && agg.getGroupingSetCount() == 1 && !agg.hasEmptyGroupingSet() && groupingKeys.containsAll(resolvedKeys.get()) - && isScanFilterProject(agg.getSource()) - && isDeterministicScanFilterProject(agg.getSource(), functionAndTypeManager)) { + && isScanFilterProject(agg.getSource())) { return Optional.of(agg.getSource()); } } @@ -305,16 +299,21 @@ public PlanNode visitJoin(JoinNode node, RewriteContext context) List rightKeyList = equiJoinClause.stream().map(EquiJoinClause::getRight).collect(toImmutableList()); Set leftKeySet = ImmutableSet.copyOf(leftKeyList); - Optional cloneableSource = findCloneableSource(rewrittenLeft, leftKeySet, functionAndTypeManager, complexEnabled); + Optional cloneableSource = findCloneableSource(rewrittenLeft, leftKeySet, complexEnabled); - if (cloneableSource.isPresent()) { + // The copy is refused when the subtree cannot be duplicated safely, e.g. it is not + // deterministic, in which case the prefilter is not applied + Map leftVarMap = new HashMap(); + Optional copiedLeftKeys = cloneableSource.flatMap( + source -> copyDeterministicScanNodes(source, metadata, idAllocator, leftKeyList, leftVarMap)); + + if (copiedLeftKeys.isPresent()) { checkState(IntStream.range(0, leftKeyList.size()).boxed().allMatch(i -> leftKeyList.get(i).getType().equals(rightKeyList.get(i).getType()))); boolean hashJoinKey = leftKeyList.size() > 1 || (leftKeyList.get(0).getType().equals(VARCHAR) || leftKeyList.get(0).getType() instanceof VarcharType); // First create a SELECT DISTINCT leftKey FROM left - Map leftVarMap = new HashMap(); - PlanNode leftKeys = clonePlanNode(cloneableSource.get(), session, metadata, idAllocator, leftKeyList, leftVarMap); + PlanNode leftKeys = copiedLeftKeys.get(); ImmutableList.Builder expressionsToProject = ImmutableList.builder(); if (hashJoinKey) { RowExpression hashExpression = getVariableHash(leftKeyList, functionAndTypeManager); diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/OptimizeTopNUsingRowId.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/OptimizeTopNUsingRowId.java index 4d53fd30481e8..1664235d45d89 100644 --- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/OptimizeTopNUsingRowId.java +++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/OptimizeTopNUsingRowId.java @@ -14,7 +14,6 @@ package com.facebook.presto.sql.planner.optimizations; import com.facebook.presto.Session; -import com.facebook.presto.metadata.FunctionAndTypeManager; import com.facebook.presto.metadata.Metadata; import com.facebook.presto.metadata.TableLayout; import com.facebook.presto.spi.ColumnHandle; @@ -51,9 +50,8 @@ import static com.facebook.presto.common.type.BooleanType.BOOLEAN; import static com.facebook.presto.sql.planner.PlannerUtils.addColumnToTableScan; import static com.facebook.presto.sql.planner.PlannerUtils.addPassThroughVariable; -import static com.facebook.presto.sql.planner.PlannerUtils.clonePlanNode; +import static com.facebook.presto.sql.planner.PlannerUtils.copyDeterministicScanNodes; import static com.facebook.presto.sql.planner.PlannerUtils.findTableScanNode; -import static com.facebook.presto.sql.planner.PlannerUtils.isDeterministicScanFilterProject; import static com.facebook.presto.sql.planner.PlannerUtils.isScanFilterProject; import static com.facebook.presto.sql.planner.PlannerUtils.restrictOutput; import static com.google.common.collect.ImmutableList.toImmutableList; @@ -114,7 +112,7 @@ public boolean isEnabled(Session session) public PlanOptimizerResult optimize(PlanNode plan, Session session, TypeProvider types, VariableAllocator variableAllocator, PlanNodeIdAllocator idAllocator, WarningCollector warningCollector) { if (isEnabled(session)) { - Rewriter rewriter = new Rewriter(session, metadata, idAllocator, variableAllocator, metadata.getFunctionAndTypeManager()); + Rewriter rewriter = new Rewriter(session, metadata, idAllocator, variableAllocator); PlanNode rewritten = SimplePlanRewriter.rewriteWith(rewriter, plan, null); return PlanOptimizerResult.optimizerResult(rewritten, rewriter.isPlanChanged()); } @@ -128,17 +126,15 @@ private static class Rewriter private final Metadata metadata; private final PlanNodeIdAllocator idAllocator; private final VariableAllocator variableAllocator; - private final FunctionAndTypeManager functionAndTypeManager; private final int minColumnSavings; private boolean planChanged; - private Rewriter(Session session, Metadata metadata, PlanNodeIdAllocator idAllocator, VariableAllocator variableAllocator, FunctionAndTypeManager functionAndTypeManager) + private Rewriter(Session session, Metadata metadata, PlanNodeIdAllocator idAllocator, VariableAllocator variableAllocator) { this.session = requireNonNull(session, "session is null"); this.metadata = requireNonNull(metadata, "metadata is null"); this.idAllocator = requireNonNull(idAllocator, "idAllocator is null"); this.variableAllocator = requireNonNull(variableAllocator, "variableAllocator is null"); - this.functionAndTypeManager = requireNonNull(functionAndTypeManager, "functionAndTypeManager is null"); this.minColumnSavings = getOptimizeTopNUsingRowIdMinColumnSavings(session); } @@ -162,11 +158,6 @@ public PlanNode visitTopN(TopNNode node, RewriteContext context) return replaceSource(node, source); } - // Guard: source must be deterministic - if (!isDeterministicScanFilterProject(source, functionAndTypeManager)) { - return replaceSource(node, source); - } - // Find the underlying TableScanNode Optional tableScanOpt = findTableScanNode(source); if (!tableScanOpt.isPresent()) { @@ -210,7 +201,12 @@ public PlanNode visitTopN(TopNNode node, RewriteContext context) // 2. Clone narrow source: sort keys only + $row_id List sortKeys = node.getOrderingScheme().getOrderByVariables(); Map varMap = new HashMap<>(); - PlanNode narrowClone = clonePlanNode(source, session, metadata, idAllocator, sortKeys, varMap); + // The copy is refused when the source cannot be duplicated safely, e.g. it is not deterministic + Optional narrowCloneCopy = copyDeterministicScanNodes(source, metadata, idAllocator, sortKeys, varMap); + if (!narrowCloneCopy.isPresent()) { + return replaceSource(node, source); + } + PlanNode narrowClone = narrowCloneCopy.get(); // Add $row_id to the cloned narrow source too Optional clonedTableScanOpt = findTableScanNode(narrowClone); @@ -320,11 +316,6 @@ public PlanNode visitTopNRowNumber(TopNRowNumberNode node, RewriteContext return replaceSource(node, source); } - // Guard: source must be deterministic - if (!isDeterministicScanFilterProject(source, functionAndTypeManager)) { - return replaceSource(node, source); - } - // Find the underlying TableScanNode Optional tableScanOpt = findTableScanNode(source); if (!tableScanOpt.isPresent()) { @@ -370,7 +361,12 @@ public PlanNode visitTopNRowNumber(TopNRowNumberNode node, RewriteContext // 2. Clone narrow source: partition/order keys only + $row_id Map varMap = new HashMap<>(); - PlanNode narrowClone = clonePlanNode(source, session, metadata, idAllocator, narrowKeys, varMap); + // The copy is refused when the source cannot be duplicated safely, e.g. it is not deterministic + Optional narrowCloneCopy = copyDeterministicScanNodes(source, metadata, idAllocator, narrowKeys, varMap); + if (!narrowCloneCopy.isPresent()) { + return replaceSource(node, source); + } + PlanNode narrowClone = narrowCloneCopy.get(); // Add $row_id to the cloned narrow source too Optional clonedTableScanOpt = findTableScanNode(narrowClone); diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/PayloadJoinOptimizer.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/PayloadJoinOptimizer.java index 27786a1be44d1..5edd1d26f03a5 100644 --- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/PayloadJoinOptimizer.java +++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/PayloadJoinOptimizer.java @@ -65,8 +65,8 @@ import static com.facebook.presto.spi.plan.JoinType.LEFT; import static com.facebook.presto.spi.relation.SpecialFormExpression.Form.IS_NULL; import static com.facebook.presto.sql.planner.PlannerUtils.addProjections; -import static com.facebook.presto.sql.planner.PlannerUtils.clonePlanNode; import static com.facebook.presto.sql.planner.PlannerUtils.coalesce; +import static com.facebook.presto.sql.planner.PlannerUtils.copyDeterministicScanNodes; import static com.facebook.presto.sql.planner.PlannerUtils.equalityPredicate; import static com.facebook.presto.sql.planner.PlannerUtils.isScanFilterProject; import static com.facebook.presto.sql.planner.PlannerUtils.restrictOutput; @@ -387,7 +387,7 @@ private PlanNode rewriteScanFilterProject(PlanNode planNode, RewriteContext context, Set joinKeys) + private PlanNode constructDistinctKeysPlan(PlanNode planNode, RewriteContext context, Set joinKeys) { List groupingKeys = joinKeys.stream().collect(toImmutableList()); AggregationNode agg = new AggregationNode( @@ -409,8 +409,14 @@ private AggregationNode constructDistinctKeysPlan(PlanNode planNode, RewriteCont } context.get().setJoinKeyMap(new HashMap<>(varMap)); - PlanNode planNodeCopy = clonePlanNode(planNode, session, metadata, planNodeIdAllocator, planNode.getOutputVariables(), varMap); - context.get().setPayloadNode(planNodeCopy); + + // The copy is refused when the payload cannot be duplicated safely, e.g. it is not + // deterministic. The payload node is then left unset, so no payload rejoin is attempted. + Optional planNodeCopy = copyDeterministicScanNodes(planNode, metadata, planNodeIdAllocator, planNode.getOutputVariables(), varMap); + if (!planNodeCopy.isPresent()) { + return planNode; + } + context.get().setPayloadNode(planNodeCopy.get()); return agg; } diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/PrefilterForLimitingAggregation.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/PrefilterForLimitingAggregation.java index 3e676fe3b1fd7..08fd738ce18fb 100644 --- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/PrefilterForLimitingAggregation.java +++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/PrefilterForLimitingAggregation.java @@ -57,8 +57,7 @@ import static com.facebook.presto.sql.analyzer.TypeSignatureProvider.fromTypes; import static com.facebook.presto.sql.planner.PlannerUtils.addAggregation; import static com.facebook.presto.sql.planner.PlannerUtils.addProjections; -import static com.facebook.presto.sql.planner.PlannerUtils.clonePlanNode; -import static com.facebook.presto.sql.planner.PlannerUtils.containsNonDeterministicExpression; +import static com.facebook.presto.sql.planner.PlannerUtils.copyDeterministicScanNodes; import static com.facebook.presto.sql.planner.PlannerUtils.createMapType; import static com.facebook.presto.sql.planner.PlannerUtils.getPartitionColumnHandles; import static com.facebook.presto.sql.planner.PlannerUtils.getTableScanNodeWithOnlyFilterAndProject; @@ -195,7 +194,7 @@ else if (source instanceof AggregationNode) { Optional scanNode = getTableScanNodeWithOnlyFilterAndProject(aggregationNode.getSource()); // Since we duplicate the source of the aggregation - we want to restrict it to simple scan/filter/project // so we can do this opportunistic optimization without too much latency/cpu overhead to support common BI usecases - if (scanNode.isPresent() && !containsNonDeterministicExpression(aggregationNode.getSource(), metadata.getFunctionAndTypeManager())) { + if (scanNode.isPresent()) { PlanNode rewrittenAggregation = addPrefilter(aggregationNode, limitNode.getCount(), scanNode.get()); if (rewrittenAggregation != aggregationNode) { planChanged = true; @@ -249,7 +248,12 @@ private PlanNode addPrefilter(AggregationNode aggregationNode, long count, Table } PlanNode originalSource = aggregationNode.getSource(); - PlanNode keySource = clonePlanNode(originalSource, session, metadata, idAllocator, distinctKeys, new HashMap<>()); + // The copy is refused when the source cannot be duplicated safely, e.g. it is not deterministic + Optional copiedKeySource = copyDeterministicScanNodes(originalSource, metadata, idAllocator, distinctKeys, new HashMap<>()); + if (!copiedKeySource.isPresent()) { + return aggregationNode; + } + PlanNode keySource = copiedKeySource.get(); // Limit the scan to avoid excessive data when distinct keys are sparse. // We scan at most SCAN_LIMIT_MULTIPLIER * LIMIT rows (e.g., 1,000,000 rows for LIMIT 1000),