[to rc/1.3.3] Add time chunk cache for batched column compaction

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/batch/BatchedFastAlignedSeriesCompactionExecutor.java

@@ -44,9 +44,11 @@
 import org.apache.tsfile.common.constant.TsFileConstant;
 import org.apache.tsfile.exception.write.PageException;
 import org.apache.tsfile.file.metadata.AlignedChunkMetadata;
+import org.apache.tsfile.file.metadata.ChunkMetadata;
 import org.apache.tsfile.file.metadata.IChunkMetadata;
 import org.apache.tsfile.file.metadata.IDeviceID;
 import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.common.Chunk;
 import org.apache.tsfile.read.common.TimeRange;
 import org.apache.tsfile.utils.Pair;
 import org.apache.tsfile.write.schema.IMeasurementSchema;
@@ -257,6 +259,17 @@ protected List<AlignedChunkMetadata> getAlignedChunkMetadataList(TsFileResource
       return getAlignedChunkMetadataListBySelectedValueColumn(resource, measurementSchemas);
     }
 
+    @Override
+    protected Chunk readChunk(TsFileSequenceReader reader, ChunkMetadata chunkMetadata)
+        throws IOException {
+      Chunk chunk = super.readChunk(reader, chunkMetadata);
+      if (AlignedSeriesBatchCompactionUtils.isTimeChunk(chunkMetadata)) {
+        batchCompactionPlan.addTimeChunkToCache(
+            reader.getFileName(), chunkMetadata.getOffsetOfChunkHeader(), chunk);
+      }
+      return chunk;
+    }
+
     @Override
     protected boolean flushChunkToCompactionWriter(ChunkMetadataElement chunkMetadataElement)
         throws IOException {
@@ -363,6 +376,15 @@ protected List<AlignedChunkMetadata> getAlignedChunkMetadataList(TsFileResource
       return getAlignedChunkMetadataListBySelectedValueColumn(resource, measurementSchemas);
     }
 
+    @Override
+    protected Chunk readChunk(TsFileSequenceReader reader, ChunkMetadata chunkMetadata)
+        throws IOException {
+      if (AlignedSeriesBatchCompactionUtils.isTimeChunk(chunkMetadata)) {
+        return batchCompactionPlan.getTimeChunkFromCache(reader, chunkMetadata);
+      }
+      return super.readChunk(reader, chunkMetadata);
+    }
+
     @Override
     protected boolean flushChunkToCompactionWriter(ChunkMetadataElement chunkMetadataElement)
         throws IOException {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/batch/BatchedReadChunkAlignedSeriesCompactionExecutor.java

@@ -32,6 +32,7 @@
 import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.batch.utils.FollowingBatchCompactionAlignedChunkWriter;
 import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.readchunk.ReadChunkAlignedSeriesCompactionExecutor;
 import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.readchunk.loader.ChunkLoader;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.readchunk.loader.InstantChunkLoader;
 import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.readchunk.loader.PageLoader;
 import org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFileWriter;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
@@ -42,6 +43,7 @@
 import org.apache.tsfile.file.metadata.IDeviceID;
 import org.apache.tsfile.file.metadata.statistics.Statistics;
 import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.common.Chunk;
 import org.apache.tsfile.read.common.TimeRange;
 import org.apache.tsfile.read.reader.IPointReader;
 import org.apache.tsfile.read.reader.page.AlignedPageReader;
@@ -235,6 +237,17 @@ protected boolean isAllValuePageEmpty(PageLoader timePage, List<PageLoader> valu
       return modifiedStatus == ModifiedStatus.ALL_DELETED;
     }
 
+    @Override
+    protected ChunkLoader getChunkLoader(TsFileSequenceReader reader, ChunkMetadata chunkMetadata)
+        throws IOException {
+      ChunkLoader chunkLoader = super.getChunkLoader(reader, chunkMetadata);
+      if (!chunkLoader.isEmpty() && AlignedSeriesBatchCompactionUtils.isTimeChunk(chunkMetadata)) {
+        batchCompactionPlan.addTimeChunkToCache(
+            reader.getFileName(), chunkMetadata.getOffsetOfChunkHeader(), chunkLoader.getChunk());
+      }
+      return chunkLoader;
+    }
+
     @Override
     protected void flushCurrentChunkWriter() throws IOException {
       chunkWriter.sealCurrentPage();
@@ -295,6 +308,16 @@ public FollowingBatchedReadChunkAlignedSeriesGroupCompactionExecutor(
               timeSchema, schemaList, batchCompactionPlan.getCompactChunkPlan(0));
     }
 
+    @Override
+    protected ChunkLoader getChunkLoader(TsFileSequenceReader reader, ChunkMetadata chunkMetadata)
+        throws IOException {
+      if (chunkMetadata != null && AlignedSeriesBatchCompactionUtils.isTimeChunk(chunkMetadata)) {
+        Chunk timeChunk = batchCompactionPlan.getTimeChunkFromCache(reader, chunkMetadata);
+        return new InstantChunkLoader(reader.getFileName(), chunkMetadata, timeChunk);
+      }
+      return super.getChunkLoader(reader, chunkMetadata);
+    }
+
     @Override
     protected void flushCurrentChunkWriter() throws IOException {
       if (chunkWriter.isEmpty() || currentCompactChunk >= batchCompactionPlan.compactedChunkNum()) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/batch/utils/AlignedSeriesBatchCompactionUtils.java

@@ -22,6 +22,7 @@
 import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.ModifiedStatus;
 
 import org.apache.tsfile.file.metadata.AlignedChunkMetadata;
+import org.apache.tsfile.file.metadata.ChunkMetadata;
 import org.apache.tsfile.file.metadata.IChunkMetadata;
 import org.apache.tsfile.read.TsFileSequenceReader;
 import org.apache.tsfile.read.common.TimeRange;
@@ -63,6 +64,10 @@ public static void markAlignedChunkHasDeletion(
     }
   }
 
+  public static boolean isTimeChunk(ChunkMetadata chunkMetadata) {
+    return chunkMetadata.getMeasurementUid().isEmpty();
+  }
+
   public static AlignedChunkMetadata filterAlignedChunkMetadataByIndex(
       AlignedChunkMetadata alignedChunkMetadata, List<Integer> selectedMeasurements) {
     IChunkMetadata[] valueChunkMetadataArr = new IChunkMetadata[selectedMeasurements.size()];

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/batch/utils/BatchCompactionPlan.java

@@ -21,18 +21,53 @@
 
 import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.ModifiedStatus;
 
+import org.apache.tsfile.file.metadata.ChunkMetadata;
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.common.Chunk;
 import org.apache.tsfile.read.common.TimeRange;
+import org.apache.tsfile.utils.Pair;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 public class BatchCompactionPlan {
+  public static final long MAX_CACHED_TIME_CHUNKS_SIZE = 2 * 1024 * 1024;
   private final List<CompactChunkPlan> compactChunkPlans = new ArrayList<>();
   private final Map<String, Map<TimeRange, ModifiedStatus>> alignedPageModifiedStatusCache =
       new HashMap<>();
+  // key is <filename, chunk offset in file>
+  private final Map<Pair<String, Long>, Chunk> cachedTimeChunks = new HashMap<>();
+  private long cachedTimeChunkSize = 0;
+
+  public Chunk getTimeChunkFromCache(TsFileSequenceReader reader, ChunkMetadata chunkMetadata)
+      throws IOException {
+    Pair<String, Long> key =
+        new Pair<>(reader.getFileName(), chunkMetadata.getOffsetOfChunkHeader());
+    Chunk chunk = cachedTimeChunks.get(key);
+    if (chunk == null) {
+      chunk = reader.readMemChunk(chunkMetadata);
+    }
+    chunk.getData().flip();
+    return chunk;
+  }
+
+  public void addTimeChunkToCache(String file, long offset, Chunk chunk) {
+    if (cachedTimeChunkSize >= MAX_CACHED_TIME_CHUNKS_SIZE) {
+      return;
+    }
+    cachedTimeChunks.put(
+        new Pair<>(file, offset),
+        new Chunk(
+            chunk.getHeader(),
+            chunk.getData(),
+            chunk.getDeleteIntervalList(),
+            chunk.getChunkStatistic()));
+    cachedTimeChunkSize += chunk.getHeader().getDataSize();
+  }
 
   public void recordCompactedChunk(CompactChunkPlan compactChunkPlan) {
     compactChunkPlans.add(compactChunkPlan);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/FastAlignedSeriesCompactionExecutor.java

@@ -338,26 +338,27 @@ void readChunk(ChunkMetadataElement chunkMetadataElement) throws IOException {
     updateSummary(chunkMetadataElement, ChunkStatus.READ_IN);
     AlignedChunkMetadata alignedChunkMetadata =
         (AlignedChunkMetadata) chunkMetadataElement.chunkMetadata;
+    TsFileSequenceReader reader = readerCacheMap.get(chunkMetadataElement.fileElement.resource);
     chunkMetadataElement.chunk =
-        readerCacheMap
-            .get(chunkMetadataElement.fileElement.resource)
-            .readMemChunk((ChunkMetadata) alignedChunkMetadata.getTimeChunkMetadata());
+        readChunk(reader, (ChunkMetadata) alignedChunkMetadata.getTimeChunkMetadata());
     List<Chunk> valueChunks = new ArrayList<>();
     for (IChunkMetadata valueChunkMetadata : alignedChunkMetadata.getValueChunkMetadataList()) {
       if (valueChunkMetadata == null || valueChunkMetadata.getStatistics().getCount() == 0) {
         // value chunk has been deleted completely or is empty value chunk
         valueChunks.add(null);
         continue;
       }
-      valueChunks.add(
-          readerCacheMap
-              .get(chunkMetadataElement.fileElement.resource)
-              .readMemChunk((ChunkMetadata) valueChunkMetadata));
+      valueChunks.add(readChunk(reader, (ChunkMetadata) valueChunkMetadata));
     }
     chunkMetadataElement.valueChunks = valueChunks;
     setForceDecoding(chunkMetadataElement);
   }
 
+  protected Chunk readChunk(TsFileSequenceReader reader, ChunkMetadata chunkMetadata)
+      throws IOException {
+    return reader.readMemChunk(chunkMetadata);
+  }
+
   @Override
   protected boolean flushChunkToCompactionWriter(ChunkMetadataElement chunkMetadataElement)
       throws IOException {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/readchunk/ReadChunkAlignedSeriesCompactionExecutor.java

@@ -237,7 +237,7 @@ private void compactWithAlignedChunk(
     }
   }
 
-  private ChunkLoader getChunkLoader(TsFileSequenceReader reader, ChunkMetadata chunkMetadata)
+  protected ChunkLoader getChunkLoader(TsFileSequenceReader reader, ChunkMetadata chunkMetadata)
       throws IOException {
     if (chunkMetadata == null || chunkMetadata.getStatistics().getCount() == 0) {
       return new InstantChunkLoader();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractCompactionEstimator.java

@@ -22,6 +22,7 @@
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.batch.utils.BatchCompactionPlan;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.DeviceTimeIndex;
@@ -62,11 +63,12 @@ public abstract class AbstractCompactionEstimator {
 
   protected IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
   protected TSFileConfig tsFileConfig = TSFileDescriptor.getInstance().getConfig();
-  protected long memoryBudgetForFileWriter =
+  protected long fixedMemoryBudget =
       (long)
-          ((double) SystemInfo.getInstance().getMemorySizeForCompaction()
-              / IoTDBDescriptor.getInstance().getConfig().getCompactionThreadCount()
-              * IoTDBDescriptor.getInstance().getConfig().getChunkMetadataSizeProportion());
+              ((double) SystemInfo.getInstance().getMemorySizeForCompaction()
+                  / IoTDBDescriptor.getInstance().getConfig().getCompactionThreadCount()
+                  * IoTDBDescriptor.getInstance().getConfig().getChunkMetadataSizeProportion())
+          + BatchCompactionPlan.MAX_CACHED_TIME_CHUNKS_SIZE;
 
   protected abstract long calculatingMetadataMemoryCost(CompactionTaskInfo taskInfo);
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/FastCompactionInnerCompactionEstimator.java

@@ -39,7 +39,7 @@ public long calculatingMetadataMemoryCost(CompactionTaskInfo taskInfo) {
                 * taskInfo.getMaxChunkMetadataSize());
 
     // add ChunkMetadata size of targetFileWriter
-    cost += memoryBudgetForFileWriter;
+    cost += fixedMemoryBudget;
 
     return cost;
   }
@@ -99,7 +99,7 @@ public long roughEstimateInnerCompactionMemory(List<TsFileResource> resources)
     // source files (chunk + uncompressed page) * overlap file num
     // target file (chunk + unsealed page writer)
     return (maxOverlapFileNum + 1) * maxConcurrentSeriesNum * (maxChunkSize + maxPageSize)
-        + memoryBudgetForFileWriter
+        + fixedMemoryBudget
         + metadataCost;
   }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/FastCrossSpaceCompactionEstimator.java

@@ -40,7 +40,7 @@ protected long calculatingMetadataMemoryCost(CompactionTaskInfo taskInfo) {
                 * taskInfo.getMaxChunkMetadataSize());
 
     // add ChunkMetadata size of targetFileWriter
-    cost += memoryBudgetForFileWriter;
+    cost += fixedMemoryBudget;
 
     return cost;
   }
@@ -103,7 +103,7 @@ public long roughEstimateCrossCompactionMemory(
     // source files (chunk + uncompressed page) * overlap file num
     // target files (chunk + unsealed page writer)
     return (maxOverlapFileNum + 1) * maxConcurrentSeriesNum * (maxChunkSize + maxPageSize)
-        + memoryBudgetForFileWriter
+        + fixedMemoryBudget
         + metadataCost;
   }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/ReadChunkInnerCompactionEstimator.java

@@ -39,7 +39,7 @@ public long calculatingMetadataMemoryCost(CompactionTaskInfo taskInfo) {
                 * taskInfo.getMaxChunkMetadataSize());
 
     // add ChunkMetadata size of targetFileWriter
-    cost += memoryBudgetForFileWriter;
+    cost += fixedMemoryBudget;
 
     return cost;
   }
@@ -89,7 +89,7 @@ public long roughEstimateInnerCompactionMemory(List<TsFileResource> resources)
     // source files (chunk + uncompressed page)
     // target file (chunk + unsealed page writer)
     return 2 * maxConcurrentSeriesNum * (maxChunkSize + maxPageSize)
-        + memoryBudgetForFileWriter
+        + fixedMemoryBudget
         + metadataCost;
   }
 }