This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 35b9b708687 Fix IoTConsensusV2 progress for split LOAD TsFiles (#18927)
35b9b708687 is described below
commit 35b9b708687cd64f8f7c8b78da8da7ba41486b47
Author: Caideyipi <[email protected]>
AuthorDate: Tue Oct 20 14:57:17 2026 +0800
Fix IoTConsensusV2 progress for split LOAD TsFiles (#18927)
---
.../db/storageengine/load/TsFileWriterManager.java | 11 +++-
.../storageengine/load/LoadTsFileManagerTest.java | 70 ++++++++++++++++++++++
2 files changed, 80 insertions(+), 1 deletion(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java
index fc8f38b8502..d3d9efb1345 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java
@@ -31,6 +31,7 @@ import org.apache.iotdb.commons.utils.RetryUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.load.LoadFileException;
import org.apache.iotdb.db.i18n.StorageEngineMessages;
+import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusOp;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
@@ -532,7 +533,7 @@ final class TsFileWriterManager {
/** Formally imports sealed staged TsFiles into the target storage engine
data regions. */
void loadAll(final boolean isGeneratedByPipe, final boolean
deleteStagedSource)
- throws LoadFileException {
+ throws IOException, LoadFileException {
for (final Map.Entry<TTimePartitionSlot, TsFileResource> entry :
timePartition2Resource.entrySet()) {
final DataRegion targetDataRegion = dataRegion;
@@ -547,6 +548,14 @@ final class TsFileWriterManager {
uuidDirName(tsFileResource.getTsFile())));
}
+ if (LoadConsensusCompat.usesIoTConsensusV2()) {
+ // V2 only replicates files carrying this node's progress. The
coordinator's index may
+ // belong to another node and precede writes between PREPARE and
COMMIT. Assign progress
+ // at import, and persist it so recovery keeps the file eligible for
replication.
+
PipeDataNodeAgent.runtime().assignProgressIndexForTsFileLoad(tsFileResource);
+ tsFileResource.serialize();
+ }
+
targetDataRegion.loadNewTsFile(
tsFileResource, deleteStagedSource, isGeneratedByPipe, false,
Optional.empty());
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java
index 360fa83c6af..a3d24733167 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java
@@ -20,16 +20,22 @@
package org.apache.iotdb.db.storageengine.load;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.load.LoadFileException;
+import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
+import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+import
org.apache.iotdb.db.pipe.processor.iotconsensusv2.IoTConsensusV2Processor;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusOp;
import
org.apache.iotdb.db.queryengine.plan.scheduler.load.ChunkOffsetCalculator;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import org.apache.iotdb.db.storageengine.dataregion.wal.node.IWALNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode;
@@ -798,6 +804,70 @@ public class LoadTsFileManagerTest {
assertFalse(manager.prepare(prepareNode("in-progress-load"),
Collections.emptyMap()));
}
+ @Test
+ public void testIoTConsensusV2CommitUsesReceivingNodeProgress() throws
Exception {
+ assertCommittedProgress(true, config.getDataNodeId() + 1);
+ }
+
+ @Test
+ public void testIoTConsensusV2CommitAdvancesPreparedProgress() throws
Exception {
+ assertCommittedProgress(true, config.getDataNodeId());
+ }
+
+ @Test
+ public void testOtherConsensusCommitPreservesCoordinatorProgress() throws
Exception {
+ assertCommittedProgress(false, config.getDataNodeId() + 1);
+ }
+
+ private void assertCommittedProgress(final boolean useIoTConsensusV2, final
int coordinatorNodeId)
+ throws Exception {
+ LoadConsensusCompat.setIoTConsensusV2ForTest(useIoTConsensusV2);
+ final String loadId = "load-progress";
+ final LoadTsFileManager manager = new LoadTsFileManager(dataRegion);
+ try {
+ emulateStagedFileImport();
+ final LoadTsFileConsensusNode piece = stagedPiece(loadId,
laidOutChunk());
+ manager.writePiece(piece);
+ final File staged = new File(tempDir,
piece.getPieceRefs().get(0).getRelativePath());
+ final RecoverProgressIndex preparedLocalProgress =
+ PipeDataNodeAgent.runtime().getNextProgressIndexForTsFileLoad();
+ final ProgressIndex coordinatorProgress =
+ new RecoverProgressIndex(
+ coordinatorNodeId,
+
preparedLocalProgress.getDataNodeId2LocalIndex().get(config.getDataNodeId()));
+ assertTrue(
+ manager.prepare(
+ prepareNode(loadId),
+ Collections.singletonMap(new TTimePartitionSlot(0L),
coordinatorProgress)));
+
+ // A local write after PREPARE must precede the progress assigned when
the file is imported.
+ final ProgressIndex interveningWrite =
+ PipeDataNodeAgent.runtime().getNextProgressIndexForTsFileLoad();
+ final LoadTsFileConsensusNode commit = commitNode(loadId);
+ assertTrue(manager.loadAll(commit, Collections.emptyMap()));
+
+ final TsFileResource imported =
+ new TsFileResource(new File(new File(tempDir, "data"),
staged.getName()));
+ imported.deserialize();
+ imported.setStatus(TsFileResourceStatus.NORMAL);
+ final ProgressIndex persistedProgress =
imported.getMaxProgressIndexAfterClose();
+ final PipeTsFileInsertionEvent event =
Mockito.mock(PipeTsFileInsertionEvent.class);
+
Mockito.when(event.forceGetProgressIndex()).thenReturn(persistedProgress);
+ assertEquals(
+ "A V2 LOAD must remain eligible for consensus replication after
resource recovery",
+ useIoTConsensusV2,
+ IoTConsensusV2Processor.isShouldReplicate(event));
+ if (useIoTConsensusV2) {
+ assertTrue(persistedProgress.isAfter(interveningWrite));
+ } else {
+ assertEquals(coordinatorProgress, persistedProgress);
+ }
+ } finally {
+ manager.stop();
+ LoadConsensusCompat.setIoTConsensusV2ForTest(null);
+ }
+ }
+
@Test
public void testCommitKeepsTheStagedBytesWhileAReplicaMayStillNeedThem()
throws Exception {
// The watermark lags behind the COMMIT command, so a replica that catches
up from this node's
