fix(pipe): reuse safe local progress on alter

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java

@@ -38,6 +38,7 @@
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMeta;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInAgent;
@@ -102,6 +103,7 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiPredicate;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
@@ -199,6 +201,8 @@ public List<TPushPipeMetaRespExceptionMessage> handlePipeMetaChangesInternal(
       return Collections.emptyList();
     }
 
+    carryOverLocalProgressIndexForAlter(pipeMetaListFromCoordinator);
+
     final List<TPushPipeMetaRespExceptionMessage> exceptionMessages =
         super.handlePipeMetaChangesInternal(pipeMetaListFromCoordinator);
 
@@ -217,6 +221,88 @@ public List<TPushPipeMetaRespExceptionMessage> handlePipeMetaChangesInternal(
     return exceptionMessages;
   }
 
+  /**
+   * Carry the committed progress of an old local task into an altered task when it is safe to do
+   * so. The old task is dropped before the new task is created, therefore this must run before
+   * {@link PipeTaskAgent#handlePipeMetaChangesInternal(List)} starts applying the metadata list.
+   *
+   * <p>We deliberately only carry progress when the old and new task stay on this DataNode and
+   * their realtime-only modes are unchanged. Mode changes have explicit progress semantics in the
+   * ConfigNode metadata (for example, realtime-only to historical resets to {@code
+   * MinimumProgressIndex}), and leader changes must use the coordinator checkpoint because the old
+   * task is not local to the new leader.
+   */
+  private void carryOverLocalProgressIndexForAlter(
+      final List<PipeMeta> pipeMetaListFromCoordinator) {
+    for (final PipeMeta droppedPipeMeta : pipeMetaListFromCoordinator) {
+      if (droppedPipeMeta.getRuntimeMeta().getStatus().get() != PipeStatus.DROPPED) {
+        continue;
+      }
+
+      final PipeStaticMeta oldStaticMeta = droppedPipeMeta.getStaticMeta();
+      final PipeMeta localOldPipeMeta = pipeMetaKeeper.getPipeMeta(oldStaticMeta);
+      if (localOldPipeMeta == null) {
+        continue;
+      }
+
+      for (final PipeMeta updatedPipeMeta : pipeMetaListFromCoordinator) {
+        if (updatedPipeMeta == droppedPipeMeta
+            || updatedPipeMeta.getRuntimeMeta().getStatus().get() == PipeStatus.DROPPED
+            || !oldStaticMeta.getPipeName().equals(updatedPipeMeta.getStaticMeta().getPipeName())
+            || oldStaticMeta.visibleUnderTableModel()
+                != updatedPipeMeta.getStaticMeta().visibleUnderTableModel()) {
+          continue;
+        }
+
+        carryOverLocalProgressIndexForAlter(
+            oldStaticMeta,
+            localOldPipeMeta,
+            updatedPipeMeta,
+            CONFIG.getDataNodeId(),
+            (staticMeta, consensusGroupId) ->
+                pipeTaskManager.getPipeTask(staticMeta, consensusGroupId) != null);
+      }
+    }
+  }
+
+  static void carryOverLocalProgressIndexForAlter(
+      final PipeStaticMeta oldStaticMeta,
+      final PipeMeta localOldPipeMeta,
+      final PipeMeta updatedPipeMeta,
+      final int localNodeId,
+      final BiPredicate<PipeStaticMeta, Integer> localTaskExists) {
+    final PipeStaticMeta updatedStaticMeta = updatedPipeMeta.getStaticMeta();
+
+    // A mode change has an explicit cutover/reset meaning in ConfigNode. In particular, a
+    // realtime-only -> historical alter must retain MinimumProgressIndex to scan old files.
+    if (PipeTaskAgent.isRealtimeOnlyPipe(oldStaticMeta.getSourceParameters())
+        != PipeTaskAgent.isRealtimeOnlyPipe(updatedStaticMeta.getSourceParameters())) {
+      return;
+    }
+
+    final Map<Integer, PipeTaskMeta> localTaskMetaMap =
+        localOldPipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();
+    final Map<Integer, PipeTaskMeta> updatedTaskMetaMap =
+        updatedPipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();
+
+    for (final Map.Entry<Integer, PipeTaskMeta> entry : updatedTaskMetaMap.entrySet()) {
+      final int consensusGroupId = entry.getKey();
+      final PipeTaskMeta updatedTaskMeta = entry.getValue();
+      final PipeTaskMeta localTaskMeta = localTaskMetaMap.get(consensusGroupId);
+
+      // Only the old task's actual leader owns an authoritative local checkpoint. Requiring the
+      // new task to stay on the same node also avoids losing the checkpoint during leader change.
+      if (localTaskMeta == null
+          || localTaskMeta.getLeaderNodeId() != localNodeId
+          || updatedTaskMeta.getLeaderNodeId() != localNodeId
+          || !localTaskExists.test(oldStaticMeta, consensusGroupId)) {
+        continue;
+      }
+
+      updatedTaskMeta.updateProgressIndex(localTaskMeta.getProgressIndex());
+    }
+  }
+
   private Set<Integer> clearSchemaRegionListeningQueueIfNecessary(
       final List<PipeMeta> pipeMetaListFromCoordinator) throws IllegalPathException {
     final Map<Integer, Long> schemaRegionId2ListeningQueueNewFirstIndex = new HashMap<>();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java

@@ -1403,6 +1403,23 @@ public TPushPipeMetaRespExceptionMessage handleDropPipe(final String pipeName) {
             return PipeDataNodeAgent.task().handleDropPipe(pipeName);
           }
 
+          @Override
+          public boolean handlePipeMetaChanges(
+              final List<ByteBuffer> pipeMetas,
+              final List<TPushPipeMetaRespExceptionMessage> exceptionMessages) {
+            final List<TPushPipeMetaRespExceptionMessage> exceptionMessagesFromAgent =
+                PipeDataNodeAgent.task()
+                    .handlePipeMetaChanges(
+                        pipeMetas.stream()
+                            .map(PipeMeta::deserialize4TaskAgent)
+                            .collect(Collectors.toList()));
+            if (exceptionMessagesFromAgent == null) {
+              return false;
+            }
+            exceptionMessages.addAll(exceptionMessagesFromAgent);
+            return true;
+          }
+
           @Override
           public TPushPipeMetaRespExceptionMessage handleSinglePipeMeta(final ByteBuffer pipeMeta) {
             return PipeDataNodeAgent.task()

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/PushMultiPipeMetaHelper.java

@@ -46,6 +46,27 @@ interface Handler {
     TPushPipeMetaRespExceptionMessage handleDropPipe(String pipeName) throws Exception;
 
     TPushPipeMetaRespExceptionMessage handleSinglePipeMeta(ByteBuffer pipeMeta) throws Exception;
+
+    /**
+     * Handles all pipe metadata in one agent invocation. Alter pipe sends the old dropped metadata
+     * and the new metadata together, so they must be visible to the agent at the same time when it
+     * decides whether the old task's local progress can be reused.
+     *
+     * <p>The default implementation preserves the per-metadata behavior for handlers that do not
+     * need batch processing.
+     */
+    default boolean handlePipeMetaChanges(
+        final List<ByteBuffer> pipeMetas,
+        final List<TPushPipeMetaRespExceptionMessage> exceptionMessages)
+        throws Exception {
+      for (final ByteBuffer pipeMeta : pipeMetas) {
+        final TPushPipeMetaRespExceptionMessage message = handleSinglePipeMeta(pipeMeta);
+        if (message != null) {
+          exceptionMessages.add(message);
+        }
+      }
+      return true;
+    }
   }
 
   static TPushPipeMetaResp pushMultiPipeMeta(
@@ -60,11 +81,9 @@ static TPushPipeMetaResp pushMultiPipeMeta(
           }
         }
       } else if (req.isSetPipeMetas()) {
-        for (final ByteBuffer pipeMeta : req.getPipeMetas()) {
-          final TPushPipeMetaRespExceptionMessage message = handler.handleSinglePipeMeta(pipeMeta);
-          if (message != null) {
-            exceptionMessages.add(message);
-          }
+        if (!handler.handlePipeMetaChanges(req.getPipeMetas(), exceptionMessages)) {
+          return new TPushPipeMetaResp()
+              .setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_TIMEOUT.getStatusCode()));
         }
       } else {
         throw new Exception(DataNodeMiscMessages.INVALID_PUSH_MULTI_PIPE_META_REQ);

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java

@@ -20,19 +20,30 @@
 package org.apache.iotdb.db.pipe.agent.task;
 
 import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
 import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
 import org.apache.iotdb.pipe.api.exception.PipeException;
 
 import org.junit.Assert;
 import org.junit.Test;
 
 import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 public class PipeDataNodeTaskAgentTest {
 
+  private static final int LOCAL_NODE_ID = 1;
+  private static final int REGION_ID = 7;
+
   @Test
   public void testCreateMemoryCheckStillRunsWhenNoPipeTasksNeedToBeCreated() throws Exception {
     final boolean originalPipeEnableMemoryCheck =
@@ -68,4 +79,152 @@ public void testCreateMemoryCheckStillRunsWhenNoPipeTasksNeedToBeCreated() throw
           .setPipeTotalFloatingMemoryProportion(originalPipeTotalFloatingMemoryProportion);
     }
   }
+
+  @Test
+  public void testCarryOverCommittedProgressForResumeAlter() {
+    final PipeStaticMeta oldStaticMeta = createStaticMeta(1, false);
+    final PipeStaticMeta updatedStaticMeta = createStaticMeta(2, false);
+    final PipeMeta localOldPipeMeta =
+        createPipeMeta(oldStaticMeta, new SimpleProgressIndex(1, 20L), LOCAL_NODE_ID);
+    final PipeMeta updatedPipeMeta =
+        createPipeMeta(updatedStaticMeta, new SimpleProgressIndex(1, 10L), LOCAL_NODE_ID);
+
+    PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter(
+        oldStaticMeta,
+        localOldPipeMeta,
+        updatedPipeMeta,
+        LOCAL_NODE_ID,
+        (staticMeta, regionId) -> regionId == REGION_ID);
+
+    Assert.assertEquals(
+        new SimpleProgressIndex(1, 20L),
+        updatedPipeMeta
+            .getRuntimeMeta()
+            .getConsensusGroupId2TaskMetaMap()
+            .get(REGION_ID)
+            .getProgressIndex());
+  }
+
+  @Test
+  public void testCarryOverDoesNotOverrideCoordinatorProgress() {
+    final PipeStaticMeta oldStaticMeta = createStaticMeta(1, false);
+    final PipeStaticMeta updatedStaticMeta = createStaticMeta(2, false);
+    final PipeMeta localOldPipeMeta =
+        createPipeMeta(oldStaticMeta, new SimpleProgressIndex(1, 10L), LOCAL_NODE_ID);
+    final PipeMeta updatedPipeMeta =
+        createPipeMeta(updatedStaticMeta, new SimpleProgressIndex(1, 20L), LOCAL_NODE_ID);
+
+    PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter(
+        oldStaticMeta,
+        localOldPipeMeta,
+        updatedPipeMeta,
+        LOCAL_NODE_ID,
+        (staticMeta, regionId) -> true);
+
+    Assert.assertEquals(
+        new SimpleProgressIndex(1, 20L),
+        updatedPipeMeta
+            .getRuntimeMeta()
+            .getConsensusGroupId2TaskMetaMap()
+            .get(REGION_ID)
+            .getProgressIndex());
+  }
+
+  @Test
+  public void testCarryOverDoesNotOverrideProgressResetOnModeChange() {
+    final PipeStaticMeta oldStaticMeta = createStaticMeta(1, false);
+    final PipeStaticMeta updatedStaticMeta = createStaticMeta(2, true);
+    final PipeMeta localOldPipeMeta =
+        createPipeMeta(oldStaticMeta, new SimpleProgressIndex(1, 20L), LOCAL_NODE_ID);
+    final PipeMeta updatedPipeMeta =
+        createPipeMeta(updatedStaticMeta, MinimumProgressIndex.INSTANCE, LOCAL_NODE_ID);
+
+    PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter(
+        oldStaticMeta,
+        localOldPipeMeta,
+        updatedPipeMeta,
+        LOCAL_NODE_ID,
+        (staticMeta, regionId) -> true);
+
+    Assert.assertSame(
+        MinimumProgressIndex.INSTANCE,
+        updatedPipeMeta
+            .getRuntimeMeta()
+            .getConsensusGroupId2TaskMetaMap()
+            .get(REGION_ID)
+            .getProgressIndex());
+  }
+
+  @Test
+  public void testCarryOverRequiresStableLeaderAndLocalTask() {
+    final PipeStaticMeta oldStaticMeta = createStaticMeta(1, false);
+    final PipeStaticMeta updatedStaticMeta = createStaticMeta(2, false);
+    final PipeMeta localOldPipeMeta =
+        createPipeMeta(oldStaticMeta, new SimpleProgressIndex(1, 20L), LOCAL_NODE_ID);
+
+    final PipeMeta localOldPipeMetaWithLeaderChange =
+        createPipeMeta(oldStaticMeta, new SimpleProgressIndex(1, 20L), 2);
+    final PipeMeta updatedWithOldLeaderChange =
+        createPipeMeta(updatedStaticMeta, new SimpleProgressIndex(1, 10L), LOCAL_NODE_ID);
+    PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter(
+        oldStaticMeta,
+        localOldPipeMetaWithLeaderChange,
+        updatedWithOldLeaderChange,
+        LOCAL_NODE_ID,
+        (staticMeta, regionId) -> true);
+    Assert.assertEquals(
+        new SimpleProgressIndex(1, 10L),
+        updatedWithOldLeaderChange
+            .getRuntimeMeta()
+            .getConsensusGroupId2TaskMetaMap()
+            .get(REGION_ID)
+            .getProgressIndex());
+
+    final PipeMeta updatedWithLeaderChange =
+        createPipeMeta(updatedStaticMeta, new SimpleProgressIndex(1, 10L), 2);
+    PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter(
+        oldStaticMeta,
+        localOldPipeMeta,
+        updatedWithLeaderChange,
+        LOCAL_NODE_ID,
+        (staticMeta, regionId) -> true);
+    Assert.assertEquals(
+        new SimpleProgressIndex(1, 10L),
+        updatedWithLeaderChange
+            .getRuntimeMeta()
+            .getConsensusGroupId2TaskMetaMap()
+            .get(REGION_ID)
+            .getProgressIndex());
+
+    final PipeMeta updatedWithoutLocalTask =
+        createPipeMeta(updatedStaticMeta, new SimpleProgressIndex(1, 10L), LOCAL_NODE_ID);
+    PipeDataNodeTaskAgent.carryOverLocalProgressIndexForAlter(
+        oldStaticMeta,
+        localOldPipeMeta,
+        updatedWithoutLocalTask,
+        LOCAL_NODE_ID,
+        (staticMeta, regionId) -> false);
+    Assert.assertEquals(
+        new SimpleProgressIndex(1, 10L),
+        updatedWithoutLocalTask
+            .getRuntimeMeta()
+            .getConsensusGroupId2TaskMetaMap()
+            .get(REGION_ID)
+            .getProgressIndex());
+  }
+
+  private PipeStaticMeta createStaticMeta(final long creationTime, final boolean historyEnabled) {
+    final Map<String, String> sourceAttributes = new HashMap<>();
+    sourceAttributes.put(
+        PipeSourceConstant.SOURCE_HISTORY_ENABLE_KEY, Boolean.toString(historyEnabled));
+    return new PipeStaticMeta(
+        String.valueOf('p'), creationTime, sourceAttributes, new HashMap<>(), new HashMap<>());
+  }
+
+  private PipeMeta createPipeMeta(
+      final PipeStaticMeta staticMeta, final ProgressIndex progressIndex, final int leaderId) {
+    final ConcurrentMap<Integer, PipeTaskMeta> taskMetaMap = new ConcurrentHashMap<>();
+    taskMetaMap.put(REGION_ID, new PipeTaskMeta(progressIndex, leaderId));
+    return new PipeMeta(staticMeta, new PipeRuntimeMeta(taskMetaMap));
+  }
 }