github-actions[bot] commented on code in PR #68390:
URL: https://github.com/apache/doris/pull/68390#discussion_r4084326150
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -732,6 +780,10 @@ private AttemptResultType
executeIvmAttempt(MTMVRefreshContext refreshContext,
}
try {
syncPartitionsIfNeeded(ctx, tableIfs);
+ // The retry can add a partition that did not exist at the
first alignment. It has to get
+ // its entry before the retried refresh reads a base table, or
an invalidation arriving
+ // in between would have nothing to land on for rows this task
is about to write.
+ mtmv.alignPartitionStates(mtmv.getPartitionNames());
Review Comment:
[P1] Re-plan the epoch decision after this retry alignment. A partition
created here gets the deliberately dirty initial state `{0,1}`, but
`dirtyPartitions` and `ivmPlannedEpochs` were frozen before the retry. The next
attempt therefore sends the new name through the delta path, and because it has
no planned ceiling, `commitCapturedEpochs` can write whatever raised epoch it
captures as satisfied. A metadata-only replace/recover around this window emits
no repairing row delta, so the new MV partition can be marked clean without
ever receiving its baseline. Refresh the state snapshot after sync and route
newly dirty names through rebuild (with matching ceilings), and add a retry
latch test.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java:
##########
@@ -1075,13 +1075,32 @@ private static void checkColumnIfChange(MTMV mtmv,
List<ColumnDefinition> analyz
+ "original length is: %s, current length is: %s",
originalColumns.size(), analyzedColumns.size()));
}
- for (int i = 0; i < originalColumns.size(); i++) {
- if (!isTypeLike(originalColumns.get(i).getType(),
analyzedColumns.get(i).getType())) {
+ // Matched by name, not by position. The order of the two lists is
decided by different passes:
+ // the physical schema is laid out when the MV is created, where
MTMVPlanUtil#applyIvmPhysicalKeyLayout
+ // puts the final key columns first, and the analysed list comes from
running that same layout again
+ // with the stored key columns as its input. The two agree except for
a chained IVM MV whose base
+ // tables carry row-id columns of their own: the create pass derives
the visible key prefix from the
+ // identity key slots, the analysed one takes it from the stored keys,
and the base tables' row-id
+ // columns end up in a different block. What this check is for is a
base-table change that makes a
+ // column disappear or change type, and where a column sits is not
part of that.
+ Map<String, Column> originalByName = Maps.newHashMap();
+ for (Column column : originalColumns) {
+ originalByName.put(column.getName().toLowerCase(), column);
Review Comment:
[P2] Make this case fold locale-independent. Both this insertion and the
lookup below use zero-argument `toLowerCase()`, so under a Turkish default
locale an original `ID` becomes `ıd` while an analyzed `id` remains `id`,
producing a spurious `column not found` even though Doris treats column
identity case-insensitively. Use `Locale.ROOT` or a
`TreeMap<>(String.CASE_INSENSITIVE_ORDER)`, and cover a Turkish-default
case-only match.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -684,33 +759,211 @@ public Map<String, MTMVPartitionState>
getPartitionStates() {
// A payload without the member carries no state at all, which is not the
same as an empty map that
// says the states are now empty: leaving them alone is the only answer
that cannot lose state.
public void alterPartitionStates(Map<String, MTMVPartitionState>
partitionStates) {
- if (partitionStates == null) {
- return;
- }
+ replayAlterPartitionStates(partitionStates, null);
+ }
+
+ /**
+ * ALTER_PARTITION_STATES replay: applies the states the payload carries,
and drops the snapshots it
+ * names. Both in one lock acquisition, because a reader that saw the new
requirement while the
+ * snapshot was still there could let a transparent rewrite serve rows the
rebuild has to replace.
+ *
+ * <p>A payload without the states carries none, which is not the same as
an empty map that says the
+ * states are now empty: leaving them alone is the only answer that cannot
lose state.
+ */
+ public void replayAlterPartitionStates(Map<String, MTMVPartitionState>
partitionStates,
+ Set<String> removedSnapshotPartitions) {
writeMvLock();
try {
- this.partitionStates = MTMVPartitionState.copyOf(partitionStates);
+ if (partitionStates != null) {
+ this.partitionStates =
MTMVPartitionState.copyOf(partitionStates);
+ }
+ refreshSnapshot.removeSnapshots(removedSnapshotPartitions);
} finally {
writeMvUnlock();
}
}
- public void invalidateIvmBaseline() {
- EditLogItem editLogItem;
+ /**
+ * The {@code latestEpoch} of the given MV partitions, taken under the MV
read lock.
+ *
+ * <p>This is the value a refresh has to remember: what it read from the
base tables is described by
+ * the requirement in force when it started reading, so writing that value
back as the new
+ * {@code refreshEpoch} is what keeps an invalidation arriving mid-refresh
from being swallowed. A
+ * partition without an entry is left out -- a caller writes an epoch only
for what it captured.
+ */
+ public Map<String, Long> getLatestEpochs(Set<String> partitionNames) {
+ if (CollectionUtils.isEmpty(partitionNames)) {
+ return Collections.emptyMap();
+ }
+ // Sized before the lock: the state map is what needs it, and building
the map is not part of that.
+ Map<String, Long> res =
Maps.newHashMapWithExpectedSize(partitionNames.size());
+ readMvLock();
+ try {
+ for (String partitionName : partitionNames) {
+ MTMVPartitionState state = partitionStates.get(partitionName);
+ if (state != null) {
+ res.put(partitionName, state.getLatestEpoch());
+ }
+ }
+ return res;
+ } finally {
+ readMvUnlock();
+ }
+ }
+
+ /**
+ * Brings the partition states in line with the MV's partitions: every
partition gets an entry, and
+ * every entry whose partition is gone is dropped.
+ *
+ * <p>Alignment is what makes "the partition exists" and "the entry
exists" the same thing, and it is
+ * why an invalidation cannot miss: rows are only written by a refresh,
and every refresh aligns
+ * before it reads a base table, so a partition that holds rows always has
an entry for the mark to
+ * land on. The other direction is what makes the criterion safe -- an
entry created here describes a
+ * partition with no rows yet, so requiring one generation of it discards
no requirement that was
+ * made earlier.
+ *
+ * <p>What it changes is journaled, because the entry has to be on disk
before the rows it describes
+ * can be: a crash between this and the task result would otherwise leave
a partition that holds rows
+ * with no entry at all, and every later invalidation of it would find
nothing to land on. That is the
+ * one shape in which the criterion cannot be read -- "no entry" is
supposed to mean "no rows" -- so
+ * the entry is made durable before any base table is read rather than
derived again on the next run.
+ *
+ * <p>It is deliberately not a hook on every path that creates or drops a
partition. An entry is
+ * derived state, and rebuilding it from the live partition set also
repairs whatever a crash left
+ * behind: the drop of a partition and the removal of its entry are two
journal records, and only
+ * their order -- partition first -- is safe, which leaves at most a stale
entry that the next
+ * alignment drops.
+ *
+ * <p>Only an IVM MV is aligned. For a non-IVM MV the map stays as it is,
and every reader treats
+ * "empty" and "no state" the same.
+ */
+ public void alignPartitionStates(Set<String> livePartitionNames) {
+ if (!isIvm()) {
+ return;
+ }
+ // Copied up front: callers pass what OlapTable holds, and that is
mutated under the table's own
+ // write lock, not this one. Iterating the live collection could see
it change.
+ Set<String> livePartitions = Sets.newHashSet(livePartitionNames);
+ EditLogItem editLogItem = null;
writeMvLock();
try {
- if (ivmInfo == null) {
- ivmInfo = new IvmInfo();
+ boolean changed =
partitionStates.keySet().retainAll(livePartitions);
Review Comment:
[P1] Do not let an unversioned topology snapshot remove live partition
state. Overlapping manual refreshes are allowed, and `getPartitionNames()`
returns a detached copy before this method takes `mvRwLock`: task A can capture
names without p2, task B can create/fill/publish p2 and its state, then A's
late `retainAll` deletes that state. A following TRUNCATE cannot mark p2 or
remove its snapshot, and A can accept the empty delta and publish a
post-truncate snapshot while old MV rows remain (missing state is treated as
publishable). Couple the name snapshot to a partition-topology generation and
retry on change, and latch-test two refreshes across capture -> align ->
metadata DDL.
--
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]
