Unaligned tvlist feat

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,35 @@ public void releaseResourceWhenAllDriversAreClosed() {
     releaseResource();
   }
 
+  private void releaseTVListOwnedByQuery() {
+    for (TVList tvList : tvListSet) {
+      tvList.lockQueryList();
+      List<QueryContext> queryContextList = tvList.getQueryContextList();
+      try {
+        queryContextList.remove(this);
+        if (tvList.getOwnerQuery() == this) {
+          tvList.setOwnerQuery(null);
+          if (queryContextList.isEmpty()) {
+            LOGGER.debug(
+                "TVList {} is released by the query, FragmentInstance Id is {}",
+                tvList,
+                this.getId());
+            tvList.clear();
+            memoryReservationManager.releaseMemoryCumulatively(tvList.calculateRamSize());
+          } else {
+            LOGGER.debug(
+                "TVList {} is 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 +699,9 @@ public synchronized void releaseResource() {
       unClosedFilePaths = null;
     }
 
+    // release TVList 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<>();
 
+  // accessed tvlists for the query
+  protected final Set<TVList> tvListSet = new HashSet<>();
+
   public QueryContext() {}
 
   public QueryContext(long queryId) {
@@ -206,4 +211,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

@@ -22,14 +22,16 @@
 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;
 import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable;
-import org.apache.iotdb.db.storageengine.dataregion.memtable.IWritableMemChunk;
 import org.apache.iotdb.db.storageengine.dataregion.memtable.IWritableMemChunkGroup;
 import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk;
+import org.apache.iotdb.db.storageengine.dataregion.memtable.WritableMemChunk;
 import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.db.utils.ModificationUtils;
@@ -47,15 +49,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;
 
@@ -84,7 +90,8 @@ 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(
@@ -181,7 +188,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();
@@ -322,6 +330,9 @@ public List<IChunkMetadata> getVisibleMetadataListFromWriter(
 
 class MeasurementResourceByPathUtils extends ResourceByPathUtils {
 
+  private static final Logger LOGGER =
+      LoggerFactory.getLogger(MeasurementResourceByPathUtils.class);
+
   NonAlignedFullPath fullPath;
 
   protected MeasurementResourceByPathUtils(IFullPath fullPath) {
@@ -359,12 +370,79 @@ public ITimeSeriesMetadata generateTimeSeriesMetadata(
     return timeSeriesMetadata;
   }
 
+  private Map<TVList, Integer> prepareTvListMapForQuery(
+      WritableMemChunk memChunk,
+      boolean isWorkMemTable,
+      QueryContext context,
+      Filter globalTimeFilter) {
+    Map<TVList, Integer> tvListQueryMap = new LinkedHashMap<>();
+    // immutable sorted lists
+    for (TVList tvList : memChunk.getSortedList()) {
+      if (globalTimeFilter != null
+          && !globalTimeFilter.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();
+    list.lockQueryList();
+    try {
+      if (!isWorkMemTable) {
+        if (globalTimeFilter == null
+            || globalTimeFilter.satisfyStartEndTime(list.getMinTime(), list.getMaxTime())) {
+          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 {
+          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
+          TVList cloneList = list.clone();
+          cloneList.getQueryContextList().add(context);
+          tvListQueryMap.put(cloneList, cloneList.rowCount());
+          memChunk.setWorkingTVList(cloneList);
+        }
+      }
+    } finally {
+      list.unlockQueryList();
+    }
+    return tvListQueryMap;
+  }
+
   @Override
   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();
@@ -373,10 +451,13 @@ public ReadOnlyMemChunk getReadOnlyMemChunkFromMemTable(
         || !memTableMap.get(deviceID).contains(fullPath.getMeasurement())) {
       return null;
     }
-    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();
+    WritableMemChunk memChunk =
+        (WritableMemChunk)
+            memTableMap.get(deviceID).getMemChunkMap().get(fullPath.getMeasurement());
+    // prepare TVList for query. It should clone and sort TVList if necessary.
+    // Also, the map keeps TVlist length at this moment.
+    Map<TVList, Integer> tvListQueryMap =
+        prepareTvListMapForQuery(memChunk, modsToMemtable == null, context, globalTimeFilter);
     List<TimeRange> deletionList = null;
     if (modsToMemtable != null) {
       deletionList =
@@ -392,7 +473,7 @@ public ReadOnlyMemChunk getReadOnlyMemChunkFromMemTable(
         fullPath.getMeasurement(),
         fullPath.getMeasurementSchema().getType(),
         fullPath.getMeasurementSchema().getEncodingType(),
-        chunkCopy,
+        tvListQueryMap,
         fullPath.getMeasurementSchema().getProps(),
         deletionList);
   }

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

@@ -2163,7 +2163,9 @@ private List<TsFileResource> getFileResourceListForQuery(
         if (tsFileResource.isClosed()) {
           tsfileResourcesForQuery.add(tsFileResource);
         } else {
-          tsFileResource.getProcessor().query(pathList, context, tsfileResourcesForQuery);
+          tsFileResource
+              .getProcessor()
+              .query(pathList, context, tsfileResourcesForQuery, globalTimeFilter);
         }
       } catch (IOException e) {
         throw new MetadataException(e);

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

@@ -58,6 +58,7 @@
 import org.apache.tsfile.file.metadata.enums.TSEncoding;
 import org.apache.tsfile.file.metadata.statistics.TimeStatistics;
 import org.apache.tsfile.read.common.TimeRange;
+import org.apache.tsfile.read.filter.basic.Filter;
 import org.apache.tsfile.utils.BitMap;
 import org.apache.tsfile.utils.Pair;
 import org.apache.tsfile.utils.ReadWriteIOUtils;
@@ -459,10 +460,12 @@ public ReadOnlyMemChunk query(
       QueryContext context,
       IFullPath fullPath,
       long ttlLowerBound,
-      List<Pair<ModEntry, IMemTable>> modsToMemtable)
+      List<Pair<ModEntry, IMemTable>> modsToMemtable,
+      Filter globalTimeFilter)
       throws IOException, QueryProcessException {
     return ResourceByPathUtils.getResourceInstance(fullPath)
-        .getReadOnlyMemChunkFromMemTable(context, this, modsToMemtable, ttlLowerBound);
+        .getReadOnlyMemChunkFromMemTable(
+            context, this, modsToMemtable, ttlLowerBound, globalTimeFilter);
   }
 
   @Override
@@ -570,7 +573,7 @@ private void getMemChunkHandleFromMemTable(
             buildChunkMetaDataForMemoryChunk(
                 measurementId,
                 timestamps[0],
-                timestamps[tvListCopy.rowCount() - 1],
+                timestamps[tvListCopy.count() - 1],
                 Collections.emptyList()));
     memChunkHandleMap
         .computeIfAbsent(measurementId, k -> new ArrayList<>())
@@ -669,7 +672,7 @@ private void getMemChunkHandleFromMemTable(
               buildChunkMetaDataForMemoryChunk(
                   measurementId,
                   timestamps[0],
-                  timestamps[tvListCopy.rowCount() - 1],
+                  timestamps[tvListCopy.count() - 1],
                   Collections.emptyList()));
       memChunkHandleMap
           .computeIfAbsent(measurementId, k -> new ArrayList<>())
@@ -759,7 +762,7 @@ private IChunkMetadata buildChunkMetaDataForMemoryChunk(
   }
 
   private long[] filterDeletedTimestamp(TVList tvList, List<TimeRange> deletionList) {
-    if (deletionList.isEmpty()) {
+    if (tvList.getBitMap() == null && deletionList.isEmpty()) {
       long[] timestamps = tvList.getTimestamps().stream().flatMapToLong(LongStream::of).toArray();
       return Arrays.copyOfRange(timestamps, 0, tvList.rowCount());
     }
@@ -771,7 +774,8 @@ private long[] filterDeletedTimestamp(TVList tvList, List<TimeRange> deletionLis
 
     for (int i = 0; i < rowCount; i++) {
       long curTime = tvList.getTime(i);
-      if (!ModificationUtils.isPointDeleted(curTime, deletionList, deletionCursor)
+      if (!tvList.isNullValue(i)
+          && !ModificationUtils.isPointDeleted(curTime, deletionList, deletionCursor)
           && (i == rowCount - 1 || curTime != lastTime)) {
         result.add(curTime);
       }

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

@@ -241,7 +241,7 @@ private Pair<Object[], BitMap[]> checkAndReorderColumnValuesInInsertPlan(
   }
 
   @Override
-  public TVList getTVList() {
+  public TVList getWorkingTVList() {
     return list;
   }
 

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

@@ -129,7 +129,7 @@ public long deleteTime(ModEntry modEntry) {
 
   @Override
   public long getCurrentTVListSize(String measurement) {
-    return memChunk.getTVList().rowCount();
+    return memChunk.getWorkingTVList().rowCount();
   }
 
   @Override

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

@@ -33,6 +33,7 @@
 
 import org.apache.tsfile.file.metadata.IChunkMetadata;
 import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.filter.basic.Filter;
 import org.apache.tsfile.utils.Pair;
 import org.apache.tsfile.write.schema.IMeasurementSchema;
 
@@ -117,7 +118,8 @@ ReadOnlyMemChunk query(
       QueryContext context,
       IFullPath fullPath,
       long ttlLowerBound,
-      List<Pair<ModEntry, IMemTable>> modsToMemtabled)
+      List<Pair<ModEntry, IMemTable>> modsToMemtabled,
+      Filter globalTimeFilter)
       throws IOException, QueryProcessException, MetadataException;
 
   void queryForSeriesRegionScan(