This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch preallocate_array_list
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/preallocate_array_list by this
push:
new e702afe02d1 add memtable cache
e702afe02d1 is described below
commit e702afe02d17af7ca64d153390e6aecdaf503957
Author: Tian Jiang <[email protected]>
AuthorDate: Wed Apr 23 09:32:23 2025 +0800
add memtable cache
---
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 19 ++++++++++++++++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 25 ++++++++++++----------
.../protocol/thrift/impl/ClientRPCServiceImpl.java | 17 ++++++++-------
.../db/storageengine/dataregion/DataRegion.java | 21 ++++++++++++++++++
.../dataregion/memtable/AbstractMemTable.java | 8 +++++++
.../memtable/AbstractWritableMemChunk.java | 16 ++++++++------
.../memtable/AlignedWritableMemChunk.java | 15 ++++++++++++-
.../memtable/AlignedWritableMemChunkGroup.java | 5 +++++
.../dataregion/memtable/IMemTable.java | 3 +++
.../dataregion/memtable/IWritableMemChunk.java | 3 +++
.../memtable/IWritableMemChunkGroup.java | 3 +++
.../dataregion/memtable/TsFileProcessor.java | 15 ++++++++-----
.../dataregion/memtable/WritableMemChunk.java | 15 ++++++++++++-
.../dataregion/memtable/WritableMemChunkGroup.java | 7 ++++++
14 files changed, 140 insertions(+), 32 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 b5d5d186953..d3594d03a37 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
@@ -1154,6 +1154,9 @@ public class IoTDBConfig {
private boolean enableGCThrottle = false;
private int gcThrottleTimeMs = 1000;
+ private boolean enableMemTableCache = true;
+ private int maxCachedMemTablePerRegion = 1;
+
IoTDBConfig() {}
public int getMaxLogEntriesNumPerBatch() {
@@ -4098,4 +4101,20 @@ public class IoTDBConfig {
public void setEnableGCThrottle(boolean enableGCThrottle) {
this.enableGCThrottle = enableGCThrottle;
}
+
+ public boolean isEnableMemTableCache() {
+ return enableMemTableCache;
+ }
+
+ public void setEnableMemTableCache(boolean enableMemTableCache) {
+ this.enableMemTableCache = enableMemTableCache;
+ }
+
+ public int getMaxCachedMemTablePerRegion() {
+ return maxCachedMemTablePerRegion;
+ }
+
+ public void setMaxCachedMemTablePerRegion(int maxCachedMemTablePerRegion) {
+ this.maxCachedMemTablePerRegion = maxCachedMemTablePerRegion;
+ }
}
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 e6e5377009d..702807607d9 100755
---
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
@@ -1074,20 +1074,23 @@ public class IoTDBDescriptor {
"detail_container_min_degrade_memory_in_bytes",
String.valueOf(conf.getDetailContainerMinDegradeMemoryInBytes()))));
- conf.setEnableGCThrottle(Boolean.parseBoolean(
- properties.getProperty(
- "enable_gc_throttle",
- String.valueOf(conf.isEnableGCThrottle())
- )
- ));
+ conf.setEnableGCThrottle(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "enable_gc_throttle",
String.valueOf(conf.isEnableGCThrottle()))));
conf.setGcThrottleTimeMs(
Integer.parseInt(
properties.getProperty(
- "gc_throttle_time_ms",
- String.valueOf(conf.getGcThrottleTimeMs())
- )
- )
- );
+ "gc_throttle_time_ms",
String.valueOf(conf.getGcThrottleTimeMs()))));
+
+ conf.setEnableMemTableCache(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "enable_memtable_cache",
String.valueOf(conf.isEnableGCThrottle()))));
+ conf.setMaxCachedMemTablePerRegion(
+ Integer.parseInt(
+ properties.getProperty(
+ "max_cached_memtable_per_region",
String.valueOf(conf.getGcThrottleTimeMs()))));
loadIoTConsensusProps(properties);
loadIoTConsensusV2Props(properties);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
index e4b9f4514dc..6f72f46ae08 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.protocol.thrift.impl;
-import java.util.Random;
import org.apache.iotdb.common.rpc.thrift.TAggregationType;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
@@ -215,6 +214,7 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
+import java.util.Random;
import java.util.concurrent.TimeUnit;
import static org.apache.iotdb.commons.partition.DataPartition.NOT_ASSIGNED;
@@ -3065,6 +3065,7 @@ public class ClientRPCServiceImpl implements
IClientRPCServiceWithHandler {
}
private static final Random throttleRandom = new Random();
+
public static void gcThrottle() {
if (!IoTDBDescriptor.getInstance().getConfig().isEnableGCThrottle()) {
return;
@@ -3072,13 +3073,13 @@ public class ClientRPCServiceImpl implements
IClientRPCServiceWithHandler {
int minimumGcTimePercentToThrottle = 20;
long gcTimePercentage =
JvmGcMonitorMetrics.getInstance().getGcData().getGcTimePercentage();
// if gc time is above 20%, pause the request with the probability equals
to gc time percent
- if (gcTimePercentage > minimumGcTimePercentToThrottle &&
throttleRandom.nextInt(100) < gcTimePercentage) {
- try {
-
Thread.sleep(IoTDBDescriptor.getInstance().getConfig().getGcThrottleTimeMs());
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
+ if (gcTimePercentage > minimumGcTimePercentToThrottle
+ && throttleRandom.nextInt(100) < gcTimePercentage) {
+ try {
+
Thread.sleep(IoTDBDescriptor.getInstance().getConfig().getGcThrottleTimeMs());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
}
-
+ }
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index 8880f4f1723..90a03098501 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -176,6 +176,8 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -333,6 +335,10 @@ public class DataRegion implements IDataRegionForQuery {
private ILoadDiskSelector ordinaryLoadDiskSelector;
private ILoadDiskSelector pipeAndIoTV2LoadDiskSelector;
+ private BlockingQueue<IMemTable> memTableCache =
+ new ArrayBlockingQueue<>(
+
IoTDBDescriptor.getInstance().getConfig().getMaxCachedMemTablePerRegion());
+
/**
* Construct a database processor.
*
@@ -3942,4 +3948,19 @@ public class DataRegion implements IDataRegionForQuery {
return DataNodeTTLCache.getInstance().getTTLForTable(databaseName,
insertNode.getTableName());
}
}
+
+ public IMemTable getCachedMemTable() {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableMemTableCache()) {
+ return memTableCache.poll();
+ } else {
+ return null;
+ }
+ }
+
+ @SuppressWarnings({"ResultOfMethodCallIgnored", "java:S899"})
+ public void cacheMemTable(IMemTable memTable) {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableMemTableCache()) {
+ memTableCache.offer(memTable);
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
index d550781858f..a664dba57cd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
@@ -411,6 +411,14 @@ public abstract class AbstractMemTable implements
IMemTable {
@Override
public void clear() {
memTableMap.clear();
+ reset();
+ }
+
+ @Override
+ public void reset() {
+ for (IWritableMemChunkGroup value : memTableMap.values()) {
+ value.reset();
+ }
memSize = 0;
seriesNumber = 0;
totalPointsNum = 0;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
index ac71e3da6bb..cdcc24978d7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
@@ -50,14 +50,13 @@ public abstract class AbstractWritableMemChunk implements
IWritableMemChunk {
* method must ensure success because it's part of flushing.
*
* @param tvList
+ * @return whether the TvList is cleared (exclusively held by the MemTable)
*/
- protected void maybeReleaseTvList(TVList tvList) {
+ protected boolean maybeReleaseTvList(TVList tvList) {
long startTimeInMs = System.currentTimeMillis();
- boolean succeed = false;
- while (!succeed) {
+ while (true) {
try {
- tryReleaseTvList(tvList);
- succeed = true;
+ return tryReleaseTvList(tvList);
} catch (MemoryNotEnoughException ex) {
long waitQueryInMs = System.currentTimeMillis() - startTimeInMs;
if (waitQueryInMs > MAX_WAIT_QUERY_MS) {
@@ -88,11 +87,15 @@ public abstract class AbstractWritableMemChunk implements
IWritableMemChunk {
}
}
- private void tryReleaseTvList(TVList tvList) {
+ /**
+ * @return whether the list is cleared.
+ */
+ private boolean tryReleaseTvList(TVList tvList) {
tvList.lockQueryList();
try {
if (tvList.getQueryContextSet().isEmpty()) {
tvList.clear();
+ return true;
} else {
QueryContext firstQuery =
tvList.getQueryContextSet().iterator().next();
// transfer memory from write process to read process. Here it
reserves read memory and
@@ -104,6 +107,7 @@ public abstract class AbstractWritableMemChunk implements
IWritableMemChunk {
}
// update current TVList owner to first query in the list
tvList.setOwnerQuery(firstQuery);
+ return false;
}
} finally {
tvList.unlockQueryList();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
index d20866d0c79..5c63b317215 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
@@ -697,12 +697,25 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
@Override
public void release() {
- maybeReleaseTvList(list);
+ boolean listCleared = maybeReleaseTvList(list);
+ if (!listCleared) {
+ // if the list is not cleared, we cannot reuse it
+ list = null;
+ }
for (AlignedTVList alignedTvList : sortedList) {
maybeReleaseTvList(alignedTvList);
}
}
+ @Override
+ public void reset() {
+ sortedList.clear();
+ if (list == null) {
+ list = AlignedTVList.newAlignedList(dataTypes);
+ }
+ sortedRowCount = 0;
+ }
+
@Override
public long getFirstPoint() {
if (rowCount() == 0) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
index 36acdedf9bc..f15fffa62e2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
@@ -72,6 +72,11 @@ public class AlignedWritableMemChunkGroup implements
IWritableMemChunkGroup {
memChunk.release();
}
+ @Override
+ public void reset() {
+ memChunk.reset();
+ }
+
@Override
public long count() {
return memChunk.count();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
index 44098b69a56..91fb831d955 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
@@ -140,6 +140,9 @@ public interface IMemTable extends WALEntryValue {
/** putBack all the memory resources. */
void clear();
+ /** Prepare the memtable for new insertions, keeping the inner structures
for reuse. */
+ void reset();
+
boolean isEmpty();
/**
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java
index 13f09a05d8a..2f2e819c9e8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java
@@ -116,6 +116,9 @@ public interface IWritableMemChunk extends WALEntryValue {
void release();
+ /** Prepare the chunk for the next round of insertion. */
+ void reset();
+
long getFirstPoint();
long getLastPoint();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
index 62bf7700218..f46c7ebd8b9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
@@ -44,6 +44,9 @@ public interface IWritableMemChunkGroup extends WALEntryValue
{
void release();
+ /** Prepare the chunk group for the next round of insertion. */
+ void reset();
+
long count();
boolean contains(String measurement);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index 861b72c923a..ce9cdf5bd23 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -453,11 +453,14 @@ public class TsFileProcessor {
}
private void createNewWorkingMemTable() {
- workMemTable =
- MemTableManager.getInstance()
- .getAvailableMemTable(
- dataRegionInfo.getDataRegion().getDatabaseName(),
- dataRegionInfo.getDataRegion().getDataRegionId());
+ workMemTable = dataRegionInfo.getDataRegion().getCachedMemTable();
+ if (workMemTable == null) {
+ workMemTable =
+ MemTableManager.getInstance()
+ .getAvailableMemTable(
+ dataRegionInfo.getDataRegion().getDatabaseName(),
+ dataRegionInfo.getDataRegion().getDataRegionId());
+ }
walNode.onMemTableCreated(workMemTable, tsFileResource.getTsFilePath());
}
@@ -1632,6 +1635,8 @@ public class TsFileProcessor {
for (FlushListener flushListener : flushListeners) {
flushListener.onMemTableFlushed(memTableToFlush);
}
+ memTableToFlush.reset();
+ dataRegionInfo.getDataRegion().cacheMemTable(memTableToFlush);
// Retry to avoid unnecessary read-only mode
int retryCnt = 0;
while (shouldClose && flushingMemTables.isEmpty() && writer != null) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
index 1f4078dfc7d..89ee4445dbd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
@@ -504,12 +504,25 @@ public class WritableMemChunk extends
AbstractWritableMemChunk {
@Override
public void release() {
- maybeReleaseTvList(list);
+ boolean listCleared = maybeReleaseTvList(list);
+ if (!listCleared) {
+ // the list cannot be reused if not cleared
+ list = null;
+ }
for (TVList tvList : sortedList) {
maybeReleaseTvList(tvList);
}
}
+ @Override
+ public void reset() {
+ sortedRowCount = 0;
+ sortedList.clear();
+ if (list == null) {
+ list = TVList.newList(schema.getType());
+ }
+ }
+
@Override
public int serializedSize() {
int serializedSize = schema.serializedSize() + list.serializedSize();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
index 854ca9c8d2c..762ca3ed13c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
@@ -80,6 +80,13 @@ public class WritableMemChunkGroup implements
IWritableMemChunkGroup {
}
}
+ @Override
+ public void reset() {
+ for (IWritableMemChunk memChunk : memChunkMap.values()) {
+ memChunk.reset();
+ }
+ }
+
@Override
public long count() {
long count = 0;
