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

@@ -167,8 +167,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

@@ -167,9 +167,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

@@ -120,7 +120,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

@@ -62,7 +62,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

@@ -430,9 +430,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;
 
@@ -491,10 +488,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;
 
   /**
@@ -2309,14 +2306,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

@@ -439,15 +439,6 @@ public void loadProperties(Properties properties) throws BadNodeUrlException, IO
                 .map(String::trim)
                 .orElse(conf.getTvListSortAlgorithm().toString())));
 
-    conf.setAvgSeriesPointNumberThreshold(
-        Integer.parseInt(
-            Optional.ofNullable(
-                    properties.getProperty(
-                        "avg_series_point_number_threshold",
-                        Integer.toString(conf.getAvgSeriesPointNumberThreshold())))
-                .map(String::trim)
-                .orElse(Integer.toString(conf.getAvgSeriesPointNumberThreshold()))));
-
     conf.setCheckPeriodWhenInsertBlocked(
         Integer.parseInt(
             Optional.ofNullable(

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/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;