Pipe: Fixed the bug that a realtime only pipe will not transfer historical data when it is altered to a historical pipe

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

@@ -570,4 +570,49 @@ public void testAlterPipeSourceAndProcessor() {
         "count(timeseries),",
         Collections.singleton("3,"));
   }
+
+  @Test
+  public void testAlterPipeRealtime() {
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    // Insert data on sender
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "insert into root.db.d1 (time, at1) values (1000, 1), (1500, 2), (2000, 3), (2500, 4), (3000, 5)",
+            "flush"),
+        null);
+
+    // Create pipe
+    final String sql =
+        String.format(
+            "create pipe a2b with source ('history.enable'='false') with sink ('node-urls'='%s')",
+            receiverDataNode.getIpAndPortString());
+
+    try (final Connection connection = senderEnv.getConnection();
+        final Statement statement = connection.createStatement()) {
+      statement.execute(sql);
+    } catch (final SQLException e) {
+      fail(e.getMessage());
+    }
+
+    TestUtils.assertDataAlwaysOnEnv(
+        receiverEnv,
+        "count timeSeries root.db.**",
+        "count(timeseries),",
+        Collections.singleton("0,"));
+
+    try (final Connection connection = senderEnv.getConnection();
+        final Statement statement = connection.createStatement()) {
+      statement.execute("alter pipe a2b modify source ('history.enable'='true')");
+    } catch (final SQLException e) {
+      fail(e.getMessage());
+    }
+
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "count timeSeries root.db.**",
+        "count(timeseries),",
+        Collections.singleton("1,"));
+  }
 }

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

@@ -21,6 +21,7 @@
 
 import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
+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;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
@@ -209,7 +210,12 @@ public void executeFromCalculateInfoForTask(final ConfigNodeProcedureEnv env) {
                   updatedConsensusGroupIdToTaskMetaMap.put(
                       regionGroupId.getId(),
                       new PipeTaskMeta(
-                          currentPipeTaskMeta.getProgressIndex(),
+                          PipeTaskAgent.isRealtimeOnlyPipe(
+                                      currentPipeStaticMeta.getSourceParameters())
+                                  && !PipeTaskAgent.isRealtimeOnlyPipe(
+                                      updatedPipeStaticMeta.getSourceParameters())
+                              ? MinimumProgressIndex.INSTANCE
+                              : currentPipeTaskMeta.getProgressIndex(),
                           PipeTaskMeta.isNewlyAdded(currentPipeTaskMeta.getLeaderNodeId())
                                   && !(!PipeTaskAgent.isHistoryOnlyPipe(
                                           currentPipeStaticMeta.getSourceParameters())