github-actions[bot] commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4360642848
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java:
##########
@@ -377,6 +383,23 @@ protected List<Plan> doRewrite(StructInfo queryStructInfo,
CascadesContext casca
}
boolean needRemoveMvPartitions =
!invalidPartitions.key().isEmpty();
boolean needBaseTableUnion =
!invalidPartitions.value().isEmpty();
+ // An uncovered base partition is read beside the MV
partitions that stay, and the two can
+ // answer the same aggregate group -- a merged LIST
component's MV partition holds a key the
+ // uncovered partition also carries. The union is below the
aggregate then, so the group is
+ // two rows rather than one. A group expression over the
partition column has the same
+ // effect even when the keys differ: date_trunc(d,'year') puts
every day of the year in one
+ // group. Answered from the base table instead of splitting
them.
+ boolean splitsAggregateGroup =
queryPlan.anyMatch(LogicalAggregate.class::isInstance)
+ && needBaseTableUnion
+ && (!uncoveredNarrowedBasePartitions.isEmpty()
+ || (!needRemoveMvPartitions &&
groupsByExpressionOverPctColumn(queryPlan, mtmv)));
Review Comment:
[P1] Keep computed-group protection when only some MV partitions are removed
Removing a stale partition does not imply that every MV partition is
removed. Extend the new `group_split_expr` fixture with a populated
`p_stale={(2037-01-01,3)}`, refresh the MV, then write `p_stale`. Query `d IN
('2020-01-01','2020-01-02','2037-01-01') GROUP BY date_trunc(d,'year')`. The
fresh merged component still holds Jan 2 with v=2; uncovered `p_old` holds Jan
1 with v=1, while the separate 2037 MV partition is stale.
`calcInvalidPartitions` removes only the 2037 partition and compensates
`{p_old,p_stale}`. Raw dates do not overlap, so
`uncoveredNarrowedBasePartitions` is empty; `needRemoveMvPartitions=true` then
disables this expression check. `canUnionRewrite` accepts the PCT slot inside
`date_trunc`, producing this reduced plan:
```text
UnionAll
Aggregate(year(d),sum(s)) -> Scan MV[fresh merged component]
Aggregate(year(d),sum(v)) -> Filter(query AND keys(old,stale)) -> Scan t
```
The two branches emit separate year-2020 rows with 2 and 1, instead of one
row with 3. Unlike #4356885832's fixed same-date case, this bypass needs
computed groups and a simultaneous stale removal. Check overlap against the MV
partitions that actually remain, even when another partition is removed, or
merge aggregate states above the union.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java:
##########
@@ -504,6 +528,42 @@ protected boolean canUnionRewrite(Plan queryPlan,
AsyncMaterializationContext co
return true;
}
+ /**
+ * Whether the query groups by an expression over a column the MV is
partitioned on, rather than by the
+ * column itself. Every value of such a group expression is a function of
several partition-column values
+ * -- {@code date_trunc(d,'year')} puts every day of the year in one group
-- so an uncovered base
+ * partition and a retained MV partition can answer the same group even
when their keys differ.
+ */
+ private static boolean groupsByExpressionOverPctColumn(Plan queryPlan,
MTMV mtmv) {
+ Optional<LogicalAggregate<Plan>> aggregate =
+ queryPlan.collectFirst(LogicalAggregate.class::isInstance);
+ if (!aggregate.isPresent()) {
+ return false;
+ }
+ List<BaseColInfo> pctInfos = mtmv.getMvPartitionInfo().getPctInfos();
+ List<? extends Expression> groupByExpressions =
ExpressionUtils.shuttleExpressionWithLineage(
+ aggregate.get().getGroupByExpressions(), queryPlan);
+ for (Expression expression : groupByExpressions) {
+ if (expression instanceof SlotReference) {
Review Comment:
[P1] Check each grouping set before treating slot grouping keys as disjoint
The normalized group-by list hides grouping sets that omit the partition
column. With the new `group_split_expr` fixture unchanged and fresh, query
`SELECT d,SUM(v) FROM group_split_expr_base WHERE d IN
('2020-01-01','2020-01-02') GROUP BY GROUPING SETS ((d),())`. The MV holds Jan
2 (v=2), and uncovered `p_old` holds Jan 1 (v=1). Their raw dates do not
overlap.
`NormalizeRepeat` builds aggregate keys `[d, grouping_id]` and preserves
`sourceRepeat=[[d],[]]`. Both keys remain `SlotReference`s through lineage, so
this loop skips them and returns false. `canUnionRewrite` accepts `d`, while
`aggregateRewriteByView` explicitly rewrites the repeat over the MV. The
reduced result is:
```text
UnionAll
Aggregate[sets(d),()] -> Repeat -> Scan MV // (Jan2,2), (NULL,2)
Aggregate[sets(d),()] -> Repeat -> Filter(keys(old)) -> Scan t
// (Jan1,1), (NULL,1)
```
There is no aggregate above this union, so it returns two grand totals
instead of `(NULL,3)`. This is separate from the stale-partition bypass: no
partition removal or computed group expression is needed. Inspect
`sourceRepeat`/each grouping set and reject compensation when a set can span
both branches, or combine its aggregate states correctly.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java:
##########
@@ -144,19 +191,198 @@ public static Pair<Map<BaseTableInfo, Set<String>>,
Map<BaseColInfo, Set<String>
if (allCompensateIsNull) {
return null;
}
+ // A base partition the query reads that no MV partition of this plan
is named from -- one the
+ // partition_sync_limit window left out, say -- is one the MV does not
answer for: a partition of the MV
+ // reads a table through the partitions it does name, and the mapping
has no entry for this one. Its
+ // rows have to come from the base table, and two things follow from
that, decided apart.
+ //
+ // The base branch of the union reads those partitions. A table the
mapping mentions not at all is
+ // uncovered in every partition the query uses, so it is read for the
query's own partitions there
+ // rather than skipped -- without this the plan would answer for that
table out of the MV alone.
+ //
+ // And the MV partitions this plan uses have to be taken out when they
hold rows of such a partition: a
+ // partition the window has since left out can have been materialized
in one of them by an earlier
+ // refresh, and the union would then count its rows twice. The refresh
snapshot names exactly the
+ // partitions a refresh read into each MV partition, so that is what
this is asked of. (It is not the
+ // keys: a list partition's default one holds the keys no other
partition of the table claims, so a
+ // partition can be read into an MV partition whose keys do not cover
it.)
+ Map<MTMVRelatedTableIf, Set<String>> uncoveredByTable =
Maps.newLinkedHashMap();
+ boolean uncoveredRecorded = false;
+ for (MTMVRelatedTableIf pctTable :
mtmv.getMvPartitionInfo().getPctTables()) {
+ Set<String> queryUsed =
queryUsedBaseTablePartitionMap.get(pctTable.getFullQualifiers());
+ if (CollectionUtils.isEmpty(queryUsed)) {
+ continue;
+ }
+ // The mapping of a table the query's filtering left out entirely
is empty, so a table it does not
+ // mention at all is a table every query-used partition of which
is uncovered.
+ Map<String, Set<String>> tableMapping =
mtmvRelatedTableIfMapMap.getOrDefault(pctTable, Maps.newHashMap());
+ Set<String> named = Sets.newHashSet();
+ for (Set<String> names : tableMapping.values()) {
+ named.addAll(names);
+ }
+ Set<String> uncovered = Sets.difference(queryUsed, named);
+ if (uncovered.isEmpty()) {
+ continue;
+ }
+ uncoveredByTable.put(pctTable, uncovered);
+ Set<String> recorded = Sets.newHashSet();
+ MTMVRefreshSnapshot refreshSnapshot = mtmv.getRefreshSnapshot();
+ if (refreshSnapshot != null) {
+ for (String mvPartitionName :
rewrittenPlanUsePartitionNameSet) {
+
recorded.addAll(refreshSnapshot.getPctSnapshots(mvPartitionName, new
BaseTableInfo(pctTable)));
+ }
+ }
+ if (!Sets.intersection(uncovered, recorded).isEmpty()) {
+ uncoveredRecorded = true;
+ }
+ }
+ // Narrowing a table's read keeps the MV partitions answering for the
rest, and that only holds while
+ // one input of the query reads the tables it narrows: the pass that
turns the compensation into a plan
+ // puts the filters it is given on one join, so two narrowed inputs
read the pairs both filters reach
+ // and lose the mixed ones -- an a_old x b_kept row, say, when A is
narrowed for a partition the
+ // mapping does not name and B for one the MV no longer holds, or when
one table is narrowed while a
+ // self-join reads it through two inputs, both of which the filter
reaches. Nor can a table be left as
+ // it is while the MV partitions that answered for it are taken out.
So the MV partitions are taken
+ // out, and every table the MV is partitioned on is read for the
query's own partitions, whenever more
+ // than one table is narrowed, a single table is narrowed while the
query reads such a table through
+ // more inputs than that, or a refresh read one of the uncovered
partitions into an MV partition.
+ int narrowedTables = 0;
+ for (MTMVRelatedTableIf pctTable :
mtmv.getMvPartitionInfo().getPctTables()) {
+ BaseColInfo pctInfo = pctInfoMap.get(new BaseTableInfo(pctTable));
+ Set<String> queryUsed =
queryUsedBaseTablePartitionMap.get(pctTable.getFullQualifiers());
+ Set<String> stale = pctInfo == null ? null :
baseTablePartitionNeedUnionNameMap.get(pctInfo);
+ boolean narrowedByStalePass = CollectionUtils.isNotEmpty(stale) &&
!stale.containsAll(queryUsed);
+ if (narrowedByStalePass || uncoveredByTable.containsKey(pctTable))
{
+ narrowedTables++;
+ }
+ }
+ boolean takeMvPartitionsOut = uncoveredRecorded || narrowedTables > 1
+ || (narrowedTables > 0 && pctInputsInQuery > narrowedTables);
+ if (!takeMvPartitionsOut) {
+ // The MV partitions stay, and the uncovered partitions are added
to the base branch beside them.
+ // Such an uncovered partition can answer a group an MV partition
also answers when its keys
+ // meet, in the MV's partition columns, the keys of the base
partitions that MV partition is
+ // recorded with: a merged LIST component {kept (2020,1), bridge
(2038,2)} projected onto d
+ // holds 2020 and 2038, and an uncovered old (2020,2) is read
beside it, so the union would
+ // answer d=2020 twice. A RANGE partition whose range no retained
MV partition covers answers
+ // its own groups and is not recorded here.
+ for (Entry<MTMVRelatedTableIf, Set<String>> entry :
uncoveredByTable.entrySet()) {
+ BaseColInfo pctInfo = pctInfoMap.get(new
BaseTableInfo(entry.getKey()));
+ if (pctInfo == null) {
+ continue;
+ }
+
uncoveredNarrowedBasePartitions.addAll(overlappingUncoveredListPartitions(
+ entry.getKey(), pctInfo, entry.getValue(),
rewrittenPlanUsePartitionNameSet,
Review Comment:
[P2] Exclude removed MV partitions from the retained-key overlap check
This passes every selected MV partition, including those already scheduled
for removal, as the helper's `retainedMvPartitions`. For LIST(d,k), use
`p_stale={(2020-01-01,1),(2037-01-01,1)}`, `p_bridge={(2037-01-01,2)}`,
`p_old={(2020-01-01,2)}`, and `p_fresh={(2038-01-01,1)}` with a 2 YEAR sync
limit. Populate both retained components, refresh `PARTITION BY(d) AS SELECT
d,SUM(v) FROM t GROUP BY d`, then write `p_stale`. Query dates 2020-01-01 and
2038-01-01, grouped by d.
The invalid merged `{2020,2037}` MV partition is removed; only the fresh
2038 partition remains. The base branch reads old plus the stale bucket (the
query filter excludes bridge's 2037 rows), so the valid target is:
```text
UnionAll
Scan MV[fresh 2038 aggregate]
Aggregate(d,sum(v)) -> Filter(d=2020 AND compensation keys) -> Scan t
```
Its groups are disjoint. But the helper sees 2020 in the stale MV partition,
flags `p_old`, and the aggregate guard rejects the rewrite before removal.
Unlike #4359578727's fixed RANGE case, this false overlap comes specifically
from including a removed LIST component. Subtract the MV removal set before
computing retained keys and add this partial-staleness case.
--
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]
