yujun777 commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4278802913
##########
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:
Fixed in 723f95fafe9. Both combinations are covered by counting the plan's
inputs rather than the tables: the narrowed tables are now the ones either pass
leaves short (the stale-partition pass counts too, which is your first
combination), and the narrowed read is kept only while one input of the plan
reads the MV's partitioned tables -- so a self-join, whose aliases are one
table, takes the MV partitions out as well (your second). With the MV
partitions taken out every table is read for the query's own partitions,
replacing what the stale pass had narrowed rather than waiting beside it.
`PartitionCompensatorTest` covers the one-narrowed-table case (narrowed read,
MV partitions kept) and the two-narrowed-table case (query partitions, MV
partitions out).
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedPartitionDescTransferGenerator.java:
##########
@@ -65,6 +75,38 @@ public void apply(MTMVPartitionInfo mvPartitionInfo,
Map<String, String> mvPrope
lastResult.setRes(res);
}
+ /**
+ * One MV partition per set of keys that meet, whichever table wrote them
down: a partition of a list
+ * partitioned base table can hold several keys of the MV's partition
column, so two partitions -- of one
+ * table or of two -- can describe keys that meet, and an MV's own
partitions cannot overlap. The one
+ * partition that holds a group's keys names every table's partitions of
them, which is what a refresh
+ * reads and records for those keys (see {@link
MTMVPartitionUtil#mergedListDescs}).
+ *
+ * <p>Merging across tables, not within each of them, is what keeps the MV
buildable: two tables of a
+ * multi-table MV have to come out with the same descs, or one table's
merged desc repeats a key another
+ * table's desc holds and `checkIntersect` (or the partition creation
itself) rejects the MV.
+ */
+ private Map<PartitionKeyDesc, Map<MTMVRelatedTableIf, Set<String>>>
mergeOverlappingListDescs(
+ Map<MTMVRelatedTableIf, Map<PartitionKeyDesc, Set<String>>> descs)
{
+ Set<PartitionKeyDesc> allDescs = Sets.newHashSet();
+ for (Map<PartitionKeyDesc, Set<String>> onePctDescs : descs.values()) {
+ allDescs.addAll(onePctDescs.keySet());
+ }
+ Map<PartitionKeyDesc, PartitionKeyDesc> mergedOfDesc =
MTMVPartitionUtil.mergedListDescs(allDescs);
+ Map<PartitionKeyDesc, Map<MTMVRelatedTableIf, Set<String>>> res =
Maps.newHashMap();
+ for (Entry<MTMVRelatedTableIf, Map<PartitionKeyDesc, Set<String>>>
entry : descs.entrySet()) {
+ for (Entry<PartitionKeyDesc, Set<String>> onePctEntry :
entry.getValue().entrySet()) {
+ PartitionKeyDesc desc =
mergedOfDesc.getOrDefault(onePctEntry.getKey(), onePctEntry.getKey());
Review Comment:
On the external mismatch: it is real, and it is not introduced by this PR --
it is the state before it, for every table. Before this PR a refresh read each
base table through the MV partition's key range and recorded the mapping, so
the read was wider than the record for external and olap tables alike; this PR
closes that for the olap tables, where the silent changes it is about
(DROP/TRUNCATE/REPLACE/RECOVER PARTITION) happen and are visible in the
metadata, and leaves external tables reading the MV partition's key range as
they were. Closing it there needs what the PR gives the olap side -- a scope
the read and the record share -- which for an external table means deciding
what its partition identity is when the table is not one the alignment can read
partitions from, and I would rather send that as its own change with a Paimon
DROP test than fold it in here. I have recorded it in the follow-up list rather
than leave it implicit.
--
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]
