fix(pipe): reuse safe local progress on alter

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAlterIT.java

@@ -56,6 +56,15 @@ public void setUp() {
     super.setUp();
   }
 
+  @Override
+  protected void setupConfig() {
+    super.setupConfig();
+    senderEnv
+        .getConfig()
+        .getCommonConfig()
+        .setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(600);
+  }
+
   @Test
   public void testBasicAlterPipe() throws Exception {
     final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
@@ -615,4 +624,43 @@ public void testAlterPipeRealtime() {
         "count(timeseries),",
         Collections.singleton("1,"));
   }
+
+  @Test
+  public void testAlterPipeDoesNotResendCommittedData() {
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList("insert into root.db.d1(time, s1) values (1, 1), (2, 2)", "flush"),
+        null);
+
+    TestUtils.executeNonQuery(
+        senderEnv,
+        String.format(
+            "create pipe a2b with source ('source.realtime.mode'='stream') with sink ('node-urls'='%s', 'sink.batch.enable'='false')",
+            receiverDataNode.getIpAndPortString()),
+        null);
+
+    final Set<String> oldData = new HashSet<>(Arrays.asList("1,1.0,", "2,2.0,"));
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", oldData);
+
+    TestUtils.executeNonQuery(
+        receiverEnv, "delete from root.db.d1.s1 where time >= 1 and time <= 2", null);
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", Collections.emptySet());
+
+    TestUtils.executeNonQuery(
+        senderEnv, "alter pipe a2b modify sink ('sink.batch.enable'='true')", null);
+    TestUtils.assertDataAlwaysOnEnv(
+        receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", Collections.emptySet());
+
+    TestUtils.executeNonQueries(
+        senderEnv, Arrays.asList("insert into root.db.d1(time, s1) values (3, 3)", "flush"), null);
+    final Set<String> newData = Collections.singleton("3,3.0,");
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", newData);
+    TestUtils.assertDataAlwaysOnEnv(
+        receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", newData);
+  }
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java

@@ -23,6 +23,7 @@
 import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
 import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
 import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp;
 import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.commons.cluster.NodeStatus;
@@ -71,6 +72,7 @@
 import org.apache.iotdb.mpp.rpc.thrift.TInactiveTriggerInstanceReq;
 import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
 import org.apache.iotdb.mpp.rpc.thrift.TNotifyRegionMigrationReq;
+import org.apache.iotdb.mpp.rpc.thrift.TPipeHeartbeatReq;
 import org.apache.iotdb.mpp.rpc.thrift.TPullCommitProgressReq;
 import org.apache.iotdb.mpp.rpc.thrift.TPullCommitProgressResp;
 import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaReq;
@@ -929,6 +931,37 @@ public Map<Integer, TPullCommitProgressResp> pullCommitProgressFromDataNodesBest
     return clientHandler.getResponseMap();
   }
 
