[fix](ivm) Refresh the surviving partitions after an IVM baseline rebuild

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

@@ -645,6 +645,41 @@ public void invalidateIvmBaseline(BaseTableInfo baseTableInfo, Map<String, Long>
         editLogItem.await();
     }
 
+    /**
+     * Release the IVM baseline barrier after the partitions it named have been rebuilt, or after
+     * partition sync removed them (a dropped partition resolves its own entry: the partition and its
+     * IVM offsets are both gone).
+     *
+     * <p>Guarded by schemaChangeVersion, like {@link #persistIvmBaselineGuard}: a base-table change
+     * landing while the rebuild runs carries its own barrier entry, and a blind clear would swallow
+     * it. Failing instead preserves that entry -- the next refresh rebuilds it together with the
+     * partitions this task handled.
+     *
+     * <p>Journals the new state right away, like every other ivmInfo mutation here. A task that dies
+     * before {@link #addTaskResult} would otherwise leave the release in memory only, and a restart
+     * would resurrect the barrier from disk.
+     */
+    public void releaseIvmBaselineRebuild(long expectedSchemaChangeVersion) throws JobException {
+        EditLogItem editLogItem;
+        writeMvLock();
+        try {
+            if (ivmInfo == null || !ivmInfo.isBaselineRebuildRequired()) {
+                // Nothing to release: skip both the mutation and the journal entry. Any base-table
+                // change that raced us in is still caught by validateIvmRefreshStart() below.
+                return;
+            }
+            if (schemaChangeVersion != expectedSchemaChangeVersion) {
+                throw new JobException("Base table metadata changed before IVM baseline refresh, mv="
+                        + getName());
+            }
+            ivmInfo.clearBaselineRebuild();
+            editLogItem = submitIvmInfoChange();
+        } finally {
+            writeMvUnlock();
+        }
+        editLogItem.await();
+    }
+
     public void persistIvmBaselineGuard(RefreshMode refreshMode, Set<String> baselinePartitions,
             long expectedSchemaChangeVersion) throws JobException {
         EditLogItem editLogItem;

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

@@ -319,9 +319,7 @@ public void run() throws JobException {
                 throw new JobException(e.getMessage(), e);
             }
             MTMVRefreshContext refreshContext = buildRefreshContext(tableIfs);
-            if (handlePendingIvmBaselineRebuild(refreshContext, request, ctx)) {
-                return;
-            }
+            handlePendingIvmBaselineRebuild(refreshContext, request, ctx, attempts);
             boolean disablePartitionRefresh = false;
             for (RefreshAttemptType attemptType : attempts) {
                 switch (attemptType) {
@@ -558,28 +556,56 @@ private void executeCompleteAttempt(MTMVRefreshContext context, ConnectContext c
         executePartitionBasedRefresh(context, RefreshMode.COMPLETE, ctx);
     }
 
-    private boolean handlePendingIvmBaselineRebuild(MTMVRefreshContext context, RefreshRequest request,
-            ConnectContext ctx)
+    /**
+     * Rebuild the MV partitions whose IVM baseline is broken, before the normal refresh runs.
+     *
+     * <p>This is a pre-step, not a terminal branch: the caller keeps running {@code attempts}
+     * afterwards, so a broken baseline no longer skips the refresh entirely. The list is rewritten
+     * in place when the baseline demands a different set of attempts.
+     *
+     * <p>Partition sync drops the MV partitions whose base partition disappeared, which is exactly
+     * what the barrier recorded when that base partition was dropped. Those partitions are resolved
+     * by the drop itself (the partition and its IVM offsets are both gone), so only the partitions
+     * that still exist need a rebuild. The barrier is released either way, otherwise the IVM attempt
+     * that follows would be rejected by {@link MTMV#validateIvmRefreshStart}.
+     */
+    private void handlePendingIvmBaselineRebuild(MTMVRefreshContext context,
+            RefreshRequest request, ConnectContext ctx, List<RefreshAttemptType> attempts)
             throws JobException, AnalysisException {
         if (!mtmv.isIvm() || request.refreshMode == RefreshMode.COMPLETE
                 || !mtmv.getIvmInfo().isBaselineRebuildRequired()) {
-            return false;
+            return;
         }
         ivmFallbackReason = IvmFailureReason.BINLOG_BROKEN.name();
         IvmInfo ivmInfo = mtmv.getIvmInfo();
+        // A lone COMPLETE attempt rebuilds every partition anyway, so a partial pre-rebuild here
+        // would be redundant; it also releases the barrier by itself once it succeeds.
+        if (attempts.size() == 1 && attempts.get(0) == RefreshAttemptType.COMPLETE) {
+            LOG.info("IVM baseline barrier is covered by the pending COMPLETE attempt, mv={}, taskId={}",
+                    mtmv.getName(), getTaskId());
+            return;
+        }
         if (ivmInfo.requiresCompleteBaselineRebuild()) {
-            executeCompleteAttempt(context, ctx);
-            return true;
+            LOG.warn("IVM baseline requires a complete rebuild, mv={}, taskId={}. "
+                    + "Continuing with COMPLETE refresh.", mtmv.getName(), getTaskId());
+            attempts.clear();
+            attempts.add(RefreshAttemptType.COMPLETE);
+            return;
         }
-        this.needRefreshPartitions = Lists.newArrayList(Sets.intersection(
+        List<String> baselinePartitions = Lists.newArrayList(Sets.intersection(
                 ivmInfo.getPendingBaselineRebuildPartitions(), mtmv.getPartitionNames()));
-        this.needRefreshPartitions.sort(String::compareTo);
-        this.refreshMode = generateRefreshMode(needRefreshPartitions);
-        if (refreshMode == MTMVTaskRefreshMode.NOT_REFRESH) {
-            return true;
+        if (baselinePartitions.isEmpty()) {
+            // Partition sync has already dropped every partition the barrier named, so there is
+            // nothing left to rebuild. The surviving partitions are picked up by the attempts below.
+            LOG.info("IVM baseline partitions were removed by partition sync, mv={}, taskId={}",
+                    mtmv.getName(), getTaskId());
+        } else {
+            baselinePartitions.sort(String::compareTo);
+            this.needRefreshPartitions = baselinePartitions;
+            this.refreshMode = generateRefreshMode(baselinePartitions);
+            executePartitionBasedRefresh(context, RefreshMode.PARTITIONS, ctx);
         }
-        executePartitionBasedRefresh(context, RefreshMode.PARTITIONS, ctx);
-        return true;
+        mtmv.releaseIvmBaselineRebuild(mtmvSchemaChangeVersion);
     }
 
     private void validateIvmBaselineBeforePartitionSync(RefreshRequest request) throws JobException {

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java

@@ -86,16 +86,16 @@ public Set<BaseTableInfo> getMtmvsByBaseTableOneLevelAndFromView(BaseTableInfo t
     }
 
     public void markIvmBaselineRebuild(BaseTableInfo baseTableInfo, String reason) {
-        markIvmBaselineRebuild(baseTableInfo, Collections.emptyMap(), reason);
+        markIvmBaselineRebuild(baseTableInfo, true, Collections.emptyMap(), reason);
     }
 
     public void markIvmBaselineRebuildForPartitionChange(BaseTableInfo baseTableInfo,
             Map<String, Long> changedPartitions, String reason) {
         Preconditions.checkArgument(!changedPartitions.isEmpty(), "changed partitions can not be empty");
-        markIvmBaselineRebuild(baseTableInfo, changedPartitions, reason);
+        markIvmBaselineRebuild(baseTableInfo, false, changedPartitions, reason);
     }
 
-    private void markIvmBaselineRebuild(BaseTableInfo baseTableInfo,
+    private void markIvmBaselineRebuild(BaseTableInfo baseTableInfo, boolean allPartitionsChanged,
             Map<String, Long> changedPartitions, String reason) {
         TableNameInfo baseTableName = new TableNameInfo(baseTableInfo.getCtlName(),
                 baseTableInfo.getDbName(), baseTableInfo.getTableName());
@@ -115,7 +115,7 @@ private void markIvmBaselineRebuild(BaseTableInfo baseTableInfo,
             if (MTMVPartitionUtil.isTableExcluded(mtmv.getExcludedTriggerTables(), baseTableName)) {
                 continue;
             }
-            if (changedPartitions.isEmpty()) {
+            if (allPartitionsChanged) {
                 mtmv.invalidateIvmBaseline();
             } else {
                 mtmv.invalidateIvmBaseline(baseTableInfo, changedPartitions);

fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java

@@ -816,12 +816,46 @@ public void testPartitionsFallbackRebuildsPendingBaselineWithComplete() throws E
         Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
 
         Deencapsulation.invoke(task, "validateIvmBaselineBeforePartitionSync", request);
-        Assertions.assertTrue((Boolean) Deencapsulation.invoke(task, "handlePendingIvmBaselineRebuild",
-                Mockito.mock(MTMVRefreshContext.class), request, new ConnectContext()));
-        Assertions.assertEquals(MTMVTask.MTMVTaskRefreshMode.NOT_REFRESH,
-                Deencapsulation.getField(task, "refreshMode"));
+        List<Object> attempts = Lists.newArrayList();
+        attempts.addAll(Deencapsulation.invoke(task, "buildAttempts", request, false));
+        Assertions.assertEquals("[PARTITIONS, COMPLETE]", attempts.toString());
+
+        Deencapsulation.invoke(task, "handlePendingIvmBaselineRebuild",
+                Mockito.mock(MTMVRefreshContext.class), request, new ConnectContext(), attempts);
+
+        // A pending COMPLETE rebuild reshapes the attempt list instead of rebuilding inline, so
+        // PARTITIONS FALLBACK rebuilds the whole MV through the COMPLETE attempt it keeps.
+        Assertions.assertEquals("[COMPLETE]", attempts.toString());
+        Assertions.assertEquals(IvmFailureReason.BINLOG_BROKEN.name(),
+                Deencapsulation.getField(task, "ivmFallbackReason"));
+        // The barrier is released by the caller once the reshaped attempts have run.
+        Mockito.verify(mtmv, Mockito.never()).releaseIvmBaselineRebuild(Mockito.anyLong());
+    }
+
+    @Test
+    public void testDroppedBaselinePartitionsReleaseBarrierWithoutRebuild() throws Exception {
+        Mockito.when(mtmv.isIvm()).thenReturn(true);
+        IvmInfo ivmInfo = new IvmInfo();
+        ivmInfo.addPendingBaselineRebuildPartitions(Sets.newHashSet(poneName));
+        Mockito.when(mtmv.getIvmInfo()).thenReturn(ivmInfo);
+        // Partition sync already dropped the partition the barrier named, so nothing is left to
+        // pre-rebuild and the surviving partitions catch up through the attempts themselves.
+        Mockito.when(mtmv.getPartitionNames()).thenReturn(Sets.newHashSet(ptwoName));
+        MTMVTask task = new MTMVTask(mtmv, relation, MTMVTaskContext.of(
+                MTMVTaskTriggerMode.MANUAL, null, RefreshMode.PARTITIONS, true, null));
+        Deencapsulation.setField(task, "mtmvSchemaChangeVersion", 7L);
+        Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
+
+        List<Object> attempts = Lists.newArrayList();
+        attempts.addAll(Deencapsulation.invoke(task, "buildAttempts", request, false));
+        Deencapsulation.invoke(task, "handlePendingIvmBaselineRebuild",
+                Mockito.mock(MTMVRefreshContext.class), request, new ConnectContext(), attempts);
+
+        Assertions.assertEquals("[PARTITIONS, COMPLETE]", attempts.toString());
+        Assertions.assertNull(Deencapsulation.getField(task, "refreshMode"));
         Assertions.assertEquals(IvmFailureReason.BINLOG_BROKEN.name(),
                 Deencapsulation.getField(task, "ivmFallbackReason"));
+        Mockito.verify(mtmv).releaseIvmBaselineRebuild(7L);
     }
 
     @Test

regression-test/data/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.out

@@ -0,0 +1,32 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !baseline_task --
+SUCCESS	NONE	NONE
+
+-- !baseline_base --
+2026-01-10	1	10
+2026-01-10	1	10
+2026-02-10	3	30
+2026-02-10	3	30
+
+-- !baseline_mv --
+2026-01-10	1	10
+2026-01-10	1	10
+2026-02-10	3	30
+2026-02-10	3	30
+
+-- !strict_task --
+FAILED	NOT_REFRESH	BINLOG_BROKEN
+
+-- !fallback_task --
+SUCCESS	PARTIAL	BINLOG_BROKEN
+
+-- !fallback_base --
+2026-02-10	3	30
+2026-02-10	3	30
+2026-02-15	4	40
+
+-- !fallback_mv --
+2026-02-10	3	30
+2026-02-10	3	30
+2026-02-15	4	40
+

regression-test/data/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.out

@@ -0,0 +1,45 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !baseline_task --
+SUCCESS	NONE
+
+-- !baseline_base --
+2026-01-10	1	10
+2026-02-10	2	20
+2026-03-10	3	30
+
+-- !baseline_mv --
+2026-01-10	1	10
+2026-02-10	2	20
+2026-03-10	3	30
+
+-- !strict_task --
+FAILED	BINLOG_BROKEN
+
+-- !fallback_task --
+SUCCESS	BINLOG_BROKEN
+
+-- !fallback_base --
+2026-02-10	2	20
+2026-02-15	4	40
+2026-03-10	3	30
+
+-- !fallback_mv --
+2026-02-10	2	20
+2026-02-15	4	40
+2026-03-10	3	30
+
+-- !resumed_task --
+SUCCESS	NONE
+
+-- !resumed_base --
+2026-02-10	2	20
+2026-02-15	4	40
+2026-03-10	3	30
+2026-03-15	5	50
+
+-- !resumed_mv --
+2026-02-10	2	20
+2026-02-15	4	40
+2026-03-10	3	30
+2026-03-15	5	50
+