yujun777 commented on code in PR #68390:
URL: https://github.com/apache/doris/pull/68390#discussion_r4089236142
##########
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:
Fixed: `Locale.ROOT` on both the insert and the lookup.
##########
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:
Fixed by moving the read rather than versioning it: `alignPartitionStates`
reads the MV's partition names itself, under the same lock as the map it edits,
so a caller's snapshot taken before the lock is gone and the caller cannot be
stale.
What remains is the window between that read and the `retainAll`, both
adjacent statements under the MV lock -- it is not closed by this lock, because
the MV's partition map is mutated under the table's lock, not this one. Closing
it needs the entry removal to happen where the partition DDL does, which is a
larger change than this one; say the word and I will take it up separately.
--
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]
