github-actions[bot] commented on code in PR #68180:
URL: https://github.com/apache/doris/pull/68180#discussion_r4058918567
##########
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());
Review Comment:
[P1] Preserve invalidation across sync-window changes. The live mapping is
window-filtered, so an empty result is not stable until the next sync. A
supported FOLLOW_BASE IVM can have populated `p_old`/`p_new`, then shrink
`partition_sync_limit` without refreshing; TRUNCATE `p_old` filters it from
this mapping while `p_new` makes `pctTableMapped` true, so the method returns
without a barrier or version bump. If the user then clears or widens the limit
(ALTER PROPERTY permits the sync limit/time-unit/date-format, but
`alterMvProperties` does not generation-guard them), alignment sees the
replacement base partition with the same range and retains the old MV
partition. The replacement is empty and emitted no row binlog, so incremental
refresh has no delta that deletes the pre-TRUNCATE rows. This is distinct from
the existing EXPR thread: it uses a supported column-partitioned FOLLOW_BASE
MV. Please derive affected live partitions from unfiltered lineage, or
conservatively invalidate f
iltered changed partitions and generation-guard mapping-property changes.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -616,26 +622,39 @@ public void invalidateIvmBaseline() {
editLogItem.await();
}
+ /**
+ * Mark the MV partitions that may hold rows read from the changed base
table partitions as needing a
+ * rebuild. When those partitions cannot be determined, the whole MV is
marked instead.
+ */
public void invalidateIvmBaseline(BaseTableInfo baseTableInfo, Map<String,
Long> changedPartitions) {
+ // Computed before the MV lock is taken, not inside it: the mapping
reads the partition items of the
+ // MV and of every PCT table, so it takes those tables' locks, and the
MV lock has to stay a leaf
+ // (nothing may be acquired under it) the way the rest of this class
assumes. The selection does not
+ // need to be atomic with the barrier it produces: the barrier is
recorded under the lock below, and
+ // the names it carries are intersected with the live partition names
when they are consumed
+ // (MTMVTask).
+ Optional<Set<String>> affectedMvPartitions =
selectAffectedMvPartitions(baseTableInfo,
+ changedPartitions);
+ if (affectedMvPartitions.isPresent() &&
affectedMvPartitions.get().isEmpty()) {
+ // No MV partition reads any of the changed base partitions, so
this change cannot leave
+ // anything behind here: there is no barrier to persist, and
skipping the version bump
+ // keeps it from discarding the result of a task that is already
running.
+ LOG.debug("No MV partition is affected by changed base partitions,
mv={}, baseTable={}, "
+ + "changedPartitions={}", name, baseTableInfo,
changedPartitions);
+ return;
Review Comment:
[P3] Keep no-op logging truthful. This no-op returns to
`MTMVRelationManager`, which then unconditionally logs `Invalidated IVM
baseline` at INFO even though neither `ivmInfo` nor `schemaChangeVersion`
changed. That makes the new intended unread-partition path claim a persisted
invalidation that did not happen. Please return an outcome (or move the
logging) so the no-affected case is logged distinctly and the INFO message
remains truthful.
##########
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:
[P2] Preserve the known-empty result for an empty MV. A partition-following
IVM can be created over an empty partitioned base; if a base partition is added
and then dropped before any sync, `calculatePartitionMappings` is empty because
the MV has never had a live partition or row. This inference leaves
`pctTableMapped` false, records a COMPLETE barrier, and makes strict
INCREMENTAL fail before partition sync even though there is nothing to rebuild.
Please represent successful table participation separately from emitted mapping
entries so the physically empty-MV case returns a known empty set. This should
not make every window-filtered result a no-op: populated filtered partitions
need the unfiltered-lineage/property-generation protection described in the
other comment.
##########
fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java:
##########
@@ -122,35 +126,172 @@ public void
testDropColumnMarksBaselineRebuildOnlyWhenReferenced() throws Except
Assertions.assertTrue(mtmv.getIvmInfo().isBaselineRebuildRequired());
}
+ /**
+ * Which MV partitions must be rebuilt is decided by the MV's partition
mapping, not by what the
+ * refresh snapshot happens to record. This test publishes no snapshot at
all: an MV whose partitions
+ * follow the base table's still narrows the rebuild down to the
partitions that read the dropped one.
+ */
@Test
- public void testPublishedPctPartitionUsesPartitionsBaselineRebuild()
throws Exception {
+ public void testDropPartitionMarksOnlyMvPartitionsThatReadIt() throws
Exception {
String db = "ivm_partitions_baseline_rebuild";
- createPartitionedIvmTableAndMv(db);
+ createPartitionedIvmTableAndPartitionedMv(db);
MTMV mtmv = getMtmv(db);
- OlapTable baseTable = getBaseTable(db);
- publishPctPartitionSnapshot(mtmv, baseTable, "p202001");
+ Assertions.assertEquals(2, mtmv.getPartitionNames().size());
+ Set<String> expected = mvPartitionsWithSameRange(mtmv,
getBaseTable(db), "p202001");
+ Assertions.assertEquals(1, expected.size());
executeSql("ALTER TABLE ivm_base DROP PARTITION p202001");
Assertions.assertFalse(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
- Assertions.assertEquals(Collections.singleton("mv_partition"),
- mtmv.getIvmInfo().getPendingBaselineRebuildPartitions());
+ Assertions.assertEquals(expected,
mtmv.getIvmInfo().getPendingBaselineRebuildPartitions());
}
+ /**
+ * A base partition that no MV partition reads: dropping it cannot leave
any of its rows in the MV, so
+ * there is nothing to rebuild. The previous selection could not tell this
apart from "the snapshot does
+ * not know this partition" and rebuilt the whole MV instead.
+ */
@Test
- public void testMissingPctSnapshotRequiresCompleteBaselineRebuild() throws
Exception {
- String db = "ivm_complete_baseline_rebuild";
- createPartitionedIvmTableAndMv(db);
+ public void testDropPartitionOutsideMvPartitionsMarksNothing() throws
Exception {
+ String db = "ivm_partition_outside_mv";
+ createPartitionedIvmTableAndPartitionedMv(db);
+ MTMV mtmv = getMtmv(db);
+ // Added after the MV was created and never synced into it, so no MV
partition reads it.
+ executeSql("ALTER TABLE ivm_base ADD PARTITION p202003 "
+ + "VALUES [('2020-03-01'), ('2020-04-01'))");
+ Assertions.assertTrue(mvPartitionsWithSameRange(mtmv,
getBaseTable(db), "p202003").isEmpty());
+
+ executeSql("ALTER TABLE ivm_base DROP PARTITION p202003");
+
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+ }
+
+ /**
+ * The partition mapping is built from the MV's PCT tables only. A changed
partition of a joined table
+ * the MV's partition column does not reach is invisible to it, and
missing such a change leaves rows
+ * of the dropped partition in the MV forever, so the whole MV has to be
rebuilt.
+ */
+ @Test
+ public void
testNonPctBaseTablePartitionChangeRequiresCompleteBaselineRebuild() throws
Exception {
+ String db = "ivm_non_pct_partition_change";
+ createPartitionedIvmTable(db);
+ createTable("CREATE TABLE " + db + ".ivm_dim (\n"
+ + " dt date NOT NULL,\n"
+ + " id int NOT NULL,\n"
+ + " v int\n"
+ + ")\n"
+ + "DUPLICATE KEY(dt, id)\n"
+ + "PARTITION BY RANGE(dt) (\n"
+ + " PARTITION d202001 VALUES [('2020-01-01'),
('2020-02-01')),\n"
+ + " PARTITION d202002 VALUES [('2020-02-01'),
('2020-03-01'))\n"
+ + ")\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1', 'binlog.enable' =
'true', "
+ + "'binlog.format' = 'ROW')");
+ // The join is on a non-partition column, so ivm_dim is a base table
of the MV but not a PCT table.
+ createMvByNereids("CREATE MATERIALIZED VIEW ivm_mv\n"
+ + "BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+ + "PARTITION BY(dt)\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "AS SELECT b.dt, b.k1, b.v1 FROM ivm_base b JOIN ivm_dim d
ON b.k1 = d.id");
+ MTMV mtmv = getMtmv(db);
+ Assertions.assertTrue(mtmv.isIvm());
+ Assertions.assertEquals(Sets.newHashSet("ivm_base"),
+ mtmv.getMvPartitionInfo().getPctInfos().stream()
+ .map(pctInfo -> pctInfo.getTableInfo().getTableName())
+ .collect(Collectors.toSet()));
+
+ executeSql("ALTER TABLE ivm_dim DROP PARTITION d202001");
+
+
Assertions.assertTrue(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
+ }
+
+ /**
+ * A join whose condition carries the MV's partition column makes both
tables PCT tables, so the mapping
+ * reads both of them. The marker already holds this table's write lock,
so it takes the other one with a
+ * bounded tryLock: while that table is free, the rebuild is still
narrowed to the partitions that read
+ * the dropped one.
+ */
+ @Test
+ public void testMultiPctTablePartitionChangeStillNarrows() throws
Exception {
+ String db = "ivm_multi_pct_partition_change";
+ createTwoPctTableIvm(db);
MTMV mtmv = getMtmv(db);
-
mtmv.getMvPartitionInfo().setPartitionType(MTMVPartitionType.FOLLOW_BASE_TABLE);
- mtmv.getMvPartitionInfo().setPctInfos(Collections.singletonList(
- new BaseColInfo("dt", new BaseTableInfo(getBaseTable(db)))));
+ Set<String> expected = mvPartitionsWithSameRange(mtmv,
getBaseTable(db), "p202001");
+ Assertions.assertEquals(1, expected.size());
executeSql("ALTER TABLE ivm_base DROP PARTITION p202001");
+
Assertions.assertFalse(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
+ Assertions.assertEquals(expected,
mtmv.getIvmInfo().getPendingBaselineRebuildPartitions());
+ }
+
+ /**
+ * The same MV, but the other PCT table is being written while the
partition DDL marks. Waiting for it
+ * would close a cycle with the DDL that holds it -- each would hold the
write lock the other one needs --
+ * so the marker gives up on the mapping and the whole MV is rebuilt.
+ */
+ @Test
+ public void testMultiPctTableBusyOtherTableRebuildsWholeMv() throws
Exception {
+ String db = "ivm_multi_pct_busy";
+ createTwoPctTableIvm(db);
+ MTMV mtmv = getMtmv(db);
+ OlapTable otherPctTable = (OlapTable)
getDb(db).getTableOrMetaException("ivm_dim");
+ CountDownLatch locked = new CountDownLatch(1);
+ CountDownLatch released = new CountDownLatch(1);
Review Comment:
[P2] Exercise rollback without a hang-prone worker. This setup fails the
read-lock batch at index 0: `ivm_dim` was created before `ivm_mv`, so sorting
by ID puts the already write-locked dimension first. The test therefore never
exercises `tryReadLockTables` releasing a previously acquired read lock, which
is the helper's defining failure guarantee. If a pre-existing reader never
releases, the worker also remains indefinitely in its uninterruptible
`writeLock()` because cleanup starts only after `locked.await` succeeds and the
raw thread is non-daemon. Please add a deterministic helper-level test with a
readable lower-ID table followed by a busy higher-ID table, use bounded or
interruptible worker acquisition, and always assert worker termination.
--
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]
