You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2025/01/23 04:23:40 UTC
(iotdb) branch master updated: Fix predicate pushdown when dealing with ProjectNode (#14754)
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 04810e27cc8 Fix predicate pushdown when dealing with ProjectNode (#14754)
04810e27cc8 is described below
commit 04810e27cc899d86d0d090d4c59fb6d29dcc3941
Author: Liao Lanyu <14...@qq.com>
AuthorDate: Thu Jan 23 12:23:34 2025 +0800
Fix predicate pushdown when dealing with ProjectNode (#14754)
---
.../optimizations/LogicalOptimizeFactory.java | 2 +-
.../optimizations/PushPredicateIntoTableScan.java | 156 ++++++++++++++++-----
.../plan/relational/planner/SubqueryTest.java | 147 +++++++++----------
3 files changed, 191 insertions(+), 114 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java
index 33450a79b86..97cd1401a55 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java
@@ -236,7 +236,7 @@ public class LogicalOptimizeFactory {
new IterativeOptimizer(
plannerContext, ruleStats, ImmutableSet.of(new PruneDistinctAggregation())),
simplifyOptimizer,
- new PushPredicateIntoTableScan(),
+ new PushPredicateIntoTableScan(plannerContext, typeAnalyzer),
// Currently, Distinct is not supported, so we cant use this rule for now.
// new IterativeOptimizer(
// plannerContext,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
index f039c09a701..dc943f5262d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
@@ -41,10 +41,13 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.NonAlignedAlignedDeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Assignments;
import org.apache.iotdb.db.queryengine.plan.relational.planner.EqualityInference;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.IrTypeAnalyzer;
import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.PlannerContext;
import org.apache.iotdb.db.queryengine.plan.relational.planner.SortOrder;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolsExtractor;
import org.apache.iotdb.db.queryengine.plan.relational.planner.ir.ReplaceSymbolInExpression;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
@@ -78,6 +81,8 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import java.util.function.Function;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkArgument;
@@ -90,6 +95,7 @@ import static org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet.PART
import static org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet.SCHEMA_FETCHER;
import static org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet.TABLE_TYPE;
import static org.apache.iotdb.db.queryengine.plan.analyze.AnalyzeVisitor.getTimePartitionSlotList;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.ExpressionSymbolInliner.inlineSymbols;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.SortOrder.ASC_NULLS_FIRST;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.SortOrder.ASC_NULLS_LAST;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolsExtractor.extractUnique;
@@ -98,6 +104,9 @@ import static org.apache.iotdb.db.queryengine.plan.relational.planner.ir.GlobalT
import static org.apache.iotdb.db.queryengine.plan.relational.planner.ir.IrUtils.combineConjuncts;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.ir.IrUtils.extractConjuncts;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.ir.IrUtils.filterDeterministicConjuncts;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.ir.IrUtils.isEffectivelyLiteral;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.CanonicalizeExpressionRewriter.canonicalizeExpression;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.ChildReplacer.replaceChildren;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.JoinNode.JoinType.FULL;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.JoinNode.JoinType.INNER;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.JoinUtils.FULL_JOIN_ONLY_SUPPORT_EQUI_JOIN;
@@ -136,6 +145,15 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
+ private final PlannerContext plannerContext;
+
+ private final IrTypeAnalyzer typeAnalyzer;
+
+ public PushPredicateIntoTableScan(PlannerContext plannerContext, IrTypeAnalyzer typeAnalyzer) {
+ this.plannerContext = plannerContext;
+ this.typeAnalyzer = typeAnalyzer;
+ }
+
@Override
public PlanNode optimize(PlanNode plan, Context context) {
return plan.accept(
@@ -143,7 +161,9 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
context.getQueryContext(),
context.getAnalysis(),
context.getMetadata(),
- context.getSymbolAllocator()),
+ context.getSymbolAllocator(),
+ plannerContext,
+ typeAnalyzer),
new RewriteContext(TRUE_LITERAL));
}
@@ -153,17 +173,23 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
private final Metadata metadata;
private final SymbolAllocator symbolAllocator;
private final QueryId queryId;
+ private final PlannerContext plannerContext;
+ private final IrTypeAnalyzer typeAnalyzer;
Rewriter(
MPPQueryContext queryContext,
Analysis analysis,
Metadata metadata,
- SymbolAllocator symbolAllocator) {
+ SymbolAllocator symbolAllocator,
+ PlannerContext plannerContext,
+ IrTypeAnalyzer typeAnalyzer) {
this.queryContext = queryContext;
this.analysis = analysis;
this.metadata = metadata;
this.symbolAllocator = symbolAllocator;
this.queryId = queryContext.getQueryId();
+ this.plannerContext = plannerContext;
+ this.typeAnalyzer = typeAnalyzer;
}
@Override
@@ -200,9 +226,98 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
}
}
- // TODO(beyyes) in some situation, predicate can not be pushed down below ProjectNode
- node.setChild(node.getChild().accept(this, context));
- return node;
+ Set<Symbol> deterministicSymbols =
+ node.getAssignments().entrySet().stream()
+ .filter(entry -> isDeterministic(entry.getValue()))
+ .map(Map.Entry::getKey)
+ .collect(Collectors.toSet());
+
+ Predicate<Expression> deterministic =
+ conjunct -> deterministicSymbols.containsAll(extractUnique(conjunct));
+
+ Expression inheritedPredicate =
+ context.inheritedPredicate != null ? context.inheritedPredicate : TRUE_LITERAL;
+ Map<Boolean, List<Expression>> conjuncts =
+ extractConjuncts(inheritedPredicate).stream()
+ .collect(Collectors.partitioningBy(deterministic));
+
+ // Push down conjuncts from the inherited predicate that only depend on deterministic
+ // assignments with
+ // certain limitations.
+ List<Expression> deterministicConjuncts = conjuncts.get(true);
+
+ // We partition the expressions in the deterministicConjuncts into two lists, and only inline
+ // the
+ // expressions that are in the inlining targets list.
+ Map<Boolean, List<Expression>> inlineConjuncts =
+ deterministicConjuncts.stream()
+ .collect(
+ Collectors.partitioningBy(expression -> isInliningCandidate(expression, node)));
+
+ List<Expression> inlinedDeterministicConjuncts =
+ inlineConjuncts.get(true).stream()
+ .map(entry -> inlineSymbols(node.getAssignments().getMap(), entry))
+ .map(
+ conjunct ->
+ canonicalizeExpression(
+ conjunct,
+ typeAnalyzer,
+ queryContext.getTypeProvider(),
+ plannerContext,
+ queryContext
+ .getSession())) // normalize expressions to a form that unwrapCasts
+ // understands
+ // no need for now
+ // .map(conjunct -> unwrapCasts(session, plannerContext, typeAnalyzer, types,
+ // conjunct))
+ .collect(Collectors.toList());
+
+ PlanNode rewrittenChild =
+ node.getChild()
+ .accept(this, new RewriteContext(combineConjuncts(inlinedDeterministicConjuncts)));
+
+ PlanNode rewrittenNode = replaceChildren(node, ImmutableList.of(rewrittenChild));
+
+ // All deterministic conjuncts that contains non-inlining targets, and non-deterministic
+ // conjuncts,
+ // if any, will be in the filter node.
+ List<Expression> nonInliningConjuncts = inlineConjuncts.get(false);
+ nonInliningConjuncts.addAll(conjuncts.get(false));
+
+ if (!nonInliningConjuncts.isEmpty()) {
+ rewrittenNode =
+ new FilterNode(
+ queryId.genPlanNodeId(), rewrittenNode, combineConjuncts(nonInliningConjuncts));
+ }
+
+ return rewrittenNode;
+ }
+
+ private boolean isInliningCandidate(Expression expression, ProjectNode node) {
+ // TryExpressions should not be pushed down. However they are now being handled as lambda
+ // passed to a FunctionCall now and should not affect predicate push down. So we want to make
+ // sure the conjuncts are not TryExpressions.
+ // verify(AstUtils.preOrder(expression).noneMatch(TryExpression.class::isInstance));
+
+ // candidate symbols for inlining are
+ // 1. references to simple constants or symbol references
+ // 2. references to complex expressions that appear only once
+ // which come from the node, as opposed to an enclosing scope.
+ Set<Symbol> childOutputSet = ImmutableSet.copyOf(node.getOutputSymbols());
+ Map<Symbol, Long> dependencies =
+ SymbolsExtractor.extractAll(expression).stream()
+ .filter(childOutputSet::contains)
+ .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));
+
+ return dependencies.entrySet().stream()
+ .allMatch(
+ entry ->
+ entry.getValue() == 1
+ || isEffectivelyLiteral(
+ node.getAssignments().get(entry.getKey()),
+ plannerContext,
+ queryContext.getSession())
+ || node.getAssignments().get(entry.getKey()) instanceof SymbolReference);
}
@Override
@@ -212,7 +327,7 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
Expression predicate = combineConjuncts(node.getPredicate(), context.inheritedPredicate);
// when exist diff function, predicate can not be pushed down into DeviceTableScanNode
- if (containsDiffFunction(predicate) || canNotPushDownBelowProjectNode(node, predicate)) {
+ if (containsDiffFunction(predicate)) {
node.setChild(node.getChild().accept(this, new RewriteContext()));
node.setPredicate(predicate);
context.inheritedPredicate = TRUE_LITERAL;
@@ -234,35 +349,6 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
return node;
}
- private boolean canNotPushDownBelowProjectNode(FilterNode node, Expression predicate) {
- PlanNode child = node.getChild();
- if (child instanceof ProjectNode) {
- // if the inheritedPredicate is not in the output of the child of ProjectNode, we can not
- // push it down for now.
- // (predicate will be computed by the ProjectNode, Trino will rewrite the predicate in
- // visitProject, but we have not implemented this for now.)
- Set<Symbol> outputSymbolsOfProjectChild =
- new HashSet<>(((ProjectNode) child).getChild().getOutputSymbols());
- return missingTermsInOutputSymbols(predicate, outputSymbolsOfProjectChild);
- }
- return false;
- }
-
- private boolean missingTermsInOutputSymbols(Expression expression, Set<Symbol> outputSymbols) {
- if (expression instanceof SymbolReference) {
- return !outputSymbols.contains(Symbol.from(expression));
- }
- if (!expression.getChildren().isEmpty()) {
- for (Node node : expression.getChildren()) {
- if (missingTermsInOutputSymbols((Expression) node, outputSymbols)) {
- return true;
- }
- }
- }
-
- return false;
- }
-
// private boolean areExpressionsEquivalent(
// Expression leftExpression, Expression rightExpression) {
// return false;
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/SubqueryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/SubqueryTest.java
index 96dbe9d2b14..3420b29719b 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/SubqueryTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/SubqueryTest.java
@@ -344,11 +344,10 @@ public class SubqueryTest {
* └──OutputNode
* └──ProjectNode
* └──FilterNode
- * └──ProjectNode
- * └──JoinNode
- * |──TableScanNode
- * ├──AggregationNode
- * │ └──TableScanNode
+ * └──JoinNode
+ * |──TableScanNode
+ * ├──AggregationNode
+ * └──TableScanNode
*/
assertPlan(
@@ -356,29 +355,26 @@ public class SubqueryTest {
output(
project(
anyTree(
- project(
- join(
- JoinNode.JoinType.INNER,
- builder ->
- builder
- .left(tableScan1)
- .right(
- aggregation(
- singleGroupingSet(),
- ImmutableMap.of(
- Optional.of("min"),
- aggregationFunction(
- "min", ImmutableList.of("s1_7")),
- Optional.of("count_all"),
- aggregationFunction(
- "count_all", ImmutableList.of("s1_7")),
- Optional.of("count_non_null"),
- aggregationFunction(
- "count", ImmutableList.of("s1_7"))),
- Collections.emptyList(),
- Optional.empty(),
- SINGLE,
- tableScan2))))))));
+ join(
+ JoinNode.JoinType.INNER,
+ builder ->
+ builder
+ .left(tableScan1)
+ .right(
+ aggregation(
+ singleGroupingSet(),
+ ImmutableMap.of(
+ Optional.of("min"),
+ aggregationFunction("min", ImmutableList.of("s1_7")),
+ Optional.of("count_all"),
+ aggregationFunction(
+ "count_all", ImmutableList.of("s1_7")),
+ Optional.of("count_non_null"),
+ aggregationFunction("count", ImmutableList.of("s1_7"))),
+ Collections.emptyList(),
+ Optional.empty(),
+ SINGLE,
+ tableScan2)))))));
// Verify DistributionPlan
assertPlan(
@@ -386,47 +382,45 @@ public class SubqueryTest {
output(
project(
anyTree(
- project(
- join(
- JoinNode.JoinType.INNER,
- builder ->
- builder
- .left(collect(exchange(), tableScan1, exchange()))
- .right(
- aggregation(
- singleGroupingSet(),
- ImmutableMap.of(
- Optional.of("min"),
- aggregationFunction(
- "min", ImmutableList.of("min_9")),
- Optional.of("count_all"),
- aggregationFunction(
- "count_all", ImmutableList.of("count_all_10")),
- Optional.of("count_non_null"),
- aggregationFunction(
- "count", ImmutableList.of("count"))),
- Collections.emptyList(),
- Optional.empty(),
- FINAL,
- collect(
- exchange(),
- aggregation(
- singleGroupingSet(),
- ImmutableMap.of(
- Optional.of("min_9"),
- aggregationFunction(
- "min", ImmutableList.of("s1_6")),
- Optional.of("count_all_10"),
- aggregationFunction(
- "count_all", ImmutableList.of("s1_6")),
- Optional.of("count"),
- aggregationFunction(
- "count", ImmutableList.of("s1_6"))),
- Collections.emptyList(),
- Optional.empty(),
- PARTIAL,
- tableScan3),
- exchange())))))))));
+ join(
+ JoinNode.JoinType.INNER,
+ builder ->
+ builder
+ .left(collect(exchange(), tableScan1, exchange()))
+ .right(
+ aggregation(
+ singleGroupingSet(),
+ ImmutableMap.of(
+ Optional.of("min"),
+ aggregationFunction("min", ImmutableList.of("min_9")),
+ Optional.of("count_all"),
+ aggregationFunction(
+ "count_all", ImmutableList.of("count_all_10")),
+ Optional.of("count_non_null"),
+ aggregationFunction(
+ "count", ImmutableList.of("count"))),
+ Collections.emptyList(),
+ Optional.empty(),
+ FINAL,
+ collect(
+ exchange(),
+ aggregation(
+ singleGroupingSet(),
+ ImmutableMap.of(
+ Optional.of("min_9"),
+ aggregationFunction(
+ "min", ImmutableList.of("s1_6")),
+ Optional.of("count_all_10"),
+ aggregationFunction(
+ "count_all", ImmutableList.of("s1_6")),
+ Optional.of("count"),
+ aggregationFunction(
+ "count", ImmutableList.of("s1_6"))),
+ Collections.emptyList(),
+ Optional.empty(),
+ PARTIAL,
+ tableScan3),
+ exchange()))))))));
assertPlan(planTester.getFragmentPlan(1), tableScan1);
@@ -519,19 +513,16 @@ public class SubqueryTest {
* └──OutputNode
* └──ProjectNode
* └──FilterNode
- * └──ProjectNode
- * └──SemiJoinNode
- * |──SortNode
- * | └──TableScanNode
- * ├──SortNode
- * │ └──TableScanNode
+ * └──SemiJoinNode
+ * |──SortNode
+ * | └──TableScanNode
+ * ├──SortNode
+ * │ └──TableScanNode
*/
assertPlan(
logicalQueryPlan,
output(
- project(
- anyTree(
- project(semiJoin("s1", "s1_6", "expr", sort(tableScan1), sort(tableScan2)))))));
+ project(anyTree(semiJoin("s1", "s1_6", "expr", sort(tableScan1), sort(tableScan2))))));
}
}