Iotdb 4251

server/src/assembly/resources/conf/iotdb-datanode.properties

@@ -460,6 +460,10 @@ timestamp_precision=ms
 # Datatype: int
 # primitive_array_size=32
 
+# the percentage of write memory for chunk metadata remains in a single file writer when flushing memtable
+# Datatype: double
+# chunk_metadata_size_proportion_in_write=0.1
+
 # Ratio of write memory for invoking flush disk, 0.4 by default
 # If you have extremely high write load (like batch=1000), it can be set lower than the default value like 0.2
 # Datatype: double
@@ -576,6 +580,10 @@ timestamp_precision=ms
 # BALANCE: alternate two compaction types
 # compaction_priority=BALANCE
 
+# size proportion for chunk metadata maintains in memory when compacting
+# Datatype: double
+# chunk_metadata_size_proportion_in_compaction=0.05
+
 # The target tsfile size in compaction
 # Datatype: long, Unit: byte
 # target_compaction_file_size=1073741824

server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -150,6 +150,8 @@ public class IoTDBConfig {
   /** The proportion of write memory for memtable */
   private double writeProportion = 0.8;
 
+  private double chunkMetadataSizeProportionInWrite = 0.1;
+
   /** The proportion of write memory for compaction */
   private double compactionProportion = 0.2;
 
@@ -434,6 +436,8 @@ public class IoTDBConfig {
    */
   private CompactionPriority compactionPriority = CompactionPriority.BALANCE;
 
+  private double chunkMetadataSizeProportionInCompaction = 0.05;
+
   /** The target tsfile size in compaction, 1 GB by default */
   private long targetCompactionFileSize = 1073741824L;
 
@@ -3163,6 +3167,23 @@ public void setThrottleThreshold(long throttleThreshold) {
     this.throttleThreshold = throttleThreshold;
   }
 
+  public double getChunkMetadataSizeProportionInWrite() {
+    return chunkMetadataSizeProportionInWrite;
+  }
+
+  public void setChunkMetadataSizeProportionInWrite(double chunkMetadataSizeProportionInWrite) {
+    this.chunkMetadataSizeProportionInWrite = chunkMetadataSizeProportionInWrite;
+  }
+
+  public double getChunkMetadataSizeProportionInCompaction() {
+    return chunkMetadataSizeProportionInCompaction;
+  }
+
+  public void setChunkMetadataSizeProportionInCompaction(
+      double chunkMetadataSizeProportionInCompaction) {
+    this.chunkMetadataSizeProportionInCompaction = chunkMetadataSizeProportionInCompaction;
+  }
+
   public long getCacheWindowTimeInMs() {
     return cacheWindowTimeInMs;
   }

server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -616,6 +616,11 @@ public void loadProperties(Properties properties) {
             properties.getProperty(
                 "concurrent_compaction_thread",
                 Integer.toString(conf.getConcurrentCompactionThread()))));
+    conf.setChunkMetadataSizeProportionInCompaction(
+        Double.parseDouble(
+            properties.getProperty(
+                "chunk_metadata_size_proportion_in_compaction",
+                Double.toString(conf.getChunkMetadataSizeProportionInCompaction()))));
     conf.setTargetCompactionFileSize(
         Long.parseLong(
             properties.getProperty(
@@ -1445,6 +1450,12 @@ public void loadHotModifiedProps(Properties properties) throws QueryProcessExcep
       // update tsfile-format config
       loadTsFileProps(properties);
 
+      conf.setChunkMetadataSizeProportionInWrite(
+          Double.parseDouble(
+              properties.getProperty(
+                  "chunk_metadata_size_proportion_in_write",
+                  Double.toString(conf.getChunkMetadataSizeProportionInWrite()))));
+
       // update max_deduplicated_path_num
       conf.setMaxQueryDeduplicatedPathNum(
           Integer.parseInt(

server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/ReadPointPerformerSubTask.java

@@ -33,7 +33,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.Callable;
 
 /**
@@ -45,7 +44,7 @@ public class ReadPointPerformerSubTask implements Callable<Void> {
   private static final Logger logger =
       LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
   private final String device;
-  private final Set<String> measurementList;
+  private final List<String> measurementList;
   private final FragmentInstanceContext fragmentInstanceContext;
   private final QueryDataSource queryDataSource;
   private final AbstractCompactionWriter compactionWriter;
@@ -54,7 +53,7 @@ public class ReadPointPerformerSubTask implements Callable<Void> {
 
   public ReadPointPerformerSubTask(
       String device,
-      Set<String> measurementList,
+      List<String> measurementList,
       FragmentInstanceContext fragmentInstanceContext,
       QueryDataSource queryDataSource,
       AbstractCompactionWriter compactionWriter,
@@ -87,7 +86,7 @@ public Void call() throws Exception {
       if (dataBlockReader.hasNextBatch()) {
         compactionWriter.startMeasurement(measurementSchemas, taskId);
         ReadPointCompactionPerformer.writeWithReader(
-            compactionWriter, dataBlockReader, taskId, false);
+            compactionWriter, dataBlockReader, device, taskId, false);
         compactionWriter.endMeasurement(taskId);
       }
     }

server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/AlignedSeriesCompactionExecutor.java

@@ -151,6 +151,7 @@ public void execute() throws IOException {
           chunkWriter.estimateMaxSeriesMemSize());
       chunkWriter.writeToFileWriter(writer);
     }
+    writer.checkMetadataSizeAndMayFlush();
   }
 
   private void compactOneAlignedChunk(AlignedChunkReader chunkReader) throws IOException {

server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/SingleSeriesCompactionExecutor.java

@@ -151,6 +151,7 @@ public void execute() throws IOException {
     } else if (pointCountInChunkWriter != 0L) {
       flushChunkWriter();
     }
+    fileWriter.checkMetadataSizeAndMayFlush();
     targetResource.updateStartTime(device, minStartTimestamp);
     targetResource.updateEndTime(device, maxEndTimestamp);
   }

server/src/main/java/org/apache/iotdb/db/engine/compaction/performer/ICompactionPerformer.java

@@ -25,6 +25,7 @@
 
 import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.ExecutionException;
 
 /**
  * CompactionPerformer is used to compact multiple files into one or multiple files. Different
@@ -35,7 +36,8 @@
 public interface ICompactionPerformer {
 
   void perform()
-      throws IOException, MetadataException, StorageEngineException, InterruptedException;
+      throws IOException, MetadataException, StorageEngineException, InterruptedException,
+          ExecutionException;
 
   void setTargetFiles(List<TsFileResource> targetFiles);
 

server/src/main/java/org/apache/iotdb/db/engine/compaction/performer/impl/ReadChunkCompactionPerformer.java

@@ -21,18 +21,19 @@
 import org.apache.iotdb.commons.conf.IoTDBConstant;
 import org.apache.iotdb.commons.exception.MetadataException;
 import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.engine.compaction.inner.utils.AlignedSeriesCompactionExecutor;
 import org.apache.iotdb.db.engine.compaction.inner.utils.MultiTsFileDeviceIterator;
 import org.apache.iotdb.db.engine.compaction.inner.utils.SingleSeriesCompactionExecutor;
 import org.apache.iotdb.db.engine.compaction.performer.ISeqCompactionPerformer;
 import org.apache.iotdb.db.engine.compaction.task.CompactionTaskSummary;
 import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
 import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.rescon.SystemInfo;
 import org.apache.iotdb.tsfile.file.metadata.AlignedChunkMetadata;
 import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
 import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
 import org.apache.iotdb.tsfile.utils.Pair;
-import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
 import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
 
 import org.slf4j.Logger;
@@ -63,8 +64,17 @@ public ReadChunkCompactionPerformer() {}
   @Override
   public void perform()
       throws IOException, MetadataException, InterruptedException, StorageEngineException {
+    // size for file writer is 5% of per compaction task memory budget
+    long sizeForFileWriter =
+        (long)
+            (SystemInfo.getInstance().getMemorySizeForCompaction()
+                / IoTDBDescriptor.getInstance().getConfig().getConcurrentCompactionThread()
+                * IoTDBDescriptor.getInstance()
+                    .getConfig()
+                    .getChunkMetadataSizeProportionInCompaction());
     try (MultiTsFileDeviceIterator deviceIterator = new MultiTsFileDeviceIterator(seqFiles);
-        TsFileIOWriter writer = new TsFileIOWriter(targetResource.getTsFile())) {
+        TsFileIOWriter writer =
+            new TsFileIOWriter(targetResource.getTsFile(), true, sizeForFileWriter)) {
       while (deviceIterator.hasNextDevice()) {
         Pair<String, Boolean> deviceInfo = deviceIterator.nextDevice();
         String device = deviceInfo.left;
@@ -138,7 +148,6 @@ private void compactNotAlignedSeries(
       checkThreadInterrupted();
       // TODO: we can provide a configuration item to enable concurrent between each series
       PartialPath p = new PartialPath(device, seriesIterator.nextSeries());
-      IMeasurementSchema measurementSchema;
       // TODO: seriesIterator needs to be refactor.
       // This statement must be called before next hasNextSeries() called, or it may be trapped in a
       // dead-loop.