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 27e9920399b feat: cache sorted WAL file lists (#18456)
27e9920399b is described below
commit 27e9920399b3200de2117b651aed235c12c9ee2f
Author: Jiang Tian <[email protected]>
AuthorDate: Thu Aug 13 15:02:26 2026 +0800
feat: cache sorted WAL file lists (#18456)
* ver1
* ver2
* add perftest
* feat: enable WAL file list cache by default
---
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 6 +
.../dataregion/wal/buffer/WALBuffer.java | 34 ++-
.../storageengine/dataregion/wal/node/WALNode.java | 108 ++++++-
.../org/apache/iotdb/db/conf/PropertiesTest.java | 28 ++
.../wal/node/ConsensusReqReaderTest.java | 51 ++++
.../wal/node/WALFileListCachePerformanceTest.java | 338 +++++++++++++++++++++
.../wal/node/WalDeleteOutdatedNewTest.java | 12 +
.../conf/iotdb-system.properties.template | 9 +
9 files changed, 585 insertions(+), 12 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 1d726366ddd..5e0ff2dec6a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -219,6 +219,9 @@ public class IoTDBConfig {
/** The period when outdated wal files are periodically deleted. Unit:
millisecond */
private volatile long deleteWalFilesPeriodInMs = 20 * 1000L;
+ /** Whether WAL nodes cache their sorted WAL file lists. */
+ private boolean walFileListCacheEnabled = true;
+
/**
* Enables or disables the automatic clearing of the WAL cache when a memory
compaction is
* triggered. When enabled, the WAL cache will be cleared to release memory
during the compaction
@@ -2220,6 +2223,14 @@ public class IoTDBConfig {
this.deleteWalFilesPeriodInMs = deleteWalFilesPeriodInMs;
}
+ public boolean isWalFileListCacheEnabled() {
+ return walFileListCacheEnabled;
+ }
+
+ public void setWalFileListCacheEnabled(boolean walFileListCacheEnabled) {
+ this.walFileListCacheEnabled = walFileListCacheEnabled;
+ }
+
public boolean getWALCacheShrinkClearEnabled() {
return WALCacheShrinkClearEnabled;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 15d2b7d003d..fedcc73dbbe 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -1414,6 +1414,12 @@ public class IoTDBDescriptor {
conf.setWALCacheShrinkClearEnabled(WALInsertNodeCacheShrinkClearEnabled);
}
+ conf.setWalFileListCacheEnabled(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "wal_file_list_cache_enabled",
+ Boolean.toString(conf.isWalFileListCacheEnabled()))));
+
loadWALHotModifiedProps(properties);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
index 25fb31ba804..6921abecf75 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
@@ -64,6 +64,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.BiConsumer;
import java.util.function.Predicate;
import static
org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode.DEFAULT_SEARCH_INDEX;
@@ -122,9 +123,16 @@ public class WALBuffer extends AbstractWALBuffer {
// manage wal files which have MemTableIds
private final Map<Long, Set<Long>> memTableIdsOfWal = new
ConcurrentHashMap<>();
+ private final BiConsumer<File, File> walFileRolledListener;
public WALBuffer(String identifier, String logDirectory) throws IOException {
- this(identifier, logDirectory, new CheckpointManager(identifier,
logDirectory), 0, 0L);
+ this(
+ identifier,
+ logDirectory,
+ new CheckpointManager(identifier, logDirectory),
+ 0,
+ 0L,
+ (sealedWalFile, currentWalFile) -> {});
}
public WALBuffer(
@@ -134,8 +142,26 @@ public class WALBuffer extends AbstractWALBuffer {
long startFileVersion,
long startSearchIndex)
throws IOException {
+ this(
+ identifier,
+ logDirectory,
+ checkpointManager,
+ startFileVersion,
+ startSearchIndex,
+ (sealedWalFile, currentWalFile) -> {});
+ }
+
+ public WALBuffer(
+ String identifier,
+ String logDirectory,
+ CheckpointManager checkpointManager,
+ long startFileVersion,
+ long startSearchIndex,
+ BiConsumer<File, File> walFileRolledListener)
+ throws IOException {
super(identifier, logDirectory, startFileVersion, startSearchIndex);
this.checkpointManager = checkpointManager;
+ this.walFileRolledListener = walFileRolledListener;
currentFileStatus = WALFileStatus.CONTAINS_NONE_SEARCH_INDEX;
allocateBuffers();
currentWALFileWriter.setCompressedByteBuffer(compressedByteBuffer);
@@ -168,8 +194,10 @@ public class WALBuffer extends AbstractWALBuffer {
@Override
protected File rollLogWriter(long searchIndex, WALFileStatus fileStatus)
throws IOException {
- File file = super.rollLogWriter(searchIndex, fileStatus);
+ File sealedWalFile = super.rollLogWriter(searchIndex, fileStatus);
currentWALFileWriter.setCompressedByteBuffer(compressedByteBuffer);
+ // Update the WAL node's ordered file index before waking readers waiting
for this roll.
+ walFileRolledListener.accept(sealedWalFile,
currentWALFileWriter.getLogFile());
buffersLock.lock();
try {
// notify WALReader that new file is generated, and it can read new file
@@ -177,7 +205,7 @@ public class WALBuffer extends AbstractWALBuffer {
} finally {
buffersLock.unlock();
}
- return file;
+ return sealedWalFile;
}
@TestOnly
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
index a59ebeaa21d..03438b60707 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
@@ -82,6 +82,7 @@ import java.util.ListIterator;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -120,6 +121,15 @@ public class WALNode implements IWALNode {
// WAL files with versionId >= this value are retained for subscription
consumers
private volatile long subscriptionRetainedMinVersionId = Long.MAX_VALUE;
+ private final boolean walFileListCacheEnabled;
+ private final Object walFileListCacheLock = new Object();
+ // Maintains the cache incrementally by version; all mutations are protected
by the cache lock.
+ private final TreeMap<Long, File> walFilesByVersion = new TreeMap<>();
+ // File changes only mark the cache dirty; the next file-list access
publishes a snapshot lazily.
+ private volatile boolean sortedWalFilesCacheDirty = true;
+ // Replaced as a never-mutated snapshot so readers cannot observe a
partially updated list.
+ private volatile File[] sortedWalFilesCache;
+
private volatile boolean deleted = false;
public WALNode(String identifier, String logDirectory) throws IOException {
@@ -135,9 +145,80 @@ public class WALNode implements IWALNode {
logger.info(StorageEngineMessages.CREATE_FOLDER_FOR_WAL_NODE,
logDirectory, identifier);
}
this.checkpointManager = new CheckpointManager(identifier, logDirectory);
+ this.walFileListCacheEnabled = config.isWalFileListCacheEnabled();
this.buffer =
new WALBuffer(
- identifier, logDirectory, checkpointManager, startFileVersion,
startSearchIndex);
+ identifier,
+ logDirectory,
+ checkpointManager,
+ startFileVersion,
+ startSearchIndex,
+ this::updateSortedWalFilesCacheAfterRoll);
+ initializeSortedWalFilesCacheIfEnabled();
+ }
+
+ private File[] getSortedWalFiles() {
+ if (!walFileListCacheEnabled) {
+ return listAndSortWalFiles();
+ }
+ File[] snapshot = sortedWalFilesCache;
+ if (!sortedWalFilesCacheDirty && snapshot != null) {
+ return snapshot;
+ }
+ synchronized (walFileListCacheLock) {
+ if (sortedWalFilesCacheDirty || sortedWalFilesCache == null) {
+ sortedWalFilesCache = walFilesByVersion.values().toArray(new File[0]);
+ sortedWalFilesCacheDirty = false;
+ }
+ return sortedWalFilesCache;
+ }
+ }
+
+ private File[] listAndSortWalFiles() {
+ final File[] walFiles = WALFileUtils.listAllWALFiles(logDirectory);
+ if (walFiles != null) {
+ WALFileUtils.ascSortByVersionId(walFiles);
+ }
+ return walFiles;
+ }
+
+ private void initializeSortedWalFilesCacheIfEnabled() {
+ if (walFileListCacheEnabled) {
+ synchronized (walFileListCacheLock) {
+ File[] walFiles = WALFileUtils.listAllWALFiles(logDirectory);
+ walFilesByVersion.clear();
+ if (walFiles != null) {
+ for (File walFile : walFiles) {
+
walFilesByVersion.put(WALFileUtils.parseVersionId(walFile.getName()), walFile);
+ }
+ }
+ sortedWalFilesCacheDirty = true;
+ }
+ }
+ }
+
+ private void updateSortedWalFilesCacheAfterRoll(File sealedWalFile, File
currentWalFile) {
+ if (!walFileListCacheEnabled) {
+ return;
+ }
+ synchronized (walFileListCacheLock) {
+
walFilesByVersion.put(WALFileUtils.parseVersionId(sealedWalFile.getName()),
sealedWalFile);
+
walFilesByVersion.put(WALFileUtils.parseVersionId(currentWalFile.getName()),
currentWalFile);
+ sortedWalFilesCacheDirty = true;
+ }
+ }
+
+ private void removeDeletedWalFilesFromCache(List<Long> deletedVersionIds) {
+ if (!walFileListCacheEnabled || deletedVersionIds.isEmpty()) {
+ return;
+ }
+ synchronized (walFileListCacheLock) {
+ for (Long deletedVersionId : deletedVersionIds) {
+ if (walFilesByVersion.remove(deletedVersionId) != null) {
+ sortedWalFilesCacheDirty = true;
+ }
+ }
+ }
}
@Override
@@ -291,7 +372,7 @@ public class WALNode implements IWALNode {
private boolean initAndCheckIfNeedContinue() {
rollWalFileIfHaveNoActiveMemTable();
- File[] allWalFilesOfOneNode = WALFileUtils.listAllWALFiles(logDirectory);
+ File[] allWalFilesOfOneNode = getSortedWalFiles();
if (allWalFilesOfOneNode == null || allWalFilesOfOneNode.length <= 1) {
if (logger.isDebugEnabled()) {
logger.debug(
@@ -301,7 +382,6 @@ public class WALNode implements IWALNode {
}
return false;
}
- WALFileUtils.ascSortByVersionId(allWalFilesOfOneNode);
this.sortedWalFilesExcludingLast =
Arrays.copyOfRange(allWalFilesOfOneNode, 0,
allWalFilesOfOneNode.length - 1);
this.activeOrPinnedMemTables =
checkpointManager.activeOrPinnedMemTables();
@@ -413,6 +493,7 @@ public class WALNode implements IWALNode {
}
buffer.subtractDiskUsage(deleteFileSize);
buffer.subtractFileNum(successfullyDeleted.size());
+ removeDeletedWalFilesFromCache(successfullyDeleted);
}
private int initFileIndexAfterFilterSafelyDeleteIndex() {
@@ -916,8 +997,7 @@ public class WALNode implements IWALNode {
}
private void updateFilesToSearch() {
- File[] filesToSearch = WALFileUtils.listAllWALFiles(logDirectory);
- WALFileUtils.ascSortByVersionId(filesToSearch);
+ File[] filesToSearch = getSortedWalFiles();
int fileIndex =
WALFileUtils.binarySearchFileBySearchIndex(filesToSearch, nextSearchIndex);
logger.debug(
StorageEngineMessages.STORAGE_LOG_SEARCHINDEX_RESULT_FILES_6151DCEB,
@@ -981,12 +1061,11 @@ public class WALNode implements IWALNode {
if (bytesToFree <= 0) {
return new Pair<>(DEFAULT_SAFELY_DELETED_SEARCH_INDEX, 0L);
}
- File[] walFiles = WALFileUtils.listAllWALFiles(logDirectory);
+ File[] walFiles = getSortedWalFiles();
if (walFiles == null || walFiles.length <= 1) {
// No files or only the current-writing file — cannot free anything
return new Pair<>(DEFAULT_SAFELY_DELETED_SEARCH_INDEX, 0L);
}
- WALFileUtils.ascSortByVersionId(walFiles);
// Exclude the last file (currently being written)
long accumulated = 0;
for (int i = 0; i < walFiles.length - 1; i++) {
@@ -1017,11 +1096,10 @@ public class WALNode implements IWALNode {
@Override
public Pair<Long, Long> getDeletionBoundBeforeTimestamp(long cutoffTimeMs) {
- File[] walFiles = WALFileUtils.listAllWALFiles(logDirectory);
+ File[] walFiles = getSortedWalFiles();
if (walFiles == null || walFiles.length <= 1) {
return new Pair<>(Long.MIN_VALUE + 1, 0L);
}
- WALFileUtils.ascSortByVersionId(walFiles);
int expiredPrefixLength = countExpiredRolledWalFiles(walFiles,
cutoffTimeMs);
if (expiredPrefixLength == 0) {
return new Pair<>(Long.MIN_VALUE + 1, 0L);
@@ -1066,6 +1144,18 @@ public class WALNode implements IWALNode {
return logDirectory;
}
+ @TestOnly
+ File[] getCachedSortedWalFiles() {
+ return sortedWalFilesCache == null
+ ? null
+ : Arrays.copyOf(sortedWalFilesCache, sortedWalFilesCache.length);
+ }
+
+ @TestOnly
+ File[] getSortedWalFilesForTest() {
+ return getSortedWalFiles();
+ }
+
/** Get the .wal file starts with the specified version id */
public File getWALFile(long versionId) throws FileNotFoundException {
return WALFileUtils.getWALFile(logDirectory, versionId);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java
index 191719ab8d4..80f8811d096 100755
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java
@@ -41,6 +41,34 @@ import java.util.Properties;
import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.noClasses;
public class PropertiesTest {
+ /**
+ * Verifies that WAL file-list caching defaults on, supports startup
override, and is
+ * restart-only.
+ */
+ @Test
+ public void testWalFileListCacheConfiguration() throws Exception {
+ final IoTDBDescriptor descriptor = IoTDBDescriptor.getInstance();
+ final boolean originalValue =
descriptor.getConfig().isWalFileListCacheEnabled();
+ final TrimProperties properties = new TrimProperties();
+
+ try {
+ Assert.assertTrue(new IoTDBConfig().isWalFileListCacheEnabled());
+ Assert.assertTrue(
+ Boolean.parseBoolean(
+
ConfigurationFileUtils.getConfigurationDefaultValue("wal_file_list_cache_enabled")));
+
+ properties.setProperty("wal_file_list_cache_enabled", "false");
+ descriptor.loadProperties(properties);
+ Assert.assertFalse(descriptor.getConfig().isWalFileListCacheEnabled());
+
+ properties.setProperty("wal_file_list_cache_enabled", "true");
+ descriptor.loadHotModifiedProps(properties);
+ Assert.assertFalse(descriptor.getConfig().isWalFileListCacheEnabled());
+ } finally {
+ descriptor.getConfig().setWalFileListCacheEnabled(originalValue);
+ }
+ }
+
@Test
public void testHotReloadNegativeWalThrottleThresholdUsesDefault() throws
Exception {
final String key = "wal_throttle_threshold_in_byte";
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/ConsensusReqReaderTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/ConsensusReqReaderTest.java
index 761f9f16025..e1a48eb73f0 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/ConsensusReqReaderTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/ConsensusReqReaderTest.java
@@ -33,6 +33,8 @@ import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNod
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus;
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.db.utils.constant.TestConstant;
@@ -60,13 +62,16 @@ public class ConsensusReqReaderTest {
private static final String logDirectory =
TestConstant.BASE_OUTPUT_PATH.concat("wal-test");
private static final String devicePath = "root.test_sg.test_d";
private WALMode prevMode;
+ private boolean prevWalFileListCacheEnabled;
private WALNode walNode;
@Before
public void setUp() throws Exception {
EnvironmentUtils.cleanDir(logDirectory);
prevMode = config.getWalMode();
+ prevWalFileListCacheEnabled = config.isWalFileListCacheEnabled();
config.setWalMode(WALMode.SYNC);
+ config.setWalFileListCacheEnabled(true);
walNode = new WALNode(identifier, logDirectory);
}
@@ -74,9 +79,53 @@ public class ConsensusReqReaderTest {
public void tearDown() throws Exception {
walNode.close();
config.setWalMode(prevMode);
+ config.setWalFileListCacheEnabled(prevWalFileListCacheEnabled);
EnvironmentUtils.cleanDir(logDirectory);
}
+ /** Verifies that reader access publishes an incrementally maintained file
snapshot. */
+ @Test
+ public void testIncrementallyUpdateCachedWalFilesAfterRoll() throws
IOException {
+ Assert.assertNull(walNode.getCachedSortedWalFiles());
+
+ Assert.assertFalse(walNode.getReqIterator(0).hasNext());
+ File[] initialCachedWalFiles = walNode.getCachedSortedWalFiles();
+ Assert.assertEquals(1, initialCachedWalFiles.length);
+ Assert.assertEquals(0,
WALFileUtils.parseVersionId(initialCachedWalFiles[0].getName()));
+
+ // A directory rescan would incorrectly add this external sentinel to the
cached snapshot.
+ Assert.assertTrue(new File(logDirectory,
"_100-100-1.wal").createNewFile());
+ walNode.rollWALFile();
+ Assert.assertArrayEquals(initialCachedWalFiles,
walNode.getCachedSortedWalFiles());
+
+ Assert.assertFalse(walNode.getReqIterator(0).hasNext());
+ File[] cachedWalFilesAfterFirstRoll = walNode.getCachedSortedWalFiles();
+ Assert.assertEquals(2, cachedWalFilesAfterFirstRoll.length);
+ Assert.assertEquals(
+ WALFileStatus.CONTAINS_NONE_SEARCH_INDEX,
+
WALFileUtils.parseStatusCode(cachedWalFilesAfterFirstRoll[0].getName()));
+ Assert.assertEquals(1,
WALFileUtils.parseVersionId(cachedWalFilesAfterFirstRoll[1].getName()));
+
+ walNode.rollWALFile();
+ Assert.assertArrayEquals(cachedWalFilesAfterFirstRoll,
walNode.getCachedSortedWalFiles());
+
+ Assert.assertFalse(walNode.getReqIterator(0).hasNext());
+ File[] cachedWalFilesAfterSecondRoll = walNode.getCachedSortedWalFiles();
+ Assert.assertEquals(3, cachedWalFilesAfterSecondRoll.length);
+ for (int i = 0; i < cachedWalFilesAfterSecondRoll.length; i++) {
+ Assert.assertEquals(
+ i,
WALFileUtils.parseVersionId(cachedWalFilesAfterSecondRoll[i].getName()));
+ }
+ }
+
+ private void recreateWalNodeWithoutFileListCache() throws IOException {
+ // These corruption tests replace WAL files outside WALNode, so use the
directory-scanning mode
+ // that is designed to observe such external test-fixture mutations.
+ walNode.close();
+ config.setWalFileListCacheEnabled(false);
+ walNode = new WALNode(identifier, logDirectory);
+ }
+
/**
* Generate wal files as below: <br>
* _0-0-1.wal: 1,-1 <br>
@@ -554,6 +603,7 @@ public class ConsensusReqReaderTest {
@Test
public void scenario03TestGetReqIterator01() throws Exception {
+ recreateWalNodeWithoutFileListCache();
simulateFileScenario03();
walNode.rollWALFile();
@@ -608,6 +658,7 @@ public class ConsensusReqReaderTest {
@Test
public void scenario03TestGetReqIterator02() throws Exception {
+ recreateWalNodeWithoutFileListCache();
simulateFileScenario03();
walNode.rollWALFile();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFileListCachePerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFileListCachePerformanceTest.java
new file mode 100644
index 00000000000..340dfa8f300
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFileListCachePerformanceTest.java
@@ -0,0 +1,338 @@
+/*
+ * 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.storageengine.dataregion.wal.node;
+
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus;
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.db.utils.constant.TestConstant;
+
+import com.sun.management.HotSpotDiagnosticMXBean;
+import org.apache.tsfile.utils.RamUsageEstimator;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.util.Arrays;
+import java.util.Locale;
+import java.util.TreeMap;
+
+public class WALFileListCachePerformanceTest {
+
+ private static final String ENABLED_PROPERTY =
"iotdb.wal.file-list-cache.perf.enabled";
+ private static final String FILE_COUNTS_PROPERTY =
"iotdb.wal.file-list-cache.perf.file-counts";
+ private static final String WARMUP_READS_PROPERTY =
"iotdb.wal.file-list-cache.perf.warmup-reads";
+ private static final String READS_PROPERTY =
"iotdb.wal.file-list-cache.perf.reads";
+ private static final String ROUNDS_PROPERTY =
"iotdb.wal.file-list-cache.perf.rounds";
+
+ private static final String BASE_DIRECTORY =
+ TestConstant.BASE_OUTPUT_PATH.concat("wal-file-list-cache-performance");
+ private static final int[] DEFAULT_FILE_COUNTS = {1, 10, 100, 1000};
+
+ private static final IoTDBConfig CONFIG =
IoTDBDescriptor.getInstance().getConfig();
+
+ private static final long TREE_MAP_ENTRY_SHALLOW_SIZE =
getTreeMapEntryShallowSize();
+ private static final long FILE_SHALLOW_SIZE =
RamUsageEstimator.shallowSizeOfInstance(File.class);
+ private static final long LONG_SHALLOW_SIZE =
RamUsageEstimator.shallowSizeOfInstance(Long.class);
+ private static final long STRING_SHALLOW_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(String.class);
+ private static final boolean COMPACT_STRINGS_ENABLED =
isCompactStringsEnabled();
+
+ private static volatile long benchmarkBlackhole;
+
+ /**
+ * Compares repeated sorted-WAL-file reads with the cache disabled and
enabled for increasing
+ * directory sizes. The cache-disabled path must rescan and sort on every
read, while the enabled
+ * path should reuse the immutable snapshot after warmup.
+ */
+ @Test
+ public void benchmarkRepeatedSortedWalFileReads() throws Exception {
+ assumePerformanceTestEnabled();
+
+ final int[] fileCounts = parseFileCounts();
+ final int warmupReads = Integer.getInteger(WARMUP_READS_PROPERTY, 500);
+ final int reads = Integer.getInteger(READS_PROPERTY, 2000);
+ final int rounds = Integer.getInteger(ROUNDS_PROPERTY, 5);
+ Assert.assertTrue(warmupReads > 0);
+ Assert.assertTrue(reads > 0);
+ Assert.assertTrue(rounds > 0);
+
+ final boolean originalCacheEnabled = CONFIG.isWalFileListCacheEnabled();
+ EnvironmentUtils.cleanDir(BASE_DIRECTORY);
+ try {
+ for (int fileCount : fileCounts) {
+ runScenario(fileCount, warmupReads, reads, rounds);
+ }
+ } finally {
+ CONFIG.setWalFileListCacheEnabled(originalCacheEnabled);
+ EnvironmentUtils.cleanDir(BASE_DIRECTORY);
+ }
+ }
+
+ /**
+ * Estimates the retained heap added by enabling the cache for increasing
WAL directory sizes. The
+ * estimate uses the current JVM object layout and the actual cached file
paths, and separates the
+ * incrementally maintained index from the snapshot that is retained only
after first access.
+ */
+ @Test
+ public void measureRetainedWalFileListCacheMemory() throws Exception {
+ assumePerformanceTestEnabled();
+
+ final boolean originalCacheEnabled = CONFIG.isWalFileListCacheEnabled();
+ EnvironmentUtils.cleanDir(BASE_DIRECTORY);
+ try {
+ for (int fileCount : parseFileCounts()) {
+ runMemoryScenario(fileCount);
+ }
+ } finally {
+ CONFIG.setWalFileListCacheEnabled(originalCacheEnabled);
+ EnvironmentUtils.cleanDir(BASE_DIRECTORY);
+ }
+ }
+
+ private static void runMemoryScenario(int fileCount) throws Exception {
+ Assert.assertTrue(fileCount > 0);
+ final WALNode cacheDisabled = createWalNode(fileCount, false);
+ final WALNode cacheEnabled = createWalNode(fileCount, true);
+ try {
+ Assert.assertNull(cacheDisabled.getCachedSortedWalFiles());
+ Assert.assertNull(cacheEnabled.getCachedSortedWalFiles());
+ assertFileList(cacheDisabled.getSortedWalFilesForTest(), fileCount);
+ assertFileList(cacheEnabled.getSortedWalFilesForTest(), fileCount);
+ Assert.assertNull(cacheDisabled.getCachedSortedWalFiles());
+
+ final File[] cachedWalFiles = cacheEnabled.getCachedSortedWalFiles();
+ Assert.assertNotNull(cachedWalFiles);
+ assertFileList(cachedWalFiles, fileCount);
+
+ final long entryBytes = TREE_MAP_ENTRY_SHALLOW_SIZE *
cachedWalFiles.length;
+ final long fileBytes = FILE_SHALLOW_SIZE * cachedWalFiles.length;
+ long keyBytes = 0;
+ long pathBytes = 0;
+ long pathCharacters = 0;
+ for (File walFile : cachedWalFiles) {
+ pathBytes += estimateStringBytes(walFile.getPath());
+ pathCharacters += walFile.getPath().length();
+ final long versionId = WALFileUtils.parseVersionId(walFile.getName());
+ // Long values in this range come from the JVM-wide cache and are not
retained by this
+ // feature.
+ if (versionId < -128 || versionId > 127) {
+ keyBytes += LONG_SHALLOW_SIZE;
+ }
+ }
+ final long indexBytes = entryBytes + fileBytes + keyBytes + pathBytes;
+ // TreeMap.values() is only a temporary view while publishing the array
snapshot.
+ final long snapshotBytes =
RamUsageEstimator.sizeOfObjectArray(cachedWalFiles.length);
+ final long totalBytes = indexBytes + snapshotBytes;
+ System.out.printf(
+ Locale.ROOT,
+ "WAL file-list cache retained-memory estimate: files=%d%n"
+ + " cache=false retained=0 B%n"
+ + " cache=true index=%d B (%.3f KiB), lazy-snapshot=%d B (%.3f
KiB), "
+ + "total=%d B (%.3f KiB), total/file=%.2f B%n"
+ + " breakdown entries=%d B, keys=%d B, files=%d B, paths=%d
B, "
+ + "avg-path=%.1f chars, compact-strings=%s%n",
+ fileCount,
+ indexBytes,
+ bytesToKiB(indexBytes),
+ snapshotBytes,
+ bytesToKiB(snapshotBytes),
+ totalBytes,
+ bytesToKiB(totalBytes),
+ (double) totalBytes / fileCount,
+ entryBytes,
+ keyBytes,
+ fileBytes,
+ pathBytes,
+ (double) pathCharacters / fileCount,
+ COMPACT_STRINGS_ENABLED);
+ } finally {
+ cacheDisabled.close();
+ cacheEnabled.close();
+ }
+ }
+
+ private static void runScenario(int fileCount, int warmupReads, int reads,
int rounds)
+ throws Exception {
+ Assert.assertTrue(fileCount > 0);
+ final WALNode cacheDisabled = createWalNode(fileCount, false);
+ final WALNode cacheEnabled = createWalNode(fileCount, true);
+ try {
+ assertFileList(cacheDisabled.getSortedWalFilesForTest(), fileCount);
+ assertFileList(cacheEnabled.getSortedWalFilesForTest(), fileCount);
+
+ runReads(cacheDisabled, warmupReads);
+ runReads(cacheEnabled, warmupReads);
+
+ final long[] cacheDisabledNanos = new long[rounds];
+ final long[] cacheEnabledNanos = new long[rounds];
+ for (int round = 0; round < rounds; round++) {
+ if ((round & 1) == 0) {
+ cacheDisabledNanos[round] = measureReads(cacheDisabled, reads);
+ cacheEnabledNanos[round] = measureReads(cacheEnabled, reads);
+ } else {
+ cacheEnabledNanos[round] = measureReads(cacheEnabled, reads);
+ cacheDisabledNanos[round] = measureReads(cacheDisabled, reads);
+ }
+ }
+
+ final double cacheDisabledNanosPerRead = median(cacheDisabledNanos) /
reads;
+ final double cacheEnabledNanosPerRead = median(cacheEnabledNanos) /
reads;
+ System.out.printf(
+ Locale.ROOT,
+ "WAL sorted-file-list benchmark: files=%d, warmup-reads=%d,
reads/round=%d, rounds=%d%n",
+ fileCount,
+ warmupReads,
+ reads,
+ rounds);
+ printResult("cache=false", cacheDisabledNanosPerRead);
+ printResult("cache=true", cacheEnabledNanosPerRead);
+ System.out.printf(
+ Locale.ROOT,
+ " speedup=%.2fx, latency reduction=%.2f%%%n",
+ cacheDisabledNanosPerRead / cacheEnabledNanosPerRead,
+ (cacheDisabledNanosPerRead - cacheEnabledNanosPerRead)
+ * 100.0
+ / cacheDisabledNanosPerRead);
+ } finally {
+ cacheDisabled.close();
+ cacheEnabled.close();
+ }
+ }
+
+ private static WALNode createWalNode(int fileCount, boolean cacheEnabled)
throws IOException {
+ final String state = cacheEnabled ? "enabled" : "disabled";
+ final String directory = BASE_DIRECTORY + File.separator + fileCount +
File.separator + state;
+ EnvironmentUtils.cleanDir(directory);
+ final File directoryFile = new File(directory);
+ Assert.assertTrue(directoryFile.mkdirs() || directoryFile.isDirectory());
+ for (int version = 0; version < fileCount - 1; version++) {
+ final File walFile =
+ new File(
+ directoryFile,
+ WALFileUtils.getLogFileName(version, version,
WALFileStatus.CONTAINS_SEARCH_INDEX));
+ Assert.assertTrue(walFile.createNewFile());
+ }
+
+ CONFIG.setWalFileListCacheEnabled(cacheEnabled);
+ return new WALNode(
+ "wal-file-list-cache-performance-" + fileCount + '-' + state,
+ directory,
+ fileCount - 1L,
+ fileCount - 1L);
+ }
+
+ private static long measureReads(WALNode walNode, int reads) {
+ final long startNanos = System.nanoTime();
+ runReads(walNode, reads);
+ return System.nanoTime() - startNanos;
+ }
+
+ private static void runReads(WALNode walNode, int reads) {
+ long checksum = 0;
+ for (int i = 0; i < reads; i++) {
+ final File[] walFiles = walNode.getSortedWalFilesForTest();
+ checksum += walFiles.length;
+ checksum += walFiles[walFiles.length - 1].getName().length();
+ }
+ benchmarkBlackhole = checksum;
+ }
+
+ private static void assertFileList(File[] walFiles, int expectedFileCount) {
+ Assert.assertEquals(expectedFileCount, walFiles.length);
+ for (int i = 0; i < walFiles.length; i++) {
+ Assert.assertEquals(i,
WALFileUtils.parseVersionId(walFiles[i].getName()));
+ }
+ }
+
+ private static long getTreeMapEntryShallowSize() {
+ final TreeMap<Long, File> sample = new TreeMap<>();
+ sample.put(0L, new File("sample.wal"));
+ return
RamUsageEstimator.shallowSizeOf(sample.entrySet().iterator().next());
+ }
+
+ private static long estimateStringBytes(String value) {
+ final boolean latin1 = value.chars().allMatch(character -> character <=
0xFF);
+ final int bytesPerCharacter = COMPACT_STRINGS_ENABLED && latin1 ? 1 : 2;
+ return STRING_SHALLOW_SIZE
+ + RamUsageEstimator.sizeOfByteArray(value.length() *
bytesPerCharacter);
+ }
+
+ private static boolean isCompactStringsEnabled() {
+ try {
+ return Boolean.parseBoolean(
+ ManagementFactory.getPlatformMXBean(HotSpotDiagnosticMXBean.class)
+ .getVMOption("CompactStrings")
+ .getValue());
+ } catch (RuntimeException ignored) {
+ // The conservative fallback models two bytes per UTF-16 code unit.
+ return false;
+ }
+ }
+
+ private static double bytesToKiB(long bytes) {
+ return bytes / 1024.0;
+ }
+
+ private static void assumePerformanceTestEnabled() {
+ Assume.assumeTrue(
+ String.format(
+ Locale.ROOT,
+ "Manual performance UT. Enable with -D%s=true; optionally tune
-D%s, -D%s, -D%s, and -D%s.",
+ ENABLED_PROPERTY,
+ FILE_COUNTS_PROPERTY,
+ WARMUP_READS_PROPERTY,
+ READS_PROPERTY,
+ ROUNDS_PROPERTY),
+ Boolean.getBoolean(ENABLED_PROPERTY));
+ }
+
+ private static int[] parseFileCounts() {
+ final String configured = System.getProperty(FILE_COUNTS_PROPERTY);
+ if (configured == null || configured.trim().isEmpty()) {
+ return DEFAULT_FILE_COUNTS;
+ }
+ return Arrays.stream(configured.split(","))
+ .map(String::trim)
+ .mapToInt(Integer::parseInt)
+ .toArray();
+ }
+
+ private static double median(long[] values) {
+ Arrays.sort(values);
+ final int middle = values.length / 2;
+ return (values.length & 1) == 1
+ ? values[middle]
+ : values[middle - 1] + (values[middle] - values[middle - 1]) / 2.0;
+ }
+
+ private static void printResult(String label, double nanosPerRead) {
+ System.out.printf(
+ Locale.ROOT,
+ " %-12s latency=%.3f us/read, throughput=%.0f reads/s%n",
+ label,
+ nanosPerRead / 1000.0,
+ 1_000_000_000.0 / nanosPerRead);
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WalDeleteOutdatedNewTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WalDeleteOutdatedNewTest.java
index 2f57db238b0..ab95d7f067e 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WalDeleteOutdatedNewTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WalDeleteOutdatedNewTest.java
@@ -62,6 +62,7 @@ public class WalDeleteOutdatedNewTest {
private static final String dataRegionId = "1";
private WALMode prevMode;
private String prevConsensus;
+ private boolean prevWalFileListCacheEnabled;
private WALNode walNode1;
@Before
@@ -69,8 +70,10 @@ public class WalDeleteOutdatedNewTest {
EnvironmentUtils.cleanDir(logDirectory1);
prevMode = config.getWalMode();
prevConsensus = config.getDataRegionConsensusProtocolClass();
+ prevWalFileListCacheEnabled = config.isWalFileListCacheEnabled();
config.setWalMode(WALMode.SYNC);
config.setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
+ config.setWalFileListCacheEnabled(true);
walNode1 = new WALNode(identifier1, logDirectory1);
DataRegion dataRegion = new DataRegionTest.DummyDataRegion(logDirectory1,
databasePath);
dataRegion.updatePartitionFileVersion(2911, 0);
@@ -82,6 +85,7 @@ public class WalDeleteOutdatedNewTest {
walNode1.close();
config.setWalMode(prevMode);
config.setDataRegionConsensusProtocolClass(prevConsensus);
+ config.setWalFileListCacheEnabled(prevWalFileListCacheEnabled);
EnvironmentUtils.cleanDir(logDirectory1);
StorageEngine.getInstance().reset();
}
@@ -214,6 +218,10 @@ public class WalDeleteOutdatedNewTest {
Assert.assertEquals(2, memTableIdsOfWal.get(0L).size());
File[] files = WALFileUtils.listAllWALFiles(new File(logDirectory1));
Assert.assertEquals(2, files.length);
+ Assert.assertNull(walNode1.getCachedSortedWalFiles());
+ walNode1.getDeletionBoundToFreeAtLeast(1);
+ File[] cachedWalFilesBeforeDeletion = walNode1.getCachedSortedWalFiles();
+ Assert.assertEquals(2, cachedWalFilesBeforeDeletion.length);
walNode1.deleteOutdatedFiles();
Map<Long, Set<Long>> memTableIdsOfWalAfter =
walNode1.getWALBuffer().getMemTableIdsOfWal();
@@ -222,6 +230,10 @@ public class WalDeleteOutdatedNewTest {
Assert.assertEquals(0, memTableIdsOfWalAfter.size());
File[] filesAfter = WALFileUtils.listAllWALFiles(new File(logDirectory1));
Assert.assertEquals(1, filesAfter.length);
+ Assert.assertEquals(1, walNode1.getCachedSortedWalFiles().length);
+ Assert.assertEquals(
+ walNode1.getCurrentWALFileVersion(),
+
WALFileUtils.parseVersionId(walNode1.getCachedSortedWalFiles()[0].getName()));
}
/**
diff --git
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
index c4a06b68b14..e325ff42573 100644
---
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
+++
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
@@ -1705,6 +1705,15 @@ wal_sync_mode_fsync_delay_in_ms=3
# Datatype: int
wal_buffer_size_in_byte=33554432
+# Whether to cache the sorted WAL file list in each WAL node.
+# Enabling this avoids scanning and sorting the WAL directory whenever the
file list is accessed,
+# including when a consensus request reader updates its files to search.
+# Keeping this enabled is especially recommended when IoTConsensus
synchronizes data through WAL
+# and synchronization latency is observed.
+# effectiveMode: restart
+# Datatype: boolean
+wal_file_list_cache_enabled=true
+
# Size threshold of each wal file
# When a wal file's size exceeds this, the wal file will be closed and a new
wal file will be created.
# If it's a value smaller than 0, use the default value 30 * 1024 * 1024
(30MB).
