Memtable enhancement for query

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

@@ -309,8 +309,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);
@@ -321,6 +319,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)");
@@ -343,7 +342,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

@@ -335,8 +335,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);
@@ -372,7 +370,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

@@ -433,8 +433,11 @@ 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;
+  /**
+   * the threshold when working TVList is sorted and added into immutable TVList list in the
+   * writable memtable
+   */
+  private int tvListSortThreshold = 0;
 
   /** Enable inner space compaction for sequence files */
   private volatile boolean enableSeqSpaceCompaction = true;
@@ -494,10 +497,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;
 
   /**
@@ -2307,12 +2310,12 @@ public void setTvListSortAlgorithm(TVListSortAlgorithm tvListSortAlgorithm) {
     this.tvListSortAlgorithm = tvListSortAlgorithm;
   }
 
-  public int getAvgSeriesPointNumberThreshold() {
-    return avgSeriesPointNumberThreshold;
+  public int getTvListSortThreshold() {
+    return tvListSortThreshold;
   }
 
-  public void setAvgSeriesPointNumberThreshold(int avgSeriesPointNumberThreshold) {
-    this.avgSeriesPointNumberThreshold = avgSeriesPointNumberThreshold;
+  public void setTVListSortThreshold(int tvListSortThreshold) {
+    this.tvListSortThreshold = tvListSortThreshold;
   }
 
   public boolean isRpcThriftCompressionEnable() {

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

@@ -386,11 +386,10 @@ public void loadProperties(TrimProperties properties) throws BadNodeUrlException
             properties.getProperty(
                 "tvlist_sort_algorithm", conf.getTvListSortAlgorithm().toString())));
 
-    conf.setAvgSeriesPointNumberThreshold(
+    conf.setTVListSortThreshold(
         Integer.parseInt(
             properties.getProperty(
-                "avg_series_point_number_threshold",
-                Integer.toString(conf.getAvgSeriesPointNumberThreshold()))));
+                "tvlist_sort_threshold", Integer.toString(conf.getTvListSortThreshold()))));
 
     conf.setCheckPeriodWhenInsertBlocked(
         Integer.parseInt(
@@ -2075,6 +2074,13 @@ public synchronized void loadHotModifiedProps(TrimProperties properties)
       loadQuerySampleThroughput(properties);
       // update trusted_uri_pattern
       loadTrustedUriPattern(properties);
+
+      // tvlist_sort_threshold
+      conf.setTVListSortThreshold(
+          Integer.parseInt(
+              properties.getProperty(
+                  "tvlist_sort_threshold",
+                  ConfigurationFileUtils.getConfigurationDefaultValue("tvlist_sort_threshold"))));
     } catch (Exception e) {
       if (e instanceof InterruptedException) {
         Thread.currentThread().interrupt();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java

@@ -42,6 +42,7 @@
 import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSourceType;
 import org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.utils.datastructure.TVList;
 import org.apache.iotdb.mpp.rpc.thrift.TFetchFragmentInstanceStatisticsResp;
 
 import org.apache.tsfile.file.metadata.IDeviceID;
@@ -653,6 +654,40 @@ public void releaseResourceWhenAllDriversAreClosed() {
     releaseResource();
   }
 
+  /**
+   * It checks all referenced TVList by the query: 1. If current is not the owner, just remove
+   * itself from query context list 2. If current query is the owner and no other query use it now,
+   * release the TVList 3. If current query is the owner and other queries still use it, set the
+   * next query as owner
+   */
+  private void releaseTVListOwnedByQuery() {
+    for (TVList tvList : tvListSet) {
+      tvList.lockQueryList();
+      List<QueryContext> queryContextList = tvList.getQueryContextList();
+      try {
+        queryContextList.remove(this);
+        if (tvList.getOwnerQuery() == this) {
+          if (queryContextList.isEmpty()) {
+            LOGGER.debug(
+                "TVList {} is released by the query, FragmentInstance Id is {}",
+                tvList,
+                this.getId());
+            memoryReservationManager.releaseMemoryCumulatively(tvList.calculateRamSize());
+            tvList.clear();
+          } else {
+            LOGGER.debug(
+                "TVList {} is now owned by another query, FragmentInstance Id is {}",
+                tvList,
+                ((FragmentInstanceContext) queryContextList.get(0)).getId());
+            tvList.setOwnerQuery(queryContextList.get(0));
+          }
+        }
+      } finally {
+        tvList.unlockQueryList();
+      }
+    }
+  }
+
   /**
    * All file paths used by this fragment instance must be cleared and thus the usage reference must
    * be decreased.
@@ -673,6 +708,9 @@ public synchronized void releaseResource() {
       unClosedFilePaths = null;
     }
 
+    // release TVList/AlignedTVList owned by current query
+    releaseTVListOwnedByQuery();
+
     dataRegion = null;
     globalTimeFilter = null;
     sharedQueryDataSource = null;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryContext.java

@@ -28,13 +28,15 @@
 import org.apache.iotdb.db.utils.ModificationUtils;
 import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
 import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory.ModsSerializer;
+import org.apache.iotdb.db.utils.datastructure.TVList;
 
 import org.apache.tsfile.file.metadata.IDeviceID;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -71,6 +73,9 @@ public class QueryContext {
 
   private final Set<TsFileID> nonExistentModFiles = new CopyOnWriteArraySet<>();
 
+  // referenced TVLists for the query
+  protected final Set<TVList> tvListSet = new HashSet<>();
+
   public QueryContext() {}
 
   public QueryContext(long queryId) {
@@ -214,4 +219,8 @@ public boolean isIgnoreAllNullRows() {
   public void setIgnoreAllNullRows(boolean ignoreAllNullRows) {
     this.ignoreAllNullRows = ignoreAllNullRows;
   }
+
+  public void addTVListToSet(Map<TVList, Integer> tvListMap) {
+    tvListSet.addAll(tvListMap.keySet());
+  }
 }