yujun777 commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4365789904
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java:
##########
@@ -144,15 +170,112 @@ 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);
Review Comment:
Fixed in `a3cdee32fd1` (the guard itself landed in `1de7a0f0504`).
`PartitionCompensator#calcInvalidPartitions` records an uncovered partition
in `uncoveredNarrowedBasePartitions` when its keys, projected onto the MV's
partition columns, meet the keys of the base partitions a retained MV partition
is recorded with, and `AbstractMaterializedViewRule` declines the candidate
when that set is not empty -- it never reaches the union, so the query is
aggregated once, from the base table.
Your shape is the `group_split` fixture of
`mtmv_p0/test_mtmv_base_partition_read_scope`: `LIST(d,k)` with
`p_kept={(2020-01-01,1),(2038-01-01,1)}`, `p_bridge={(2038-01-01,2)}`,
`p_old={(2020-01-01,2)}`, a 2 YEAR window, and the MV `PARTITION BY (d) AS
SELECT d, SUM(v) FROM t GROUP BY d`. `p_old` meets the retained merged
component on `d=2020-01-01`, and before the guard the plan was exactly the
`UnionAll` you wrote -- two rows, `(2020-01-01,2)` and `(2020-01-01,1)`. It now
holds no `UNION` and returns the single `(2020-01-01,3)` (the
`order_qt_group_split` golden).
The two shapes reported later in this thread -- a group expression over the
partition column, and a grouping set that leaves it out -- are no longer
decided from the partition keys at all: they are decided by the grouping, in
the aggregate rule, and are answered in the replies there.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewRule.java:
##########
@@ -377,6 +382,18 @@ 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. Answered from the base table
instead of splitting it.
+ if (needBaseTableUnion &&
!uncoveredNarrowedBasePartitions.isEmpty()
+ &&
queryPlan.anyMatch(LogicalAggregate.class::isInstance)) {
Review Comment:
Fixed in `a3cdee32fd1`.
The fallback you pointed at is now only the key-overlap question, and that
set is filled from a `ListPartitionInfo`: `overlappingUncoveredListPartitions`
returns empty for any other partitioning, because a RANGE partition is either
covered by a retained MV partition -- then it is not uncovered -- or lies
outside its range, where it answers groups of its own. Your windowed RANGE case
keeps its `UnionAll(Scan MV[kept], Aggregate(d,SUM(v)) -> Scan t[old])`, and so
does every ordinary TTL-shaped query.
What I took out of the base rule is the group-expression test, which is what
over-declined here; it now lives in
`AbstractMaterializedViewAggregateRule#canUnionRewrite`, which asks the
grouping itself: the rewrite stands when the grouping holds every column the MV
is partitioned on, each of them by itself, and is declined otherwise. A bare
`GROUP BY d` therefore keeps the rewrite -- which is what
`nereids_rules_p0/mv/partition_union_rewrite` and
`nereids_rules_p0/mv/union_rewrite_grace_big` need. Those two suites were
failing on the branch before this commit (the roll-up query of
`partition_mv_rewrite`, which groups by `date_trunc(l_shipdate,'day')` *and*
bare `l_shipdate`, was declined by the expression test); both pass now, along
with `union_rewrite`, `grouping_sets`, `agg_with_roll_up`,
`agg_without_roll_up` and `date_trunc` -- 14 suites, none failed.
One thing to flag rather than bury: the grouping test declines an expression
over the partition column even when the expression happens not to coarsen.
`union_all_compensate`'s query5 groups by `to_date(data_date)` over daily RANGE
partitions, where the branches do come out disjoint; telling that apart needs
the expression evaluated per partition value, which is not something the
grouping can answer, so that candidate is now declined and the suite's
expectation moved from `mv_rewrite_success` to `mv_rewrite_fail`. Its rows are
unchanged -- the `.out` compares the base-table answer against the same values
-- only the plan differs.
##########
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:
Fixed in `a3cdee32fd1`.
The bypass was the `!needRemoveMvPartitions` term, and the whole expression
test has left the base rule: what the union of two finished aggregates needs is
a question about the grouping, not about which partitions were dropped, and the
aggregate rule is where the grouping is known.
`AbstractMaterializedViewAggregateRule#canUnionRewrite` now requires every
column the MV is partitioned on to appear in the grouping by itself, and
`AbstractMaterializedViewRule` consults it for any compensation union --
`needBaseTableUnion` alone, with removal beside the point -- so a stale
partition removed beside the compensation no longer switches the question off.
Your shape is the new `group_split_stale_expr` fixture of
`mtmv_p0/test_mtmv_base_partition_read_scope`:
`p_kept={(2020-01-01,1),(2038-01-01,1)}`, `p_bridge={(2038-01-01,2)}`,
`p_stale={(2037-01-01,3)}`, `p_old={(2020-01-01,2)}`, 2 YEAR window, MV
`PARTITION BY (d) AS SELECT d, SUM(v) FROM t GROUP BY d`; after the refresh,
100 more go into `p_stale`. The query over the three dates grouped by
`date_trunc(d,'year')` removes only the 2037 MV partition and compensates
`{p_old, p_stale}`, so raw dates do not overlap and the key-overlap set stays
empty. Before the change the plan carried your `UnionAll` and year 2020 came
out twice (2 and 1); the `explain` now asserts no `UNION` and the golden is one
row per year, `2020-01-01 -> 3` and `2037-01-01 -> 104`.
##########
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:
Fixed in `a3cdee32fd1`.
Done as you describe, and in the place that can ask it: the method at line
547 you pointed at is gone, and
`AbstractMaterializedViewAggregateRule#canUnionRewrite` now takes the grouping
apart itself. When the aggregate has a source Repeat it walks
`getGroupingSets()` and requires, of *every* set, that each column the MV is
partitioned on appears by itself; the normalized key list is no longer trusted
to say which set a slot came from. Your `[[d],[]]` case fails on the empty set,
and a set that groups only by an expression over the column fails on the same
rule as the plain case.
`group_split_sets` in `mtmv_p0/test_mtmv_base_partition_read_scope` is the
case: the `group_split_expr_base` fixture unchanged, `SELECT d, SUM(v) ...
WHERE d IN ('2020-01-01','2020-01-02') GROUP BY GROUPING SETS ((d),())`.
Before, the plan was the `UnionAll` of two `Aggregate -> Repeat -> ...`
branches and the grand total came out as two rows; the `explain` now asserts no
`UNION`, and the golden is the three rows with one total: `\N 3`, `2020-01-01
1`, `2020-01-02 2`.
`nereids_rules_p0/mv/grouping_sets` and `mtmv_p0/test_multi_pct_rollup_mtmv`
/ `test_rollup_partition_mtmv` / `test_create_rollup_mtmv` pass with it, so the
tightening is limited to the shapes that can actually span the two branches.
##########
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:
Fixed in `a3cdee32fd1`.
`calcInvalidPartitions` now computes the set the overlap is projected
against as the plan's MV partitions minus everything the removal pass has named:
```java
Set<String> removedMvPartitions = Sets.newHashSet();
mvPartitionNeedRemoveNameMap.values().forEach(removedMvPartitions::addAll);
Set<String> retainedMvPartitions =
Sets.difference(rewrittenPlanUsePartitionNameSet, removedMvPartitions);
```
A partition scheduled for removal is read back through the compensation
branch, so its keys are the compensation's own and cannot be the retained group
an uncovered partition would meet.
`partial_stale` in `mtmv_p0/test_mtmv_base_partition_read_scope` is your
case: `p_stale={(2020-01-01,1),(2037-01-01,1)}`, `p_bridge={(2037-01-01,2)}`,
`p_old={(2020-01-01,2)}`, `p_fresh={(2038-01-01,1)}`, 2 YEAR window, both
retained components populated before the refresh and `p_stale` written after
it. The merged `{2020,2037}` MV partition is removed and only the fresh 2038
partition remains, so `p_old` is no longer flagged for a 2020 it shares with a
partition that is going away; `SELECT d, SUM(v) ... WHERE d IN
('2020-01-01','2038-01-01') GROUP BY d` keeps the union, and its `explain`
asserts both `UNION` and `partial_stale_mv` (rows `2020-01-01 103`, `2038-01-01
0`). Note it needs the bare-column grouping too: this is the case where the
partition keys are the only thing separating the two branches' groups.
--
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]
