From 9d7cd4df7a064328362f62ae32e6a5228afb9b8d Mon Sep 17 00:00:00 2001 From: morrySnow Date: Wed, 9 Sep 2026 18:38:08 +0800 Subject: [PATCH] [fix](aggregate) Guard cast aggregate pushdown by nullability Storage-layer aggregation operates on source column values, so a cast that can introduce NULL must not be pushed through the scan aggregate. Use the cast type pair to detect conversion-induced nullability while retaining pushdown for safe numeric casts. Cover direct, projected, OLAP, and file-scan paths for CAST and TRY_CAST.\n\nIssue Number: None --- .../implementation/AggregateStrategies.java | 66 ++++------- .../PhysicalStorageLayerAggregateTest.java | 107 ++++++++++++++++++ 2 files changed, 130 insertions(+), 43 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java index 906ba73052b0d9..52b9502a9a0031 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java @@ -571,7 +571,6 @@ private LogicalAggregate storageLayerAggregate( boolean containsCount = false; boolean containsCountStar = false; - boolean countHasCastArgument = false; Set checkNullSlots = new HashSet<>(); Set expressionAfterProject = new HashSet<>(); @@ -594,13 +593,9 @@ private LogicalAggregate storageLayerAggregate( 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); - } + } else if (isSupportedStorageLayerAggregateArgument(arg0)) { + checkNullSlots.add((SlotReference) arg0.child(0)); + expressionAfterProject.add(arg0); } } } @@ -622,22 +617,13 @@ private LogicalAggregate storageLayerAggregate( } } - // TODO: refactor this to process slot reference or expression together - boolean onlyContainsSlotOrNumericCastSlot = aggregateFunctions.stream() + // Storage-layer aggregation operates on source column values. It is safe to push through + // a numeric cast only when the cast cannot introduce NULL for a non-null source value. + boolean onlyContainsSupportedArgument = 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) { + .allMatch(this::isSupportedStorageLayerAggregateArgument); + if (!onlyContainsSupportedArgument) { return canNotPush; } @@ -664,17 +650,9 @@ private LogicalAggregate storageLayerAggregate( 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 if (isSupportedStorageLayerAggregateArgument(argument)) { + if (needCheckSlotNull) { + checkNullSlots.add((SlotReference) argument.child(0)); } } else { return canNotPush; @@ -683,16 +661,6 @@ private LogicalAggregate storageLayerAggregate( 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; - } - Set pushDownAggOps = functionClasses.stream() .map(supportedAgg::get) .collect(Collectors.toSet()); @@ -806,6 +774,18 @@ private boolean enablePushDownStringMinMax() { return connectContext != null && connectContext.getSessionVariable().isEnablePushDownStringMinMax(); } + private boolean isSupportedStorageLayerAggregateArgument(Expression argument) { + if (argument instanceof SlotReference) { + return true; + } + if (!(argument instanceof Cast) || !(argument.child(0) instanceof SlotReference) + || !argument.getDataType().isNumericType() + || !argument.child(0).getDataType().isNumericType()) { + return false; + } + return !Cast.castNullable(false, argument.child(0).getDataType(), argument.getDataType()); + } + private boolean enablePushDownNoGroupAgg() { ConnectContext connectContext = ConnectContext.get(); return connectContext == null || connectContext.getSessionVariable().enablePushDownNoGroupAgg(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PhysicalStorageLayerAggregateTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PhysicalStorageLayerAggregateTest.java index 74708104e35357..15c245ea33918e 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PhysicalStorageLayerAggregateTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PhysicalStorageLayerAggregateTest.java @@ -31,7 +31,10 @@ import org.apache.doris.nereids.rules.RuleType; import org.apache.doris.nereids.rules.implementation.AggregateStrategies; import org.apache.doris.nereids.trees.expressions.Alias; +import org.apache.doris.nereids.trees.expressions.Cast; import org.apache.doris.nereids.trees.expressions.IsNull; +import org.apache.doris.nereids.trees.expressions.TryCast; +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; @@ -44,6 +47,8 @@ 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.physical.PhysicalStorageLayerAggregate.PushDownAggOp; +import org.apache.doris.nereids.types.BigIntType; +import org.apache.doris.nereids.types.TinyIntType; import org.apache.doris.nereids.util.MemoPatternMatchSupported; import org.apache.doris.nereids.util.MemoTestUtils; import org.apache.doris.nereids.util.PlanChecker; @@ -55,6 +60,7 @@ import org.mockito.Mockito; import java.util.Collections; +import java.util.List; import java.util.Optional; public class PhysicalStorageLayerAggregateTest implements MemoPatternMatchSupported { @@ -143,6 +149,35 @@ public void testNullableFileCountUsesStorageLayerAggregate() { ImmutableList.of(fileScan.getOutput().get(0).getExprId()))))); } + @Test + public void testFileCountCastNullabilityControlsStorageLayerAggregate() { + LogicalFileScan fileScan = newNullableFileCountAggregate().child(); + List unsafeCasts = ImmutableList.of( + new Cast(fileScan.getOutput().get(0), TinyIntType.INSTANCE), + new TryCast(fileScan.getOutput().get(0), TinyIntType.INSTANCE)); + for (Cast cast : unsafeCasts) { + LogicalAggregate aggregate = new LogicalAggregate<>( + Collections.emptyList(), ImmutableList.of(new Alias(new Count(cast), "count")), + true, Optional.empty(), fileScan); + PlanChecker.from(MemoTestUtils.createCascadesContext(aggregate)) + .applyImplementation(storageLayerAggregateWithoutProjectForFileScan()) + .nonMatch(physicalStorageLayerAggregate()); + } + + List safeCasts = ImmutableList.of( + new Cast(fileScan.getOutput().get(0), BigIntType.INSTANCE), + new TryCast(fileScan.getOutput().get(0), BigIntType.INSTANCE)); + for (Cast cast : safeCasts) { + LogicalAggregate aggregate = new LogicalAggregate<>( + Collections.emptyList(), ImmutableList.of(new Alias(new Count(cast), "count")), + true, Optional.empty(), fileScan); + PlanChecker.from(MemoTestUtils.createCascadesContext(aggregate)) + .applyImplementation(storageLayerAggregateWithoutProjectForFileScan()) + .matches(logicalAggregate( + physicalStorageLayerAggregate().when(agg -> agg.getAggOp() == PushDownAggOp.COUNT))); + } + } + @Test public void testNullableFileCountDoesNotUseV1StorageLayerAggregate() { LogicalAggregate aggregate = newNullableFileCountAggregate(); @@ -289,6 +324,78 @@ public void testCountOnIndexRejectsIsNullOnProjectedCountSlot() { .matches(logicalAggregate(logicalProject(logicalFilter(logicalOlapScan())))); } + @Test + public void testCastThatMayProduceNullDoesNotUseStorageLayerAggregate() { + LogicalOlapScan olapScan = PlanConstructor.newLogicalOlapScan(2, "cast_aggregate", 0); + Cast cast = new Cast(olapScan.getOutput().get(0), TinyIntType.INSTANCE); + TryCast tryCast = new TryCast(olapScan.getOutput().get(0), TinyIntType.INSTANCE); + List castAggregates = ImmutableList.of( + new Count(cast), new Count(tryCast), new Min(cast), new Max(tryCast)); + + for (AggregateFunction function : castAggregates) { + LogicalAggregate aggregate = new LogicalAggregate<>( + Collections.emptyList(), ImmutableList.of(new Alias(function, "aggregate")), + true, Optional.empty(), olapScan); + + PlanChecker.from(MemoTestUtils.createCascadesContext(aggregate)) + .applyImplementation(storageLayerAggregateWithoutProject()) + .matches(logicalAggregate(logicalOlapScan())); + } + } + + @Test + public void testSafeCastAggregateUsesStorageLayerAggregate() { + LogicalOlapScan olapScan = PlanConstructor.newLogicalOlapScan(3, "safe_cast_aggregate", 0); + Cast cast = new Cast(olapScan.getOutput().get(0), BigIntType.INSTANCE); + TryCast tryCast = new TryCast(olapScan.getOutput().get(0), BigIntType.INSTANCE); + List castAggregates = ImmutableList.of( + new Count(cast), new Count(tryCast), new Min(cast), new Max(tryCast)); + + for (AggregateFunction function : castAggregates) { + LogicalAggregate aggregate = new LogicalAggregate<>( + Collections.emptyList(), ImmutableList.of(new Alias(function, "aggregate")), + true, Optional.empty(), olapScan); + + PlanChecker.from(MemoTestUtils.createCascadesContext(aggregate)) + .applyImplementation(storageLayerAggregateWithoutProject()) + .matches(logicalAggregate(physicalStorageLayerAggregate())); + } + } + + @Test + public void testProjectedCastThatMayProduceNullDoesNotUseStorageLayerAggregate() { + LogicalOlapScan olapScan = PlanConstructor.newLogicalOlapScan(3, "projected_cast_count", 0); + LogicalProject project = new LogicalProject<>( + ImmutableList.of(new Alias( + new Cast(olapScan.getOutput().get(0), TinyIntType.INSTANCE), "cast_value")), + olapScan); + LogicalAggregate> aggregate = new LogicalAggregate<>( + Collections.emptyList(), + ImmutableList.of(new Alias(new Count(project.getOutput().get(0)), "count")), + true, Optional.empty(), project); + + PlanChecker.from(MemoTestUtils.createCascadesContext(aggregate)) + .applyImplementation(storageLayerAggregateWithProject()) + .matches(logicalAggregate(logicalProject(logicalOlapScan()))); + } + + @Test + public void testProjectedSafeCastCountUsesStorageLayerAggregate() { + LogicalOlapScan olapScan = PlanConstructor.newLogicalOlapScan(4, "projected_safe_cast_count", 0); + LogicalProject project = new LogicalProject<>( + ImmutableList.of(new Alias( + new Cast(olapScan.getOutput().get(0), BigIntType.INSTANCE), "cast_value")), + olapScan); + LogicalAggregate> aggregate = new LogicalAggregate<>( + Collections.emptyList(), + ImmutableList.of(new Alias(new Count(project.getOutput().get(0)), "count")), + true, Optional.empty(), project); + + PlanChecker.from(MemoTestUtils.createCascadesContext(aggregate)) + .applyImplementation(storageLayerAggregateWithProject()) + .matches(logicalAggregate(logicalProject(physicalStorageLayerAggregate()))); + } + @Test void testProjectionCheck() { LogicalOlapScan olapScan = PlanConstructor.newLogicalOlapScan(1, "tbl", 0);