This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch codex/ratis-historical-replay-20261029 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1722b533d1c296dd10a804dc73922112b88434ef Author: Caideyipi <[email protected]> AuthorDate: Thu Oct 29 14:54:11 2026 +0800 [Bugfix] Replay Ratis historical TsFiles without Pipe progress Treat MinimumProgressIndex as unknown for Ratis historical extraction so successful writes remain eligible for replay after a leader change. Preserve valid-index and other consensus-protocol coverage filtering. Add a regression test for initial and resumed checkpoints and the unaffected filtering behavior. --- ...istoricalDataRegionTsFileAndDeletionSource.java | 9 +++++ ...ricalDataRegionTsFileAndDeletionSourceTest.java | 46 ++++++++++++++++++++++ 2 files changed, 55 insertions(+) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java index 041d33c2faf..c47fb606017 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java @@ -42,6 +42,7 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; import org.apache.iotdb.commons.queryengine.utils.DateTimeUtils; import org.apache.iotdb.commons.utils.PathUtils; +import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.consensus.pipe.IoTConsensusV2; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.consensus.DataRegionConsensusImpl; @@ -919,6 +920,14 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource private boolean isProgressIndexNotCoveredByStartIndex( PersistentResource resource, ProgressIndex progressIndex) { + // Ratis writes do not carry Pipe progress indexes. Minimum therefore means unknown here, + // rather than an empty file already covered by the initial checkpoint. Replay conservatively + // after a leader change, or successful writes can disappear from the subscription backlog. + if (progressIndex instanceof MinimumProgressIndex + && ConsensusFactory.RATIS_CONSENSUS.equals( + IoTDBDescriptor.getInstance().getConfig().getDataRegionConsensusProtocolClass())) { + return true; + } final ProgressIndex innerStartIndex = getInnerProgressIndex(startIndex); if (innerStartIndex.isEqualOrAfter(progressIndex) || isProgressIndexCoveredByTimePartitionProgressIndex( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java index 991dab795f9..090124c7fc8 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java @@ -35,6 +35,8 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern; import org.apache.iotdb.commons.pipe.datastructure.resource.PersistentResource; import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; import org.apache.iotdb.commons.utils.FileUtils; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.pipe.consensus.ReplicateProgressDataNodeManager; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; @@ -688,6 +690,50 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSourceTest { } } + @Test + public void testRatisFilesWithoutPipeProgressAreNotCoveredByCheckpoint() throws Exception { + final String originalConsensus = + IoTDBDescriptor.getInstance().getConfig().getDataRegionConsensusProtocolClass(); + final File tempDir = Files.createTempDirectory("pipeHistoricalRatisCoverage").toFile(); + try { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + assertMayTsFileContainUnprocessedData( + tempDir, + "unknown-initial.tsfile", + MinimumProgressIndex.INSTANCE, + MinimumProgressIndex.INSTANCE, + true); + assertMayTsFileContainUnprocessedData( + tempDir, + "unknown-recovery.tsfile", + new SimpleProgressIndex(0, 100), + MinimumProgressIndex.INSTANCE, + true); + assertMayTsFileContainUnprocessedData( + tempDir, + "indexed-covered.tsfile", + new SimpleProgressIndex(0, 100), + new SimpleProgressIndex(0, 50), + false); + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS); + assertMayTsFileContainUnprocessedData( + tempDir, + "other-consensus.tsfile", + MinimumProgressIndex.INSTANCE, + MinimumProgressIndex.INSTANCE, + false); + } finally { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(originalConsensus); + FileUtils.deleteFileOrDirectory(tempDir); + } + } + @Test public void testMayTsFileContainUnprocessedDataUsesTimePartitionProgressCoverage() throws Exception {