+  /**
+   * Collect the current pipe metadata from the specified DataNodes before a metadata-changing
+   * procedure. The caller can use the returned task progress to avoid basing a replacement pipe on
+   * a stale ConfigNode heartbeat.
+   *
+   * <p>This is deliberately best effort. A DataNode that is unavailable cannot contribute a newer
+   * checkpoint, but the alter procedure can still use the checkpoint already stored by ConfigNode.
+   */
+  public Map<Integer, TPipeHeartbeatResp> collectPipeMetaFromDataNodes(
+      final Set<Integer> dataNodeIds) {
+    if (dataNodeIds.isEmpty()) {
+      return Collections.emptyMap();
+    }
+
+    final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+        configManager.getNodeManager().getRegisteredDataNodeLocations().entrySet().stream()
+            .filter(entry -> dataNodeIds.contains(entry.getKey()))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    if (dataNodeLocationMap.isEmpty()) {
+      return Collections.emptyMap();
+    }
+
+    final DataNodeAsyncRequestContext<TPipeHeartbeatReq, TPipeHeartbeatResp> clientHandler =
+        new DataNodeAsyncRequestContext<>(
+            CnToDnAsyncRequestType.PIPE_HEARTBEAT,
+            new TPipeHeartbeatReq(System.currentTimeMillis()),
+            dataNodeLocationMap);
+    sendRuntimeMetaRequest(clientHandler, true, getRuntimeMetaPushTimeoutInMs());
+    return clientHandler.getResponseMap();
+  }
+
   public Map<Integer, TSStatus> pushSubscriptionRuntimeStatesToDataNodes(
       final Map<TConsensusGroupId, Pair<Integer, Integer>> regionGroupToOldAndNewLeaderPairMap,
       final long runtimeVersion) {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java

@@ -20,7 +20,9 @@
 package org.apache.iotdb.confignode.procedure.impl.pipe.task;
 
 import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp;
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.consensus.index.ProgressIndex;
 import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
 import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
@@ -39,6 +41,7 @@
 import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
 import org.apache.iotdb.confignode.i18n.ProcedureMessages;
 import org.apache.iotdb.confignode.manager.pipe.coordinator.PipeManager;
+import org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.heartbeat.PipeHeartbeat;
 import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
 import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
 import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
@@ -60,9 +63,11 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -178,15 +183,29 @@ public void executeFromCalculateInfoForTask(final ConfigNodeProcedureEnv env) {
             new HashMap<>(alterPipeRequest.getProcessorAttributes()),
             new HashMap<>(alterPipeRequest.getConnectorAttributes()));
 
+    // The periodic heartbeat may not have reached ConfigNode immediately before this alter. Pull
+    // the current leader checkpoints first so a leader migration does not make the replacement
+    // task start from an older coordinator checkpoint.
+    final Map<Integer, ProgressIndex> latestProgressIndexMap =
+        PipeTaskAgent.isRealtimeOnlyPipe(currentPipeStaticMeta.getSourceParameters())
+                == PipeTaskAgent.isRealtimeOnlyPipe(updatedPipeStaticMeta.getSourceParameters())
+            ? collectLatestProgressIndexes(
+                env, currentPipeStaticMeta, currentConsensusGroupId2PipeTaskMeta)
+            : Collections.emptyMap();
+
     final ConcurrentMap<Integer, PipeTaskMeta> updatedConsensusGroupIdToTaskMetaMap =
         new ConcurrentHashMap<>();
     if (currentPipeStaticMeta.isSourceExternal()) {
       currentConsensusGroupId2PipeTaskMeta.forEach(
-          (taskId, pipeTaskMeta) ->
-              updatedConsensusGroupIdToTaskMetaMap.put(
-                  taskId,
-                  new PipeTaskMeta(
-                      pipeTaskMeta.getProgressIndex(), pipeTaskMeta.getLeaderNodeId())));
+          (taskId, pipeTaskMeta) -> {
+            final PipeTaskMeta updatedPipeTaskMeta =
+                new PipeTaskMeta(pipeTaskMeta.getProgressIndex(), pipeTaskMeta.getLeaderNodeId());
+            final ProgressIndex latestProgressIndex = latestProgressIndexMap.get(taskId);
+            if (latestProgressIndex != null) {
+              updatedPipeTaskMeta.updateProgressIndex(latestProgressIndex);
+            }
+            updatedConsensusGroupIdToTaskMetaMap.put(taskId, updatedPipeTaskMeta);
+          });
     } else {
       // data regions & schema regions
       env.getConfigManager()
@@ -217,8 +236,7 @@ public void executeFromCalculateInfoForTask(final ConfigNodeProcedureEnv env) {
                   // then it will extract all existing data now, not existing data since the
                   // original pipe was created
                   // Similar for "pure realtime"
-                  updatedConsensusGroupIdToTaskMetaMap.put(
-                      regionGroupId.getId(),
+                  final PipeTaskMeta updatedPipeTaskMeta =
                       new PipeTaskMeta(
                           PipeTaskAgent.isRealtimeOnlyPipe(
                                       currentPipeStaticMeta.getSourceParameters())
@@ -236,7 +254,14 @@ public void executeFromCalculateInfoForTask(final ConfigNodeProcedureEnv env) {
                                       && PipeTaskAgent.isRealtimeOnlyPipe(
                                           updatedPipeStaticMeta.getSourceParameters()))
                               ? PipeTaskMeta.getRevertedLeader(regionLeaderNodeId)
-                              : regionLeaderNodeId));
+                              : regionLeaderNodeId);
+                  final ProgressIndex latestProgressIndex =
+                      latestProgressIndexMap.get(regionGroupId.getId());
+                  if (latestProgressIndex != null) {
+                    updatedPipeTaskMeta.updateProgressIndex(latestProgressIndex);
+                  }
+                  updatedConsensusGroupIdToTaskMetaMap.put(
+                      regionGroupId.getId(), updatedPipeTaskMeta);
                 }
               });
 
@@ -267,6 +292,61 @@ public void executeFromCalculateInfoForTask(final ConfigNodeProcedureEnv env) {
     }
   }
 
+  private Map<Integer, ProgressIndex> collectLatestProgressIndexes(
+      final ConfigNodeProcedureEnv env,
+      final PipeStaticMeta pipeStaticMeta,
+      final Map<Integer, PipeTaskMeta> taskMetaMap) {
+    final Set<Integer> leaderNodeIds = new HashSet<>();
+    final Set<Integer> registeredDataNodeIds =
+        env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations().keySet();
+    taskMetaMap.forEach(
+        (consensusGroupId, taskMeta) -> {
+          // The ConfigRegion task is led by a ConfigNode, not by a DataNode.
+          if (consensusGroupId != Integer.MIN_VALUE
+              && registeredDataNodeIds.contains(taskMeta.getLeaderNodeId())) {
+            leaderNodeIds.add(taskMeta.getLeaderNodeId());
+          }
+        });
+
+    if (leaderNodeIds.isEmpty()) {
+      return Collections.emptyMap();
+    }
+
+    final Map<Integer, TPipeHeartbeatResp> responseMap =
+        env.collectPipeMetaFromDataNodes(leaderNodeIds);
+    final Map<Integer, ProgressIndex> latestProgressIndexMap = new HashMap<>();
+    responseMap.forEach(
+        (dataNodeId, response) -> {
+          if (response == null || !response.isSetPipeMetaList()) {
+            return;
+          }
+
+          final PipeMeta pipeMetaFromDataNode =
+              new PipeHeartbeat(response.getPipeMetaList(), null, null, null, null)
+                  .getPipeMeta(pipeStaticMeta);
+          if (pipeMetaFromDataNode == null) {
+            return;
+          }
+
+          pipeMetaFromDataNode
+              .getRuntimeMeta()
+              .getConsensusGroupId2TaskMetaMap()
+              .forEach(
+                  (consensusGroupId, taskMetaFromDataNode) -> {
+                    final PipeTaskMeta taskMetaFromCoordinator = taskMetaMap.get(consensusGroupId);
+                    if (taskMetaFromCoordinator == null
+                        || taskMetaFromCoordinator.getLeaderNodeId() != dataNodeId) {
+                      return;
+                    }
+                    latestProgressIndexMap.merge(
+                        consensusGroupId,
+                        taskMetaFromDataNode.getProgressIndex(),
+                        ProgressIndex::updateToMinimumEqualOrIsAfterProgressIndex);
+                  });
+        });
+    return latestProgressIndexMap;
+  }
+
   @Override
   public void executeFromWriteConfigNodeConsensus(final ConfigNodeProcedureEnv env)
       throws PipeException {

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<>();