github-actions[bot] commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4356885832
##########
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:
[P1] Keep aggregate groups whole across uncovered compensation
The single-input case can retain overlapping aggregate groups even when the
row scopes are disjoint. Use LIST(d,k) partitions
`p_kept={(2020-01-01,1),(2038-01-01,1)}`, `p_bridge={(2038-01-01,2)}`, and
`p_old={(2020-01-01,2)}`, with a 2 YEAR sync limit. The new merge admits the
overlapping retained descriptors from kept/bridge. Put v=2 in kept's 2020 row
and v=1 in old, then refresh `PARTITION BY(d) AS SELECT d,SUM(v) s FROM t GROUP
BY d`.
For `SELECT d,SUM(v) FROM t WHERE d='2020-01-01' GROUP BY d`, pruning
selects kept+old. The fresh MV records kept+bridge, so old is uncovered but
unrecorded; `narrowedTables == pctInputsInQuery == 1` leaves
`takeMvPartitionsOut=false`. The reduced rewritten plan is:
```text
UnionAll
Project(d,s) -> Filter(d=2020-01-01) -> Scan MV[M] // (2020-01-01,2)
Aggregate(d,SUM(v)) -> Filter(d=2020-01-01 AND k=2) -> Scan t //
(2020-01-01,1)
```
`canUnionRewrite` accepts GROUP BY d, and `AbstractMaterializedViewRule`
unions these finished aggregates without an aggregate above the union. The
query returns two groups instead of the required single `(2020-01-01,3)` row.
This differs from the existing full-tuple compensation finding: the base branch
now reads exactly the right rows, but its group overlaps the MV's. Fall back
for this case or combine the aggregate states correctly, and add a single-PCT
grouped-result regression with the compensation plan asserted.
--
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]
