Remove the logic of memtable flush controlled by memory series size

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java

@@ -149,8 +149,8 @@ public CommonConfig setPrimitiveArraySize(int primitiveArraySize) {
   }
 
   @Override
-  public CommonConfig setAvgSeriesPointNumberThreshold(int avgSeriesPointNumberThreshold) {
-    setProperty("avg_series_point_number_threshold", String.valueOf(avgSeriesPointNumberThreshold));
+  public CommonConfig setTargetChunkPointNum(int targetChunkPointNum) {
+    setProperty("target_chunk_point_num", String.valueOf(targetChunkPointNum));
     return this;
   }
 

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java

@@ -146,9 +146,9 @@ public CommonConfig setPrimitiveArraySize(int primitiveArraySize) {
   }
 
   @Override
-  public CommonConfig setAvgSeriesPointNumberThreshold(int avgSeriesPointNumberThreshold) {
-    cnConfig.setAvgSeriesPointNumberThreshold(avgSeriesPointNumberThreshold);
-    dnConfig.setAvgSeriesPointNumberThreshold(avgSeriesPointNumberThreshold);
+  public CommonConfig setTargetChunkPointNum(int targetChunkPointNum) {
+    cnConfig.setTargetChunkPointNum(targetChunkPointNum);
+    dnConfig.setTargetChunkPointNum(targetChunkPointNum);
     return this;
   }
 

integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java

@@ -105,7 +105,7 @@ public CommonConfig setPrimitiveArraySize(int primitiveArraySize) {
   }
 
   @Override
-  public CommonConfig setAvgSeriesPointNumberThreshold(int avgSeriesPointNumberThreshold) {
+  public CommonConfig setTargetChunkPointNum(int targetChunkPointNum) {
     return this;
   }
 

integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java

@@ -56,7 +56,7 @@ public interface CommonConfig {
 
   CommonConfig setPrimitiveArraySize(int primitiveArraySize);
 
-  CommonConfig setAvgSeriesPointNumberThreshold(int avgSeriesPointNumberThreshold);
+  CommonConfig setTargetChunkPointNum(int targetChunkPointNum);
 
   CommonConfig setMaxTsBlockLineNumber(int maxTsBlockLineNumber);
 

integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBDuplicateTimeIT.java

@@ -43,8 +43,6 @@ public class IoTDBDuplicateTimeIT {
 
   @Before
   public void setUp() throws Exception {
-    EnvFactory.getEnv().getConfig().getCommonConfig().setAvgSeriesPointNumberThreshold(2);
-    // Adjust memstable threshold size to make it flush automatically
     EnvFactory.getEnv().initClusterEnvironment();
   }
 
@@ -62,16 +60,19 @@ public void testDuplicateTime() throws SQLException {
       // version-1 tsfile
       statement.execute("insert into root.db.d1(time,s1) values (2,2)");
       statement.execute("insert into root.db.d1(time,s1) values (3,3)");
+      statement.execute("flush");
 
       // version-2 unseq work memtable
       statement.execute("insert into root.db.d1(time,s1) values (2,20)");
 
       // version-3 tsfile
       statement.execute("insert into root.db.d1(time,s1) values (5,5)");
       statement.execute("insert into root.db.d1(time,s1) values (6,6)");
+      statement.execute("flush root.db true");
 
       // version-2 unseq work memtable -> unseq tsfile
       statement.execute("insert into root.db.d1(time,s1) values (5,50)");
+      statement.execute("flush");
 
       try (ResultSet set = statement.executeQuery("SELECT s1 FROM root.db.d1 where time = 5")) {
         int cnt = 0;

integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestartIT.java

@@ -308,8 +308,6 @@ public void testRecoverWALDeleteSchema() throws Exception {
   @Test
   public void testRecoverWALDeleteSchemaCheckResourceTime() throws Exception {
     IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
-    int avgSeriesPointNumberThreshold = config.getAvgSeriesPointNumberThreshold();
-    config.setAvgSeriesPointNumberThreshold(2);
     long tsFileSize = config.getSeqTsFileSize();
     long unFsFileSize = config.getSeqTsFileSize();
     config.setSeqTsFileSize(10000000);
@@ -320,6 +318,7 @@ public void testRecoverWALDeleteSchemaCheckResourceTime() throws Exception {
       statement.execute("create timeseries root.turbine1.d1.s1 with datatype=INT64");
       statement.execute("insert into root.turbine1.d1(timestamp,s1) values(1,1)");
       statement.execute("insert into root.turbine1.d1(timestamp,s1) values(2,1)");
+      statement.execute("flush");
       statement.execute("create timeseries root.turbine1.d1.s2 with datatype=BOOLEAN");
       statement.execute("insert into root.turbine1.d1(timestamp,s2) values(3,true)");
       statement.execute("insert into root.turbine1.d1(timestamp,s2) values(4,true)");
@@ -342,7 +341,6 @@ public void testRecoverWALDeleteSchemaCheckResourceTime() throws Exception {
       assertEquals(2, cnt);
     }
 
-    config.setAvgSeriesPointNumberThreshold(avgSeriesPointNumberThreshold);
     config.setSeqTsFileSize(tsFileSize);
     config.setUnSeqTsFileSize(unFsFileSize);
   }

integration-test/src/test/java/org/apache/iotdb/db/it/groupby/IoTDBGroupByUnseqIT.java

@@ -110,7 +110,7 @@ public void test2() {
         .getConfig()
         .getCommonConfig()
         .setMaxNumberOfPointsInPage(4)
-        .setAvgSeriesPointNumberThreshold(2);
+        .setTargetChunkPointNum(2);
     EnvFactory.getEnv().initClusterEnvironment();
     String[] expectedHeader = new String[] {TIMESTAMP_STR, count("root.sg2.d1.s1")};
     String[] retArray = new String[] {"5,1,", "10,1,", "15,2,", "20,0,", "25,1,"};

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBRestartTableIT.java

@@ -336,8 +336,6 @@ public void testRecoverWALDeleteSchema() throws Exception {
   @Test
   public void testRecoverWALDeleteSchemaCheckResourceTime() throws Exception {
     IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
-    int avgSeriesPointNumberThreshold = config.getAvgSeriesPointNumberThreshold();
-    config.setAvgSeriesPointNumberThreshold(2);
     long tsFileSize = config.getSeqTsFileSize();
     long unFsFileSize = config.getSeqTsFileSize();
     config.setSeqTsFileSize(10000000);
@@ -351,6 +349,7 @@ public void testRecoverWALDeleteSchemaCheckResourceTime() throws Exception {
           "create table turbine1 (id1 string id, s1 int64 measurement, s2 boolean measurement)");
       statement.execute("insert into turbine1(time,id1,s1) values(1,\'d1\',1)");
       statement.execute("insert into turbine1(time,id1,s1) values(2,\'d1\',2)");
+      statement.execute("flush");
       statement.execute("insert into turbine1(time,id1,s2) values(3,\'d1\',true)");
       statement.execute("insert into turbine1(time,id1,s2) values(4,\'d1\',true)");
     }
@@ -373,7 +372,6 @@ public void testRecoverWALDeleteSchemaCheckResourceTime() throws Exception {
       assertEquals(2, cnt);
     }
 
-    config.setAvgSeriesPointNumberThreshold(avgSeriesPointNumberThreshold);
     config.setSeqTsFileSize(tsFileSize);
     config.setUnSeqTsFileSize(unFsFileSize);
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -427,9 +427,6 @@ public class IoTDBConfig {
   /** The sort algorithm used in TVList */
   private TVListSortAlgorithm tvListSortAlgorithm = TVListSortAlgorithm.TIM;
 
-  /** When average series point number reaches this, flush the memtable to disk */
-  private int avgSeriesPointNumberThreshold = 100000;
-
   /** Enable inner space compaction for sequence files */
   private volatile boolean enableSeqSpaceCompaction = true;
 
@@ -488,10 +485,10 @@ public class IoTDBConfig {
   /** The target tsfile size in compaction, 2 GB by default */
   private long targetCompactionFileSize = 2147483648L;
 
-  /** The target chunk size in compaction. */
-  private long targetChunkSize = 1048576L;
+  /** The target chunk size in compaction and flushing. */
+  private long targetChunkSize = 1600000L;
 
-  /** The target chunk point num in compaction. */
+  /** The target chunk point num in compaction and flushing. */
   private long targetChunkPointNum = 100000L;
 
   /**
@@ -2279,14 +2276,6 @@ public void setTvListSortAlgorithm(TVListSortAlgorithm tvListSortAlgorithm) {
     this.tvListSortAlgorithm = tvListSortAlgorithm;
   }
 
-  public int getAvgSeriesPointNumberThreshold() {
-    return avgSeriesPointNumberThreshold;
-  }
-
-  public void setAvgSeriesPointNumberThreshold(int avgSeriesPointNumberThreshold) {
-    this.avgSeriesPointNumberThreshold = avgSeriesPointNumberThreshold;
-  }
-
   public boolean isRpcThriftCompressionEnable() {
     return rpcThriftCompressionEnable;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -398,12 +398,6 @@ public void loadProperties(Properties properties) throws BadNodeUrlException, IO
             properties.getProperty(
                 "tvlist_sort_algorithm", conf.getTvListSortAlgorithm().toString())));
 
-    conf.setAvgSeriesPointNumberThreshold(
-        Integer.parseInt(
-            properties.getProperty(
-                "avg_series_point_number_threshold",
-                Integer.toString(conf.getAvgSeriesPointNumberThreshold()))));
-
     conf.setCheckPeriodWhenInsertBlocked(
         Integer.parseInt(
             properties.getProperty(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java

@@ -39,6 +39,7 @@
 import org.apache.iotdb.metrics.utils.MetricType;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 
 public class WritingMetrics implements IMetricSet {
@@ -363,7 +364,7 @@ private void unbindWALCostMetrics(AbstractMetricService metricService) {
                     MAKE_CHECKPOINT,
                     Tag.TYPE.toString(),
                     type));
-    Arrays.asList(SERIALIZE_WAL_ENTRY_TOTAL)
+    Collections.singletonList(SERIALIZE_WAL_ENTRY_TOTAL)
         .forEach(
             type ->
                 metricService.remove(
@@ -407,7 +408,6 @@ private void unbindWALCostMetrics(AbstractMetricService metricService) {
   public static final String WAL_FLUSH_MEMTABLE_COUNT = "wal_flush_memtable_count";
   public static final String MANUAL_FLUSH_MEMTABLE_COUNT = "manual_flush_memtable_count";
   public static final String MEM_CONTROL_FLUSH_MEMTABLE_COUNT = "mem_control_flush_memtable_count";
-  public static final String SERIES_FULL_FLUSH_MEMTABLE = "series_full_flush_memtable";
 
   private Gauge flushThreholdGauge = DoNothingMetricManager.DO_NOTHING_GAUGE;
   private Gauge rejectThreholdGauge = DoNothingMetricManager.DO_NOTHING_GAUGE;
@@ -416,7 +416,6 @@ private void unbindWALCostMetrics(AbstractMetricService metricService) {
 
   private Counter walFlushMemtableCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
   private Counter timedFlushMemtableCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
-  private Counter seriesFullFlushMemtableCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
   private Counter manualFlushMemtableCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
   private Counter memControlFlushMemtableCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
 
@@ -429,7 +428,6 @@ public void bindDataRegionMetrics() {
     createActiveTimePartitionCounterMetrics();
     walFlushMemtableCounter = createWalFlushMemTableCounterMetrics();
     timedFlushMemtableCounter = createTimedFlushMemTableCounterMetrics();
-    seriesFullFlushMemtableCounter = createSeriesFullFlushMemTableCounterMetrics();
     manualFlushMemtableCounter = createManualFlushMemTableCounterMetrics();
     memControlFlushMemtableCounter = createMemControlFlushMemTableCounterMetrics();
 
@@ -462,7 +460,6 @@ public void unbindDataRegionMetrics() {
           removeActiveMemtableCounterMetrics(dataRegionId);
         });
     removeActiveTimePartitionCounterMetrics();
-    removeSeriesFullFlushMemTableCounterMetrics();
     removeTimedFlushMemTableCounterMetrics();
     removeWalFlushMemTableCounterMetrics();
     removeManualFlushMemTableCounterMetrics();
@@ -575,15 +572,6 @@ public Counter createTimedFlushMemTableCounterMetrics() {
             TIMED_FLUSH_MEMTABLE_COUNT);
   }
 
-  public Counter createSeriesFullFlushMemTableCounterMetrics() {
-    return MetricService.getInstance()
-        .getOrCreateCounter(
-            Metric.FLUSH_MEMTABLE_COUNT.toString(),
-            MetricLevel.IMPORTANT,
-            Tag.TYPE.toString(),
-            SERIES_FULL_FLUSH_MEMTABLE);
-  }
-
   public Counter createManualFlushMemTableCounterMetrics() {
     return MetricService.getInstance()
         .getOrCreateCounter(
@@ -616,15 +604,6 @@ public void createActiveTimePartitionCounterMetrics() {
         .getOrCreateCounter(Metric.ACTIVE_TIME_PARTITION_COUNT.toString(), MetricLevel.IMPORTANT);
   }
 
-  public void removeSeriesFullFlushMemTableCounterMetrics() {
-    MetricService.getInstance()
-        .remove(
-            MetricType.COUNTER,
-            Metric.FLUSH_MEMTABLE_COUNT.toString(),
-            Tag.TYPE.toString(),
-            SERIES_FULL_FLUSH_MEMTABLE);
-  }
-
   public void removeTimedFlushMemTableCounterMetrics() {
     MetricService.getInstance()
         .remove(
@@ -929,10 +908,6 @@ public void recordWalFlushMemTableCount(int number) {
     walFlushMemtableCounter.inc(number);
   }
 
-  public void recordSeriesFullFlushMemTableCount(int number) {
-    seriesFullFlushMemtableCounter.inc(number);
-  }
-
   public void recordManualFlushMemTableCount(int number) {
     manualFlushMemtableCounter.inc(number);
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java

@@ -3063,86 +3063,6 @@ private long computeMaxVersion(Long oldVersion, Long newVersion) {
     return Math.max(oldVersion, newVersion);
   }
 
-  /**
-   * If the historical versions of a file is a sub-set of the given file's, (close and) remove it to
-   * reduce unnecessary merge. Only used when the file sender and the receiver share the same file
-   * close policy. Warning: DO NOT REMOVE
-   */
-  @SuppressWarnings("unused")
-  public void removeFullyOverlapFiles(TsFileResource resource) {
-    writeLock("removeFullyOverlapFiles");
-    try {
-      Iterator<TsFileResource> iterator = tsFileManager.getIterator(true);
-      removeFullyOverlapFiles(resource, iterator, true);
-
-      iterator = tsFileManager.getIterator(false);
-      removeFullyOverlapFiles(resource, iterator, false);
-    } finally {
-      writeUnlock();
-    }
-  }
-
-  private void removeFullyOverlapFiles(
-      TsFileResource newTsFile, Iterator<TsFileResource> iterator, boolean isSeq) {
-    while (iterator.hasNext()) {
-      TsFileResource existingTsFile = iterator.next();
-      if (newTsFile.isPlanRangeCovers(existingTsFile)
-          && !newTsFile.getTsFile().equals(existingTsFile.getTsFile())
-          && existingTsFile.tryWriteLock()) {
-        logger.info(
-            "{} is covered by {}: [{}, {}], [{}, {}], remove it",
-            existingTsFile,
-            newTsFile,
-            existingTsFile.minPlanIndex,
-            existingTsFile.maxPlanIndex,
-            newTsFile.minPlanIndex,
-            newTsFile.maxPlanIndex);
-        // if we fail to lock the file, it means it is being queried or merged and we will not
-        // wait until it is free, we will just leave it to the next merge
-        try {
-          removeFullyOverlapFile(existingTsFile, iterator, isSeq);
-        } catch (Exception e) {
-          logger.error(
-              "Something gets wrong while removing FullyOverlapFiles: {}",
-              existingTsFile.getTsFile().getAbsolutePath(),
-              e);
-        } finally {
-          existingTsFile.writeUnlock();
-        }
-      }
-    }
-  }
-
-  /**
-   * Remove the given {@link TsFileResource}. If the corresponding {@link TsFileProcessor} is in the
-   * working status, close it before remove the related resource files. maybe time-consuming for
-   * closing a tsfile.
-   */
-  private void removeFullyOverlapFile(
-      TsFileResource tsFileResource, Iterator<TsFileResource> iterator, boolean isSeq) {
-    logger.info(
-        "Removing a covered file {}, closed: {}", tsFileResource, tsFileResource.isClosed());
-    if (!tsFileResource.isClosed()) {
-      try {
-        // also remove the TsFileProcessor if the overlapped file is not closed
-        long timePartition = tsFileResource.getTimePartition();
-        Map<Long, TsFileProcessor> fileProcessorMap =
-            isSeq ? workSequenceTsFileProcessors : workUnsequenceTsFileProcessors;
-        TsFileProcessor tsFileProcessor = fileProcessorMap.get(timePartition);
-        if (tsFileProcessor != null && tsFileProcessor.getTsFileResource() == tsFileResource) {
-          // have to take some time to close the tsFileProcessor
-          tsFileProcessor.syncClose();
-          fileProcessorMap.remove(timePartition);
-        }
-      } catch (Exception e) {
-        logger.error("Cannot close {}", tsFileResource, e);
-      }
-    }
-    tsFileManager.remove(tsFileResource, isSeq);
-    iterator.remove();
-    tsFileResource.remove();
-  }
-
   private long getAndSetNewVersion(long timePartitionId, TsFileResource tsFileResource) {
     long version =
         partitionMaxFileVersions.compute(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java

@@ -44,6 +44,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -67,8 +68,8 @@ public class MemTableFlushTask {
   private final Future<?> ioTaskFuture;
   private RestorableTsFileIOWriter writer;
 
-  private final LinkedBlockingQueue<Object> encodingTaskQueue = new LinkedBlockingQueue<>();
-  private final LinkedBlockingQueue<Object> ioTaskQueue =
+  private final BlockingQueue<Object> encodingTaskQueue = new LinkedBlockingQueue<>();
+  private final BlockingQueue<Object> ioTaskQueue =
       (SystemInfo.getInstance().isEncodingFasterThanIo())
           ? new LinkedBlockingQueue<>(config.getIoTaskQueueSizeForFlushing())
           : new LinkedBlockingQueue<>();
@@ -247,16 +248,7 @@ public void run() {
             } else {
               long starTime = System.currentTimeMillis();
               IWritableMemChunk writableMemChunk = (IWritableMemChunk) task;
-              IChunkWriter seriesWriter = writableMemChunk.createIChunkWriter();
-              writableMemChunk.encode(seriesWriter);
-              seriesWriter.sealCurrentPage();
-              seriesWriter.clearPageWriter();
-              try {
-                ioTaskQueue.put(seriesWriter);
-              } catch (InterruptedException e) {
-                LOGGER.error("Put task into ioTaskQueue Interrupted");
-                Thread.currentThread().interrupt();
-              }
+              writableMemChunk.encode(ioTaskQueue);
               long subTaskTime = System.currentTimeMillis() - starTime;
               WRITING_METRICS.recordFlushSubTaskCost(WritingMetrics.ENCODING_TASK, subTaskTime);
               memSerializeTime += subTaskTime;