This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch to-2011-n in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 6ac9891874701850abe4bdaf1cd60ece6aa3551d Author: Caideyipi <[email protected]> AuthorDate: Thu Aug 6 12:23:21 2026 +0800 [Subscription] Reduce consensus WAL replay and ACK contention (#18402) * [Subscription] Reduce consensus WAL replay and ACK contention * [Subscription] Avoid redundant consensus WAL refresh work * [Subscription] Fast-forward WAL iterator across pending delivery (cherry picked from commit 5d2a0a58774db5359e133b89270f1216a77f3c02) --- .../consensus/ConsensusPrefetchingQueue.java | 74 ++++- .../broker/consensus/ProgressWALIterator.java | 251 ++++++++++++--- .../consensus/ConsensusPrefetchingQueueTest.java | 342 +++++++++++++++++++++ .../broker/consensus/ProgressWALIteratorTest.java | 285 +++++++++++++++++ 4 files changed, 895 insertions(+), 57 deletions(-) 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 e00a3c5b21d..c89e9d3f39c 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 @@ -193,9 +193,9 @@ public class ConsensusPrefetchingQueue { private volatile ProgressWALIterator subscriptionWALIterator; /** - * Seek requests must not close/reset the WAL iterator from RPC threads because the prefetch - * worker may be reading it concurrently. Instead, seek only records the latest desired reset and - * the queue's next prefetch round applies it after observing the new seek generation. + * WAL cursor changes outside the iterator must not close/reset it from RPC threads because the + * prefetch worker may be reading it concurrently. Instead, the latest desired reset is recorded + * and applied by the next prefetch round after observing the expected seek generation. */ private volatile long pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE; @@ -1567,12 +1567,41 @@ public class ConsensusPrefetchingQueue { return hasLocalSearchIndex(request) && request.getSearchIndex() < nextExpectedSearchIndex.get(); } - private void advanceLocalCursorIfPresent(final IndexedConsensusRequest request) { + private boolean advanceLocalCursorIfPresent(final IndexedConsensusRequest request) { if (hasLocalSearchIndex(request)) { nextExpectedSearchIndex.set(request.getSearchIndex() + 1); + return true; + } + return false; + } + + private void advanceLocalCursorFromPendingIfPresent( + final IndexedConsensusRequest request, final long expectedSeekGeneration) { + if (advanceLocalCursorIfPresent(request)) { + // Pending delivery advances independently of the WAL reader. Raise its local lower bound in + // place so stale local requests are filtered without rebuilding and rescanning retained WAL. + final ProgressWALIterator iterator = subscriptionWALIterator; + if (Objects.nonNull(iterator) && seekGeneration.get() == expectedSeekGeneration) { + iterator.advanceTo( + nextExpectedSearchIndex.get(), this::isWriterProgressCoveredForWalFastForward); + } } } + private boolean isWriterProgressCoveredForWalFastForward( + final long physicalTime, final int nodeId, final long localSeq) { + final WriterProgress candidate = new WriterProgress(physicalTime, localSeq); + final WriterProgress recoveryProgress = + recoveryWriterProgressByWriter.get(new WriterId(consensusGroupId.toString(), nodeId)); + if (Objects.nonNull(recoveryProgress) + && compareWriterProgress(candidate, recoveryProgress) <= 0) { + return true; + } + final WriterProgress materializedProgress = materializedProgressByWriter.get(nodeId); + return Objects.nonNull(materializedProgress) + && compareWriterProgress(candidate, materializedProgress) <= 0; + } + private MaterializationResult appendRealtimeRequest( final IndexedConsensusRequest request, final DeliveryBatchState batchState, @@ -1648,12 +1677,12 @@ public class ConsensusPrefetchingQueue { if (shouldSkipForRecoveryProgress(request)) { skippedCount++; - advanceLocalCursorIfPresent(request); + advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration); continue; } if (shouldSkipForMaterializedProgress(request)) { skippedCount++; - advanceLocalCursorIfPresent(request); + advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration); continue; } @@ -1665,7 +1694,7 @@ public class ConsensusPrefetchingQueue { } markMaterializedProgress(request); processedCount++; - advanceLocalCursorIfPresent(request); + advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration); if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) { break; } @@ -1757,7 +1786,9 @@ public class ConsensusPrefetchingQueue { // Use the persistent linger batch so an unexpected runtime failure cannot orphan already // reserved Tablets or advance replay progress past data that has become unreachable. final DeliveryBatchState batchState = lingerBatch; - resetSubscriptionWALPosition(nextExpectedSearchIndex.get()); + // Keep the iterator and its buffered next request across rounds. Reopening it here discards the + // request prepared by hasNext() and repeatedly re-reads, skips, and decompresses the same WAL + // segment. Pending-path cursor advances and seek operations request explicit realignment. final MaterializationResult materializationResult = pumpFromSubscriptionWAL( batchState, expectedSeekGeneration, maxWalEntries, maxTablets, maxBatchBytes); @@ -1790,7 +1821,6 @@ public class ConsensusPrefetchingQueue { return MaterializationResult.SUCCESS; } - subscriptionWALIterator.refresh(); ensureSubscriptionWalReadable(); int entriesRead = 0; @@ -1846,9 +1876,14 @@ public class ConsensusPrefetchingQueue { } private void ensureSubscriptionWalReadable() { - if (Objects.isNull(subscriptionWALIterator) - || subscriptionWALIterator.hasNext() - || !(consensusReqReader instanceof WALNode)) { + if (Objects.isNull(subscriptionWALIterator) || subscriptionWALIterator.hasNext()) { + return; + } + + // Listing and sorting all retained WAL files is only necessary after the iterator is + // exhausted. While it still has a readable request, refreshing cannot affect the next result. + subscriptionWALIterator.refresh(); + if (subscriptionWALIterator.hasNext() || !(consensusReqReader instanceof WALNode)) { return; } @@ -1865,9 +1900,6 @@ public class ConsensusPrefetchingQueue { currentWalIndex); ((WALNode) consensusReqReader).rollWALFile(); resetSubscriptionWALPosition(nextExpectedSearchIndex.get()); - if (Objects.nonNull(subscriptionWALIterator)) { - subscriptionWALIterator.refresh(); - } } private void resetSubscriptionWALPosition(final long startSearchIndex) { @@ -1885,6 +1917,11 @@ public class ConsensusPrefetchingQueue { protected void onWalGapRetryScheduled() {} private boolean hasReadableWalEntries() { + if (pendingSubscriptionWalResetSearchIndex != Long.MIN_VALUE) { + // Do not advance the stale iterator only to discard its buffered request when the next round + // applies the pending realignment. Returning true keeps the worker scheduled for that round. + return true; + } return Objects.nonNull(subscriptionWALIterator) && subscriptionWALIterator.hasNext(); } @@ -2179,7 +2216,10 @@ public class ConsensusPrefetchingQueue { private boolean ackMissingInFlightEvent( final SubscriptionCommitContext commitContext, final boolean silent) { - acquireWriteLock(); + // Late or duplicate ACKs touch the same concurrent lifecycle indexes and commit manager as the + // regular in-flight ACK path. A read lock is sufficient to fence seek/close transitions while + // allowing ACKs to proceed concurrently with a long-running WAL prefetch round. + acquireReadLock(); try { if (!canAcceptCommitContext(commitContext, "ack", silent)) { return false; @@ -2226,7 +2266,7 @@ public class ConsensusPrefetchingQueue { } return true; } finally { - releaseWriteLock(); + releaseReadLock(); } } 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 8fd23042ea9..3e6b351eed0 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 @@ -42,6 +42,7 @@ import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashSet; import java.util.Iterator; import java.util.List; @@ -56,6 +57,12 @@ import java.util.Set; */ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensusRequest> { + @FunctionalInterface + public interface WriterProgressCoverage { + + boolean isCovered(long physicalTime, int nodeId, long localSeq); + } + private static final Logger LOGGER = LoggerFactory.getLogger(ProgressWALIterator.class); private static final int SEARCH_INDEX_OFFSET = @@ -65,9 +72,10 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus WALFileVersion.V2.getVersionBytes().length, WALFileVersion.V3.getVersionBytes().length); private final File logDirectory; - private final long startSearchIndex; + private long minimumSearchIndex; private final WALNode liveWalNode; private File[] walFiles; + private long[] walFileVersionIds; private int currentFileIndex = -1; private ProgressWALReader currentReader; private long currentReaderVersionId = -1L; @@ -105,7 +113,7 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus private ProgressWALIterator( final File logDirectory, final long startSearchIndex, final WALNode liveWalNode) { this.logDirectory = logDirectory; - this.startSearchIndex = startSearchIndex; + this.minimumSearchIndex = startSearchIndex; this.liveWalNode = liveWalNode; refreshFileList(); } @@ -114,23 +122,29 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus final File[] discoveredWalFiles = WALFileUtils.listAllWALFiles(logDirectory); if (discoveredWalFiles == null) { walFiles = new File[0]; + walFileVersionIds = new long[0]; return; } WALFileUtils.ascSortByVersionId(discoveredWalFiles); - final List<File> filteredWalFiles = new ArrayList<>(discoveredWalFiles.length); + final File[] filteredWalFiles = new File[discoveredWalFiles.length]; + final long[] filteredWalFileVersionIds = new long[discoveredWalFiles.length]; + int filteredWalFileCount = 0; for (int i = 0; i < discoveredWalFiles.length; i++) { final File walFile = discoveredWalFiles[i]; + final long versionId = WALFileUtils.parseVersionId(walFile.getName()); final boolean isLastWalFile = i == discoveredWalFiles.length - 1; - if (!isLastWalFile && shouldSkipWalFile(walFile)) { + if (!isLastWalFile && shouldSkipWalFile(walFile, versionId)) { continue; } - filteredWalFiles.add(walFile); + filteredWalFiles[filteredWalFileCount] = walFile; + filteredWalFileVersionIds[filteredWalFileCount] = versionId; + filteredWalFileCount++; } - walFiles = filteredWalFiles.toArray(new File[0]); + walFiles = Arrays.copyOf(filteredWalFiles, filteredWalFileCount); + walFileVersionIds = Arrays.copyOf(filteredWalFileVersionIds, filteredWalFileCount); } - private boolean shouldSkipWalFile(final File walFile) { - final long versionId = WALFileUtils.parseVersionId(walFile.getName()); + private boolean shouldSkipWalFile(final File walFile, final long versionId) { return skippedBrokenWalVersionIds.contains(versionId) || isHeaderOnlyWalFile(walFile); } @@ -139,42 +153,188 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus } public void refresh() { + final boolean exhaustedKnownFiles = currentFileIndex >= walFiles.length; final long currentVersionId = (currentFileIndex >= 0 && currentFileIndex < walFiles.length) - ? WALFileUtils.parseVersionId(walFiles[currentFileIndex].getName()) - : -1; + ? walFileVersionIds[currentFileIndex] + : exhaustedKnownFiles && walFileVersionIds.length > 0 + ? walFileVersionIds[walFileVersionIds.length - 1] + : -1; refreshFileList(); if (currentVersionId >= 0) { + final int refreshedIndex = Arrays.binarySearch(walFileVersionIds, currentVersionId); + currentFileIndex = refreshedIndex >= 0 ? refreshedIndex : -refreshedIndex - 2; + } else if (exhaustedKnownFiles) { currentFileIndex = -1; - for (int i = 0; i < walFiles.length; i++) { - if (WALFileUtils.parseVersionId(walFiles[i].getName()) >= currentVersionId) { - currentFileIndex = i; - break; + } + } + + public boolean hasNext() { + while (true) { + if (nextReady != null) { + if (!shouldSkip(nextReady)) { + return true; } + nextReady = null; } - if (currentFileIndex < 0) { - currentFileIndex = walFiles.length; + try { + nextReady = advance(); + if (nextReady != null) { + lastError = null; + } + } catch (IOException e) { + lastError = e; + LOGGER.warn( + DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41, e); + return false; + } + if (nextReady == null) { + return false; } } } - public boolean hasNext() { - if (nextReady != null) { - return true; + /** + * Advances the local search-index lower bound without rebuilding the iterator. Whole WAL files + * are skipped only when every writer progress tuple in them is already covered by queue state. + */ + public void advanceTo( + final long targetSearchIndex, final WriterProgressCoverage writerProgressCoverage) { + if (targetSearchIndex <= minimumSearchIndex) { + return; } + minimumSearchIndex = targetSearchIndex; + + if (writerProgressCoverage == null || !canDiscardBufferedRequests(writerProgressCoverage)) { + return; + } + + refreshForNewerLiveWalFile(); + int targetFileIndex = locateTargetFile(targetSearchIndex); + if (targetFileIndex < 0 + && liveWalNode != null + && targetSearchIndex <= liveWalNode.getCurrentSearchIndex()) { + refresh(); + targetFileIndex = locateTargetFile(targetSearchIndex); + } + if (targetFileIndex < 0 || targetFileIndex <= currentFileIndex) { + return; + } + + final int firstFileToSkip = Math.max(0, currentFileIndex); try { - nextReady = advance(); - if (nextReady != null) { - lastError = null; + for (int fileIndex = firstFileToSkip; fileIndex < targetFileIndex; fileIndex++) { + if (!isWalFileCovered(fileIndex, writerProgressCoverage)) { + return; + } } - } catch (IOException e) { - lastError = e; - LOGGER.warn(DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41, e); + + closeCurrentReader(); + nextReady = null; + pendingRequests.clear(); + pendingSearchIndex = Long.MIN_VALUE; + pendingLocalSeq = Long.MIN_VALUE; + currentFileIndex = targetFileIndex - 1; + resetCurrentFileTracking(); + } catch (final IOException ignored) { + // Fast-forward is opportunistic. Sequential replay remains the correctness fallback. + } + } + + private void refreshForNewerLiveWalFile() { + if (walFileVersionIds.length == 0 + || (liveWalNode != null + && walFileVersionIds[walFileVersionIds.length - 1] + < liveWalNode.getCurrentWALFileVersion())) { + refresh(); + } + } + + private int locateTargetFile(final long targetSearchIndex) { + if (walFiles.length == 0 + || (liveWalNode != null && targetSearchIndex > liveWalNode.getCurrentSearchIndex())) { + return -1; + } + return WALFileUtils.binarySearchFileBySearchIndex(walFiles, targetSearchIndex); + } + + private boolean canDiscardBufferedRequests(final WriterProgressCoverage writerProgressCoverage) { + return (nextReady == null || isCovered(nextReady, writerProgressCoverage)) + && (pendingRequests.isEmpty() + || isCovered( + pendingSearchIndex, + pendingPhysicalTime, + pendingNodeId, + pendingLocalSeq, + writerProgressCoverage)); + } + + private boolean isCovered( + final IndexedConsensusRequest request, final WriterProgressCoverage writerProgressCoverage) { + return isCovered( + request.getSearchIndex(), + request.getPhysicalTime(), + request.getNodeId(), + request.getProgressLocalSeq(), + writerProgressCoverage); + } + + private boolean isCovered( + final long searchIndex, + final long physicalTime, + final int nodeId, + final long localSeq, + final WriterProgressCoverage writerProgressCoverage) { + if (searchIndex >= 0 && searchIndex < minimumSearchIndex) { + return true; + } + return nodeId >= 0 + && physicalTime >= 0 + && localSeq >= 0 + && writerProgressCoverage.isCovered(physicalTime, nodeId, localSeq); + } + + private boolean isWalFileCovered( + final int fileIndex, final WriterProgressCoverage writerProgressCoverage) throws IOException { + final File walFile = walFiles[fileIndex]; + if (WALFileVersion.getVersion(walFile) != WALFileVersion.V3) { return false; } - return nextReady != null; + + final WALMetaData metadata; + final long versionId = walFileVersionIds[fileIndex]; + if (liveWalNode != null && versionId == liveWalNode.getCurrentWALFileVersion()) { + metadata = liveWalNode.getCurrentWALMetaDataSnapshot(); + } else { + try (final ProgressWALReader reader = new ProgressWALReader(walFile)) { + metadata = reader.getMetaData(); + } + } + + final List<Integer> bufferSizes = metadata.getBuffersSize(); + final List<Long> physicalTimes = metadata.getPhysicalTimes(); + final List<Short> nodeIds = metadata.getNodeIds(); + final List<Long> localSeqs = metadata.getLocalSeqs(); + if (physicalTimes.size() != bufferSizes.size() + || nodeIds.size() != bufferSizes.size() + || localSeqs.size() != bufferSizes.size()) { + return false; + } + + for (int entryIndex = 0; entryIndex < bufferSizes.size(); entryIndex++) { + final long physicalTime = physicalTimes.get(entryIndex); + final int nodeId = nodeIds.get(entryIndex); + final long localSeq = localSeqs.get(entryIndex); + if (nodeId < 0 + || physicalTime < 0 + || localSeq < 0 + || !writerProgressCoverage.isCovered(physicalTime, nodeId, localSeq)) { + return false; + } + } + return true; } public IndexedConsensusRequest next() { @@ -337,7 +497,7 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus if (fileIndex < 0) { return false; } - return openReaderAtIndex(fileIndex, consumedEntryCountInCurrentFile); + return openReaderAtIndex(fileIndex, consumedEntryCountInCurrentFile, true, snapshot); } final int previousFileIndex = findFileIndexByVersion(currentReaderVersionId); @@ -351,21 +511,34 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus } private boolean openReaderAtIndex(final int fileIndex, final int skipEntries) throws IOException { - return openReaderAtIndex(fileIndex, skipEntries, true); + return openReaderAtIndex(fileIndex, skipEntries, true, null); } private boolean openReaderAtIndex( final int fileIndex, final int skipEntries, final boolean allowNearLiveRetry) throws IOException { + return openReaderAtIndex(fileIndex, skipEntries, allowNearLiveRetry, null); + } + + private boolean openReaderAtIndex( + final int fileIndex, + final int skipEntries, + final boolean allowNearLiveRetry, + final WALMetaData liveMetaDataSnapshot) + throws IOException { final File walFile = walFiles[fileIndex]; - final long versionId = WALFileUtils.parseVersionId(walFile.getName()); + final long versionId = walFileVersionIds[fileIndex]; final boolean useLiveSnapshot = liveWalNode != null && versionId == liveWalNode.getCurrentWALFileVersion(); try { final ProgressWALReader reader = useLiveSnapshot - ? new ProgressWALReader(walFile, liveWalNode.getCurrentWALMetaDataSnapshot()) + ? new ProgressWALReader( + walFile, + liveMetaDataSnapshot != null + ? liveMetaDataSnapshot + : liveWalNode.getCurrentWALMetaDataSnapshot()) : new ProgressWALReader(walFile); if (!skipEntries(reader, skipEntries)) { reader.close(); @@ -422,18 +595,16 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus } private int findFileIndexByVersion(final long versionId) { - for (int i = 0; i < walFiles.length; i++) { - if (WALFileUtils.parseVersionId(walFiles[i].getName()) == versionId) { - return i; - } - } - return -1; + final int fileIndex = Arrays.binarySearch(walFileVersionIds, versionId); + return fileIndex >= 0 ? fileIndex : -1; } private boolean openFirstReaderAfterVersion(final long versionId) throws IOException { - for (int i = 0; i < walFiles.length; i++) { - if (WALFileUtils.parseVersionId(walFiles[i].getName()) > versionId - && openReaderAtIndex(i, 0)) { + final int matchedFileIndex = Arrays.binarySearch(walFileVersionIds, versionId); + final int firstFileIndexAfterVersion = + matchedFileIndex >= 0 ? matchedFileIndex + 1 : -matchedFileIndex - 1; + for (int i = firstFileIndexAfterVersion; i < walFiles.length; i++) { + if (openReaderAtIndex(i, 0)) { return true; } } @@ -495,7 +666,7 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus } private boolean shouldSkip(final IndexedConsensusRequest request) { - return request.getSearchIndex() >= 0 && request.getSearchIndex() < startSearchIndex; + return request.getSearchIndex() >= 0 && request.getSearchIndex() < minimumSearchIndex; } private void closeCurrentReader() throws IOException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java index 8460fe319f8..d2a8fba8951 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java @@ -28,6 +28,7 @@ import org.apache.iotdb.consensus.iot.WriterSafeFrontierTracker; import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; +import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode; import org.apache.iotdb.db.subscription.event.SubscriptionEvent; import org.apache.iotdb.db.subscription.resource.SubscriptionMemoryManager; import org.apache.iotdb.rpc.subscription.config.TopicConstant; @@ -57,15 +58,21 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.BooleanSupplier; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class ConsensusPrefetchingQueueTest { @@ -450,6 +457,312 @@ public class ConsensusPrefetchingQueueTest { } } + @Test + public void testLateAckDoesNotWaitForPrefetchReadLock() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("late-ack-read-lock"); + ConsensusPrefetchingQueue queue = null; + try { + final DataRegionId regionId = new DataRegionId(1); + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(reader); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.convert(any())) + .thenReturn(Collections.singletonList(createTablet()), Collections.emptyList()); + when(converter.getDatabaseName()).thenReturn("db"); + + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + regionId, + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + converter, + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true); + + final IndexedConsensusRequest dataRequest = + new IndexedConsensusRequest( + 1L, Collections.singletonList(StatementTestUtils.genInsertRowNode(1))) + .setPhysicalTime(1000L) + .setNodeId(7); + final IndexedConsensusRequest requestWithEmptyConversionResult = + new IndexedConsensusRequest( + 2L, Collections.singletonList(StatementTestUtils.genInsertRowNode(2))) + .setPhysicalTime(1001L) + .setNodeId(7); + reader.currentSearchIndex = 2L; + assertTrue(pendingEntries(queue).offer(dataRequest)); + assertTrue(pendingEntries(queue).offer(requestWithEmptyConversionResult)); + assertNull(queue.poll("consumer")); + queue.drivePrefetchOnce(); + + final SubscriptionEvent event = queue.poll("consumer"); + assertNotNull(event); + assertTrue(queue.ackSilent("consumer", event.getCommitContext())); + + final ConsensusPrefetchingQueue activeQueue = queue; + final ReentrantReadWriteLock queueLock = queueLock(queue); + final CountDownLatch ackStarted = new CountDownLatch(1); + final CountDownLatch ackCompleted = new CountDownLatch(1); + final AtomicReference<Boolean> lateAckResult = new AtomicReference<>(); + final AtomicReference<Throwable> asyncFailure = new AtomicReference<>(); + final Thread lateAckThread = + new Thread( + () -> { + ackStarted.countDown(); + try { + lateAckResult.set(activeQueue.ackSilent("consumer", event.getCommitContext())); + } catch (final Throwable t) { + asyncFailure.set(t); + } finally { + ackCompleted.countDown(); + } + }); + lateAckThread.setDaemon(true); + + final boolean completedWhileReadLocked; + queueLock.readLock().lock(); + try { + lateAckThread.start(); + assertTrue(ackStarted.await(5, TimeUnit.SECONDS)); + completedWhileReadLocked = ackCompleted.await(5, TimeUnit.SECONDS); + } finally { + queueLock.readLock().unlock(); + } + lateAckThread.join(TimeUnit.SECONDS.toMillis(5)); + + assertTrue(completedWhileReadLocked); + assertFalse(lateAckThread.isAlive()); + if (asyncFailure.get() != null) { + throw new AssertionError(asyncFailure.get()); + } + assertTrue(Boolean.TRUE.equals(lateAckResult.get())); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testWalCatchUpReusesIteratorAcrossRounds() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("reuse-wal-iterator"); + ConsensusPrefetchingQueue queue = null; + try { + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(reader); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + new DataRegionId(1), + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + mock(ConsensusLogToTabletConverter.class), + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true); + + final ProgressWALIterator iterator = mock(ProgressWALIterator.class); + when(iterator.hasNext()).thenReturn(false); + setSubscriptionWalIterator(queue, iterator); + + final Method tryCatchUpFromWAL = + ConsensusPrefetchingQueue.class.getDeclaredMethod("tryCatchUpFromWAL", long.class); + tryCatchUpFromWAL.setAccessible(true); + tryCatchUpFromWAL.invoke(queue, queue.getCurrentSeekGeneration()); + tryCatchUpFromWAL.invoke(queue, queue.getCurrentSeekGeneration()); + + assertSame(iterator, subscriptionWalIterator(queue)); + verify(iterator, times(2)).refresh(); + verify(iterator, never()).close(); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testReadableWalIteratorSkipsFileListRefresh() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("skip-readable-wal-refresh"); + ConsensusPrefetchingQueue queue = null; + try { + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(reader); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + new DataRegionId(1), + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + mock(ConsensusLogToTabletConverter.class), + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true); + + final ProgressWALIterator iterator = mock(ProgressWALIterator.class); + when(iterator.hasNext()).thenReturn(true); + setSubscriptionWalIterator(queue, iterator); + + invokeEnsureSubscriptionWalReadable(queue); + + verify(iterator).hasNext(); + verify(iterator, never()).refresh(); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testWalRollDoesNotRefreshNewIteratorTwice() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("skip-new-iterator-refresh"); + final File walDirectory = temporaryFolder.newFolder("skip-new-iterator-refresh-wal"); + ConsensusPrefetchingQueue queue = null; + try { + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(walDirectory); + when(walNode.getCurrentSearchIndex()).thenReturn(1L); + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(walNode); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + + final ProgressWALIterator replacementIterator = mock(ProgressWALIterator.class); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + new DataRegionId(1), + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + mock(ConsensusLogToTabletConverter.class), + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true) { + @Override + protected ProgressWALIterator createSubscriptionWALIterator( + final long startSearchIndex) { + assertEquals(1L, startSearchIndex); + return replacementIterator; + } + }; + + final ProgressWALIterator exhaustedIterator = mock(ProgressWALIterator.class); + when(exhaustedIterator.hasNext()).thenReturn(false); + setSubscriptionWalIterator(queue, exhaustedIterator); + + invokeEnsureSubscriptionWalReadable(queue); + + verify(exhaustedIterator, times(2)).hasNext(); + verify(exhaustedIterator).refresh(); + verify(exhaustedIterator).close(); + verify(walNode).rollWALFile(); + verify(replacementIterator, never()).refresh(); + assertSame(replacementIterator, subscriptionWalIterator(queue)); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testPendingCursorAdvanceFastForwardsWalIteratorInPlace() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("deferred-wal-realignment"); + ConsensusPrefetchingQueue queue = null; + try { + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(reader); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + new DataRegionId(1), + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + mock(ConsensusLogToTabletConverter.class), + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true); + + final ProgressWALIterator staleIterator = mock(ProgressWALIterator.class); + setSubscriptionWalIterator(queue, staleIterator); + + final Method advancePendingCursor = + ConsensusPrefetchingQueue.class.getDeclaredMethod( + "advanceLocalCursorFromPendingIfPresent", IndexedConsensusRequest.class, long.class); + advancePendingCursor.setAccessible(true); + advancePendingCursor.invoke( + queue, + new IndexedConsensusRequest(7L, Collections.emptyList()), + queue.getCurrentSeekGeneration()); + + assertEquals(8L, queue.getCurrentReadSearchIndex()); + verify(staleIterator) + .advanceTo(anyLong(), any(ProgressWALIterator.WriterProgressCoverage.class)); + verify(staleIterator, never()).close(); + assertSame(staleIterator, subscriptionWalIterator(queue)); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testCleanupReconcilesUnindexedTabletReservation() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -937,6 +1250,35 @@ public class ConsensusPrefetchingQueueTest { return (BlockingQueue<IndexedConsensusRequest>) field.get(queue); } + private static ReentrantReadWriteLock queueLock(final ConsensusPrefetchingQueue queue) + throws Exception { + final Field field = ConsensusPrefetchingQueue.class.getDeclaredField("lock"); + field.setAccessible(true); + return (ReentrantReadWriteLock) field.get(queue); + } + + private static ProgressWALIterator subscriptionWalIterator(final ConsensusPrefetchingQueue queue) + throws Exception { + final Field field = ConsensusPrefetchingQueue.class.getDeclaredField("subscriptionWALIterator"); + field.setAccessible(true); + return (ProgressWALIterator) field.get(queue); + } + + private static void setSubscriptionWalIterator( + final ConsensusPrefetchingQueue queue, final ProgressWALIterator iterator) throws Exception { + final Field field = ConsensusPrefetchingQueue.class.getDeclaredField("subscriptionWALIterator"); + field.setAccessible(true); + field.set(queue, iterator); + } + + private static void invokeEnsureSubscriptionWalReadable(final ConsensusPrefetchingQueue queue) + throws Exception { + final Method method = + ConsensusPrefetchingQueue.class.getDeclaredMethod("ensureSubscriptionWalReadable"); + method.setAccessible(true); + method.invoke(queue); + } + private static Tablet createTablet() { final List<String> columnNames = Arrays.asList("device", "temperature"); final List<TSDataType> dataTypes = Arrays.asList(TSDataType.STRING, TSDataType.DOUBLE); 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 b37542e936d..92928663126 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 @@ -30,6 +30,7 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; +import org.junit.Assume; import org.junit.Test; import java.io.File; @@ -41,6 +42,8 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class ProgressWALIteratorTest { @@ -272,6 +275,41 @@ public class ProgressWALIteratorTest { } } + @Test + public void testLiveWalReopenReusesMetadataSnapshot() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-live-snapshot"); + final File liveWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + + try { + try (WALWriter writer = new WALWriter(liveWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 1000L, 7, 1L)); + writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 1001L, 7, 2L)); + } + + final WALMetaData firstSnapshot = singleEntryMeta(19, 1L, 1L, 1000L, 7, 1L); + final WALMetaData secondSnapshot = firstSnapshot.copy(); + secondSnapshot.add(19, 2L, 1L, 1001L, 7, 2L); + + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(dir.toFile()); + when(walNode.getCurrentWALFileVersion()).thenReturn(0L); + when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(firstSnapshot, secondSnapshot); + + try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { + assertTrue(iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); + assertTrue(iterator.hasNext()); + assertEquals(2L, iterator.next().getSearchIndex()); + verify(walNode, times(2)).getCurrentWALMetaDataSnapshot(); + } + } finally { + Files.deleteIfExists(liveWal.toPath()); + Files.deleteIfExists(dir); + } + } + @Test public void testIteratorMarksIncompleteScanWhenNearLiveWalCannotBeOpened() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-incomplete-scan"); @@ -299,6 +337,253 @@ public class ProgressWALIteratorTest { } } + @Test + public void testAdvanceToKeepsUncoveredFollowerRequestInCurrentFile() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-current-file-advance"); + final File dataWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File successorWal = + dir.resolve(WALFileUtils.getLogFileName(1, 3, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + + try { + try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + } + try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { + writer.write(searchableEntry(-1L), singleEntryMeta(19, -1L, 1L, 200L, 8, 20L)); + writer.write(searchableEntry(3L), singleEntryMeta(19, 3L, 1L, 300L, 7, 3L)); + } + + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) { + assertTrue(iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); + + iterator.advanceTo(3L, (physicalTime, nodeId, localSeq) -> false); + + assertTrue(iterator.hasNext()); + final IndexedConsensusRequest followerRequest = iterator.next(); + assertEquals(-1L, followerRequest.getSearchIndex()); + assertEquals(8, followerRequest.getNodeId()); + assertTrue(iterator.hasNext()); + assertEquals(3L, iterator.next().getSearchIndex()); + } + } finally { + Files.deleteIfExists(dataWal.toPath()); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + + @Test + public void testAdvanceToSkipsOnlyFullyCoveredWalFiles() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-covered-file-advance"); + final File localWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File followerWal = + dir.resolve(WALFileUtils.getLogFileName(1, 2, WALFileStatus.CONTAINS_NONE_SEARCH_INDEX)) + .toFile(); + final File targetWal = + dir.resolve(WALFileUtils.getLogFileName(2, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + + try { + try (WALWriter writer = new WALWriter(localWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L)); + } + try (WALWriter writer = new WALWriter(followerWal, WALFileVersion.V3)) { + writer.write(searchableEntry(-1L), singleEntryMeta(19, -1L, 1L, 300L, 8, 30L)); + } + try (WALWriter writer = new WALWriter(targetWal, WALFileVersion.V3)) { + writer.write(searchableEntry(6L), singleEntryMeta(19, 6L, 1L, 600L, 7, 6L)); + writer.write(searchableEntry(7L), singleEntryMeta(19, 7L, 1L, 700L, 7, 7L)); + } + + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) { + assertTrue(iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); + + iterator.advanceTo(6L, (physicalTime, nodeId, localSeq) -> nodeId == 7); + + assertTrue(iterator.hasNext()); + final IndexedConsensusRequest uncoveredFollower = iterator.next(); + assertEquals(-1L, uncoveredFollower.getSearchIndex()); + assertEquals(8, uncoveredFollower.getNodeId()); + } + + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) { + assertTrue(iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); + + iterator.advanceTo(6L, (physicalTime, nodeId, localSeq) -> true); + + assertTrue(iterator.hasNext()); + assertEquals(6L, iterator.next().getSearchIndex()); + } + } finally { + Files.deleteIfExists(localWal.toPath()); + Files.deleteIfExists(followerWal.toPath()); + Files.deleteIfExists(targetWal.toPath()); + Files.deleteIfExists(dir); + } + } + + @Test + public void testRefreshAfterExhaustionDiscoversNextWalFile() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-refresh-after-exhaustion"); + final File firstWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File secondWal = + dir.resolve(WALFileUtils.getLogFileName(1, 1, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + + try { + try (WALWriter writer = new WALWriter(firstWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + } + + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) { + assertTrue(iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasNext()); + + try (WALWriter writer = new WALWriter(secondWal, WALFileVersion.V3)) { + writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L)); + } + iterator.refresh(); + + assertTrue(iterator.hasNext()); + assertEquals(2L, iterator.next().getSearchIndex()); + } + } finally { + Files.deleteIfExists(firstWal.toPath()); + Files.deleteIfExists(secondWal.toPath()); + Files.deleteIfExists(dir); + } + } + + @Test + public void testIteratorReusePerformance() throws Exception { + Assume.assumeTrue( + "Enable with -Diotdb.test.subscription.performance=true", + Boolean.getBoolean("iotdb.test.subscription.performance")); + + final int entryCount = Integer.getInteger("iotdb.test.subscription.performance.entries", 4096); + final int batchSize = Integer.getInteger("iotdb.test.subscription.performance.batch-size", 64); + assertTrue("entry count must be positive", entryCount > 0); + assertTrue("batch size must be positive", batchSize > 0); + final Path dir = Files.createTempDirectory("progress-wal-iterator-performance"); + final File dataWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File successorWal = + dir.resolve( + WALFileUtils.getLogFileName( + 1, entryCount + 1L, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + + try { + try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) { + for (long index = 1; index <= entryCount; index++) { + writer.write(searchableEntry(index), singleEntryMeta(19, index, 1L, index, 7, index)); + } + } + try (WALWriter ignored = new WALWriter(successorWal, WALFileVersion.V3)) { + // Seal the data WAL so both benchmark variants read the same historical file. + } + + assertEquals(entryCount, consumeWithReusedIterator(dir.toFile(), entryCount)); + + final long reopenStartNanos = System.nanoTime(); + assertEquals(entryCount, consumeWithReopenedIterator(dir.toFile(), entryCount, batchSize)); + final long reopenNanos = System.nanoTime() - reopenStartNanos; + + final long refreshStartNanos = System.nanoTime(); + assertEquals(entryCount, consumeWithRefreshedIterator(dir.toFile(), entryCount, batchSize)); + final long refreshNanos = System.nanoTime() - refreshStartNanos; + + final long reuseStartNanos = System.nanoTime(); + assertEquals(entryCount, consumeWithReusedIterator(dir.toFile(), entryCount)); + final long reuseNanos = System.nanoTime() - reuseStartNanos; + + final double reopenSpeedup = (double) reopenNanos / Math.max(1L, reuseNanos); + final double refreshSpeedup = (double) refreshNanos / Math.max(1L, reuseNanos); + System.out.printf( + "Subscription WAL iterator benchmark: entries=%d, batchSize=%d, " + + "reopen=%.3f ms, refreshEachBatch=%.3f ms, reuse=%.3f ms, " + + "reopenSpeedup=%.2fx, refreshSpeedup=%.2fx%n", + entryCount, + batchSize, + reopenNanos / 1_000_000.0, + refreshNanos / 1_000_000.0, + reuseNanos / 1_000_000.0, + reopenSpeedup, + refreshSpeedup); + assertTrue( + "Reusing the iterator should be faster than reopening each batch", reopenSpeedup > 1.0); + } finally { + Files.deleteIfExists(dataWal.toPath()); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + + private static int consumeWithReopenedIterator( + final File walDirectory, final int entryCount, final int batchSize) throws Exception { + int consumed = 0; + long nextSearchIndex = 1L; + while (consumed < entryCount) { + int batchCount = 0; + try (ProgressWALIterator iterator = new ProgressWALIterator(walDirectory, nextSearchIndex)) { + while (batchCount < batchSize && iterator.hasNext()) { + nextSearchIndex = iterator.next().getSearchIndex() + 1L; + batchCount++; + consumed++; + } + } + if (batchCount == 0) { + break; + } + } + return consumed; + } + + private static int consumeWithReusedIterator(final File walDirectory, final int entryCount) + throws Exception { + int consumed = 0; + try (ProgressWALIterator iterator = new ProgressWALIterator(walDirectory, 1L)) { + while (consumed < entryCount && iterator.hasNext()) { + iterator.next(); + consumed++; + } + } + return consumed; + } + + private static int consumeWithRefreshedIterator( + final File walDirectory, final int entryCount, final int batchSize) throws Exception { + int consumed = 0; + try (ProgressWALIterator iterator = new ProgressWALIterator(walDirectory, 1L)) { + while (consumed < entryCount) { + iterator.refresh(); + int batchCount = 0; + while (batchCount < batchSize && iterator.hasNext()) { + iterator.next(); + batchCount++; + consumed++; + } + if (batchCount == 0) { + break; + } + } + } + return consumed; + } + private static ByteBuffer searchableEntry(final long bodySearchIndex) { final ByteBuffer buffer = ByteBuffer.allocate(WALInfoEntry.FIXED_SERIALIZED_SIZE + PlanNodeType.BYTES + Long.BYTES);
