Tvlist feat new

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

@@ -433,6 +433,12 @@ public class IoTDBConfig {
   /** The sort algorithm used in TVList */
   private TVListSortAlgorithm tvListSortAlgorithm = TVListSortAlgorithm.TIM;
 
+  /**
+   * 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;
 
@@ -2306,6 +2312,14 @@ public void setTvListSortAlgorithm(TVListSortAlgorithm tvListSortAlgorithm) {
     this.tvListSortAlgorithm = tvListSortAlgorithm;
   }
 
+  public int getTvListSortThreshold() {
+    return tvListSortThreshold;
+  }
+
+  public void setTVListSortThreshold(int tvListSortThreshold) {
+    this.tvListSortThreshold = tvListSortThreshold;
+  }
+
   public boolean isRpcThriftCompressionEnable() {
     return rpcThriftCompressionEnable;
   }

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

@@ -386,6 +386,11 @@ public void loadProperties(TrimProperties properties) throws BadNodeUrlException
             properties.getProperty(
                 "tvlist_sort_algorithm", conf.getTvListSortAlgorithm().toString())));
 
+    conf.setTVListSortThreshold(
+        Integer.parseInt(
+            properties.getProperty(
+                "tvlist_sort_threshold", Integer.toString(conf.getTvListSortThreshold()))));
+
     conf.setCheckPeriodWhenInsertBlocked(
         Integer.parseInt(
             properties.getProperty(
@@ -2059,6 +2064,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

@@ -41,6 +41,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;
@@ -649,6 +650,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.
@@ -669,6 +704,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());
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java

@@ -23,7 +23,9 @@
 import org.apache.iotdb.commons.path.IFullPath;
 import org.apache.iotdb.commons.path.NonAlignedFullPath;
 import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
 import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
+import org.apache.iotdb.db.queryengine.plan.planner.memory.MemoryReservationManager;
 import org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedReadOnlyMemChunk;
 import org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedWritableMemChunk;
 import org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedWritableMemChunkGroup;
@@ -48,15 +50,19 @@
 import org.apache.tsfile.file.metadata.enums.TSEncoding;
 import org.apache.tsfile.file.metadata.statistics.Statistics;
 import org.apache.tsfile.read.common.TimeRange;
+import org.apache.tsfile.read.filter.basic.Filter;
 import org.apache.tsfile.utils.Pair;
 import org.apache.tsfile.write.schema.IMeasurementSchema;
 import org.apache.tsfile.write.schema.VectorMeasurementSchema;
 import org.apache.tsfile.write.writer.RestorableTsFileIOWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -67,6 +73,7 @@
  * MeasurementPath have different implementations, and the default PartialPath should not use it.
  */
 public abstract class ResourceByPathUtils {
+  private static final Logger LOGGER = LoggerFactory.getLogger(ResourceByPathUtils.class);
 
   public static ResourceByPathUtils getResourceInstance(IFullPath path) {
     if (path instanceof AlignedFullPath) {
@@ -85,14 +92,118 @@ public abstract ReadOnlyMemChunk getReadOnlyMemChunkFromMemTable(
       QueryContext context,
       IMemTable memTable,
       List<Pair<ModEntry, IMemTable>> modsToMemtable,
-      long timeLowerBound)
+      long timeLowerBound,
+      Filter globalTimeFilter)
       throws QueryProcessException, IOException;
 
   public abstract List<IChunkMetadata> getVisibleMetadataListFromWriter(
       RestorableTsFileIOWriter writer,
       TsFileResource tsFileResource,
       QueryContext context,
       long timeLowerBound);
+
+  /**
+   * Prepare the TVList references for the query. We remember TVLists' row count here and determine
+   * whether the TVLists needs sorting later during operator execution based on it. It need not
+   * protect sorted list. Sorted list is changed in the handover process of inserting, which holds
+   * the data region write lock. At this moment, query thread holds the data region read lock.
+   *
+   * @param context query context
+   * @param memChunk writable memchunk
+   * @param isWorkMemTable in working or flushing memtable
+   * @param globalTimeFilter global time filter
+   * @return Map<TVList, Integer>
+   */
+  protected Map<TVList, Integer> prepareTvListMapForQuery(
+      QueryContext context,
+      IWritableMemChunk memChunk,
+      boolean isWorkMemTable,
+      Filter globalTimeFilter) {
+    // should copy globalTimeFilter because GroupByMonthFilter is stateful
+    Filter copyTimeFilter = null;
+    if (globalTimeFilter != null) {
+      copyTimeFilter = globalTimeFilter.copy();
+    }
+
+    Map<TVList, Integer> tvListQueryMap = new LinkedHashMap<>();
+    // immutable sorted lists
+    for (TVList tvList : memChunk.getSortedList()) {
+      if (copyTimeFilter != null
+          && !copyTimeFilter.satisfyStartEndTime(tvList.getMinTime(), tvList.getMaxTime())) {
+        continue;
+      }
+      tvList.lockQueryList();
+      try {
+        LOGGER.debug(
+            "Flushing/Working MemTable - add current query context to immutable TVList's query list");
+        tvList.getQueryContextList().add(context);
+        tvListQueryMap.put(tvList, tvList.rowCount());
+      } finally {
+        tvList.unlockQueryList();
+      }
+    }
+
+    // mutable tvlist
+    TVList list = memChunk.getWorkingTVList();
+    TVList cloneList = null;
+    list.lockQueryList();
+    try {
+      if (copyTimeFilter != null
+          && !copyTimeFilter.satisfyStartEndTime(list.getMinTime(), list.getMaxTime())) {
+        return tvListQueryMap;
+      }
+
+      if (!isWorkMemTable) {
+        LOGGER.debug(
+            "Flushing MemTable - add current query context to mutable TVList's query list");
+        list.getQueryContextList().add(context);
+        tvListQueryMap.put(list, list.rowCount());
+      } else {
+        if (list.isSorted() || list.getQueryContextList().isEmpty()) {
+          LOGGER.debug(
+              "Working MemTable - add current query context to mutable TVList's query list when it's sorted or no other query on it");
+          list.getQueryContextList().add(context);
+          tvListQueryMap.put(list, list.rowCount());
+        } else {
+          /*
+           * +----------------------+
+           * |      MemTable        |
+           * |                      |
+           * |    +------------+    |          +-----------------+
+           * |    |   TVList   |<---+--+   +---+  Previous Query |
+           * |    +-----^------+    |  |   |   +-----------------+
+           * |          |           |  |   |
+           * +----------+-----------+  |   |   +----------------+
+           *            | Clone        +---+---+  Current Query |
+           *      +-----+------+           |   +----------------+
+           *      |   TVList   | <---------+
+           *      +------------+
+           */
+          LOGGER.debug(
+              "Working MemTable - clone mutable TVList and replace old TVList in working MemTable");
+          QueryContext firstQuery = list.getQueryContextList().get(0);
+          // reserve query memory
+          if (firstQuery instanceof FragmentInstanceContext) {
+            MemoryReservationManager memoryReservationManager =
+                ((FragmentInstanceContext) firstQuery).getMemoryReservationContext();
+            memoryReservationManager.reserveMemoryCumulatively(list.calculateRamSize());
+          }
+          list.setOwnerQuery(firstQuery);
+
+          // clone TVList
+          cloneList = list.clone();
+          cloneList.getQueryContextList().add(context);
+          tvListQueryMap.put(cloneList, cloneList.rowCount());
+        }
+      }
+    } finally {
+      list.unlockQueryList();
+    }
+    if (cloneList != null) {
+      memChunk.setWorkingTVList(cloneList);
+    }
+    return tvListQueryMap;
+  }
 }
 
 class AlignedResourceByPathUtils extends ResourceByPathUtils {
@@ -149,6 +260,8 @@ public AlignedTimeSeriesMetadata generateTimeSeriesMetadata(
 
     for (ReadOnlyMemChunk memChunk : readOnlyMemChunk) {
       if (!memChunk.isEmpty()) {
+        memChunk.sortTvLists();
+        memChunk.initChunkMetaFromTvLists();
         AlignedChunkMetadata alignedChunkMetadata =
             (AlignedChunkMetadata) memChunk.getChunkMetaData();
         timeStatistics.mergeStatistics(alignedChunkMetadata.getTimeChunkMetadata().getStatistics());
@@ -182,7 +295,8 @@ public ReadOnlyMemChunk getReadOnlyMemChunkFromMemTable(
       QueryContext context,
       IMemTable memTable,
       List<Pair<ModEntry, IMemTable>> modsToMemtable,
-      long timeLowerBound)
+      long timeLowerBound,
+      Filter globalTimeFilter)
       throws QueryProcessException {
     Map<IDeviceID, IWritableMemChunkGroup> memTableMap = memTable.getMemTableMap();
     IDeviceID deviceID = alignedFullPath.getDeviceId();
@@ -207,10 +321,15 @@ public ReadOnlyMemChunk getReadOnlyMemChunkFromMemTable(
       }
     }
 
-    // get sorted tv list is synchronized so different query can get right sorted list reference
-    TVList alignedTvListCopy =
-        alignedMemChunk.getSortedTvListForQuery(
-            alignedFullPath.getSchemaList(), context.isIgnoreAllNullRows());
+    // prepare AlignedTVList for query. It should clone TVList if necessary.
+    Map<TVList, Integer> alignedTvListQueryMap =
+        prepareTvListMapForQuery(
+            context, alignedMemChunk, modsToMemtable == null, globalTimeFilter);
+
+    // column index list for the query
+    List<Integer> columnIndexList =
+        alignedMemChunk.buildColumnIndexList(alignedFullPath.getSchemaList());
+
     List<TimeRange> timeColumnDeletion = null;
     List<List<TimeRange>> valueColumnsDeletionList = null;
     if (modsToMemtable != null) {
@@ -232,8 +351,9 @@ public ReadOnlyMemChunk getReadOnlyMemChunkFromMemTable(
     }
     return new AlignedReadOnlyMemChunk(
         context,
+        columnIndexList,
         getMeasurementSchema(),
-        alignedTvListCopy,
+        alignedTvListQueryMap,
         timeColumnDeletion,
         valueColumnsDeletionList);
   }
@@ -352,6 +472,8 @@ public ITimeSeriesMetadata generateTimeSeriesMetadata(
 
     for (ReadOnlyMemChunk memChunk : readOnlyMemChunk) {
       if (!memChunk.isEmpty()) {
+        memChunk.sortTvLists();
+        memChunk.initChunkMetaFromTvLists();
         seriesStatistics.mergeStatistics(memChunk.getChunkMetaData().getStatistics());
       }
     }
@@ -365,7 +487,8 @@ public ReadOnlyMemChunk getReadOnlyMemChunkFromMemTable(
       QueryContext context,
       IMemTable memTable,
       List<Pair<ModEntry, IMemTable>> modsToMemtable,
-      long timeLowerBound)
+      long timeLowerBound,
+      Filter globalTimeFilter)
       throws QueryProcessException, IOException {
     Map<IDeviceID, IWritableMemChunkGroup> memTableMap = memTable.getMemTableMap();
     IDeviceID deviceID = fullPath.getDeviceId();
@@ -376,8 +499,9 @@ public ReadOnlyMemChunk getReadOnlyMemChunkFromMemTable(
     }
     IWritableMemChunk memChunk =
         memTableMap.get(deviceID).getMemChunkMap().get(fullPath.getMeasurement());
-    // get sorted tv list is synchronized so different query can get right sorted list reference
-    TVList chunkCopy = memChunk.getSortedTvListForQuery();
+    // prepare TVList for query. It should clone TVList if necessary.
+    Map<TVList, Integer> tvListQueryMap =
+        prepareTvListMapForQuery(context, memChunk, modsToMemtable == null, globalTimeFilter);
     List<TimeRange> deletionList = null;
     if (modsToMemtable != null) {
       deletionList =
@@ -393,7 +517,7 @@ public ReadOnlyMemChunk getReadOnlyMemChunkFromMemTable(
         fullPath.getMeasurement(),
         fullPath.getMeasurementSchema().getType(),
         fullPath.getMeasurementSchema().getEncodingType(),
-        chunkCopy,
+        tvListQueryMap,
         fullPath.getMeasurementSchema().getProps(),
         deletionList);
   }