use compression rate to update wal disk usage

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java

@@ -520,8 +520,9 @@ public void run() {
           forceFlag, syncingBuffer.position(), syncingBuffer.capacity(), usedRatio * 100);
 
       // flush buffer to os
+      double compressionRate = 1.0;
       try {
-        currentWALFileWriter.write(syncingBuffer, info.metaData);
+        compressionRate = currentWALFileWriter.write(syncingBuffer, info.metaData);
       } catch (Throwable e) {
         logger.error(
             "Fail to sync wal node-{}'s buffer, change system mode to error.", identifier, e);
@@ -534,7 +535,7 @@ public void run() {
       memTableIdsOfWal
           .computeIfAbsent(currentWALFileVersion, memTableIds -> new HashSet<>())
           .addAll(info.metaData.getMemTablesId());
-      checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage);
+      checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage, compressionRate);
 
       boolean forceSuccess = false;
       // try to roll log writer

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManager.java

@@ -344,12 +344,13 @@ public long getFirstValidWALVersionId() {
   }
 
   /** Update wal disk cost of active memTables. */
-  public void updateCostOfActiveMemTables(Map<Long, Long> memTableId2WalDiskUsage) {
+  public void updateCostOfActiveMemTables(
+      Map<Long, Long> memTableId2WalDiskUsage, double compressionRate) {
     for (Map.Entry<Long, Long> memTableWalUsage : memTableId2WalDiskUsage.entrySet()) {
       memTableId2Info.computeIfPresent(
           memTableWalUsage.getKey(),
           (k, v) -> {
-            v.addWalDiskUsage(memTableWalUsage.getValue());
+            v.addWalDiskUsage((long) (memTableWalUsage.getValue() * compressionRate));
             return v;
           });
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/ILogWriter.java

@@ -33,8 +33,9 @@ public interface ILogWriter extends Closeable {
    *
    * @param buffer content that have been converted to bytes
    * @throws IOException if an I/O error occurs
+   * @return Compression rate of the buffer after compression
    */
-  void write(ByteBuffer buffer) throws IOException;
+  double write(ByteBuffer buffer) throws IOException;
 
   /**
    * Forces any updates to this file to be written to the storage device that contains it.

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java

@@ -72,13 +72,14 @@ protected LogWriter(File logFile) throws IOException {
   }
 
   @Override
-  public void write(ByteBuffer buffer) throws IOException {
+  public double write(ByteBuffer buffer) throws IOException {
     int bufferSize = buffer.position();
     buffer.flip();
     boolean compressed = false;
     int uncompressedSize = bufferSize;
     if (compressionAlg != CompressionType.UNCOMPRESSED
-        && bufferSize > MIN_COMPRESS_SIZE /* Do not compress buffer that is less than 512KB */) {
+        /* Do not compress buffer that is less than min size */
+        && bufferSize > MIN_COMPRESS_SIZE) {
       compressedByteBuffer.clear();
       compressor.compress(buffer, compressedByteBuffer);
       buffer = compressedByteBuffer;
@@ -103,12 +104,12 @@ public void write(ByteBuffer buffer) throws IOException {
     size += 5;
     try {
       headerBuffer.flip();
-      long before = logChannel.position();
       logChannel.write(headerBuffer);
       logChannel.write(buffer);
     } catch (ClosedChannelException e) {
       logger.warn("Cannot write to {}", logFile, e);
     }
+    return ((double) bufferSize / uncompressedSize);
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java

@@ -51,11 +51,11 @@ public WALWriter(File logFile) throws IOException {
    *
    * @throws IOException when failing to write
    */
-  public void write(ByteBuffer buffer, WALMetaData metaData) throws IOException {
+  public double write(ByteBuffer buffer, WALMetaData metaData) throws IOException {
     // update metadata
     updateMetaData(metaData);
     // flush buffer
-    write(buffer);
+    return write(buffer);
   }
 
   public void updateMetaData(WALMetaData metaData) {