[opt](rbo) Fold external partition min max aggregate to constant #67136
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,202 @@ | ||
| // Licensed to the Apache Software Foundation (ASF) under one | ||
| // or more contributor license agreements. See the NOTICE file | ||
| // distributed with this work for additional information | ||
| // regarding copyright ownership. The ASF licenses this file | ||
| // to you under the Apache License, Version 2.0 (the | ||
| // "License"); you may not use this file except in compliance | ||
| // with the License. You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, | ||
| // software distributed under the License is distributed on an | ||
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| // KIND, either express or implied. See the License for the | ||
| // specific language governing permissions and limitations | ||
| // under the License. | ||
|
|
||
| package org.apache.doris.nereids.rules.rewrite; | ||
|
|
||
| import org.apache.doris.catalog.Column; | ||
| import org.apache.doris.catalog.ListPartitionItem; | ||
| import org.apache.doris.catalog.PartitionItem; | ||
| import org.apache.doris.catalog.PartitionKey; | ||
| import org.apache.doris.catalog.Type; | ||
| import org.apache.doris.datasource.ExternalTable; | ||
| import org.apache.doris.nereids.StatementContext; | ||
| import org.apache.doris.nereids.rules.Rule; | ||
| import org.apache.doris.nereids.rules.RuleType; | ||
| import org.apache.doris.nereids.trees.expressions.Alias; | ||
| import org.apache.doris.nereids.trees.expressions.Expression; | ||
| import org.apache.doris.nereids.trees.expressions.NamedExpression; | ||
| import org.apache.doris.nereids.trees.expressions.Slot; | ||
| import org.apache.doris.nereids.trees.expressions.SlotReference; | ||
| import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction; | ||
| import org.apache.doris.nereids.trees.expressions.functions.agg.Max; | ||
| import org.apache.doris.nereids.trees.expressions.functions.agg.Min; | ||
| import org.apache.doris.nereids.trees.expressions.literal.Literal; | ||
| import org.apache.doris.nereids.trees.expressions.literal.NullLiteral; | ||
| import org.apache.doris.nereids.trees.plans.Plan; | ||
| import org.apache.doris.nereids.trees.plans.algebra.Project; | ||
| import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate; | ||
| import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan; | ||
| import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation; | ||
| import org.apache.doris.nereids.trees.plans.logical.LogicalProject; | ||
| import org.apache.doris.nereids.types.DataType; | ||
|
|
||
| import com.google.common.collect.ImmutableList; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import java.util.Set; | ||
|
|
||
| /** | ||
| * Rewrite MIN/MAX on a single external list partition column to constants from partition metadata. | ||
| * | ||
| * <p>For queries like {@code dt = (select max(dt) from hive_table)}, evaluating MAX(dt) by scanning | ||
| * every partition blocks partition pruning for the outer scan. The selected partition map already | ||
| * contains the exact list partition values, so this rule replaces the scalar aggregate with a | ||
| * one-row constant relation before file-scan partition pruning runs. | ||
| */ | ||
| public class RewritePartitionColumnMinMaxToConstantRule implements RewriteRuleFactory { | ||
|
|
||
| @Override | ||
|
Contributor
There was a problem hiding this comment. [P1] Add tests before enabling this semantic rewrite. The PR changes only production files, and there is no existing FE or regression reference to this rule or RuleType. Please cover direct and scalar-subquery plans, the normalized alias shape, rule disabling, all applicability gates, null/default/empty partitions, connector transforms and scan selectors, and parameterized output types; assert both result value/type and whether the file scan is removed. |
||
| public List<Rule> buildRules() { | ||
| return ImmutableList.of( | ||
| logicalAggregate(logicalFileScan()) | ||
| .thenApply(ctx -> { | ||
| LogicalAggregate<LogicalFileScan> agg = ctx.root; | ||
| LogicalFileScan scan = agg.child(); | ||
| return tryRewrite(agg, scan, Optional.empty(), ctx.statementContext); | ||
| }) | ||
| .toRule(RuleType.REWRITE_PARTITION_COLUMN_MIN_MAX_TO_CONSTANT), | ||
| logicalAggregate(logicalProject(logicalFileScan())) | ||
| .thenApply(ctx -> { | ||
| LogicalAggregate<LogicalProject<LogicalFileScan>> agg = ctx.root; | ||
| LogicalProject<LogicalFileScan> project = agg.child(); | ||
| LogicalFileScan scan = project.child(); | ||
| return tryRewrite(agg, scan, Optional.of(project), ctx.statementContext); | ||
| }) | ||
| .toRule(RuleType.REWRITE_PARTITION_COLUMN_MIN_MAX_TO_CONSTANT) | ||
| ); | ||
| } | ||
|
|
||
| private Plan tryRewrite(LogicalAggregate<?> agg, LogicalFileScan scan, | ||
| Optional<LogicalProject<LogicalFileScan>> project, StatementContext statementContext) { | ||
| if (scan.getTableSample().isPresent() || !agg.getGroupByExpressions().isEmpty()) { | ||
|
Contributor
There was a problem hiding this comment. [P1] Skip this fold for incremental scan params. The Paimon |
||
| return null; | ||
| } | ||
|
|
||
| ExternalTable table = scan.getTable(); | ||
| if (!table.supportInternalPartitionPruned()) { | ||
|
Contributor
There was a problem hiding this comment. [P1] Require an exact identity-partition contract here. |
||
| return null; | ||
| } | ||
|
|
||
| List<Column> partitionColumns = table.getPartitionColumns( | ||
| statementContext.getSnapshot(table, scan.getTableSnapshot(), scan.getScanParams())); | ||
| if (partitionColumns.size() != 1) { | ||
| return null; | ||
| } | ||
| Column partitionColumn = partitionColumns.get(0); | ||
|
|
||
| Set<AggregateFunction> funcs = agg.getAggregateFunctions(); | ||
| if (funcs.isEmpty()) { | ||
| return null; | ||
| } | ||
| for (AggregateFunction func : funcs) { | ||
| if (!(func instanceof Min) && !(func instanceof Max)) { | ||
| return null; | ||
| } | ||
| } | ||
|
|
||
| List<NamedExpression> newOutputExprs = new ArrayList<>(); | ||
| for (NamedExpression outputExpr : agg.getOutputExpressions()) { | ||
| if (!(outputExpr instanceof Alias)) { | ||
| return null; | ||
| } | ||
| Alias alias = (Alias) outputExpr; | ||
| Expression child = alias.child(); | ||
| if (!(child instanceof AggregateFunction)) { | ||
| return null; | ||
| } | ||
| Optional<Literal> constant = tryGetConstant( | ||
| (AggregateFunction) child, partitionColumn, scan, project); | ||
| if (!constant.isPresent()) { | ||
| return null; | ||
| } | ||
| newOutputExprs.add(new Alias(alias.getExprId(), constant.get(), alias.getName())); | ||
| } | ||
|
|
||
| if (newOutputExprs.isEmpty()) { | ||
| return null; | ||
| } | ||
|
|
||
| LogicalOneRowRelation oneRowRelation = new LogicalOneRowRelation( | ||
| statementContext.getNextRelationId(), | ||
| ImmutableList.of(new Alias(new NullLiteral(), "__dummy__"))); | ||
| return new LogicalProject<>(newOutputExprs, oneRowRelation); | ||
| } | ||
|
|
||
| private Optional<Literal> tryGetConstant(AggregateFunction func, Column partitionColumn, LogicalFileScan scan, | ||
| Optional<LogicalProject<LogicalFileScan>> project) { | ||
| if (func.isDistinct() || func.getArguments().size() != 1) { | ||
| return Optional.empty(); | ||
| } | ||
| Optional<SlotReference> slot = resolveSlot(func.getArguments().get(0), project); | ||
| if (!slot.isPresent() || !isPartitionColumn(slot.get(), partitionColumn)) { | ||
| return Optional.empty(); | ||
| } | ||
|
|
||
| return findPartitionMinMaxLiteral(func instanceof Min, scan, partitionColumn); | ||
| } | ||
|
|
||
| private Optional<SlotReference> resolveSlot(Expression expression, | ||
| Optional<LogicalProject<LogicalFileScan>> project) { | ||
| Expression resolved = expression; | ||
| if (project.isPresent() && expression instanceof Slot) { | ||
| Map<Slot, Expression> aliasToProducer = ((Project) project.get()).getAliasToProducer(); | ||
| resolved = aliasToProducer.getOrDefault(expression, expression); | ||
| } | ||
| if (resolved instanceof SlotReference) { | ||
| return Optional.of((SlotReference) resolved); | ||
| } | ||
| return Optional.empty(); | ||
| } | ||
|
|
||
| private boolean isPartitionColumn(SlotReference slot, Column partitionColumn) { | ||
| Optional<Column> originalColumn = slot.getOriginalColumn(); | ||
| if (originalColumn.isPresent()) { | ||
| return originalColumn.get().getName().equalsIgnoreCase(partitionColumn.getName()); | ||
| } | ||
| return slot.getName().equalsIgnoreCase(partitionColumn.getName()); | ||
| } | ||
|
|
||
| private Optional<Literal> findPartitionMinMaxLiteral(boolean isMin, LogicalFileScan scan, Column partitionColumn) { | ||
|
Contributor
There was a problem hiding this comment. [P1] Do not derive query results from a partition generation that the scan does not use. MaxCompute serves this map from a cross-query cache with a default 600-second TTL, but |
||
| PartitionKey selectedKey = null; | ||
| for (PartitionItem item : scan.getSelectedPartitions().selectedPartitions.values()) { | ||
|
Contributor
There was a problem hiding this comment. [P1] Partition existence is not row-existence evidence. Hive lists every HMS partition name while its |
||
| if (item.isDefaultPartition() || !(item instanceof ListPartitionItem)) { | ||
| return Optional.empty(); | ||
| } | ||
| for (PartitionKey key : ((ListPartitionItem) item).getItems()) { | ||
| if (key.isDefaultListPartitionKey()) { | ||
| return Optional.empty(); | ||
| } | ||
| org.apache.doris.analysis.LiteralExpr literalExpr = key.getKeys().get(0); | ||
| if (literalExpr instanceof org.apache.doris.analysis.NullLiteral) { | ||
| continue; | ||
| } | ||
| if (selectedKey == null || (isMin ? key.compareTo(selectedKey) < 0 : key.compareTo(selectedKey) > 0)) { | ||
| selectedKey = key; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if (selectedKey == null) { | ||
|
Contributor
There was a problem hiding this comment. [P1] Do not treat an unenumerated or unresolved partition universe as an empty relation. Explicit |
||
| return Optional.of(new NullLiteral(DataType.fromCatalogType(partitionColumn.getType()))); | ||
| } | ||
| Type literalType = Type.fromPrimitiveType(selectedKey.getTypes().get(0)); | ||
|
Contributor
There was a problem hiding this comment. [P1] Preserve the normalized aggregate return type instead of rebuilding it from |
||
| return Optional.of(Literal.fromLegacyLiteral(selectedKey.getKeys().get(0), literalType)); | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
[P2] Revisit this rule after project merging. For
SELECT max(q) FROM (SELECT p AS q FROM ext) v,NormalizeAggregatefirst adds its own bottom project, so the operative whole-plan job here seesAggregate -> Project -> Project -> FileScan; the factory matches at most one project.MergeProjectableruns only in the following traversal, which never reruns this rule, so the intended alias case still scans files. Move/rerun the fold after project merging in both pipelines, or handle the normalized project chain directly.