github-actions[bot] commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4278511568
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java:
##########
@@ -144,15 +148,85 @@ 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;
+ }
+ }
+ if (!uncoveredByTable.isEmpty()) {
+ for (Entry<MTMVRelatedTableIf, Set<String>> uncoveredEntry :
uncoveredByTable.entrySet()) {
Review Comment:
[P1] Restore every PCT input when removing all selected MV partitions
The full-query expansion visits only `uncoveredByTable`, but the branch
below removes every selected MV partition. Consider an inner-join MV with
selected partitions `{M1,M2}`: M1 is recently refreshed/in grace, M2 is
stale/outside grace. Current mappings are `A[M1]={a_kept}`, `A[M2]={a2}`,
`B[M1]={b_kept}`, `B[M2]={b2}`; the query reads `A={a_kept,a_old,a2}` and
`B={b_kept,b2}`. A narrowed sync window has just excluded `a_old`, while M1's
refresh snapshot still records it (a mixed old/future `a_kept` keeps M1's
descriptor unchanged).
`getNeedCompensatePartitions` initially makes the base scopes
`A={a_old,a2}`, `B={b2}` for stale M2. The recorded `a_old` then sets
`uncoveredRecorded=true`: this loop expands A alone, and lines 212-215 remove
both M1 and M2. The resulting reduced plan is:
```text
UnionAll
Scan MV[]
Join(a.d=b.d)
Scan A[a_kept,a_old,a2]
Scan B[b2]
```
Rows from `b_kept`'s day are lost because B's existing compensation filter
remains restricted to M2. This differs from the earlier narrowed-window
duplication thread: removal now succeeds, but the other table's covered input
is not restored. When discarding all selected MV partitions, reset the
compensation scope of every participating PCT table to the query's full scope,
or abandon this rewrite.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java:
##########
@@ -144,15 +148,85 @@ 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;
+ }
+ }
+ if (!uncoveredByTable.isEmpty()) {
+ for (Entry<MTMVRelatedTableIf, Set<String>> uncoveredEntry :
uncoveredByTable.entrySet()) {
+ BaseColInfo pctInfo = pctInfoMap.get(new
BaseTableInfo(uncoveredEntry.getKey()));
+ if (pctInfo == null) {
+ continue;
+ }
+ // With the MV partitions taken out the MV answers none of
what the query reads, so the whole of
+ // the query's partitions comes from the base; otherwise the
MV answers the partitions the
+ // mapping names and the base branch reads the ones it does
not.
+ Set<String> toRead = uncoveredRecorded
+ ? queryUsedBaseTablePartitionMap.getOrDefault(
+ uncoveredEntry.getKey().getFullQualifiers(),
Sets.newHashSet())
+ : uncoveredEntry.getValue();
Review Comment:
[P1] Preserve mixed join pairs when compensating two PCT tables
Give both OLAP A and B `LIST(d,k)` partitions
`p_kept=((2020-01-01,1),(2038-01-01,1))`, `p_old=((2020-01-01,2))`, plus an
unrelated partition, with one populated 2020 row in each named partition.
Refresh a `PARTITION BY(a_d)` MV of `SELECT a.d AS a_d,a.k AS ak,b.k AS bk FROM
A a JOIN B b ON a.d=b.d` under a 2 YEAR sync window. Only each `p_kept` is
read/recorded, and their merged MV partition is synchronized even with zero
grace. A query for `a.d=b.d='2020-01-01'` reads both named partitions from each
table and must return all four `(ak,bk)` pairs.
Here `uncoveredRecorded` is false and both compensation sets are `{p_old}`.
`StructInfo.addFilterOnTableScan`/`PredicateAdder` applies both filters to one
join, producing the reduced plan:
```text
UnionAll
Scan MV[M] // (1,1)
Join(a.d=b.d)
Filter(A.k=2 AND A.d=D0) -> Scan A
Filter(B.k=2 AND B.d=D0) -> Scan B // (2,2)
```
The committed `(1,2)` and `(2,1)` pairs disappear. Both tables are present
in the mapping; this differs from the existing entirely-unmapped-table and
partition-name-identity threads. Include the mixed combinations with a disjoint
compensation plan, or fall back to the full base query for this shape, and
assert all four pairs with an MV/UNION plan check.
--
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]
