AggregateStrategies.java
// 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.implementation;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Index;
import org.apache.doris.catalog.KeysType;
import org.apache.doris.catalog.MaterializedIndexMeta;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.catalog.RowBinlogTableWrapper;
import org.apache.doris.catalog.info.IndexType;
import org.apache.doris.datasource.mvcc.MvccUtil;
import org.apache.doris.datasource.plugin.PluginDrivenExternalTable;
import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.annotation.DependsRules;
import org.apache.doris.nereids.rules.Rule;
import org.apache.doris.nereids.rules.RuleType;
import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
import org.apache.doris.nereids.rules.expression.rules.FoldConstantRuleOnFE;
import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.Cast;
import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.IsNull;
import org.apache.doris.nereids.trees.expressions.Or;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.functions.ExpressionTrait;
import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
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.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.LogicalFilter;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
import org.apache.doris.nereids.trees.plans.physical.PhysicalFileScan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate.PushDownAggOp;
import org.apache.doris.nereids.util.ExpressionUtils;
import org.apache.doris.qe.ConnectContext;
import com.google.common.collect.ImmutableList;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
/** AggregateStrategies */
@DependsRules({
NormalizeAggregate.class,
FoldConstantRuleOnFE.class
})
public class AggregateStrategies implements ImplementationRuleFactory {
@Override
public List<Rule> buildRules() {
return ImmutableList.of(
RuleType.COUNT_ON_INDEX_WITHOUT_PROJECT.build(
logicalAggregate(
logicalFilter(
logicalOlapScan().when(this::isDupOrMowKeyTable).when(this::isInvertedIndexEnabledOnTable)
)
)
.when(agg -> enablePushDownCountOnIndex())
.when(agg -> agg.getGroupByExpressions().isEmpty())
.when(agg -> {
Set<AggregateFunction> funcs = agg.getAggregateFunctions();
if (funcs.isEmpty() || !funcs.stream().allMatch(f -> f instanceof Count && !f.isDistinct()
&& (((Count) f).isCountStar() || f.child(0) instanceof Slot))) {
return false;
}
Set<Expression> conjuncts = agg.child().getConjuncts();
if (conjuncts.isEmpty()) {
return false;
}
Set<Slot> aggSlots = funcs.stream()
.flatMap(f -> f.getInputSlots().stream())
.collect(Collectors.toSet());
return aggSlots.isEmpty() || conjuncts.stream().allMatch(expr ->
checkSlotInOrExpression(expr, aggSlots) && checkIsNullExpr(expr, aggSlots));
})
.thenApply(ctx -> {
LogicalAggregate<LogicalFilter<LogicalOlapScan>> agg = ctx.root;
LogicalFilter<LogicalOlapScan> filter = agg.child();
LogicalOlapScan olapScan = filter.child();
return pushdownCountOnIndex(agg, null, filter, olapScan, ctx.cascadesContext);
})
),
RuleType.COUNT_ON_INDEX.build(
logicalAggregate(
logicalProject(
logicalFilter(
logicalOlapScan().when(this::isDupOrMowKeyTable).when(this::isInvertedIndexEnabledOnTable)
)
)
)
.when(agg -> enablePushDownCountOnIndex())
.when(agg -> agg.getGroupByExpressions().isEmpty())
.when(agg -> {
Set<AggregateFunction> funcs = agg.getAggregateFunctions();
if (funcs.isEmpty() || !funcs.stream().allMatch(f -> f instanceof Count && !f.isDistinct()
&& (((Count) f).isCountStar() || f.child(0) instanceof Slot))) {
return false;
}
Set<Expression> conjuncts = agg.child().child().getConjuncts();
if (conjuncts.isEmpty()) {
return false;
}
Set<Slot> aggSlots = normalizeArguments(funcs, agg.child()).stream()
.flatMap(argument -> argument.getInputSlots().stream())
.collect(Collectors.toSet());
return aggSlots.isEmpty() || conjuncts.stream().allMatch(expr ->
checkSlotInOrExpression(expr, aggSlots) && checkIsNullExpr(expr, aggSlots));
})
.thenApply(ctx -> {
LogicalAggregate<LogicalProject<LogicalFilter<LogicalOlapScan>>> agg = ctx.root;
LogicalProject<LogicalFilter<LogicalOlapScan>> project = agg.child();
LogicalFilter<LogicalOlapScan> filter = project.child();
LogicalOlapScan olapScan = filter.child();
return pushdownCountOnIndex(agg, project, filter, olapScan, ctx.cascadesContext);
})
),
RuleType.STORAGE_LAYER_AGGREGATE_MINMAX_ON_UNIQUE_WITHOUT_PROJECT.build(
logicalAggregate(
logicalFilter(
logicalOlapScan().when(this::isUniqueKeyTable))
.when(filter -> {
if (filter.getConjuncts().size() != 1) {
return false;
}
Expression childExpr = filter.getConjuncts().iterator().next().children().get(0);
if (childExpr instanceof SlotReference) {
Optional<Column> column = ((SlotReference) childExpr).getOriginalColumn();
return column.map(Column::isDeleteSignColumn).orElse(false);
}
return false;
})
)
.when(agg -> enablePushDownMinMaxOnUnique())
.when(agg -> agg.getGroupByExpressions().isEmpty())
.when(agg -> {
Set<AggregateFunction> funcs = agg.getAggregateFunctions();
return !funcs.isEmpty() && funcs.stream()
.allMatch(f -> (f instanceof Min) || (f instanceof Max));
})
.thenApply(ctx -> {
LogicalAggregate<LogicalFilter<LogicalOlapScan>> agg = ctx.root;
LogicalFilter<LogicalOlapScan> filter = agg.child();
LogicalOlapScan olapScan = filter.child();
return pushdownMinMaxOnUniqueTable(agg, null, filter, olapScan,
ctx.cascadesContext);
})
),
RuleType.STORAGE_LAYER_AGGREGATE_MINMAX_ON_UNIQUE.build(
logicalAggregate(logicalProject(logicalFilter(logicalOlapScan().when(this::isUniqueKeyTable))
.when(filter -> {
if (filter.getConjuncts().size() != 1) {
return false;
}
Expression childExpr = filter.getConjuncts().iterator().next()
.children().get(0);
if (childExpr instanceof SlotReference) {
Optional<Column> column = ((SlotReference) childExpr).getOriginalColumn();
return column.map(Column::isDeleteSignColumn).orElse(false);
}
return false;
})))
.when(agg -> enablePushDownMinMaxOnUnique())
.when(agg -> agg.getGroupByExpressions().isEmpty())
.when(agg -> {
Set<AggregateFunction> funcs = agg.getAggregateFunctions();
return !funcs.isEmpty()
&& funcs.stream().allMatch(f -> (f instanceof Min) || (f instanceof Max));
})
.thenApply(ctx -> {
LogicalAggregate<LogicalProject<LogicalFilter<LogicalOlapScan>>> agg = ctx.root;
LogicalProject<LogicalFilter<LogicalOlapScan>> project = agg.child();
LogicalFilter<LogicalOlapScan> filter = project.child();
LogicalOlapScan olapScan = filter.child();
return pushdownMinMaxOnUniqueTable(agg, project, filter, olapScan,
ctx.cascadesContext);
})
),
RuleType.STORAGE_LAYER_AGGREGATE_WITHOUT_PROJECT.build(
logicalAggregate(
logicalOlapScan()
)
.when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
.thenApply(ctx -> storageLayerAggregate(ctx.root, null, ctx.root.child(), ctx.cascadesContext))
),
RuleType.STORAGE_LAYER_WITH_PROJECT_NO_SLOT_REF.build(
logicalProject(
logicalOlapScan()
)
.thenApply(ctx -> {
LogicalProject<LogicalOlapScan> project = ctx.root;
LogicalOlapScan olapScan = project.child();
return pushDownCountWithoutSlotRef(project, olapScan, ctx.cascadesContext);
})
),
RuleType.STORAGE_LAYER_AGGREGATE_WITH_PROJECT.build(
logicalAggregate(
logicalProject(
logicalOlapScan()
)
)
.when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
.thenApply(ctx -> {
LogicalAggregate<LogicalProject<LogicalOlapScan>> agg = ctx.root;
LogicalProject<LogicalOlapScan> project = agg.child();
LogicalOlapScan olapScan = project.child();
return storageLayerAggregate(agg, project, olapScan, ctx.cascadesContext);
})
),
RuleType.STORAGE_LAYER_AGGREGATE_WITHOUT_PROJECT_FOR_FILE_SCAN.build(
logicalAggregate(
logicalFileScan()
)
.when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
.thenApply(ctx -> storageLayerAggregate(ctx.root, null, ctx.root.child(), ctx.cascadesContext))
),
RuleType.STORAGE_LAYER_AGGREGATE_WITH_PROJECT_FOR_FILE_SCAN.build(
logicalAggregate(
logicalProject(
logicalFileScan()
)
).when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
.thenApply(ctx -> {
LogicalAggregate<LogicalProject<LogicalFileScan>> agg = ctx.root;
LogicalProject<LogicalFileScan> project = agg.child();
LogicalFileScan fileScan = project.child();
return storageLayerAggregate(agg, project, fileScan, ctx.cascadesContext);
})
),
// The two patterns above deliberately do not contain a LogicalFilter, so any query with
// a WHERE clause never reaches storageLayerAggregate: PruneFileScanPartition keeps the
// LogicalFilter above the scan after partition pruning (see PruneFileScanPartition#build),
// which leaves the plan shaped as Agg(Project(Filter(FileScan))).
//
// Nereids keeps the filter as a separate node until PhysicalPlanTranslator turns it into
// scan conjuncts, so we must walk through it explicitly here.
//
// Only PARTITION_VALUE is allowed to cross a filter. COUNT would return the raw row count
// of each file (ignoring the predicate) and MIN_MAX is derived from zone maps, so neither
// stays correct once an unapplied predicate sits above the scan. PARTITION_VALUE is safe
// because the emitted rows carry the partition column values that the filter re-evaluates.
RuleType.STORAGE_LAYER_PARTITION_VALUE_WITH_FILTER_FOR_FILE_SCAN.build(
logicalAggregate(
logicalFilter(
logicalFileScan()
)
).when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
.thenApply(ctx -> {
LogicalAggregate<LogicalFilter<LogicalFileScan>> agg = ctx.root;
LogicalFilter<LogicalFileScan> filter = agg.child();
return partitionValueThroughFilter(
agg, null, filter, filter.child(), ctx.cascadesContext);
})
),
RuleType.STORAGE_LAYER_PARTITION_VALUE_WITH_PROJECT_FILTER_FOR_FILE_SCAN.build(
logicalAggregate(
logicalProject(
logicalFilter(
logicalFileScan()
)
)
).when(agg -> agg.isNormalized() && enablePushDownNoGroupAgg())
.thenApply(ctx -> {
LogicalAggregate<LogicalProject<LogicalFilter<LogicalFileScan>>> agg = ctx.root;
LogicalProject<LogicalFilter<LogicalFileScan>> project = agg.child();
LogicalFilter<LogicalFileScan> filter = project.child();
return partitionValueThroughFilter(
agg, project, filter, filter.child(), ctx.cascadesContext);
})
)
);
}
/*
* select 66 from baseall_dup; could use pushAggOp=COUNT to not scan real data.
*/
private LogicalProject<? extends Plan> pushDownCountWithoutSlotRef(
LogicalProject<? extends Plan> project,
LogicalOlapScan logicalScan,
CascadesContext cascadesContext) {
final LogicalProject<? extends Plan> canNotPush = project;
if (!enablePushDownNoGroupAgg()) {
return canNotPush;
}
if (logicalScan != null) {
KeysType keysType = logicalScan.getTable().getKeysType();
if (keysType != KeysType.DUP_KEYS) {
return canNotPush;
}
}
for (Expression e : project.getProjects()) {
if (e.anyMatch(SlotReference.class::isInstance)) {
return canNotPush;
}
}
PhysicalOlapScan physicalOlapScan
= (PhysicalOlapScan) new LogicalOlapScanToPhysicalOlapScan()
.build()
.transform(logicalScan, cascadesContext)
.get(0);
return project.withChildren(ImmutableList.of(new PhysicalStorageLayerAggregate(
physicalOlapScan, PushDownAggOp.COUNT)));
}
private boolean enablePushDownMinMaxOnUnique() {
ConnectContext connectContext = ConnectContext.get();
return connectContext != null && connectContext.getSessionVariable().isEnablePushDownMinMaxOnUnique();
}
private boolean isUniqueKeyTable(LogicalOlapScan logicalScan) {
if (logicalScan != null) {
KeysType keysType = logicalScan.getTable().getKeysType();
return keysType == KeysType.UNIQUE_KEYS;
}
return false;
}
private boolean enablePushDownCountOnIndex() {
ConnectContext connectContext = ConnectContext.get();
return connectContext != null && connectContext.getSessionVariable().isEnablePushDownCountOnIndex();
}
private boolean checkSlotInOrExpression(Expression expr, Set<Slot> aggSlots) {
if (expr instanceof Or) {
Set<Slot> slots = expr.getInputSlots();
if (!slots.stream().allMatch(aggSlots::contains)) {
return false;
}
} else {
for (Expression child : expr.children()) {
if (!checkSlotInOrExpression(child, aggSlots)) {
return false;
}
}
}
return true;
}
private boolean checkIsNullExpr(Expression expr, Set<Slot> aggSlots) {
if (expr instanceof IsNull) {
Set<Slot> slots = expr.getInputSlots();
if (slots.stream().anyMatch(aggSlots::contains)) {
return false;
}
} else {
for (Expression child : expr.children()) {
if (!checkIsNullExpr(child, aggSlots)) {
return false;
}
}
}
return true;
}
private boolean isDupOrMowKeyTable(LogicalOlapScan logicalScan) {
if (logicalScan != null) {
KeysType keysType = logicalScan.getTable().getKeysType();
return (keysType == KeysType.DUP_KEYS)
|| (keysType == KeysType.UNIQUE_KEYS && logicalScan.getTable().getEnableUniqueKeyMergeOnWrite());
}
return false;
}
private boolean isInvertedIndexEnabledOnTable(LogicalOlapScan logicalScan) {
if (logicalScan == null) {
return false;
}
OlapTable olapTable = logicalScan.getTable();
Map<Long, MaterializedIndexMeta> indexIdToMeta = olapTable.getIndexIdToMeta();
for (MaterializedIndexMeta indexMeta : indexIdToMeta.values()) {
for (Index index : indexMeta.getIndexes()) {
IndexType indexType = index.getIndexType();
switch (indexType) {
case INVERTED:
case BITMAP:
return true;
default: {
}
}
}
}
return false;
}
/**
* sql: select count(*) from tbl where column match 'token'
* <p>
* before:
* <p>
* LogicalAggregate(groupBy=[], output=[count(*)])
* |
* LogicalFilter(column match 'token')
* |
* LogicalOlapScan(table=tbl)
* <p>
* after:
* <p>
* LogicalAggregate(groupBy=[], output=[count(*)])
* |
* LogicalFilter(column match 'token')
* |
* PhysicalStorageLayerAggregate(pushAggOp=COUNT_ON_INDEX, table=PhysicalOlapScan(table=tbl))
*
*/
private LogicalAggregate<? extends Plan> pushdownCountOnIndex(
LogicalAggregate<? extends Plan> agg,
@Nullable LogicalProject<? extends Plan> project,
LogicalFilter<? extends Plan> filter,
LogicalOlapScan olapScan,
CascadesContext cascadesContext) {
PhysicalOlapScan physicalOlapScan = (PhysicalOlapScan) new LogicalOlapScanToPhysicalOlapScan()
.build()
.transform(olapScan, cascadesContext)
.get(0);
List<Expression> argumentsOfAggregateFunction = normalizeArguments(agg.getAggregateFunctions(), project);
if (!onlyContainsSlotOrLiteral(argumentsOfAggregateFunction)) {
return agg;
}
return agg.withChildren(ImmutableList.of(
project != null
? project.withChildren(ImmutableList.of(
filter.withChildren(ImmutableList.of(
new PhysicalStorageLayerAggregate(
physicalOlapScan, PushDownAggOp.COUNT_ON_MATCH)))))
: filter.withChildren(ImmutableList.of(
new PhysicalStorageLayerAggregate(
physicalOlapScan, PushDownAggOp.COUNT_ON_MATCH)))
));
}
private List<Expression> normalizeArguments(Set<AggregateFunction> aggregateFunctions,
@Nullable LogicalProject<? extends Plan> project) {
List<Expression> arguments = aggregateFunctions.stream()
.flatMap(aggregateFunction -> aggregateFunction.getArguments().stream())
.collect(ImmutableList.toImmutableList());
if (project != null) {
arguments = Project.findProject(arguments, project.getProjects())
.stream()
.map(p -> p instanceof Alias ? p.child(0) : p)
.collect(ImmutableList.toImmutableList());
}
return arguments;
}
private boolean onlyContainsSlotOrLiteral(List<Expression> arguments) {
return arguments.stream().allMatch(argument -> {
if (argument instanceof SlotReference || argument instanceof Literal) {
return true;
}
return false;
});
}
//select /*+SET_VAR(enable_pushdown_minmax_on_unique=true) */min(user_id) from table_unique;
//push pushAggOp=MINMAX to scan node
private LogicalAggregate<? extends Plan> pushdownMinMaxOnUniqueTable(
LogicalAggregate<? extends Plan> aggregate,
@Nullable LogicalProject<? extends Plan> project,
LogicalFilter<? extends Plan> filter,
LogicalOlapScan olapScan,
CascadesContext cascadesContext) {
final LogicalAggregate<? extends Plan> canNotPush = aggregate;
Set<AggregateFunction> aggregateFunctions = aggregate.getAggregateFunctions();
if (checkWhetherPushDownMinMax(aggregateFunctions, project, olapScan.getOutput())) {
PhysicalOlapScan physicalOlapScan = (PhysicalOlapScan) new LogicalOlapScanToPhysicalOlapScan()
.build()
.transform(olapScan, cascadesContext)
.get(0);
if (project != null) {
return aggregate.withChildren(ImmutableList.of(
project.withChildren(ImmutableList.of(
filter.withChildren(ImmutableList.of(
new PhysicalStorageLayerAggregate(
physicalOlapScan,
PushDownAggOp.MIN_MAX)))))));
} else {
return aggregate.withChildren(ImmutableList.of(
filter.withChildren(ImmutableList.of(
new PhysicalStorageLayerAggregate(
physicalOlapScan,
PushDownAggOp.MIN_MAX)))));
}
} else {
return canNotPush;
}
}
private boolean checkWhetherPushDownMinMax(Set<AggregateFunction> aggregateFunctions,
@Nullable LogicalProject<? extends Plan> project, List<Slot> outPutSlots) {
boolean onlyContainsSlotOrNumericCastSlot = aggregateFunctions.stream()
.map(ExpressionTrait::getArguments)
.flatMap(List::stream)
.allMatch(argument -> argument instanceof SlotReference);
if (!onlyContainsSlotOrNumericCastSlot) {
return false;
}
List<Expression> argumentsOfAggregateFunction = aggregateFunctions.stream()
.flatMap(aggregateFunction -> aggregateFunction.getArguments().stream())
.collect(ImmutableList.toImmutableList());
if (project != null) {
argumentsOfAggregateFunction = Project.findProject(
argumentsOfAggregateFunction, project.getProjects())
.stream()
.map(p -> p instanceof Alias ? p.child(0) : p)
.collect(ImmutableList.toImmutableList());
}
onlyContainsSlotOrNumericCastSlot = argumentsOfAggregateFunction
.stream()
.allMatch(argument -> argument instanceof SlotReference);
if (!onlyContainsSlotOrNumericCastSlot) {
return false;
}
Set<SlotReference> aggUsedSlots = ExpressionUtils.collect(argumentsOfAggregateFunction,
SlotReference.class::isInstance);
List<SlotReference> usedSlotInTable = (List<SlotReference>) Project.findProject(aggUsedSlots, outPutSlots);
for (SlotReference slot : usedSlotInTable) {
Optional<Column> optionalColumn = slot.getOriginalColumn();
if (!optionalColumn.isPresent()) {
return false;
}
Column column = optionalColumn.get();
PrimitiveType colType = column.getType().getPrimitiveType();
if (colType.isComplexType() || colType.isHllType() || colType.isBitmapType()) {
return false;
}
}
return true;
}
/**
* sql: select count(*) from tbl
* <p>
* before:
* <p>
* LogicalAggregate(groupBy=[], output=[count(*)])
* |
* LogicalOlapScan(table=tbl)
* <p>
* after:
* <p>
* LogicalAggregate(groupBy=[], output=[count(*)])
* |
* PhysicalStorageLayerAggregate(pushAggOp=COUNT, table=PhysicalOlapScan(table=tbl))
*
*/
private LogicalAggregate<? extends Plan> storageLayerAggregate(
LogicalAggregate<? extends Plan> aggregate,
@Nullable LogicalProject<? extends Plan> project,
LogicalRelation logicalScan, CascadesContext cascadesContext) {
final LogicalAggregate<? extends Plan> canNotPush = aggregate;
if (!(logicalScan instanceof LogicalOlapScan) && !(logicalScan instanceof LogicalFileScan)) {
return canNotPush;
}
if (logicalScan instanceof LogicalOlapScan) {
KeysType keysType = ((LogicalOlapScan) logicalScan).getTable().getKeysType();
if (keysType != KeysType.AGG_KEYS && keysType != KeysType.DUP_KEYS) {
return canNotPush;
}
}
List<Expression> groupByExpressions = aggregate.getGroupByExpressions();
if (logicalScan instanceof LogicalFileScan
&& canUsePartitionValueOnly(aggregate, project, null, (LogicalFileScan) logicalScan)) {
PhysicalFileScan physicalScan = toPhysicalFileScan(
(LogicalFileScan) logicalScan, cascadesContext);
PhysicalStorageLayerAggregate storageLayerAgg = new PhysicalStorageLayerAggregate(
physicalScan, PushDownAggOp.PARTITION_VALUE);
if (project != null) {
return aggregate.withChildren(ImmutableList.of(
project.withChildren(ImmutableList.of(storageLayerAgg))));
} else {
return aggregate.withChildren(ImmutableList.of(storageLayerAgg));
}
}
if (!groupByExpressions.isEmpty() || !aggregate.getDistinctArguments().isEmpty()) {
return canNotPush;
}
Set<AggregateFunction> aggregateFunctions = aggregate.getAggregateFunctions();
// Use for loop to replace Stream API
Set<Class<? extends AggregateFunction>> functionClasses = new HashSet<>();
Map<Class<? extends AggregateFunction>, PushDownAggOp> supportedAgg = PushDownAggOp.supportedFunctions();
boolean containsCount = false;
boolean containsCountStar = false;
boolean countHasCastArgument = false;
Set<SlotReference> checkNullSlots = new HashSet<>();
Set<Expression> expressionAfterProject = new HashSet<>();
// Single loop through aggregateFunctions to handle multiple logic
for (AggregateFunction function : aggregateFunctions) {
Class<? extends AggregateFunction> functionClass = function.getClass();
functionClasses.add(functionClass);
// Check if any function has arity > 1
if (function.arity() > 1) {
return canNotPush;
}
// Check if contains Count function
if (functionClass.equals(Count.class)) {
containsCount = true;
if (function.getArguments().isEmpty()) {
containsCountStar = true;
} else {
Expression arg0 = function.getArguments().get(0);
if (arg0 instanceof SlotReference) {
checkNullSlots.add((SlotReference) arg0);
expressionAfterProject.add(arg0);
} else if (arg0 instanceof Cast) {
countHasCastArgument = true;
Expression child0 = arg0.child(0);
if (child0 instanceof SlotReference) {
checkNullSlots.add((SlotReference) child0);
expressionAfterProject.add(arg0);
}
}
}
}
// Check if function is supported by supportedAgg
if (!supportedAgg.containsKey(functionClass)) {
return canNotPush;
}
}
if (logicalScan instanceof LogicalOlapScan) {
LogicalOlapScan logicalOlapScan = (LogicalOlapScan) logicalScan;
KeysType keysType = logicalOlapScan.getTable().getKeysType();
if (containsCount && keysType != KeysType.DUP_KEYS) {
return canNotPush;
}
if (containsCount && logicalOlapScan.isDirectMvScan()) {
return canNotPush;
}
}
// TODO: refactor this to process slot reference or expression together
boolean onlyContainsSlotOrNumericCastSlot = aggregateFunctions.stream()
.map(ExpressionTrait::getArguments)
.flatMap(List::stream)
.allMatch(argument -> {
if (argument instanceof SlotReference) {
return true;
}
if (argument instanceof Cast) {
return argument.child(0) instanceof SlotReference
&& argument.getDataType().isNumericType()
&& argument.child(0).getDataType().isNumericType();
}
return false;
});
if (!onlyContainsSlotOrNumericCastSlot) {
return canNotPush;
}
// we already normalize the arguments to slotReference
List<Expression> argumentsOfAggregateFunction = aggregateFunctions.stream()
.flatMap(aggregateFunction -> aggregateFunction.getArguments().stream())
.collect(ImmutableList.toImmutableList());
if (project != null) {
List<Expression> processedExpressions = new ArrayList<>();
List<? extends Expression> projections = Project.findProject(argumentsOfAggregateFunction,
project.getProjects());
for (int i = 0, size = projections.size(); i < size; i++) {
// Process the expression (replace Alias with its child)
boolean needCheckSlotNull = expressionAfterProject.contains(argumentsOfAggregateFunction.get(i));
Expression p = projections.get(i);
Expression argument = p instanceof Alias ? p.child(0) : p;
processedExpressions.add(argument);
// Check if the argument matches the required pattern
if (argument instanceof SlotReference) {
// Argument is valid, continue
if (needCheckSlotNull) {
checkNullSlots.add((SlotReference) argument);
}
} else if (argument instanceof Cast) {
boolean castMatch = argument.child(0) instanceof SlotReference
&& argument.getDataType().isNumericType()
&& argument.child(0).getDataType().isNumericType();
if (!castMatch) {
return canNotPush;
} else {
if (needCheckSlotNull) {
countHasCastArgument = true;
checkNullSlots.add((SlotReference) argument.child(0));
}
}
} else {
return canNotPush;
}
}
argumentsOfAggregateFunction = processedExpressions;
}
// File aggregate metadata can describe COUNT(*) or COUNT(file_column), but it cannot
// describe the CAST wrapped around a COUNT argument. Dropping that CAST is incorrect even
// when the source column is NOT NULL. For example, a non-null DOUBLE value outside the INT
// range becomes NULL for CAST(double_col AS INT), so COUNT(CAST(double_col AS INT)) must
// exclude it while a footer-level COUNT(double_col) would include it. Keep OLAP's existing
// storage-layer behavior unchanged, and make external files evaluate the CAST normally.
if (logicalScan instanceof LogicalFileScan && countHasCastArgument) {
return canNotPush;
}
// File footers and OLAP zone maps retain only source endpoints. Casts that introduce NULL
// can discard a valid interior value. Check the cast independently of source nullability
// so safe widening casts over nullable columns remain eligible. Floating sources may have
// NaNs omitted by file statistics; DOUBLE/DECIMAL-to-FLOAT can also underflow to signed
// zero and change the MIN/MAX representative even without introducing NULL.
if ((functionClasses.contains(Min.class) || functionClasses.contains(Max.class))
&& argumentsOfAggregateFunction.stream().anyMatch(argument -> argument instanceof Cast
&& (Cast.castNullable(false, argument.child(0).getDataType(), argument.getDataType())
|| argument.child(0).getDataType().isFloatLikeType()
|| (argument.child(0).getDataType().isDecimalLikeType()
&& argument.getDataType().isFloatType())))) {
return canNotPush;
}
Set<PushDownAggOp> pushDownAggOps = functionClasses.stream()
.map(supportedAgg::get)
.collect(Collectors.toSet());
PushDownAggOp mergeOp = pushDownAggOps.size() == 1
? pushDownAggOps.iterator().next()
: PushDownAggOp.MIX;
Set<SlotReference> aggUsedSlots =
ExpressionUtils.collect(argumentsOfAggregateFunction, SlotReference.class::isInstance);
List<SlotReference> usedSlotInTable = (List<SlotReference>) Project.findProject(aggUsedSlots,
logicalScan.getOutput());
// COUNT(*) has no aggregate arguments, even though later column pruning retains one
// arbitrary scan slot. Preserve the semantic arguments here so the BE never needs to infer
// COUNT(col) from the post-pruning scan shape.
List<ExprId> countArgumentExprIds = mergeOp == PushDownAggOp.COUNT
? usedSlotInTable.stream().map(SlotReference::getExprId).collect(Collectors.toList())
: ImmutableList.of();
for (SlotReference slot : usedSlotInTable) {
Optional<Column> optionalColumn = slot.getOriginalColumn();
if (!optionalColumn.isPresent()) {
// virtual columns (e.g., generated from MATCH_ALL expressions) do not have
// an original column and cannot be pushed down to storage layer aggregate
return canNotPush;
}
Column column = optionalColumn.get();
if (column.isAggregated()) {
return canNotPush;
}
if (mergeOp == PushDownAggOp.MIN_MAX || mergeOp == PushDownAggOp.MIX) {
if (logicalScan instanceof LogicalOlapScan
&& ((LogicalOlapScan) logicalScan).getTable() instanceof RowBinlogTableWrapper
&& RowBinlogTableWrapper.isRowBinlogSyntheticColumn(column)) {
return canNotPush;
}
PrimitiveType colType = column.getType().getPrimitiveType();
if (colType.isComplexType() || colType.isHllType() || colType.isBitmapType()) {
return canNotPush;
}
}
if (mergeOp == PushDownAggOp.COUNT || mergeOp == PushDownAggOp.MIX) {
if (logicalScan instanceof LogicalFileScan && mergeOp == PushDownAggOp.COUNT
&& containsCountStar && column.isAllowNull() && checkNullSlots.contains(slot)) {
// One metadata cardinality cannot represent both COUNT(*) and the smaller
// COUNT(nullable_col); synthetic rows would make one upper aggregate wrong.
return canNotPush;
}
// Nullable file COUNT is exact only when this query is routed to FileScannerV2,
// which carries the semantic argument and counts definition levels. Gating on the
// session switch keeps V1 on its original full-column evaluation path.
boolean supportsNullableFileCount = logicalScan instanceof LogicalFileScan
&& mergeOp == PushDownAggOp.COUNT
&& cascadesContext.getConnectContext() != null
&& cascadesContext.getConnectContext().getSessionVariable().enableFileScannerV2;
if (column.isAllowNull() && checkNullSlots.contains(slot)
&& !supportsNullableFileCount) {
return canNotPush;
}
}
}
if (logicalScan instanceof LogicalOlapScan) {
PhysicalOlapScan physicalScan = (PhysicalOlapScan) new LogicalOlapScanToPhysicalOlapScan()
.build()
.transform(logicalScan, cascadesContext)
.get(0);
if (project != null) {
return aggregate.withChildren(ImmutableList.of(
project.withChildren(
ImmutableList.of(new PhysicalStorageLayerAggregate(
physicalScan, mergeOp, countArgumentExprIds)))
));
} else {
return aggregate.withChildren(ImmutableList.of(
new PhysicalStorageLayerAggregate(physicalScan, mergeOp, countArgumentExprIds)
));
}
} else if (logicalScan instanceof LogicalFileScan) {
PhysicalFileScan physicalScan =
toPhysicalFileScan((LogicalFileScan) logicalScan, cascadesContext);
if (project != null) {
return aggregate.withChildren(ImmutableList.of(
project.withChildren(
ImmutableList.of(new PhysicalStorageLayerAggregate(
physicalScan, mergeOp, countArgumentExprIds)))
));
} else {
return aggregate.withChildren(ImmutableList.of(
new PhysicalStorageLayerAggregate(physicalScan, mergeOp, countArgumentExprIds)
));
}
} else {
return canNotPush;
}
}
private boolean enablePushDownNoGroupAgg() {
ConnectContext connectContext = ConnectContext.get();
return connectContext == null || connectContext.getSessionVariable().enablePushDownNoGroupAgg();
}
/**
* Retain the partition predicate above the reduced scan. Operative slots include the filter's
* inputs, and the shared eligibility check rejects volatile predicates. Return the original
* aggregate by reference on a miss, as required by ApplyRuleJob.
*/
private Plan partitionValueThroughFilter(
LogicalAggregate<? extends Plan> aggregate,
@Nullable LogicalProject<? extends Plan> project,
LogicalFilter<? extends Plan> filter,
LogicalFileScan logicalScan,
CascadesContext cascadesContext) {
if (!canUsePartitionValueOnly(aggregate, project, filter, logicalScan)) {
return aggregate;
}
PhysicalFileScan physicalScan = toPhysicalFileScan(logicalScan, cascadesContext);
Plan storageLayerAgg = new PhysicalStorageLayerAggregate(physicalScan, PushDownAggOp.PARTITION_VALUE);
// Keep the LogicalFilter: its conjuncts are still needed and will be translated onto the
// ScanNode by PhysicalPlanTranslator#visitPhysicalFilter. This mirrors the existing
// pushdownCountOnIndex / pushdownMinMaxOnUniqueTable rules, which also return a logical
// filter wrapping a PhysicalStorageLayerAggregate.
Plan newFilter = filter.withChildren(ImmutableList.of(storageLayerAgg));
if (project != null) {
return aggregate.withChildren(ImmutableList.of(
project.withChildren(ImmutableList.of(newFilter))));
}
return aggregate.withChildren(ImmutableList.of(newFilter));
}
/**
* Shared eligibility check for every PARTITION_VALUE rewrite. The reader emits one partition row
* only when metadata proves visible nonempty input; unsupported ranges are scanned normally.
* MIN/MAX and grouping tolerate duplicates, but volatile and non-movable expressions keep cardinality.
*
* <p>Check operative scan slots rather than aggregate arguments: constant propagation can turn
* {@code max(dt)} into {@code max('2026-08-11')} while a retained filter still reads {@code dt}.
*/
private boolean canUsePartitionValueOnly(LogicalAggregate<? extends Plan> aggregate,
@Nullable LogicalProject<? extends Plan> project,
@Nullable LogicalFilter<? extends Plan> filter, LogicalFileScan logicalScan) {
if (!enablePartitionColumnValueOnly() || logicalScan.getTableSample().isPresent()) {
return false;
}
if (aggregate.getExpressions().stream().anyMatch(Expression::containsVolatileOrNoneMovableExpression)
|| (project != null && project.getExpressions().stream()
.anyMatch(Expression::containsVolatileOrNoneMovableExpression))
|| (filter != null && filter.getExpressions().stream()
.anyMatch(Expression::containsVolatileOrNoneMovableExpression))) {
return false;
}
if (filter != null && !logicalScan.getSelectedPartitions().isPruned) {
return false;
}
// This optimization supports MIN/MAX and pure grouping, not distinct aggregate functions.
if (!aggregate.getDistinctArguments().isEmpty()) {
return false;
}
Set<AggregateFunction> aggregateFunctions = aggregate.getAggregateFunctions();
// A LogicalAggregate always has at least a group by key or an aggregate function; require it
// explicitly so a degenerate aggregate never reaches the fast path.
if (aggregateFunctions.isEmpty() && aggregate.getGroupByExpressions().isEmpty()) {
return false;
}
for (AggregateFunction function : aggregateFunctions) {
if (!(function instanceof Min) && !(function instanceof Max)) {
return false;
}
}
return isAllPartitionColumns(scanOutputSlots(logicalScan), logicalScan);
}
/**
* The columns the scan actually has to read (its OPERATIVE slots), or an empty list if any of
* them is not a plain SlotReference (an empty list makes {@link #isAllPartitionColumns} bail out).
*
* <p>Must use {@code getOperativeSlots()}, NOT {@code getOutput()}: {@code getOutput()} is the
* scan's full nominal schema (e.g. {@code [id, name, dt]}) even when
* a Project above only needs {@code dt}, whereas column pruning trims the operative slots to the
* columns really materialized ({@code [dt]}). This also matches BE, whose PARTITION_VALUE fast
* path only fires when no non-partition file slot is materialized
* ({@code _file_slot_descs.empty()}). Keying off {@code getOutput()} makes the check fail for
* every table that has non-partition columns, which is the common case.
*/
private List<SlotReference> scanOutputSlots(LogicalFileScan logicalScan) {
List<Slot> operative = logicalScan.getOperativeSlots();
if (operative.isEmpty()) {
return ImmutableList.of();
}
ImmutableList.Builder<SlotReference> slots =
ImmutableList.builderWithExpectedSize(operative.size());
for (Slot slot : operative) {
if (!(slot instanceof SlotReference)) {
return ImmutableList.of();
}
slots.add((SlotReference) slot);
}
return slots.build();
}
/** Implement a LogicalFileScan into its PhysicalFileScan. */
private PhysicalFileScan toPhysicalFileScan(
LogicalFileScan logicalScan, CascadesContext cascadesContext) {
Rule rule = new LogicalFileScanToPhysicalFileScan().build();
return (PhysicalFileScan) rule.transform(logicalScan, cascadesContext).get(0);
}
private boolean enablePartitionColumnValueOnly() {
ConnectContext connectContext = ConnectContext.get();
return connectContext == null
|| connectContext.getSessionVariable().isEnablePartitionColumnValueOnlyOptimization();
}
/** Check operative slots against partition columns at this scan reference's statement snapshot. */
private boolean isAllPartitionColumns(List<SlotReference> usedSlotInTable, LogicalFileScan fileScan) {
if (usedSlotInTable.isEmpty() || !(fileScan.getTable() instanceof PluginDrivenExternalTable)) {
return false;
}
PluginDrivenExternalTable table = (PluginDrivenExternalTable) fileScan.getTable();
// The connector limits this capability to nontransactional Hive Parquet/ORC tables, not
// delegated Hudi/Iceberg/Paimon tables sharing the same PluginDrivenExternalTable class.
if (!table.supportsPartitionValueOnly()) {
return false;
}
List<Column> partitionColumns = table.getPartitionColumns(MvccUtil.getSnapshotFromContext(
table, fileScan.getTableSnapshot(), fileScan.getScanParams()));
Set<String> partitionColumnNames = new HashSet<>();
for (Column column : partitionColumns) {
partitionColumnNames.add(column.getName().toLowerCase(Locale.ROOT));
}
for (SlotReference slot : usedSlotInTable) {
Optional<Column> optionalColumn = slot.getOriginalColumn();
if (!optionalColumn.isPresent()) {
return false;
}
if (!partitionColumnNames.contains(optionalColumn.get().getName().toLowerCase(Locale.ROOT))) {
return false;
}
}
return true;
}
}