Caideyipi · Oct 21, 2026
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeDataSinkIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeDataSinkIT.java
@@ -39,15 +39,18 @@
 
 import java.nio.charset.Charset;
 import java.sql.Connection;
+import java.sql.ResultSet;
 import java.sql.Statement;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 import static org.apache.iotdb.db.it.IoTDBRestServiceIT.getHttpPost;
+import static org.awaitility.Awaitility.await;
 
 @RunWith(IoTDBTestRunner.class)
 @Category({MultiClusterIT2AutoCreateSchema.class})
@@ -566,4 +569,92 @@ public void testSpecialPartialInsert() throws Exception {
                 "1635232151960,null,null,2.0,2.1,null,",
                 "1635232143960,6.0,4.0,null,null,null,")));
   }
+
+  @Test
+  public void testTransferMods() {
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "create database root.sg_nonAligned",
+            "create TIMESERIES root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s0 with datatype=boolean, encoding=RLE,compressor=snappy",
+            "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s1 with datatype=int32, encoding=PLAIN,compressor=LZ4",
+            "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s2 with datatype=int64,encoding=gorilla,compressor=uncompressed",
+            "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s3 with datatype=float,encoding=chimp,compressor=gzip",
+            "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s4 with datatype=double,encoding=ts_2diff,compressor=zstd",
+            "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s5 with datatype=text,encoding=dictionary,compressor=lzma2",
+            "insert into root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`(time,s1, s2,s3,s4,s0,s5) values(1,1,10,5.39,5.51234,true,''),(11,null,20,5.39,15.51234,false,'第2条 device_nonAligned'),(21,3,null,5.39,25.51234,true,'第3条device_nonAligned'),(31,4,40,null,35.51234,true,'第4条device_nonAligned'),(41,5,50,5.39,null,false,'第5条device_nonAligned'),(51,6,60,5.39,55.51234,null,'第6条device_nonAligned'),(61,7,70,5.39,65.51234,false,null),(71,8,80,5.39,75.51234,false,'第8条device_nonAligned'),(81,9,90,5.39,85.51234,false,'第9条device_nonAligned'),(91,10,100,5.39,95.51234,false,'第10条device_nonAligned'),(101,11,110,5.39,105.51234,false,'第11条device_nonAligned'),(111,12,120,5.39,115.51234,false,'第12条device_nonAligned'),(121,13,130,5.39,125.51234,false,'第13条device_nonAligned'),(131,14,140,5.39,135.51234,false,'第14条device_nonAligned'),(141,15,150,5.39,145.51234,false,'第15条'),(151,16,160,5.39,155.51234,false,'第16条'),(161,17,170,5.39,165.51234,false,'第17条'),(171,18,180,5.39,175.51234,false,'第18条'),(181,19,190,5.39,185.51234,false,'第19条'),(191,20,200,5.39,195.51234,false,'第20条'),(201,21,210,5.39,null,false,'第21条')",
+            "flush",
+            "delete timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.s0",
+            String.format(
+                "create pipe test with source ('source.realtime.mode'='stream','inclusion'='data','path'='root.sg_nonAligned.非对齐序列带有encoding和压缩方式.**','source.realtime.enable'='true','mods.enable'='true') with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
+                receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString())));
+
+    awaitHistoricalPipeCompletion("test");
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*",
+        "count(timeseries),",
+        Collections.singleton("5,"));
+
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList("drop pipe if exists test_history", "drop pipe if exists test_realtime"));
+
+    TestUtils.executeNonQuery(receiverEnv, "drop database root.**");
+
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "delete timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.s1",
+            String.format(
+                "create pipe test with source ('source.realtime.mode'='stream','inclusion'='data','path'='root.sg_nonAligned.非对齐序列带有encoding和压缩方式.**','source.realtime.enable'='true','mods.enable'='true') with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
+                receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString())));
+
+    awaitHistoricalPipeCompletion("test");
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*",
+        "count(timeseries),",
+        Collections.singleton("4,"),
+        15);
+    TestUtils.assertDataAlwaysOnEnv(
+        receiverEnv,
+        "count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*",
+        "count(timeseries),",
+        Collections.singleton("4,"));
+  }
+
+  private void awaitHistoricalPipeCompletion(final String pipeName) {
+    await()
+        .pollInSameThread()
+        .pollDelay(1L, TimeUnit.SECONDS)
+        .pollInterval(1L, TimeUnit.SECONDS)
+        .atMost(600L, TimeUnit.SECONDS)
+        .untilAsserted(
+            () -> {
+              boolean sourcePipeFound = false;
+              boolean sourcePipeCompleted = true;
+              boolean realtimePipeFound = false;
+
+              for (final DataNodeWrapper dataNode : senderEnv.getDataNodeWrapperList()) {
+                try (final Connection connection = senderEnv.getConnection(dataNode);
+                    final Statement statement = connection.createStatement();
+                    final ResultSet resultSet = statement.executeQuery("show pipes")) {
+                  while (resultSet.next()) {
+                    final String id = resultSet.getString(1);
+                    if (id.equals(pipeName) || id.equals(pipeName + "_history")) {
+                      sourcePipeFound = true;
+                      sourcePipeCompleted &= "0".equals(resultSet.getString(8));
+                    } else if (id.equals(pipeName + "_realtime")) {
+                      realtimePipeFound = true;
+                    }
+                  }
+                }
+              }
+
+              // The historical sub-pipe may be auto-dropped after all events are committed.
+              Assert.assertTrue(sourcePipeFound || realtimePipeFound);
+              Assert.assertTrue(!sourcePipeFound || sourcePipeCompleted);
+            });
+  }
 }
