yujun777 commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4371620470
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java:
##########
@@ -303,6 +306,236 @@ public static Optional<Map<BaseTableInfo, Set<Long>>>
generateRelatedBasePartiti
return Optional.of(res);
}
+ /**
+ * Whether the table has a list partition's default partition, the
partition that takes the rows no other
+ * partition of it claims.
+ *
+ * <p>Such a table cannot be represented completely by a materialized view
partitioned on it: a key that
+ * only that partition holds has no MV partition to be read into -- the
MV's partitions are the keys the
+ * explicit partitions project to -- so its rows are not in the MV, and no
rewrite of a query that may
+ * read them can be answered from it. The partitions are walked under the
table's read lock, so a
+ * concurrent ADD or DROP PARTITION cannot be seen half applied.
+ */
+ public static boolean hasDefaultListPartition(MTMVRelatedTableIf table) {
+ if (!(table instanceof OlapTable)) {
+ return false;
+ }
+ OlapTable olapTable = (OlapTable) table;
+ if (!(olapTable.getPartitionInfo() instanceof ListPartitionInfo)) {
+ return false;
+ }
+ olapTable.readLock();
+ try {
+ for (PartitionItem item :
olapTable.getPartitionInfo().getIdToItem(false).values()) {
+ if (item.isDefaultPartition()) {
+ return true;
+ }
+ }
+ } finally {
+ olapTable.readUnlock();
+ }
+ return false;
+ }
+
+ /**
+ * The desc of the MV partition each of these descs belongs to, grouped:
the keys of a list partitioned
+ * base table can meet at the MV's partition column -- a partition holding
several keys of it, one of them
+ * shared with another partition -- and an MV's own partitions cannot
overlap, so descs whose keys meet
+ * are one partition whose keys are the union of theirs. A desc whose keys
meet nothing is answered with
+ * itself, and so is every desc that is not a list of keys.
+ *
+ * <p>The keys of a merged desc are written out in the order
+ * {@link PartitionKeyDesc#sortedInValues} gives the same key set, which
is the order
+ * {@link ListPartitionItem#toPartitionKeyDesc(int)} writes it in too,
because
+ * {@link PartitionKeyDesc#equals} compares that list: the same keys in
another order are another desc,
+ * which would leave the MV partition an alignment computes unmatchable to
the one it holds.
+ *
+ * <p>Answered as groups -- the merged desc, and the descs whose keys it
holds -- rather than one merged
+ * desc per source desc: a caller mapping every source desc would
otherwise hash the group's merged desc,
+ * N+1 keys, once per source desc in it, which is quadratic in a group of
N.
+ */
+ public static Map<PartitionKeyDesc, Set<PartitionKeyDesc>>
groupedListDescs(
+ Collection<PartitionKeyDesc> descs) {
+ // A union-find over the keys: two descs whose keys meet end up in one
group, transitively, and each
+ // key is looked up once -- walking the groups per desc would be
quadratic in the number of partitions.
+ Map<List<PartitionValue>, List<PartitionValue>> groupOfKey =
Maps.newHashMap();
+ for (PartitionKeyDesc desc : descs) {
+ if (!desc.hasInValues()) {
+ continue;
+ }
+ List<PartitionValue> first = desc.getInValues().iterator().next();
+ groupOfKey.putIfAbsent(first, first);
+ for (List<PartitionValue> key : desc.getInValues()) {
+ groupOfKey.putIfAbsent(key, key);
+ unionKey(groupOfKey, first, key);
+ }
+ }
+ Map<List<PartitionValue>, Set<List<PartitionValue>>> keysOfGroup =
Maps.newHashMap();
+ for (List<PartitionValue> key : groupOfKey.keySet()) {
+ keysOfGroup.computeIfAbsent(findKey(groupOfKey, key), k ->
Sets.newHashSet()).add(key);
+ }
+ Map<List<PartitionValue>, Set<PartitionKeyDesc>> descsOfGroup =
Maps.newHashMap();
+ Map<List<PartitionValue>, PartitionKeyDesc> onlyDescOfGroup =
Maps.newHashMap();
+ for (PartitionKeyDesc desc : descs) {
+ if (!desc.hasInValues()) {
+ continue;
+ }
+ List<PartitionValue> group = findKey(groupOfKey,
desc.getInValues().iterator().next());
+ descsOfGroup.computeIfAbsent(group, k ->
Sets.newHashSet()).add(desc);
+ onlyDescOfGroup.putIfAbsent(group, desc);
+ }
+ Map<PartitionKeyDesc, Set<PartitionKeyDesc>> res = Maps.newHashMap();
+ for (Entry<List<PartitionValue>, Set<PartitionKeyDesc>> entry :
descsOfGroup.entrySet()) {
+ Set<PartitionKeyDesc> groupDescs = entry.getValue();
+ PartitionKeyDesc mergedDesc = groupDescs.size() == 1
+ ? onlyDescOfGroup.get(entry.getKey())
+ : PartitionKeyDesc.createIn(
+
PartitionKeyDesc.sortedInValues(keysOfGroup.get(entry.getKey())));
+ res.put(mergedDesc, groupDescs);
+ }
+ // A desc that is not a list of keys meets nothing and is its own
group.
+ for (PartitionKeyDesc desc : descs) {
+ if (!desc.hasInValues()) {
+ res.put(desc, Sets.newHashSet(desc));
+ }
+ }
+ return res;
+ }
+
+ private static void unionKey(Map<List<PartitionValue>,
List<PartitionValue>> groupOfKey,
+ List<PartitionValue> left, List<PartitionValue> right) {
+ List<PartitionValue> leftGroup = findKey(groupOfKey, left);
+ List<PartitionValue> rightGroup = findKey(groupOfKey, right);
+ if (leftGroup != rightGroup) {
+ groupOfKey.put(rightGroup, leftGroup);
+ }
+ }
+
+ private static List<PartitionValue> findKey(Map<List<PartitionValue>,
List<PartitionValue>> groupOfKey,
+ List<PartitionValue> key) {
+ List<PartitionValue> group =
Preconditions.checkNotNull(groupOfKey.get(key),
+ "a key is registered before it is looked up: %s", key);
+ while (group != groupOfKey.get(group)) {
+ group = groupOfKey.get(group);
+ }
+ List<PartitionValue> root = group;
+ // Path compression, so that the walk is not repeated for the rest of
this group's keys.
+ group = groupOfKey.get(key);
+ while (group != root) {
+ List<PartitionValue> next = groupOfKey.get(group);
+ groupOfKey.put(group, root);
+ group = next;
+ }
+ return root;
+ }
+
+ /**
+ * The base partitions each MV partition is read from, by MV partition and
then by base table: the ones
+ * that MV partition is recorded with, which is the same set the refresh
reads and the same one a rebuild
+ * of it has to read again. A table named with an empty set is not read
for that MV partition, and a
+ * table absent from an MV partition's map keeps that MV partition's own
key range.
+ *
+ * <p>Kept per MV partition rather than unioned for the batch: a batch
writes the rows of each MV
+ * partition, and those are the rows of that partition's own recorded base
partitions. A base partition
+ * of a range table can end past the MV partition it is rolled up into, see
+ * UpdateMvByPartitionCommand#constructTableWithPredicates, so a row of it
can lie inside another MV
+ * partition's key range. Unioned, the batch reads that row through the
other partition's range although
+ * the partition it comes from is not the one recorded there, and a batch
that does not write that
+ * partition reads a row it has no target for. Paired, each MV partition
is read from exactly its own
+ * record, which is what the snapshot written next to the read claims.
+ *
+ * <p>Every table the MV is partitioned on is named for each MV partition,
olap or external: a table the
+ * mapping does not name for an MV partition feeds it nothing -- the sync
window dropped its partitions,
+ * say -- and its MV key range reads rows the snapshot records no
partition for. The empty set says that
+ * outright, which is what separates it from a table that is not one of
the MV partition column's
+ * sources at all.
+ *
+ * <p>An MV partition no base partition describes keeps the MV key ranges
for the whole batch instead:
+ * nothing is recorded for it, so there is no scope its read could match,
and the rows it holds are
+ * read through its own key range rather than not at all.
+ *
+ * <p>The tables are named by {@link BaseTableInfo} rather than by the
table object: the mapping is keyed
+ * by the tables the MV's partition info holds and this reads them by the
name it is given, so what
+ * identifies a table here is the table it names, not which of the two
objects it was read from.
+ */
+ public static Map<String, Map<BaseTableInfo, Set<String>>>
mappedBasePartitionsByMvPartition(
+ Map<TableIf, String> tableWithPartKey, MTMVRefreshContext context,
Set<String> mvPartitionNames) {
+ for (String mvPartitionName : mvPartitionNames) {
+ if
(context.getUndescribedMvPartitions().contains(mvPartitionName)) {
+ // An MV partition no base partition describes has nothing to
scope the read to: the mapping
+ // names no base partition for it, so a scoped read would read
no rows over the rows it
+ // holds. The olap cause of it -- an alignment that has not
caught up yet -- is refused by a
+ // refresh rather than read
(MTMVTask#requireDescribedPartitions), because there it can still
+ // catch up; what a read reaches this way is an external
partition table's partition the
+ // mapping cannot describe at all -- a row whose partition key
is null, say -- which no
+ // alignment will ever catch up with either. It keeps the read
it had before partitions were
+ // scoped: every table is left unscoped for the batch, and the
MV partitions' own key ranges
+ // select the rows they hold.
+ return Maps.newHashMap();
Review Comment:
Fixed in 90957ae20db.
- `mappedBasePartitionsByMvPartition` now leaves out only the undescribed MV
partition instead of returning an empty map for the whole batch. The omitted
partition keeps its own key range (the behaviour `UpdateMvByPartitionCommand`
gives a name the map does not contain), while the described partitions keep
their records — so October keeps its October 25 scope even while September is
undescribed.
- `EXPLAIN REFRESH ... COMPLETE` now computes the alignment the next refresh
would run (against the preview's pinned snapshots) and previews the
post-alignment partition set: dropped names are removed, a name the alignment
drops and recreates stays, and an empty result falls back to the original set.
Self-managed MVs skip it, as execution does.
- The explain context now turns the transparent MV rewrite off, exactly as
`MTMVPlanUtil.executeCommand` does before running a refresh; previously the
preview could be rewritten to read the MV itself, hiding the base reads
entirely.
Regression: `test_mtmv_base_partition_read_scope` gains `explain_realign` —
an MV built without a window, then `ALTER ... partition_sync_limit = 2 DAY`;
the explain must show `partitions=2/3 (p_kept,p_today)` and not `p_expired`,
and the real refresh must drop the expired MV partition. Also ran
`test_mtmv_external_pct_read_scope`, `test_ivm_explain_refresh`,
`test_ivm_refresh_dry_run`, and the focused FE UTs
(`MTMVRefreshContextBatchTest`, `RefreshMTMVCommandTest`,
`UpdateMvByPartitionCommandTest`, `ExplainRefreshMTMVCommandTest`).
--
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]
