(iotdb) branch master updated: DAL: Fix deletion buffer oom & enable DAL direct memory management

[tanxinyu]

This is an automated email from the ASF dual-hosted git repository.

tanxinyu 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 a7ebfa82873 DAL: Fix deletion buffer oom & enable DAL direct memory 
management
a7ebfa82873 is described below

commit a7ebfa8287393bdb054519ded1d25d45a3841f31
Author: Peng Junzhi <78788603+peng...@users.noreply.github.com>
AuthorDate: Wed Mar 12 15:47:51 2025 +0800

    DAL: Fix deletion buffer oom & enable DAL direct memory management
---
 .../pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java | 9 +++++----
 .../org/apache/iotdb/db/storageengine/dataregion/DataRegion.java | 4 ++++
 2 files changed, 9 insertions(+), 4 deletions(-)

diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
index f9f6965f40d..6ff090475bc 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
@@ -58,12 +58,13 @@ import java.util.concurrent.locks.ReentrantLock;
 public class PageCacheDeletionBuffer implements DeletionBuffer {
   private static final Logger LOGGER = 
LoggerFactory.getLogger(PageCacheDeletionBuffer.class);
   private static final IoTDBConfig config = 
IoTDBDescriptor.getInstance().getConfig();
-  // Buffer config keep consistent with WAL.
-  private static final int ONE_THIRD_WAL_BUFFER_SIZE = 
config.getWalBufferSize() / 3;
   private static final double FSYNC_BUFFER_RATIO = 0.95;
   private static final int QUEUE_CAPACITY = 
config.getDeletionAheadLogBufferQueueCapacity();
   private static final long MAX_WAIT_CLOSE_TIME_IN_MS = 10000;
 
+  // Buffer config keep consistent with WAL.
+  public static final int DAL_BUFFER_SIZE = config.getWalBufferSize() / 3;
+
   // DeletionResources received from storage engine, which is waiting to be 
persisted.
   private final BlockingQueue<DeletionResource> deletionResources =
       new PriorityBlockingQueue<>(
@@ -152,7 +153,7 @@ public class PageCacheDeletionBuffer implements 
DeletionBuffer {
 
   private void allocateBuffers() {
     try {
-      serializeBuffer = ByteBuffer.allocate(ONE_THIRD_WAL_BUFFER_SIZE);
+      serializeBuffer = ByteBuffer.allocateDirect(DAL_BUFFER_SIZE);
     } catch (OutOfMemoryError e) {
       LOGGER.error(
           "Fail to allocate deletionBuffer-group-{}'s buffer because out of 
memory.",
@@ -334,7 +335,7 @@ public class PageCacheDeletionBuffer implements 
DeletionBuffer {
 
       // For further deletion, we use non-blocking poll() method to persist 
existing deletion of
       // current batch in time.
-      while (totalSize.get() < ONE_THIRD_WAL_BUFFER_SIZE * FSYNC_BUFFER_RATIO) 
{
+      while (totalSize.get() < DAL_BUFFER_SIZE * FSYNC_BUFFER_RATIO) {
         DeletionResource deletionResource = null;
         try {
           // Timeout config keep consistent with WAL async mode.
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 263de360061..b1a2f056204 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
@@ -52,6 +52,7 @@ import 
org.apache.iotdb.db.exception.query.QueryProcessException;
 import org.apache.iotdb.db.exception.quota.ExceedQuotaException;
 import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResource;
 import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResource.Status;
+import 
org.apache.iotdb.db.pipe.consensus.deletion.persist.PageCacheDeletionBuffer;
 import 
org.apache.iotdb.db.pipe.extractor.dataregion.realtime.listener.PipeInsertionDataNodeListener;
 import org.apache.iotdb.db.queryengine.common.DeviceContext;
 import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
@@ -3834,6 +3835,9 @@ public class DataRegion implements IDataRegionForQuery {
         .equals(ConsensusFactory.RATIS_CONSENSUS)) {
       acquireDirectBufferMemCost = 
config.getDataRatisConsensusLogAppenderBufferSizeMax();
     }
+    if 
(config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS_V2))
 {
+      acquireDirectBufferMemCost += PageCacheDeletionBuffer.DAL_BUFFER_SIZE;
+    }
     return acquireDirectBufferMemCost;
   }