diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SetPreAggStatus.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SetPreAggStatus.java index aef5f660056d57..33b023a1f91d70 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SetPreAggStatus.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SetPreAggStatus.java @@ -21,6 +21,7 @@ import org.apache.doris.catalog.KeysType; import org.apache.doris.catalog.MaterializedIndexMeta; import org.apache.doris.common.Pair; +import org.apache.doris.nereids.exceptions.AnalysisException; import org.apache.doris.nereids.jobs.JobContext; import org.apache.doris.nereids.trees.expressions.CaseWhen; import org.apache.doris.nereids.trees.expressions.Cast; @@ -91,11 +92,45 @@ public static class PreAggInfoContext { private List groupingScalarFunctionExpresssions = new ArrayList<>(); private Set aggregateFunctions = new HashSet<>(); private Set olapScanIds = new HashSet<>(); + private boolean hasUnresolvedExpression = false; private Map replaceMap = new HashMap<>(); - private void setReplaceMap(Map replaceMap) { - this.replaceMap = replaceMap; + private void setReplaceMap(Map newReplaceMap) { + // merge instead of replace: sibling projects under a join share one + // PreAggInfoContext, and a full replacement would lose mappings from + // the sibling. merge keeps all entries; new entries shadow old ones + // so a chain of projects still resolves correctly. + // + // Before merging, resolve the new aliases' producers through the + // existing replaceMap so that upper-layer aliases reference base table + // columns directly instead of intermediate computed aliases. + // Resolve only the new entries into a small temp map against the + // unchanged old map, then putAll them to keep accumulation linear. + if (newReplaceMap.isEmpty()) { + return; + } + Map resolved = new HashMap<>( + com.google.common.collect.Maps.newHashMapWithExpectedSize( + newReplaceMap.size())); + for (Map.Entry entry : newReplaceMap.entrySet()) { + Expression resolvedProducer; + try { + resolvedProducer = ExpressionUtils.replace( + entry.getValue(), this.replaceMap); + } catch (AnalysisException e) { + if (e.getErrorCode() == AnalysisException.ErrorCode.EXPRESSION_EXCEEDS_LIMIT) { + // Eager composition hit the depth/width limit (e.g. deep + // xN = x(N-1) + x(N-1) chains expand width exponentially). + // Keep the raw producer so the query still plans. + resolvedProducer = entry.getValue(); + } else { + throw e; + } + } + resolved.put(entry.getKey(), resolvedProducer); + } + this.replaceMap.putAll(resolved); } private void addRelationId(RelationId id) { @@ -104,18 +139,45 @@ private void addRelationId(RelationId id) { private void addJoinInfo(LogicalJoin logicalJoin) { joinConjuncts.addAll(logicalJoin.getExpressions()); - joinConjuncts = Lists.newArrayList(ExpressionUtils.replace(joinConjuncts, replaceMap)); + try { + joinConjuncts = Lists.newArrayList( + ExpressionUtils.replace(joinConjuncts, replaceMap)); + } catch (AnalysisException e) { + if (e.getErrorCode() == AnalysisException.ErrorCode.EXPRESSION_EXCEEDS_LIMIT) { + hasUnresolvedExpression = true; + } else { + throw e; + } + } } private void addFilterConjuncts(List conjuncts) { filterConjuncts.addAll(conjuncts); - filterConjuncts = Lists.newArrayList(ExpressionUtils.replace(filterConjuncts, replaceMap)); + try { + filterConjuncts = Lists.newArrayList( + ExpressionUtils.replace(filterConjuncts, replaceMap)); + } catch (AnalysisException e) { + if (e.getErrorCode() == AnalysisException.ErrorCode.EXPRESSION_EXCEEDS_LIMIT) { + hasUnresolvedExpression = true; + } else { + throw e; + } + } } private void addGroupByExpresssions(List expressions) { groupByExpresssions.addAll(expressions); groupByExpresssions.removeAll(groupingScalarFunctionExpresssions); - groupByExpresssions = Lists.newArrayList(ExpressionUtils.replace(groupByExpresssions, replaceMap)); + try { + groupByExpresssions = Lists.newArrayList( + ExpressionUtils.replace(groupByExpresssions, replaceMap)); + } catch (AnalysisException e) { + if (e.getErrorCode() == AnalysisException.ErrorCode.EXPRESSION_EXCEEDS_LIMIT) { + hasUnresolvedExpression = true; + } else { + throw e; + } + } } private void addGroupingScalarFunctionExpresssions(List expressions) { @@ -130,8 +192,17 @@ private void addAggregateFunctions(Set functions) { aggregateFunctions.addAll(functions); Set newAggregateFunctions = Sets.newHashSet(); for (AggregateFunction aggregateFunction : aggregateFunctions) { - newAggregateFunctions - .add((AggregateFunction) ExpressionUtils.replace(aggregateFunction, replaceMap)); + try { + newAggregateFunctions + .add((AggregateFunction) ExpressionUtils.replace( + aggregateFunction, replaceMap)); + } catch (AnalysisException e) { + if (e.getErrorCode() == AnalysisException.ErrorCode.EXPRESSION_EXCEEDS_LIMIT) { + hasUnresolvedExpression = true; + } else { + throw e; + } + } } aggregateFunctions = newAggregateFunctions; } @@ -265,6 +336,9 @@ public Plan visitLogicalOlapScan(LogicalOlapScan olapScan, Map filterConjuncts = context.filterConjuncts; List joinConjuncts = context.joinConjuncts; Set aggregateFuncs = context.aggregateFunctions; @@ -294,6 +368,41 @@ private PreAggStatus createPreAggStatus(LogicalOlapScan logicalOlapScan, PreAggI return PreAggStatus.off(String.format("Join conjuncts %s contains non-key column %s", joinConjuncts, joinInputSlots)); } + + // Row-stability check: volatile expressions evaluated per partial row + // produce different results than per merged logical row, even when + // their input slots are all key columns or empty. Check centrally + // before per-scan candidate filtering so the guard also covers + // other-table aggregates, slot-less filters, joins, and grouping. + for (AggregateFunction aggFunc : aggregateFuncs) { + if (aggFunc.containsVolatileExpression()) { + return PreAggStatus.off( + String.format("aggregate function %s contains volatile expression", + aggFunc)); + } + } + for (Expression conjunct : filterConjuncts) { + if (conjunct.containsVolatileExpression()) { + return PreAggStatus.off( + String.format("filter conjunct %s contains volatile expression", + conjunct)); + } + } + for (Expression conjunct : joinConjuncts) { + if (conjunct.containsVolatileExpression()) { + return PreAggStatus.off( + String.format("join conjunct %s contains volatile expression", + conjunct)); + } + } + for (Expression expr : groupingExprs) { + if (expr.containsVolatileExpression()) { + return PreAggStatus.off( + String.format("grouping expression %s contains volatile expression", + expr)); + } + } + Set candidateAggFuncs = Sets.newHashSet(); for (AggregateFunction aggregateFunction : aggregateFuncs) { if (!Sets.intersection(aggregateFunction.getInputSlots(), outputSlots).isEmpty()) { @@ -315,39 +424,53 @@ private PreAggStatus createPreAggStatus(LogicalOlapScan logicalOlapScan, PreAggI return !aggregateFuncs.isEmpty() || !groupingExprs.isEmpty() ? PreAggStatus.on() : PreAggStatus.off("No aggregate on scan."); } else { - return checkAggregateFunctions(candidateAggFuncs, candidateGroupByInputSlots); + return checkAggregateFunctions(candidateAggFuncs, candidateGroupByInputSlots, outputSlots); } } private PreAggStatus checkAggregateFunctions(Set aggregateFuncs, - Set groupingExprsInputSlots) { + Set groupingExprsInputSlots, Set outputSlots) { if (aggregateFuncs.isEmpty() && groupingExprsInputSlots.isEmpty()) { return PreAggStatus.off("No aggregate on scan."); } PreAggStatus preAggStatus = PreAggStatus.on(); for (AggregateFunction aggFunc : aggregateFuncs) { - if (aggFunc.children().isEmpty()) { + Set aggSlots = Sets.intersection( + aggFunc.getInputSlots(), outputSlots); + if (aggSlots.isEmpty()) { preAggStatus = PreAggStatus.off( String.format("can't turn preAgg on for aggregate function %s", aggFunc)); - } else if (aggFunc.children().size() == 1 && aggFunc.child(0) instanceof Slot) { - Slot aggSlot = (Slot) aggFunc.child(0); - if (aggSlot instanceof SlotReference - && ((SlotReference) aggSlot).getOriginalColumn().isPresent()) { - if (((SlotReference) aggSlot).getOriginalColumn().get().isKey()) { - preAggStatus = OneKeySlotAggChecker.INSTANCE.check(aggFunc); + } else { + Pair, Set> splitSlots = splitKeyValueSlots(aggSlots); + if (splitSlots.first.isEmpty()) { + // only value slots + if (aggFunc.children().size() == 1 && aggFunc.child(0) instanceof SlotReference) { + SlotReference slotRef = (SlotReference) aggFunc.child(0); + if (slotRef.getOriginalColumn().isPresent()) { + preAggStatus = OneValueSlotAggChecker.INSTANCE.check(aggFunc, + slotRef.getOriginalColumn().get().getAggregationType()); + } else { + preAggStatus = PreAggStatus.off( + String.format("can't turn preAgg on for aggregate function %s", aggFunc)); + } } else { - preAggStatus = OneValueSlotAggChecker.INSTANCE.check(aggFunc, - ((SlotReference) aggSlot).getOriginalColumn().get().getAggregationType()); + preAggStatus = PreAggStatus.off( + String.format("can't turn preAgg on for aggregate function %s", aggFunc)); } + } else if (splitSlots.second.isEmpty()) { + // only key slots + preAggStatus = KeySlotAggChecker.INSTANCE.check(aggFunc); } else { - preAggStatus = PreAggStatus.off( - String.format("aggregate function %s use unknown slot %s from scan", - aggFunc, aggSlot)); + // checkAggWithKeyAndValueSlots only inspects child(0) for IF/CaseWhen patterns. + // For multi-argument aggregate functions, child(0) inspection is insufficient + // as later arguments may contain value columns that are not validated. + if (aggFunc.children().size() > 1) { + preAggStatus = PreAggStatus.off( + String.format("can't turn preAgg on for aggregate function %s", aggFunc)); + } else { + preAggStatus = checkAggWithKeyAndValueSlots(aggFunc, splitSlots.first, splitSlots.second); + } } - } else { - Set aggSlots = aggFunc.getInputSlots(); - Pair, Set> splitSlots = splitKeyValueSlots(aggSlots); - preAggStatus = checkAggWithKeyAndValueSlots(aggFunc, splitSlots.first, splitSlots.second); } if (preAggStatus.isOff()) { return preAggStatus; @@ -402,6 +525,10 @@ private PreAggStatus checkAggWithKeyAndValueSlots(AggregateFunction aggFunc, // currently, only IF and CASE WHEN are supported returnExps.add(removeCast(child)); } + if (conditionExps.isEmpty()) { + return PreAggStatus.off( + String.format("can't turn preAgg on for aggregate function %s", aggFunc)); + } // step 2: check condition expressions Set inputSlots = ExpressionUtils.getInputSlotSet(conditionExps); @@ -511,8 +638,8 @@ public PreAggStatus visitHllUnion(HllUnion hllUnion, AggregateType aggregateType } } - private static class OneKeySlotAggChecker extends ExpressionVisitor { - public static final OneKeySlotAggChecker INSTANCE = new OneKeySlotAggChecker(); + private static class KeySlotAggChecker extends ExpressionVisitor { + public static final KeySlotAggChecker INSTANCE = new KeySlotAggChecker(); public PreAggStatus check(AggregateFunction aggFun) { return aggFun.accept(INSTANCE, null); diff --git a/regression-test/suites/nereids_rules_p0/set_preagg/set_preagg.groovy b/regression-test/suites/nereids_rules_p0/set_preagg/set_preagg.groovy index f47b2b40fc261a..fd8e371f9dbc5f 100644 --- a/regression-test/suites/nereids_rules_p0/set_preagg/set_preagg.groovy +++ b/regression-test/suites/nereids_rules_p0/set_preagg/set_preagg.groovy @@ -250,7 +250,7 @@ suite("set_preagg") { order by 1, 2; """) contains "(preagg_t1), PREAGGREGATION: ON" - contains "(preagg_t2), PREAGGREGATION: OFF. Reason: count(" + contains "(preagg_t2), PREAGGREGATION: OFF. Reason: some columns in condition" contains "(preagg_t3), PREAGGREGATION: OFF. Reason: can't turn preAgg on because aggregate function sum" } @@ -349,4 +349,136 @@ suite("set_preagg") { select count(*) from (select * from numbers("number"="10")) t; """) } + + explain { + sql("""select count(distinct k6, v7) from preagg_t1;""") + notContains "(preagg_t1), PREAGGREGATION: ON" + } + + explain { + sql("""select count(distinct k6, k5) from preagg_t1;""") + contains "(preagg_t1), PREAGGREGATION: ON" + } + + // Negative: count(DISTINCT IF(...), value_col) — checkAggWithKeyAndValueSlots + // only inspects child(0) (the IF). Without a multi-arg guard, it would + // miss v7 in child(1) and incorrectly return ON. + explain { + sql(""" + select count(distinct if(k6 > 0, k5, 0), v7) from preagg_t1; + """) + notContains "(preagg_t1), PREAGGREGATION: ON" + } + + explain { + sql(""" + select count(distinct case when k6 > 0 then k5 else 0 end, v7) from preagg_t1; + """) + notContains "(preagg_t1), PREAGGREGATION: ON" + } + + // Negative: count(DISTINCT key + random()) — volatile in the expression + // argument. With pre-agg ON, random() would be evaluated per partial row + // instead of per merged logical row, changing the distinct count. + explain { + sql("""select count(distinct k6 + random()) from preagg_t1;""") + notContains "(preagg_t1), PREAGGREGATION: ON" + } + + // max/min(key + random()) have the same volatile concern. + explain { + sql("""select max(k6 + random()) from preagg_t1;""") + notContains "(preagg_t1), PREAGGREGATION: ON" + } + + explain { + sql("""select min(k6 + random()) from preagg_t1;""") + notContains "(preagg_t1), PREAGGREGATION: ON" + } + + // Volatile in an IF condition inside a mixed-key-value aggregate. + // The condition k6 + random() > 0 uses only key input slots but is + // volatile, so pre-agg must be OFF. + explain { + sql(""" + select sum(if(k6 + random() > 0, v7, 0)) from preagg_t1; + """) + notContains "(preagg_t1), PREAGGREGATION: ON" + } + + // Positive: two-project join where both aliases resolve to key-only + // expressions. With merge+resolve, x = k5 + 1 resolves fully to base + // key columns, so pre-agg can be ON for both scans. + explain { + sql(""" + select count(distinct a) + from ( + select l.k1 + l.x as a + from ( + select t1.k1, t1.k5 + 1 as x from preagg_t1 t1 + ) l + inner join ( + select abs(t2.k1) as rk from preagg_t2 t2 + ) r on l.k1 = r.rk + ) t; + """) + contains "(preagg_t1), PREAGGREGATION: ON" + contains "(preagg_t2), PREAGGREGATION: ON" + } + + // Negative: two-project join; x carries v7 + 1 (a value column). Even + // with merge+resolve, the fully resolved expression contains v7 so + // pre-agg is correctly OFF. + explain { + sql(""" + select count(distinct a) + from ( + select l.k1 + l.x as a + from ( + select t1.k1, t1.v7 + 1 as x from preagg_t1 t1 + ) l + inner join ( + select abs(t2.k1) as rk from preagg_t2 t2 + ) r on l.k1 = r.rk + ) t; + """) + notContains "(preagg_t1), PREAGGREGATION: ON" + } + + // Bypass 1: volatile in an other-table aggregate function. max(r.k1 + random()) + // is whitelisted as a duplicate-insensitive MAX for scan l; the candidate set + // for l is empty, so without a central volatile check it returns ON before + // reaching the per-function guard. Both scans must be OFF. + explain { + sql(""" + select max(r.k1 + random()) + from preagg_t1 l + inner join preagg_t2 r on l.k1 = r.k1; + """) + notContains "(preagg_t1), PREAGGREGATION: ON" + notContains "(preagg_t2), PREAGGREGATION: ON" + } + + // Bypass 2: volatile filter with no input slots. random() < 0.5 has an + // empty input-slot set, so the slot-based value-column check bypasses it. + // The central volatile guard must reject pre-agg on this scan. + explain { + sql(""" + select sum(v7) from preagg_t1 where random() < 0.5; + """) + notContains "(preagg_t1), PREAGGREGATION: ON" + } + + // Foreign value column in mixed aggregate: sum(if(abs(l.k1) > 0, r.v7, 0)) + // references l.k1 (local key) and r.v7 (foreign value). The mixed helper + // must not use r.v7's SUM type to justify pre-agg on l — r.v7 is not + // a column of l. l must be OFF; r can be ON. + explain { + sql(""" + select sum(if(t.a > 0, r.v7, 0)) + from (select abs(k1) as a from preagg_t1) t + inner join preagg_t2 r on t.a = r.k1; + """) + notContains "(preagg_t1), PREAGGREGATION: ON" + } }