diff --git a/.../consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java b/.../consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java
@@ -41,6 +41,27 @@ default long getMemorySize() {
     return 0;
   }
 
+  /**
+   * Whether a consensus implementation may forward this write to another leader. Requests based on
+   * the local leader's runtime state must not be forwarded after that leader steps down.
+   */
+  default boolean isForwardable() {
+    return true;
+  }
+
+  /**
+   * Returns whether this request materializes its bytes only when it is about to be sent, instead
+   * of when it is handed over to the replication layer.
+   *
+   * <p>A deferred request keeps just its source objects while it waits in the replication queues,
+   * so it must be able to rebuild the very same bytes on demand and must stay small. Consensus
+   * implementations may postpone {@link #serializeToByteBuffer()} for such requests and keep the
+   * source objects alive until then.
+   */
+  default boolean isSerializationDeferred() {
+    return false;
+  }
+
   default void markAsGeneratedByRemoteConsensusLeader() {
     // do nothing by default
   }

diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
@@ -366,6 +366,12 @@ && waitUntilLeaderReady(raftGroupId)) {
       }
     }
 
+    // A leader-local decision may already be stale after a leadership change. Let the current
+    // leader make its own decision instead of replaying the old leader's request there.
+    if (!request.isForwardable()) {
+      return StatusUtils.getStatus(TSStatusCode.REDIRECTION_RECOMMEND);
+    }
+
     // 2. try raft client
     TSStatus writeResult;
     try (AutoCloseable ignored =

diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
@@ -23,12 +23,14 @@
 import org.apache.iotdb.consensus.IStateMachine;
 import org.apache.iotdb.consensus.common.ConsensusGroup;
 import org.apache.iotdb.consensus.common.Peer;
+import org.apache.iotdb.consensus.common.request.IConsensusRequest;
 import org.apache.iotdb.consensus.config.RatisConfig;
 import org.apache.iotdb.consensus.exception.ConsensusException;
 import org.apache.iotdb.consensus.exception.ConsensusGroupAlreadyExistException;
 import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException;
 import org.apache.iotdb.consensus.exception.PeerAlreadyInConsensusGroupException;
 import org.apache.iotdb.consensus.exception.PeerNotInConsensusGroupException;
+import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.ratis.util.TimeDuration;
 import org.junit.After;
@@ -113,6 +115,44 @@ public void basicConsensus3Copy() throws Exception {
     doConsensus(0, 10, 10);
   }
 
+  @Test
+  public void testNonForwardableWriteAfterLeaderTransfer() throws Exception {
+    for (RatisConsensus server : servers) {
+      server.createLocalPeer(gid, peers);
+    }
+    miniCluster.waitUntilActiveLeaderElectedAndReady();
+
+    final int oldLeaderId = servers.get(0).getLeader(gid).getNodeId();
+    final int newLeaderId = (oldLeaderId + 1) % servers.size();
+    final RatisConsensus oldLeader = servers.get(oldLeaderId);
+    final IConsensusRequest leaderLocalRequest =
+        new TestUtils.TestRequest(TestUtils.TestRequest.incrRequest().serializeToByteBuffer()) {
+          @Override
+          public boolean isForwardable() {
+            return false;
+          }
+        };
+
+    oldLeader.transferLeader(gid, peers.get(newLeaderId));
+    miniCluster.waitUntilActiveLeaderElectedAndReady();
+    Assert.assertEquals(newLeaderId, oldLeader.getLeader(gid).getNodeId());
+
+    // The new leader has already made its own decision before the old request arrives.
+    Assert.assertEquals(
+        TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+        servers.get(newLeaderId).write(gid, leaderLocalRequest).getCode());
+    Assert.assertEquals(
+        TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode(),
+        oldLeader.write(gid, leaderLocalRequest).getCode());
+    Assert.assertEquals(1, miniCluster.mustRead(newLeaderId));
+
+    // Ordinary writes issued to a follower must still reach the current leader.
+    Assert.assertEquals(
+        TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+        oldLeader.write(gid, TestUtils.TestRequest.incrRequest()).getCode());
+    Assert.assertEquals(2, miniCluster.mustRead(newLeaderId));
+  }
+
   @Test
   public void addMemberToGroup() throws Exception {
     List<Peer> original = peers.subList(0, 1);

diff --git a/...g/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNode.java b/...g/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNode.java
@@ -53,6 +53,13 @@ public boolean isOpen() {
     return isOpen;
   }
 
+  @Override
+  public boolean isForwardable() {
+    // A close is decided from the local pipe metadata. After a leader change the new leader may
+    // already have newer metadata and an active source, so an old close must not be forwarded.
+    return isOpen;
+  }
+
   @Override
   public List<PlanNode> getChildren() {
     return Collections.emptyList();

diff --git a/...ache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNodeTest.java b/...ache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeOperateSchemaQueueNodeTest.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe;
+
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class PipeOperateSchemaQueueNodeTest {
+
+  @Test
+  public void testCloseCannotBeForwardedToNewLeader() {
+    final PipeOperateSchemaQueueNode close =
+        new PipeOperateSchemaQueueNode(new PlanNodeId("close"), false);
+    Assert.assertFalse(close.isForwardable());
+    Assert.assertFalse(close.clone().isForwardable());
+    Assert.assertFalse(PlanNodeType.deserialize(close.serializeToByteBuffer()).isForwardable());
+
+    final PipeOperateSchemaQueueNode open =
+        new PipeOperateSchemaQueueNode(new PlanNodeId("open"), true);
+    Assert.assertTrue(open.isForwardable());
+    Assert.assertTrue(PlanNodeType.deserialize(open.serializeToByteBuffer()).isForwardable());
+  }
+}