github-actions[bot] commented on code in PR #68936:
URL: https://github.com/apache/doris/pull/68936#discussion_r4396432848
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DecomposeRepeatWithPreAggregation.java:
##########
@@ -168,27 +222,61 @@ public Plan visitLogicalAggregate(LogicalAggregate<?
extends Plan> aggregate, Di
Set<Expression> needRemovedExprSet = getNeedAddNullExpressions(repeat,
newGroupingSets, maxGroupIndex);
Map<AggregateFunction, Slot> aggFuncToSlot = new HashMap<>();
LogicalAggregate<Plan> topAgg = constructAgg(aggregate,
originToConsumerMap, newRepeat, groupingFunctionSlots,
- aggFuncToSlot);
+ aggFuncToSlot, combiners);
LogicalProject<Plan> project = constructProject(aggregate,
originToConsumerMap, needRemovedExprSet,
groupingFunctionSlots, newRepeat.getGroupingId().get(),
topAgg, aggFuncToSlot);
- LogicalPlan directChild = getDirectChild(directConsumer,
groupingFunctionsValues, maxGroupIndex);
+ LogicalPlan directChild = getDirectChild(directConsumer,
producer.child().getOutputExpressions(),
+ groupingFunctionsValues, maxGroupIndex);
return constructUnion(project, directChild, aggregate);
}
+ private Map<AggregateFunction, CombineCombinator> constructCombiners(
+ LogicalAggregate<? extends Plan> aggregate) {
+ LogicalRepeat<?> repeat = (LogicalRepeat<?>) aggregate.child();
+ // Repeat makes grouping keys nullable. State types must use the
actual input nullability.
+ Map<Slot, Slot> inputSlots = new HashMap<>();
+ for (Slot slot : repeat.child().getOutput()) {
+ inputSlots.put(slot, slot);
+ }
+ Map<AggregateFunction, CombineCombinator> combiners = new HashMap<>();
+ for (AggregateFunction function : aggregate.getAggregateFunctions()) {
+ if (RESULT_ROLLUP_FUNCTIONS.contains(function.getClass())) {
+ continue;
+ }
+ AggregateFunction nested = (AggregateFunction)
ExpressionUtils.replace(function, inputSlots);
+ combiners.put(function, new CombineCombinator(nested.children(),
nested));
Review Comment:
[P1] Preserve Decimal256 AVG result types when introducing states
With `enable_decimal256=true` and `v DECIMAL(18,2)`, `SELECT a,b,c,AVG(v)
FROM t GROUP BY ROLLUP(a,b,c)` is now rewritten from
`Aggregate(AVG(v):DECIMAL(76,4)) -> Repeat -> input` into a producer
`AVG_COMBINE(v)` and consumers `AVG_MERGE(s)` / `AVG_FINALIZE(s)`.
`Avg.computePrecision` preserves the narrow input type while producing
DECIMAL(76,4), but this state records only that input signature.
`DataTypeAggState` reconstructs its nested AVG with precision 38 unless the
input itself is DECIMAL256. Both consumers still declare DECIMAL(76,4), so the
exact return-type checks in `vectorized_agg_fn.cpp` and
`create_agg_state_finalize_function` reject the query instead of returning its
previously valid result. Preserve the original result signature in the state
path, or exclude this configuration until it can be represented, and add a
differential decimal256 AVG execution case.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DecomposeRepeatWithPreAggregation.java:
##########
@@ -70,34 +107,43 @@
import java.util.TreeMap;
/**
- * This rule will rewrite grouping sets. eg:
- * select a, b, c, d, e sum(f) from t group by rollup(a, b, c, d, e);
- * rewrite to:
- * with cte1 as (select a, b, c, d, e, sum(f) x from t group by rollup(a,
b, c, d, e))
- * select * fom cte1
- * union all
- * select a, b, c, d, null, sum(x) x from t group by rollup(a, b, c, d)
+ * Pre-aggregate the grouping set that contains all other grouping sets and
share it through a CTE.
+ * SUM, SUM0, MIN, MAX, ANY_VALUE and COUNT use their final results for
further aggregation;
+ * COUNT results are rolled up by SUM;
+ * other eligible built-in aggregates use combine/merge/finalize to preserve
their intermediate states.
+ * Both kinds of aggregate can share the same producer, Repeat and union.
+ *
+ * For example, SUM(d), COUNT(d) and AVG(d) over ROLLUP(a,b,c) are rewritten
as:
*
- * LogicalAggregate(gby: a,b,c,d,e,grouping_id
output:a,b,c,d,e,grouping_id,sum(f))
- * +--LogicalRepeat(grouping sets:
(a,b,c,d,e),(a,b,c,d),(a,b,c),(a,b),(a),())
- * ->
* LogicalCTEAnchor
* +--LogicalCTEProducer(cte)
- * +--LogicalAggregate(gby: a,b,c,d,e; aggFunc: sum(f) as x)
+ * +--LogicalAggregate(gby: a,b,c; aggFunc: sum(d) as s, count(d) as n,
avg_combine(d) as x)
* +--LogicalUnionAll
- * +--LogicalProject(a,b,c,d, null as e, sum(x))
- * +--LogicalAggregate(gby:a,b,c,d,grouping_id; aggFunc: sum(x))
- * +--LogicalRepeat(grouping sets: (a,b,c,d),(a,b,c),(a,b),(a),())
+ * +--LogicalProject(a,b, null as c, sum_result, count_result, avg_result,
grouping_id)
+ * +--LogicalAggregate(gby: a,b,grouping_id;
+ * aggFunc: sum(s) as sum_result, sum(n) as count_result,
avg_merge(x) as avg_result)
+ * +--LogicalRepeat(grouping sets: (a,b),(a),())
* +--LogicalCTEConsumer(aggregateConsumer)
- * +--LogicalCTEConsumer(directConsumer)
+ * +--LogicalProject(a,b,c, s, n, avg_finalize(x), grouping_id constant)
+ * +--LogicalCTEConsumer(directConsumer)
+ *
+ * The direct branch needs no aggregation: it passes through ordinary results
and finalizes states
+ * introduced by this rule. Existing aggregate combinators (such as
avg_combine) are not eligible;
+ * if any original aggregate function is unsupported or DISTINCT, the entire
aggregate is left unchanged.
*/
public class DecomposeRepeatWithPreAggregation extends
DefaultPlanRewriter<DistinctSelectorContext>
implements CustomRewriter {
public static final DecomposeRepeatWithPreAggregation INSTANCE = new
DecomposeRepeatWithPreAggregation();
- private static final Set<Class<? extends AggregateFunction>>
SUPPORT_AGG_FUNCTIONS =
+ private static final Set<Class<? extends AggregateFunction>>
RESULT_ROLLUP_FUNCTIONS =
ImmutableSet.of(Sum.class, Sum0.class, Min.class, Max.class,
AnyValue.class, Count.class);
private static final int DECOMPOSE_REPEAT_THRESHOLD = 3;
private static final int BALANCE_MULTIPLIER = 128;
+ private static final Set<Class<? extends AggregateFunction>>
AGG_STATE_REPEAT_REWRITE_WHITELIST = ImmutableSet.of(
+ Avg.class, Variance.class, Corr.class, CorrWelford.class,
Covar.class, CovarSamp.class, AvgWeighted.class,
+ BoolAnd.class, BoolOr.class, BoolXor.class, GroupBitAnd.class,
GroupBitOr.class, GroupBitXor.class,
+ RegrAvgx.class, RegrAvgy.class, RegrCount.class,
RegrIntercept.class, RegrR2.class, RegrSlope.class,
Review Comment:
[P2] Correct the REGR_COUNT state result hint before enabling it
`SELECT a,b,c,REGR_COUNT(d,a) FROM t GROUP BY ROLLUP(a,b,c)` now produces
`Aggregate(REGR_COUNT_COMBINE(...))` in the CTE, followed by merge/finalize
consumers. REGR_COUNT has signature `BIGINT(DOUBLE,DOUBLE)`, so its generated
state starts with a DOUBLE subtype. `DataTypeAggState` passes that first
subtype as `result_type` to the aggregate factory, but
`create_aggregate_function_regr<regr_count>` requires
`DCHECK(result_type->get_primitive_type() == TYPE_BIGINT)`. Constructing the
generated state therefore terminates a DCHECK-enabled backend (including the
repository's DEBUG/ASAN builds). The ordinary aggregate supplied BIGINT before
this rewrite. Correct the state factory's result metadata, or exclude
REGR_COUNT until supported, and cover the rewritten query with a BE execution
test; the new EXPLAIN-only case does not instantiate this factory. This
assertion is disabled in RELEASE builds.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
