yujun777 commented on code in PR #68180:
URL: https://github.com/apache/doris/pull/68180#discussion_r4058981885
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -645,6 +664,143 @@ public void invalidateIvmBaseline(BaseTableInfo
baseTableInfo, Map<String, Long>
editLogItem.await();
}
+ /**
+ * Select the MV partitions that may hold rows read from the changed base
table partitions.
+ *
+ * <p>This asks which MV partitions read the changed base partitions at
all, instead of (as the
+ * refresh snapshot based selection did) which of them had already seen
them. The snapshot is a lower
+ * bound that is allowed to lag: a base partition that was added after the
snapshot was captured never
+ * appears in it, so it can report "this partition never read the changed
base partition" about a
+ * partition that does hold its rows. Missing a partition here is not
repaired by a later refresh --
+ * dropping or truncating a base partition emits no row binlog, so the
incremental path never learns
+ * about those orphan rows and they stay in the MV forever.
+ *
+ * <p>Three cases have no answer in the mapping, and each of them must
rebuild the whole MV instead:
+ * a SELF_MANAGE MV (the mapping API answers nothing for it, although its
single partition reads every
+ * base partition); a base table that is not one of the MV's PCT tables
(the mapping is seeded from
+ * {@code getPctTables()} and never gains a table later, so a joined
partition table that the MV's
+ * partition column does not reach is not described at all); and a changed
partition that is not in
+ * the base table's metadata right now, which is how RECOVER PARTITION
arrives here -- it marks before
+ * the partition is added back, so at this point the partition is still in
the recycle bin.
+ *
+ * <p>Locking is the fourth way to end up rebuilding everything, but it is
contention rather than a
+ * property of the MV: the tables whose partition items the mapping reads
are locked with a bounded
+ * tryLock, and the MV is rebuilt only while one of them is being written.
See the comment at that
+ * loop.
+ *
+ * <p>An empty result is meaningful, on the other hand: the mapping lists
every base partition read by
+ * the MV, so a changed base partition that no MV partition maps to is
read by none of them.
+ *
+ * @param changedBasePartitions base partition name to partition id, never
empty
+ * @return {@link Optional#empty()} when the affected MV partitions cannot
be determined, otherwise the
+ * (possibly empty) set of MV partition names that must be rebuilt
+ */
+ private Optional<Set<String>> selectAffectedMvPartitions(BaseTableInfo
baseTableInfo,
+ Map<String, Long> changedBasePartitions) {
+ if (mvPartitionInfo.getPartitionType() ==
MTMVPartitionType.SELF_MANAGE) {
+ return Optional.empty();
+ }
+ MTMVRelatedTableIf pctTable = findPctTable(baseTableInfo);
+ if (pctTable == null) {
+ return Optional.empty();
+ }
+ // Computing the mapping reads the partition items of the MV and of
every PCT table, which means
+ // taking their read locks. The caller already holds the changed
table's write lock (a partition DDL
+ // marks before it releases it), so these other reads must not block:
two partition DDLs on two PCT
+ // tables of this MV would otherwise each hold the write lock the
other one needs, and acquiring in
+ // id order cannot break a cycle whose first lock is already held.
They are taken with a bounded
+ // tryLock instead, the way the stream cleanup treats a busy table: a
busy table means a writer is
+ // involved, and then the whole MV is rebuilt. The list is still
sorted by id so that the acquisition
+ // order matches the rest of the code base.
+ List<TableIf> tablesToRead =
Lists.newArrayListWithCapacity(mvPartitionInfo.getPctInfos().size() + 1);
+ tablesToRead.add(this);
+ for (BaseColInfo pctInfo : mvPartitionInfo.getPctInfos()) {
+ if (pctInfo.getTableInfo().equals(baseTableInfo)) {
+ continue;
+ }
+ try {
+ tablesToRead.add(MTMVUtil.getTable(pctInfo.getTableInfo()));
+ } catch (Exception e) {
+ LOG.warn("Failed to resolve PCT table {}, rebuild the whole
MV. mv={}",
+ pctInfo.getTableInfo(), name, e);
+ return Optional.empty();
+ }
+ }
+ tablesToRead.sort(Comparator.comparing(TableIf::getId));
+ if (!MetaLockUtils.tryReadLockTables(tablesToRead,
Table.TRY_LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
+ LOG.warn("A PCT table is busy, rebuild the whole MV {} instead of
selecting part of it", name);
+ return Optional.empty();
+ }
+ try {
+ // A partition that is missing from the metadata here is invisible
to the mapping as well, so
+ // an empty answer below would be indistinguishable from "no MV
partition reads it". The match
+ // is deliberately exact: the mapping is keyed by the metadata's
spelling, so a name that only
+ // differs in case must take the whole-MV path too, or the lookup
below would quietly select
+ // nothing for a partition that some MV partition does read. Base
tables that do not implement
+ // getPartitionNames -- the external ones -- report no partition
at all, so a partition change
+ // on them always rebuilds the whole MV. That matches what the
refresh-snapshot selection
+ // answered for them, and the mapping has never been exercised for
external tables (IVM does
+ // not support them as base tables yet): revisit before taking the
narrow path for them.
+ if
(!pctTable.getPartitionNames().containsAll(changedBasePartitions.keySet())) {
+ return Optional.empty();
+ }
+ Map<String, Map<MTMVRelatedTableIf, Set<String>>>
partitionMappings =
+ calculatePartitionMappings(Maps.newHashMap());
+ Set<String> res = Sets.newHashSet();
+ boolean pctTableMapped = false;
+ for (Entry<String, Map<MTMVRelatedTableIf, Set<String>>> mapping :
partitionMappings.entrySet()) {
+ for (Entry<MTMVRelatedTableIf, Set<String>> tableMapping :
mapping.getValue().entrySet()) {
+ if (!tableMapping.getKey().equals(pctTable)) {
+ continue;
+ }
+ pctTableMapped = true;
+ if (!Collections.disjoint(tableMapping.getValue(),
changedBasePartitions.keySet())) {
+ res.add(mapping.getKey());
+ }
+ }
+ }
+ // The mapping does not describe this base table at all. That
contradicts the PCT check above,
+ // so it is safer to rebuild everything than to trust a selection
that never saw the table.
+ if (!pctTableMapped) {
Review Comment:
Addressed in 7ad093ee957.
When the mapping describes no entry for the PCT table, the marker now asks
whether the MV has partitions of its own. With none, the answer is known to be
empty -- an MV without partitions holds no rows -- and it returns an empty
selection, so no COMPLETE barrier is recorded and a strict INCREMENTAL is not
rejected before partition sync. When the MV does have partitions, the
conservative path stays as it was: a table that the mapping does not describe
at all is still a contradiction worth rebuilding for.
Going through the cases again, the two empties are genuinely different
things and are now treated that way: an MV with no partitions is empty by
construction, while a populated MV whose changed partition the window filters
is not, and that one is what the lineage mapping in the other comment answers.
--
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]
