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/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/storageengine/dataregion/memtable/AbstractMemTable.java

@@ -29,7 +29,6 @@
 import org.apache.iotdb.commons.service.metric.MetricService;
 import org.apache.iotdb.commons.service.metric.enums.Metric;
 import org.apache.iotdb.commons.service.metric.enums.Tag;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.exception.WriteProcessException;
 import org.apache.iotdb.db.exception.query.QueryProcessException;
 import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
@@ -96,8 +95,6 @@ public abstract class AbstractMemTable implements IMemTable {
 
   private boolean shouldFlush = false;
   private volatile FlushStatus flushStatus = FlushStatus.WORKING;
-  private final int avgSeriesPointNumThreshold =
-      IoTDBDescriptor.getInstance().getConfig().getAvgSeriesPointNumberThreshold();
 
   /** Memory size of data points, including TEXT values. */
   private long memSize = 0;
@@ -176,7 +173,6 @@ private IWritableMemChunkGroup createMemChunkGroupIfNotExistAndGet(
     for (IMeasurementSchema schema : schemaList) {
       if (schema != null && !memChunkGroup.contains(schema.getMeasurementId())) {
         seriesNumber++;
-        totalPointsNumThreshold += avgSeriesPointNumThreshold;
       }
     }
     return memChunkGroup;
@@ -189,14 +185,12 @@ private IWritableMemChunkGroup createAlignedMemChunkGroupIfNotExistAndGet(
             deviceId,
             k -> {
               seriesNumber += schemaList.size();
-              totalPointsNumThreshold += ((long) avgSeriesPointNumThreshold) * schemaList.size();
               return new AlignedWritableMemChunkGroup(
                   schemaList.stream().filter(Objects::nonNull).collect(Collectors.toList()));
             });
     for (IMeasurementSchema schema : schemaList) {
       if (schema != null && !memChunkGroup.contains(schema.getMeasurementId())) {
         seriesNumber++;
-        totalPointsNumThreshold += avgSeriesPointNumThreshold;
       }
     }
     return memChunkGroup;

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

@@ -58,7 +58,7 @@ public class AlignedWritableMemChunk implements IWritableMemChunk {
 
   private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
   private final long TARGET_CHUNK_SIZE = CONFIG.getTargetChunkSize();
-  private final long MAX_NUMBER_OF_POINTS_IN_CHUNK = CONFIG.getAvgSeriesPointNumberThreshold();
+  private long MAX_NUMBER_OF_POINTS_IN_CHUNK = CONFIG.getTargetChunkPointNum();
   private final int MAX_NUMBER_OF_POINTS_IN_PAGE =
       TSFileDescriptor.getInstance().getConfig().getMaxNumberOfPointsInPage();
 
@@ -330,9 +330,9 @@ public IChunkWriter createIChunkWriter() {
   @Override
   public void encode(BlockingQueue<Object> ioTaskQueue) {
     BitMap rowBitMap = list.getRowBitMap();
-    int avgBinaryPointSize = list.getAvgBinaryPointSize();
-    int maxNumberOfPointsInBinaryChunk =
-        avgBinaryPointSize > 0 ? (int) (TARGET_CHUNK_SIZE / avgBinaryPointSize) : Integer.MAX_VALUE;
+    int avgPointSizeOfLargestColumn = list.getAvgPointSizeOfLargestColumn();
+    MAX_NUMBER_OF_POINTS_IN_CHUNK =
+        Math.min(MAX_NUMBER_OF_POINTS_IN_CHUNK, (TARGET_CHUNK_SIZE / avgPointSizeOfLargestColumn));
 
     boolean[] timeDuplicateInfo = null;
 
@@ -355,8 +355,7 @@ public void encode(BlockingQueue<Object> ioTaskQueue) {
         pageRange.add(sortedRowIndex);
         pointNumInPage = 0;
       }
-      if (pointNumInChunk == MAX_NUMBER_OF_POINTS_IN_CHUNK
-          || pointNumInChunk >= maxNumberOfPointsInBinaryChunk) {
+      if (pointNumInChunk >= MAX_NUMBER_OF_POINTS_IN_CHUNK) {
         if (pointNumInPage != 0) {
           pageRange.add(sortedRowIndex);
           pointNumInPage = 0;

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

@@ -52,7 +52,7 @@ public class WritableMemChunk implements IWritableMemChunk {
 
   private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
   private final long TARGET_CHUNK_SIZE = CONFIG.getTargetChunkSize();
-  private final long MAX_NUMBER_OF_POINTS_IN_CHUNK = CONFIG.getAvgSeriesPointNumberThreshold();
+  private final long MAX_NUMBER_OF_POINTS_IN_CHUNK = CONFIG.getTargetChunkPointNum();
 
   public WritableMemChunk(IMeasurementSchema schema) {
     this.schema = schema;
@@ -337,7 +337,7 @@ public void encode(BlockingQueue<Object> ioTaskQueue) {
 
     TSDataType tsDataType = schema.getType();
     ChunkWriterImpl chunkWriterImpl = createIChunkWriter();
-    long binarySizeInCurrentChunk = 0;
+    long dataSizeInCurrentChunk = 0;
     int pointNumInCurrentChunk = 0;
     for (int sortedRowIndex = 0; sortedRowIndex < list.rowCount(); sortedRowIndex++) {
       long time = list.getTime(sortedRowIndex);
@@ -355,35 +355,40 @@ public void encode(BlockingQueue<Object> ioTaskQueue) {
       switch (tsDataType) {
         case BOOLEAN:
           chunkWriterImpl.write(time, list.getBoolean(sortedRowIndex));
+          dataSizeInCurrentChunk += 8L + 1L;
           break;
         case INT32:
         case DATE:
           chunkWriterImpl.write(time, list.getInt(sortedRowIndex));
+          dataSizeInCurrentChunk += 8L + 4L;
           break;
         case INT64:
         case TIMESTAMP:
           chunkWriterImpl.write(time, list.getLong(sortedRowIndex));
+          dataSizeInCurrentChunk += 8L + 8L;
           break;
         case FLOAT:
           chunkWriterImpl.write(time, list.getFloat(sortedRowIndex));
+          dataSizeInCurrentChunk += 8L + 4L;
           break;
         case DOUBLE:
           chunkWriterImpl.write(time, list.getDouble(sortedRowIndex));
+          dataSizeInCurrentChunk += 8L + 8L;
           break;
         case TEXT:
         case BLOB:
         case STRING:
           Binary value = list.getBinary(sortedRowIndex);
           chunkWriterImpl.write(time, value);
-          binarySizeInCurrentChunk += getBinarySize(value);
+          dataSizeInCurrentChunk += 8L + getBinarySize(value);
           break;
         default:
           LOGGER.error("WritableMemChunk does not support data type: {}", tsDataType);
           break;
       }
       pointNumInCurrentChunk++;
       if (pointNumInCurrentChunk > MAX_NUMBER_OF_POINTS_IN_CHUNK
-          || binarySizeInCurrentChunk > TARGET_CHUNK_SIZE) {
+          || dataSizeInCurrentChunk > TARGET_CHUNK_SIZE) {
         chunkWriterImpl.sealCurrentPage();
         chunkWriterImpl.clearPageWriter();
         try {
@@ -392,7 +397,7 @@ public void encode(BlockingQueue<Object> ioTaskQueue) {
           Thread.currentThread().interrupt();
         }
         chunkWriterImpl = createIChunkWriter();
-        binarySizeInCurrentChunk = 0;
+        dataSizeInCurrentChunk = 0;
         pointNumInCurrentChunk = 0;
       }
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java

@@ -1365,20 +1365,22 @@ public BitMap getRowBitMap() {
     return new BitMap(rowCount, rowBitsArr);
   }
 
-  public int getAvgBinaryPointSize() {
-    long maxSize = 0;
-    int index = 0;
+  public int getAvgPointSizeOfLargestColumn() {
+    int largestPrimitivePointSize = 8; // TimeColumn or int64,double ValueColumn
+    long largestBinaryChunkSize = 0;
+    int largestBinaryColumnIndex = 0;
     for (int i = 0; i < memoryBinaryChunkSize.length; i++) {
-      if (memoryBinaryChunkSize[i] > maxSize) {
-        maxSize = memoryBinaryChunkSize[i];
-        index = i;
+      if (memoryBinaryChunkSize[i] > largestBinaryChunkSize) {
+        largestBinaryChunkSize = memoryBinaryChunkSize[i];
+        largestBinaryColumnIndex = i;
       }
     }
-    if (maxSize == 0) {
-      return 0;
+    if (largestBinaryChunkSize == 0) {
+      return largestPrimitivePointSize;
     }
-    int pointNum = getColumnValueCnt(index);
-    return (int) (maxSize / pointNum);
+    int avgPointSizeOfLargestBinaryColumn =
+        (int) largestBinaryChunkSize / getColumnValueCnt(largestBinaryColumnIndex);
+    return Math.max(avgPointSizeOfLargestBinaryColumn, largestPrimitivePointSize);
   }
 
   public int getColumnValueCnt(int columnIndex) {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java

@@ -1050,8 +1050,6 @@ public void testInsertUnSequenceRows()
           QueryProcessException,
           DataRegionException,
           TsFileProcessorException {
-    int defaultAvgSeriesPointNumberThreshold = config.getAvgSeriesPointNumberThreshold();
-    config.setAvgSeriesPointNumberThreshold(2);
     DataRegion dataRegion1 = new DummyDataRegion(systemDir, "root.Rows");
     long[] time = new long[] {3, 4, 1, 2};
     List<Integer> indexList = new ArrayList<>();
@@ -1081,7 +1079,6 @@ tmpDeviceId, new MeasurementSchema(measurementId, TSDataType.INT32))),
       Assert.assertTrue(resource.isClosed());
     }
     dataRegion1.syncDeleteDataFiles();
-    config.setAvgSeriesPointNumberThreshold(defaultAvgSeriesPointNumberThreshold);
   }
 
   @Test

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java

@@ -96,7 +96,7 @@ public class TsFileProcessorTest {
   private final Map<String, String> props = Collections.emptyMap();
   private QueryContext context;
   private final String systemDir = TestConstant.OUTPUT_DATA_DIR.concat("info");
-  private int defaultAvgSeriesPointNumberThreshold;
+  private long defaultTargetChunkPointNum;
   private long defaultTargetChunkSize;
   private static final Logger logger = LoggerFactory.getLogger(TsFileProcessorTest.class);
 
@@ -108,7 +108,7 @@ public void setUp() throws DataRegionException {
     if (!file.getParentFile().exists()) {
       Assert.assertTrue(file.getParentFile().mkdirs());
     }
-    defaultAvgSeriesPointNumberThreshold = config.getAvgSeriesPointNumberThreshold();
+    defaultTargetChunkPointNum = config.getTargetChunkPointNum();
     defaultTargetChunkSize = config.getTargetChunkSize();
     EnvironmentUtils.envSetUp();
     sgInfo = new DataRegionInfo(new DataRegionTest.DummyDataRegion(systemDir, storageGroup));
@@ -128,7 +128,7 @@ public void tearDown() throws Exception {
       }
     } catch (IOException ignored) {
     }
-    config.setAvgSeriesPointNumberThreshold(defaultAvgSeriesPointNumberThreshold);
+    config.setTargetChunkPointNum(defaultTargetChunkPointNum);
     config.setTargetChunkSize(defaultTargetChunkSize);
   }
 
@@ -210,7 +210,7 @@ public void testWriteAndFlush()
   @Test
   public void testFlushMultiChunks()
       throws IOException, WriteProcessException, MetadataException, ExecutionException {
-    config.setAvgSeriesPointNumberThreshold(40);
+    config.setTargetChunkPointNum(40);
     processor =
         new TsFileProcessor(
             storageGroup,
@@ -364,7 +364,7 @@ public void testFlushMultiBinaryChunks()
   @Test
   public void testFlushMultiAlignedChunks()
       throws IOException, WriteProcessException, MetadataException, ExecutionException {
-    config.setAvgSeriesPointNumberThreshold(40);
+    config.setTargetChunkPointNum(40);
     processor =
         new TsFileProcessor(
             storageGroup,