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 99a3e06c76b [Fix] Preserve absent local search index during
subscription WAL replay (#18805)
99a3e06c76b is described below
commit 99a3e06c76b6b47c6d369632cc576a958d99e311
Author: Caideyipi <[email protected]>
AuthorDate: Sat Oct 10 10:35:21 2026 +0800
[Fix] Preserve absent local search index during subscription WAL replay
(#18805)
* Fix subscription WAL replay gaps across seek
* Fix subscription replay across transient WAL rotation visibility
* Hide rotated WAL under a non-WAL filename in regression test
* Fix subscription WAL gap recovery behavior
* Preserve absent local index in subscription WAL replay
* Remove speculative WAL gap recovery from subscription fix
* Remove unproven seek boundary changes from subscription fix
* test: cover replicated WAL index after fragmented local requests
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 7 +-
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 6 +-
.../consensus/ConsensusPrefetchingQueue.java | 2 +-
.../broker/consensus/ProgressWALIterator.java | 4 +-
.../broker/consensus/ProgressWALIteratorTest.java | 76 +++++++++++++++++++++-
5 files changed, 83 insertions(+), 12 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 1c0362b627b..5e92afb5f9c 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -2216,10 +2216,9 @@ public final class DataNodePipeMessages {
"ProgressWALIterator: skipped {} unreadable retained WAL files in
directory {}, "
+ "firstFile={}, lastFile={}, firstError={}; historical subscription
data in these "
+ "files cannot be replayed";
- public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64
=
- "ConsensusPrefetchingQueue {}: WAL replay skipped unavailable search
indexes [{}, {}), "
- + "skippedEntries={}, totalWalGapSkippedEntries={}; the missing WAL
data may have been "
- + "reclaimed before subscription consumption";
+ public static final String
+
LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A
=
+ "{}: WAL gap [{}, {}), skipped {}, total skipped {}; cause
undetermined";
public static final String
PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28
=
"Pipe {}@{}: terminate event committed for historical transfer.
creationTime: {}, "
+ "shouldMark: {}. {}";
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 49096659ade..14bc9cd4b26 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -2057,9 +2057,9 @@ public final class DataNodePipeMessages {
public static final String
PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E =
"ProgressWALIterator：跳过了 {} 个无法读取的保留 WAL 文件，directory={}，firstFile={}，"
+ "lastFile={}，firstError={}；这些文件中的历史订阅数据无法重放";
- public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64
=
- "ConsensusPrefetchingQueue {}：WAL 重放跳过了不可用的 searchIndex 区间 [{}, {})，"
- + "skippedEntries={}，totalWalGapSkippedEntries={}；缺失的 WAL
数据可能已在订阅消费前被回收";
+ public static final String
+
LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A
=
+ "{}：WAL 缺口 [{}, {})，跳过 {} 条，累计跳过 {} 条；原因尚未确定";
public static final String
PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28
=
"Pipe {}@{}：历史传输的终止事件已提交。creationTime：{}，shouldMark：{}。{}";
public static final String
PIPE_LOG_PIPE_HISTORICAL_SOURCE_HAS_SUPPLIED_ALL_EVENTS_EMITTING_8B58DE19 =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
index b692ec94ed0..8d2f0af7cec 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
@@ -2043,7 +2043,7 @@ public class ConsensusPrefetchingQueue {
final long totalSkippedEntries =
walGapSkippedEntries.addAndGet(skippedEntries);
LOGGER.warn(
DataNodePipeMessages
-
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64,
+
.LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A,
this,
expectedSearchIndex,
actualSearchIndex,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
index ee61c105c15..6572473cb58 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
@@ -734,7 +734,9 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
buffer.position(SEARCH_INDEX_OFFSET);
final long bodySearchIndex =
SearchNode.extractSearchIndex(buffer.getLong());
- return bodySearchIndex >= 0 ? bodySearchIndex : metadataSearchIndex;
+ // A replicated request has no local search index. The metadata fallback
is based on the
+ // WAL entry offset, which also counts fragments and cannot assign a local
index to it.
+ return bodySearchIndex;
}
private void startPending(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
index 45bff39d6ba..c800a2327e9 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.subscription.broker.consensus;
import
org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.SearchNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion;
@@ -87,7 +88,7 @@ public class ProgressWALIteratorTest {
}
@Test
- public void testIteratorUsesMetadataSearchIndexForStartFiltering() throws
Exception {
+ public void testIteratorUsesBodySearchIndexForStartFiltering() throws
Exception {
final Path dir =
Files.createTempDirectory("progress-wal-iterator-metadata-search-index");
final File firstWal =
dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
@@ -98,8 +99,8 @@ public class ProgressWALIteratorTest {
try {
try (WALWriter writer = new WALWriter(firstWal, WALFileVersion.V3)) {
- writer.write(searchableEntry(-1L), singleEntryMeta(19, 5L, 1L, 1000L,
7, 105L));
- writer.write(searchableEntry(-1L), singleEntryMeta(19, 6L, 1L, 2000L,
7, 106L));
+ writer.write(searchableEntry(5L), singleEntryMeta(19, 5L, 1L, 1000L,
7, 105L));
+ writer.write(searchableEntry(6L), singleEntryMeta(19, 6L, 1L, 2000L,
7, 106L));
}
try (WALWriter ignored = new WALWriter(lastWal, WALFileVersion.V3)) {
// Create a sealed successor so the first WAL becomes historical and
readable.
@@ -121,6 +122,75 @@ public class ProgressWALIteratorTest {
}
}
+ @Test
+ public void testIteratorPreservesAbsentLocalIndexAfterManyFragments() throws
Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-mixed-fragments");
+ final File dataWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File successorWal =
+ dir.resolve(WALFileUtils.getLogFileName(1, 1878,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+
+ try {
+ try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) {
+ // Physical WAL entries count fragments, not local consensus requests.
The old fallback
+ // assigns the replicated entry 1877 + 550 = 2427, creating a false
[1878, 2427) gap.
+ for (int fragment = 0; fragment < 550; fragment++) {
+ writer.write(searchableEntry(1877L), singleEntryMeta(19, 1877L, 1L,
1000L, 7, 1877L));
+ }
+ writer.write(
+ searchableEntry(SearchNode.NO_CONSENSUS_INDEX),
+ singleEntryMeta(19, SearchNode.NO_CONSENSUS_INDEX, 1L, 2000L, 8,
10000L));
+ writer.write(searchableEntry(1878L), singleEntryMeta(19, 1878L, 1L,
3000L, 7, 1878L));
+ }
+ try (WALWriter ignored = new WALWriter(successorWal, WALFileVersion.V3))
{
+ // Keep both files retained and sealed; there is no deletion or
concurrent write.
+ }
+
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) {
+ assertTrue(iterator.hasNext());
+ final IndexedConsensusRequest fragmented = iterator.next();
+ assertEquals(1877L, fragmented.getSearchIndex());
+ assertEquals(1877L, fragmented.getProgressLocalSeq());
+ assertEquals(7, fragmented.getNodeId());
+ assertEquals(550, fragmented.getRequests().size());
+
+ assertTrue(iterator.hasNext());
+ final IndexedConsensusRequest replicated = iterator.next();
+ assertEquals(SearchNode.NO_CONSENSUS_INDEX,
replicated.getSearchIndex());
+ assertEquals(10000L, replicated.getProgressLocalSeq());
+ assertEquals(2000L, replicated.getPhysicalTime());
+ assertEquals(8, replicated.getNodeId());
+ assertEquals(1, replicated.getRequests().size());
+
+ assertTrue(iterator.hasNext());
+ final IndexedConsensusRequest nextLocal = iterator.next();
+ assertEquals(1878L, nextLocal.getSearchIndex());
+ assertEquals(1878L, nextLocal.getProgressLocalSeq());
+ assertEquals(7, nextLocal.getNodeId());
+ assertEquals(1, nextLocal.getRequests().size());
+ assertFalse(iterator.hasNext());
+ assertFalse(iterator.hasIncompleteScan());
+ assertTrue(dataWal.isFile());
+ }
+
+ // A local-index seek must also retain replicated requests with their
own writer progress.
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 1878L)) {
+ assertTrue(iterator.hasNext());
+ assertEquals(SearchNode.NO_CONSENSUS_INDEX,
iterator.next().getSearchIndex());
+ assertTrue(iterator.hasNext());
+ assertEquals(1878L, iterator.next().getSearchIndex());
+ assertFalse(iterator.hasNext());
+ assertFalse(iterator.hasIncompleteScan());
+ }
+ } finally {
+ Files.deleteIfExists(dataWal.toPath());
+ Files.deleteIfExists(successorWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
@Test
public void testIteratorMergesFragmentsWithSameLocalSeq() throws Exception {
final Path dir = Files.createTempDirectory("progress-wal-iterator-merge");
