[improvement](mtmv) Read the base partitions an MV partition is recorded with

fe/fe-core/src/main/java/org/apache/doris/analysis/PartitionKeyDesc.java

@@ -26,6 +26,7 @@
 import com.google.common.collect.Lists;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 
 // Describe the partition key values in create table or add partition clause
@@ -64,6 +65,33 @@ public static PartitionKeyDesc createLessThan(List<PartitionValue> upperValues)
         return desc;
     }
 
+    /**
+     * The keys of a list desc in a canonical order, so that the same set of keys is the same desc wherever it
+     * is written down. {@link #equals} compares this list, and a hash set of keys iterates in the order it was
+     * filled in whenever two of them hash alike -- "Aa" and "BB" do -- so the set a desc was read from and the
+     * set another was built by unioning cannot be relied on to produce the same list.
+     */
+    public static List<List<PartitionValue>> sortedInValues(Collection<List<PartitionValue>> inValues) {
+        List<List<PartitionValue>> res = Lists.newArrayList(inValues);
+        res.sort(PartitionKeyDesc::compareKeys);
+        return res;
+    }
+
+    private static int compareKeys(List<PartitionValue> left, List<PartitionValue> right) {
+        for (int i = 0; i < Math.min(left.size(), right.size()); i++) {
+            PartitionValue leftValue = left.get(i);
+            PartitionValue rightValue = right.get(i);
+            int cmp = Boolean.compare(leftValue.isNullPartition(), rightValue.isNullPartition());
+            if (cmp == 0) {
+                cmp = leftValue.getStringValue().compareTo(rightValue.getStringValue());
+            }
+            if (cmp != 0) {
+                return cmp;
+            }
+        }
+        return Integer.compare(left.size(), right.size());
+    }
+
     public static PartitionKeyDesc createIn(List<List<PartitionValue>> inValues) {
         PartitionKeyDesc desc = new PartitionKeyDesc();
         desc.inValues = inValues;

fe/fe-core/src/main/java/org/apache/doris/catalog/ListPartitionItem.java

@@ -84,7 +84,7 @@ public PartitionItem getIntersect(PartitionItem newItem) {
     public PartitionKeyDesc toPartitionKeyDesc() {
         List<List<PartitionValue>> inValues = partitionKeys.stream().map(PartitionInfo::toPartitionValue)
                 .collect(Collectors.toList());
-        return PartitionKeyDesc.createIn(inValues);
+        return PartitionKeyDesc.createIn(PartitionKeyDesc.sortedInValues(inValues));
     }
 
     @Override
@@ -100,7 +100,7 @@ public PartitionKeyDesc toPartitionKeyDesc(int pos) throws AnalysisException {
             }
             res.add(Lists.newArrayList(values.get(pos)));
         }
-        return PartitionKeyDesc.createIn(Lists.newArrayList(res));
+        return PartitionKeyDesc.createIn(PartitionKeyDesc.sortedInValues(res));
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java

@@ -1679,6 +1679,23 @@ public Map<String, Map<MTMVRelatedTableIf, Set<String>>> calculatePartitionMappi
     public Map<String, Map<MTMVRelatedTableIf, Set<String>>> calculatePartitionMappings(
             Map<List<String>, Set<String>> queryUsedBaseTablePartitionMap,
             Map<MvccTableInfo, MvccSnapshot> pinnedSnapshots) throws AnalysisException {
+        return calculatePartitionMappings(queryUsedBaseTablePartitionMap, pinnedSnapshots, Sets.newHashSet());
+    }
+
+    /**
+     * The same, with every MV partition no base partition describes added to {@code undescribedMvPartitions}.
+     *
+     * <p>Such a partition is one the alignment has not caught up with: a base {@code ADD PARTITION} widens the
+     * keys an MV partition covers, so the alignment replaces it with an empty partition of the wider keys, and
+     * until that is done its own descriptor is one the base partitions no longer produce. Reading it as no
+     * table feeding it would write no rows over the rows it holds, so a refresh has to tell it apart from a
+     * partition that really is fed by nothing -- and the default list partition of a table is named in every
+     * MV partition's mapping, which is why the mapping being non-empty says nothing about this.
+     */
+    public Map<String, Map<MTMVRelatedTableIf, Set<String>>> calculatePartitionMappings(
+            Map<List<String>, Set<String>> queryUsedBaseTablePartitionMap,
+            Map<MvccTableInfo, MvccSnapshot> pinnedSnapshots,
+            Set<String> undescribedMvPartitions) throws AnalysisException {
         if (mvPartitionInfo.getPartitionType() == MTMVPartitionType.SELF_MANAGE) {
             return Maps.newHashMap();
         }
@@ -1697,9 +1714,15 @@ public Map<String, Map<MTMVRelatedTableIf, Set<String>>> calculatePartitionMappi
         Map<PartitionKeyDesc, Map<MTMVRelatedTableIf, Set<String>>> pctPartitionDescs = MTMVPartitionUtil
                 .generateRelatedPartitionDescs(mvPartitionInfo, mvProperties, getPartitionColumns(),
                         effectiveFilter, pinnedSnapshots);
+        Map<MTMVRelatedTableIf, String> defaultListPartitions = defaultListPartitionsOf();
         for (Entry<String, PartitionItem> entry : mvPartitionItems.entrySet()) {
-            res.put(entry.getKey(),
-                    pctPartitionDescs.getOrDefault(entry.getValue().toPartitionKeyDesc(), Maps.newHashMap()));
+            Map<MTMVRelatedTableIf, Set<String>> matched =
+                    pctPartitionDescs.get(entry.getValue().toPartitionKeyDesc());
+            if (matched == null) {
+                undescribedMvPartitions.add(entry.getKey());
+            }
+            res.put(entry.getKey(), withDefaultListPartitions(
+                    matched == null ? Maps.newHashMap() : matched, defaultListPartitions));
         }
         if (LOG.isDebugEnabled()) {
             LOG.debug("calculatePartitionMappings use [{}] mills, mvName is [{}]",
@@ -1708,6 +1731,65 @@ public Map<String, Map<MTMVRelatedTableIf, Set<String>>> calculatePartitionMappi
         return res;
     }
 
+    /**
+     * The list partition each base table of this MV has that takes the rows no other partition of it claims,
+     * by table, or none for a table that has no such partition.
+     *
+     * <p>Read once per mapping rather than per MV partition: the mapping describes every MV partition and the
+     * answer is the table's, not the partition's. The table's partitions are read under its read lock, so
+     * that a concurrent ADD or DROP PARTITION cannot be seen half applied -- its name list and the items the
+     * walk resolves against it have to come from one state of the table -- and so that this walk is not one
+     * more reader of a tree another thread is modifying.
+     */
+    private Map<MTMVRelatedTableIf, String> defaultListPartitionsOf() throws AnalysisException {
+        Map<MTMVRelatedTableIf, String> res = Maps.newHashMap();
+        for (MTMVRelatedTableIf pctTable : mvPartitionInfo.getPctTables()) {
+            if (!(pctTable instanceof OlapTable)) {
+                continue;
+            }
+            OlapTable olapTable = (OlapTable) pctTable;
+            if (!(olapTable.getPartitionInfo() instanceof ListPartitionInfo)) {
+                continue;
+            }
+            olapTable.readLock();
+            try {
+                for (String partitionName : olapTable.getPartitionNames()) {
+                    if (olapTable.getPartitionItemOrAnalysisException(partitionName).isDefaultPartition()) {
+                        res.put(pctTable, partitionName);
+                        break;
+                    }
+                }
+            } finally {
+                olapTable.readUnlock();
+            }
+        }
+        return res;
+    }
+
+    /**
+     * One MV partition's mapping, with every base table's default list partition named in it.
+     *
+     * <p>Such a partition holds rows for every key its table can be read by, so it belongs to every MV
+     * partition that reads the table -- not only to the one its own key, the sentinel those rows were placed
+     * by, maps to. Naming it everywhere is what the read and the record have to agree on: the refresh reads
+     * the rows of it that belong to the MV partition being refreshed, and the partition is recorded among the
+     * ones that partition is read through, so an insert into it leaves that MV partition out of sync instead
+     * of changing nothing the MV compares.
+     */
+    private Map<MTMVRelatedTableIf, Set<String>> withDefaultListPartitions(
+            Map<MTMVRelatedTableIf, Set<String>> mapping, Map<MTMVRelatedTableIf, String> defaultListPartitions) {
+        if (defaultListPartitions.isEmpty()) {
+            return mapping;
+        }
+        Map<MTMVRelatedTableIf, Set<String>> res = Maps.newHashMap(mapping);
+        for (Entry<MTMVRelatedTableIf, String> entry : defaultListPartitions.entrySet()) {
+            Set<String> partitions = Sets.newHashSet(res.getOrDefault(entry.getKey(), Sets.newHashSet()));
+            partitions.add(entry.getValue());
+            res.put(entry.getKey(), partitions);
+        }
+        return res;
+    }
+
     public ConcurrentLinkedQueue<MTMVTask> getHistoryTasks() {
         return jobInfo.getHistoryTasks();
     }

fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java

@@ -1293,6 +1293,13 @@ private void executePartitionBasedRefresh(MTMVRefreshContext context, RefreshMod
             int end = start + refreshPartitionNum;
             Set<String> execPartitionNames = Sets.newHashSet(partitions
                     .subList(start, Math.min(end, partitions.size())));
+            // What this batch reads and what it records are the same set, both decided here from the one
+            // mapping the snapshots below are generated from; see MTMVPartitionUtil#mappedBasePartitions.
+            Map<BaseTableInfo, Set<String>> readableBasePartitions = mtmv.isIvm()
+                    ? null : MTMVPartitionUtil.mappedBasePartitions(tableWithPartKey, context, execPartitionNames);
+            if (!mtmv.isIvm() && hasOlapPctTable()) {
+                requireDescribedPartitions(context, execPartitionNames);
+            }
             Map<BaseTableInfo, Set<Long>> batchResetPartitionIds = useIvmFallbackStreams
                     ? collectPctResetPartitionIds(context, execPartitionNames) : Maps.newHashMap();
             Optional<IvmRewriteContext> rewriteContext = Optional.empty();
@@ -1313,7 +1320,8 @@ private void executePartitionBasedRefresh(MTMVRefreshContext context, RefreshMod
                             execPartitionNames);
             try {
                 IvmPlanSignature batchPlanSignature = refreshPartitionsWithRetry(
-                        execPartitionNames, tableWithPartKey, rewriteContext, refreshMode);
+                        execPartitionNames, tableWithPartKey, readableBasePartitions, rewriteContext,
+                        refreshMode);
                 if (capturePlanSignature) {
                     batchPlanSignature = Objects.requireNonNull(batchPlanSignature,
                             "IVM COMPLETE refresh did not produce a plan signature");
@@ -1400,10 +1408,11 @@ private boolean hasUnusableIvmStream() {
      * Whether a stream this partition refresh will read is missing or unusable.
      *
      * <p>The set is narrower than the plan, and which tables are in it depends on the partitions being
-     * refreshed: a PCT table that no refreshed partition's mapping names keeps its place in the plan as
-     * an ordinary scan, so its stream is never read, and a table outside the plan's own tables is not
-     * scanned at all. Judging the whole plan instead sends a partition refresh that would have worked to
-     * a full rebuild whenever such an unread stream is missing.
+     * refreshed: a PCT table that no refreshed partition's mapping names is read as no rows at all -- its
+     * reset set is empty, and the rewriter answers it with an empty relation rather than a stream scan --
+     * so its stream is never read, and a table outside the plan's own tables is not scanned at all.
+     * Judging the whole plan instead sends a partition refresh that would have worked to a full rebuild
+     * whenever such an unread stream is missing.
      */
     private boolean hasUnusableIvmStreamForPartitions(MTMVRefreshContext context,
             List<String> mvPartitionNames) {
@@ -1496,9 +1505,67 @@ private void reconcileIvmStreams(ConnectContext ctx) throws JobException {
         }
     }
 
+    /**
+     * Whether any of the tables the MV is partitioned on is an olap one, which is where the check below can
+     * read anything. It reads a partition no base partition describes as the alignment having fallen behind an
+     * olap table's change, and that is a state only an olap table has: an external table describes its
+     * partitions from the snapshot the read is taken at, and one it cannot describe -- a row whose partition
+     * key is null, say -- is not a state the alignment can catch up with at all. So the check is taken where
+     * its reading can be true, and an MV partitioned on external tables alone is left to the read's own scope.
+     */
+    private boolean hasOlapPctTable() throws AnalysisException {
+        for (MTMVRelatedTableIf pctTable : mtmv.getMvPartitionInfo().getPctTables()) {
+            if (pctTable instanceof OlapTable) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    /**
+     * Whether each MV partition of this batch is one the mapping describes, i.e. one that some base partition's
+     * keys name. A partition nothing describes is one the alignment has not caught up with: a base `ADD
+     * PARTITION` widens the keys an MV partition covers, which replaces it with an empty partition of the new
+     * keys, and if that lands between the alignment and this mapping the partition is read as no table feeding
+     * it and the batch writes no rows over the rows it holds. The batch is not read at all then: the task tries
+     * again, and trying again realigns.
+     */
+    private void requireDescribedPartitions(MTMVRefreshContext context, Set<String> execPartitionNames)
+            throws AnalysisException {
+        for (String mvPartitionName : execPartitionNames) {
+            if (context.getUndescribedMvPartitions().contains(mvPartitionName)) {
+                throw new AnalysisException("MV partition " + mvPartitionName + " of " + mtmv.getName()
+                        + " is described by no base partition: the alignment has not caught up with a base "
+                        + "table change. Refreshing it would write no rows over the rows it holds.");
+            }
+        }
+    }
+
+    /**
+     * The partitions each PCT input of this batch is read from, by table, with an explicit empty set for a
+     * table the batch's mapping does not name.
+     *
+     * <p>An absent table is one whose retained base partitions share no key with the MV partitions this
+     * batch refreshes -- after a sync window dropped them, say -- so the batch must read no rows of it.
+     * Leaving the entry out instead is what makes the read differ from the record: the rewriter has no
+     * reset for the table and falls back to its ordinary scan, which reads the partitions the window
+     * dropped, while the mapping records none of them. The rows land in the MV partition and the snapshot
+     * says the table was not read, so a later change to them is never compared and a later delta never
+     * looks at them.
+     *
+     * <p>Seeded the same way {@link MTMVPartitionUtil#generateRelatedBasePartitionIds} seeds the delta's
+     * scope, and for the same reason: the empty set says "this table may not be read at all", which is not
+     * the same statement as the table's absence. The pre-seeded set is not looked up for a table the MV
+     * excludes from trigger tracking -- the rewriter returns its scan before consulting the reset.
+     */
     private Map<BaseTableInfo, Set<Long>> collectPctResetPartitionIds(MTMVRefreshContext context,
             Set<String> execPartitionNames) throws AnalysisException {
         Map<BaseTableInfo, Set<Long>> resetPartitionIds = Maps.newHashMap();
+        for (MTMVRelatedTableIf pctTable : mtmv.getMvPartitionInfo().getPctTables()) {
+            if (pctTable instanceof OlapTable) {
+                resetPartitionIds.put(new BaseTableInfo(pctTable), Sets.newHashSet());
+            }
+        }
         for (String mvPartitionName : execPartitionNames) {
             for (Entry<MTMVRelatedTableIf, Set<String>> entry
                     : context.getByPartitionName(mvPartitionName).entrySet()) {
@@ -1517,11 +1584,11 @@ private Map<BaseTableInfo, Set<Long>> collectPctResetPartitionIds(MTMVRefreshCon
     }
 
     private IvmPlanSignature refreshPartitionsWithRetry(Set<String> execPartitionNames,
-            Map<TableIf, String> tableWithPartKey,
+            Map<TableIf, String> tableWithPartKey, Map<BaseTableInfo, Set<String>> readableBasePartitions,
             Optional<IvmRewriteContext> rewriteContext, RefreshMode refreshMode)
             throws Exception {
         return executeWithRetry(() -> refreshPartitions(execPartitionNames, tableWithPartKey,
-                        rewriteContext, refreshMode),
+                        readableBasePartitions, rewriteContext, refreshMode),
                 "partition refresh, execPartitionNames=" + execPartitionNames);
     }
 
@@ -1577,7 +1644,7 @@ private boolean needRetry(Exception exception) {
     }
 
     private IvmPlanSignature refreshPartitions(Set<String> refreshPartitionNames,
-            Map<TableIf, String> tableWithPartKey,
+            Map<TableIf, String> tableWithPartKey, Map<BaseTableInfo, Set<String>> readableBasePartitions,
             Optional<IvmRewriteContext> rewriteContext, RefreshMode refreshMode)
             throws Exception {
         // Create the MTMV context before parsing the MV definition SQL so SET_VAR hints
@@ -1601,7 +1668,8 @@ private IvmPlanSignature refreshPartitions(Set<String> refreshPartitionNames,
         // if SELF_MANAGE mv, only have default partition,  will not have partitionItem, so we give empty set
         UpdateMvByPartitionCommand command = UpdateMvByPartitionCommand
                 .from(mtmv, mtmv.getMvPartitionInfo().getPartitionType() != MTMVPartitionType.SELF_MANAGE
-                        ? refreshPartitionNames : Sets.newHashSet(), tableWithPartKey, statementContext);
+                        ? refreshPartitionNames : Sets.newHashSet(), tableWithPartKey, statementContext,
+                        readableBasePartitions);
         setupComputeGroup(mtmvCtx);
         AtomicReference<IvmPlanSignature> signatureRef = new AtomicReference<>();
         try {