From ac110bce20b85ee1677252c8aeb2d3aa6b4ffc6c Mon Sep 17 00:00:00 2001
From: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date: Mon, 15 Jul 2024 19:21:14 +0800
Subject: [PATCH] Use max bytes to allocate WAL compressed buffer (#12928)

* use max bytes to allocate WAL compressed buffer

* calculate max bytes with lz4 compressor
---
 .../dataregion/wal/buffer/WALBuffer.java              | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

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 304b182d923f..d2b14597c50d 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
@@ -38,6 +38,8 @@
 import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALFlushListener;
 import org.apache.iotdb.db.utils.MmapUtil;
 
+import org.apache.tsfile.compress.ICompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -149,7 +151,8 @@ private void allocateBuffers() {
     try {
       workingBuffer = ByteBuffer.allocateDirect(ONE_THIRD_WAL_BUFFER_SIZE);
       idleBuffer = ByteBuffer.allocateDirect(ONE_THIRD_WAL_BUFFER_SIZE);
-      compressedByteBuffer = ByteBuffer.allocateDirect(ONE_THIRD_WAL_BUFFER_SIZE);
+      compressedByteBuffer =
+          ByteBuffer.allocateDirect(getCompressedByteBufferSize(ONE_THIRD_WAL_BUFFER_SIZE));
     } catch (OutOfMemoryError e) {
       logger.error("Fail to allocate wal node-{}'s buffer because out of memory.", identifier, e);
       close();
@@ -157,6 +160,10 @@ private void allocateBuffers() {
     }
   }
 
+  private int getCompressedByteBufferSize(int size) {
+    return ICompressor.getCompressor(CompressionType.LZ4).getMaxBytesForCompression(size);
+  }
+
   @Override
   protected File rollLogWriter(long searchIndex, WALFileStatus fileStatus) throws IOException {
     File file = super.rollLogWriter(searchIndex, fileStatus);
@@ -175,7 +182,7 @@ public void setBufferSize(int size) {
       MmapUtil.clean(compressedByteBuffer);
       workingBuffer = ByteBuffer.allocateDirect(capacity);
       idleBuffer = ByteBuffer.allocateDirect(capacity);
-      compressedByteBuffer = ByteBuffer.allocateDirect(capacity);
+      compressedByteBuffer = ByteBuffer.allocateDirect(getCompressedByteBufferSize(capacity));
       currentWALFileWriter.setCompressedByteBuffer(compressedByteBuffer);
     } catch (OutOfMemoryError e) {
       logger.error("Fail to allocate wal node-{}'s buffer because out of memory.", identifier, e);