fix: memtable enhancement issues (#14994)

shizy818 · web-flow · commit 1adc74dffd2a · 2025-03-04T14:28:42.000+08:00
* perf: improve seq inserting

* rename MergeSortTvListIterator to MergeSortTVListIterator

* fix: concurrent indices modification during query sort and flush sort
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
@@ -375,15 +375,6 @@ public boolean chunkNotExist(IDeviceID deviceId, String measurement) {
     return !memChunkGroup.contains(measurement);
   }
 
-  @Override
-  public long getMeasurementSize(IDeviceID deviceId, String measurement) {
-    IWritableMemChunkGroup memChunkGroup = memTableMap.get(deviceId);
-    if (null == memChunkGroup) {
-      return 0;
-    }
-    return memChunkGroup.getMeasurementSize(measurement);
-  }
-
   @Override
   public IWritableMemChunk getWritableMemChunk(IDeviceID deviceId, String measurement) {
     IWritableMemChunkGroup memChunkGroup = memTableMap.get(deviceId);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
@@ -137,11 +137,6 @@ public long deleteTime(ModEntry modEntry) {
     return memChunk.deleteTime(modEntry.getStartTime(), modEntry.getEndTime());
   }
 
-  @Override
-  public long getMeasurementSize(String measurement) {
-    return memChunk.rowCount();
-  }
-
   @Override
   public IWritableMemChunk getWritableMemChunk(String measurement) {
     return memChunk;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
@@ -169,8 +169,6 @@ void queryForDeviceRegionScan(
   boolean chunkNotExist(IDeviceID deviceId, String measurement);
 
   /** only used when mem control enabled */
-  long getMeasurementSize(IDeviceID deviceId, String measurement);
-
   IWritableMemChunk getWritableMemChunk(IDeviceID deviceId, String measurement);
 
   /** only used when mem control enabled */
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
@@ -56,8 +56,6 @@ void writeTablet(
 
   long deleteTime(ModEntry modEntry);
 
-  long getMeasurementSize(String measurement);
-
   IWritableMemChunk getWritableMemChunk(String measurement);
 
   long getMaxTime();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/ReadOnlyMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/ReadOnlyMemChunk.java
@@ -24,7 +24,7 @@
 import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
 import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.MemChunkLoader;
 import org.apache.iotdb.db.utils.MathUtils;
-import org.apache.iotdb.db.utils.datastructure.MergeSortTvListIterator;
+import org.apache.iotdb.db.utils.datastructure.MergeSortTVListIterator;
 import org.apache.iotdb.db.utils.datastructure.TVList;
 
 import org.apache.tsfile.common.conf.TSFileDescriptor;
@@ -85,7 +85,7 @@ public class ReadOnlyMemChunk {
   // TVList and its rowCount during query
   private Map<TVList, Integer> tvListQueryMap;
 
-  private MergeSortTvListIterator timeValuePairIterator;
+  private MergeSortTVListIterator timeValuePairIterator;
 
   protected final int MAX_NUMBER_OF_POINTS_IN_PAGE =
       TSFileDescriptor.getInstance().getConfig().getMaxNumberOfPointsInPage();
@@ -148,7 +148,7 @@ public void initChunkMetaFromTvLists() {
     int cnt = 0;
     int[] deleteCursor = {0};
     List<TVList> tvLists = new ArrayList<>(tvListQueryMap.keySet());
-    timeValuePairIterator = new MergeSortTvListIterator(tvLists, floatPrecision, encoding);
+    timeValuePairIterator = new MergeSortTVListIterator(tvLists, floatPrecision, encoding);
     int[] tvListOffsets = timeValuePairIterator.getTVListOffsets();
     while (timeValuePairIterator.hasNextTimeValuePair()) {
       if (cnt % MAX_NUMBER_OF_POINTS_IN_PAGE == 0) {
@@ -246,7 +246,7 @@ private void writeValidValuesIntoTsBlock(TsBlockBuilder builder) throws IOExcept
     int[] deleteCursor = {0};
     List<TVList> tvLists = new ArrayList<>(tvListQueryMap.keySet());
     IPointReader timeValuePairIterator =
-        new MergeSortTvListIterator(tvLists, floatPrecision, encoding);
+        new MergeSortTVListIterator(tvLists, floatPrecision, encoding);
 
     while (timeValuePairIterator.hasNextTimeValuePair()) {
       TimeValuePair tvPair = timeValuePairIterator.nextTimeValuePair();
@@ -334,7 +334,7 @@ public TsBlock getTsBlock() {
     return null;
   }
 
-  public MergeSortTvListIterator getMergeSortTVListIterator() {
+  public MergeSortTVListIterator getMergeSortTVListIterator() {
     return timeValuePairIterator;
   }
 
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -658,8 +658,8 @@ private long[] checkMemCostAndAddToTspInfoForRow(
         memTableIncrement += TVList.tvListArrayMemCost(dataTypes[i]);
       } else {
         // here currentChunkPointNum >= 1
-        long currentChunkPointNum = workMemTable.getMeasurementSize(deviceId, measurements[i]);
         IWritableMemChunk memChunk = workMemTable.getWritableMemChunk(deviceId, measurements[i]);
+        long currentChunkPointNum = memChunk != null ? memChunk.rowCount() : 0;
         if (currentChunkPointNum % PrimitiveArrayManager.ARRAY_SIZE == 0) {
           memTableIncrement +=
               memChunk != null ? memChunk.getWorkingTVList().tvListArrayMemCost() : 0;
@@ -704,12 +704,12 @@ private long[] checkMemCostAndAddToTspInfoForRows(List<InsertRowNode> insertRowN
               .putIfAbsent(measurements[i], 1);
         } else {
           // here currentChunkPointNum >= 1
-          long currentChunkPointNum = workMemTable.getMeasurementSize(deviceId, measurements[i]);
           IWritableMemChunk memChunk = workMemTable.getWritableMemChunk(deviceId, measurements[i]);
           int addingPointNum =
               increasingMemTableInfo
                   .computeIfAbsent(deviceId, k -> new HashMap<>())
                   .computeIfAbsent(measurements[i], k -> 0);
+          long currentChunkPointNum = memChunk != null ? memChunk.rowCount() : 0;
           if ((currentChunkPointNum + addingPointNum) % PrimitiveArrayManager.ARRAY_SIZE == 0) {
             memTableIncrement +=
                 memChunk != null
@@ -976,8 +976,8 @@ private void updateMemCost(
           ((end - start) / PrimitiveArrayManager.ARRAY_SIZE + 1)
               * TVList.tvListArrayMemCost(dataType);
     } else {
-      long currentChunkPointNum = workMemTable.getMeasurementSize(deviceId, measurement);
       IWritableMemChunk memChunk = workMemTable.getWritableMemChunk(deviceId, measurement);
+      long currentChunkPointNum = memChunk != null ? memChunk.rowCount() : 0;
       if (currentChunkPointNum % PrimitiveArrayManager.ARRAY_SIZE == 0) {
         memIncrements[0] +=
             ((end - start) / PrimitiveArrayManager.ARRAY_SIZE + 1)
@@ -989,11 +989,7 @@ private void updateMemCost(
             (end - start - 1 + (currentChunkPointNum % PrimitiveArrayManager.ARRAY_SIZE))
                 / PrimitiveArrayManager.ARRAY_SIZE;
         if (acquireArray != 0) {
-          memIncrements[0] +=
-              acquireArray
-                  * (memChunk != null
-                      ? memChunk.getWorkingTVList().tvListArrayMemCost()
-                      : TVList.tvListArrayMemCost(dataType));
+          memIncrements[0] += acquireArray * memChunk.getWorkingTVList().tvListArrayMemCost();
         }
       }
     }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
@@ -27,7 +27,7 @@
 import org.apache.iotdb.db.queryengine.plan.planner.memory.MemoryReservationManager;
 import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView;
 import org.apache.iotdb.db.utils.ModificationUtils;
-import org.apache.iotdb.db.utils.datastructure.MergeSortTvListIterator;
+import org.apache.iotdb.db.utils.datastructure.MergeSortTVListIterator;
 import org.apache.iotdb.db.utils.datastructure.TVList;
 
 import org.apache.tsfile.enums.TSDataType;
@@ -641,7 +641,7 @@ public synchronized void encode(BlockingQueue<Object> ioTaskQueue) {
     // create MergeSortTvListIterator. It need not handle float/double precision here.
     List<TVList> tvLists = new ArrayList<>(sortedList);
     tvLists.add(list);
-    MergeSortTvListIterator timeValuePairIterator = new MergeSortTvListIterator(tvLists);
+    MergeSortTVListIterator timeValuePairIterator = new MergeSortTVListIterator(tvLists);
 
     TimeValuePair prevTvPair = null;
     while (timeValuePairIterator.hasNextTimeValuePair()) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
@@ -145,14 +145,6 @@ public long deleteTime(ModEntry modEntry) {
     return delete(modEntry);
   }
 
-  @Override
-  public long getMeasurementSize(String measurement) {
-    if (!memChunkMap.containsKey(measurement)) {
-      return 0;
-    }
-    return memChunkMap.get(measurement).rowCount();
-  }
-
   @Override
   public IWritableMemChunk getWritableMemChunk(String measurement) {
     if (!memChunkMap.containsKey(measurement)) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkReader.java
@@ -20,7 +20,7 @@
 package org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk;
 
 import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk;
-import org.apache.iotdb.db.utils.datastructure.MergeSortTvListIterator;
+import org.apache.iotdb.db.utils.datastructure.MergeSortTVListIterator;
 
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.file.metadata.IChunkMetadata;
@@ -47,7 +47,7 @@
 public class MemChunkReader implements IChunkReader, IPointReader {
 
   private final ReadOnlyMemChunk readableChunk;
-  private final MergeSortTvListIterator timeValuePairIterator;
+  private final MergeSortTVListIterator timeValuePairIterator;
   private final Filter globalTimeFilter;
   private final List<IPageReader> pageReaderList;
 
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemPageReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemPageReader.java
@@ -20,7 +20,7 @@
 package org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk;
 
 import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.metadata.PageMetadata;
-import org.apache.iotdb.db.utils.datastructure.MergeSortTvListIterator;
+import org.apache.iotdb.db.utils.datastructure.MergeSortTVListIterator;
 
 import org.apache.tsfile.block.column.Column;
 import org.apache.tsfile.block.column.ColumnBuilder;
@@ -51,7 +51,7 @@ public class MemPageReader implements IPageReader {
   private TsBlock tsBlock;
   private Filter recordFilter;
 
-  private final MergeSortTvListIterator mergeSortTvListIterator;
+  private final MergeSortTVListIterator mergeSortTvListIterator;
   // MemPage range - [pageStartOffsets, pageEndOffsets)
   private final int[] pageStartOffsets;
   private final int[] pageEndOffsets;
@@ -63,7 +63,7 @@ public class MemPageReader implements IPageReader {
 
   public MemPageReader(
       Supplier<TsBlock> tsBlockSupplier,
-      MergeSortTvListIterator mergeSortTvListIterator,
+      MergeSortTVListIterator mergeSortTvListIterator,
       int[] pageStartOffsets,
       int[] pageEndOffSets,
       TSDataType tsDataType,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortTVListIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortTVListIterator.java
@@ -30,7 +30,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
-public class MergeSortTvListIterator implements IPointReader {
+public class MergeSortTVListIterator implements IPointReader {
   private final TVList.TVListIterator[] tvListIterators;
   private final int[] tvListOffsets;
 
@@ -42,7 +42,7 @@ public class MergeSortTvListIterator implements IPointReader {
       new PriorityQueue<>(
           (a, b) -> a.left.equals(b.left) ? b.right.compareTo(a.right) : a.left.compareTo(b.left));
 
-  public MergeSortTvListIterator(List<TVList> tvLists) {
+  public MergeSortTVListIterator(List<TVList> tvLists) {
     tvListIterators = new TVList.TVListIterator[tvLists.size()];
     for (int i = 0; i < tvLists.size(); i++) {
       tvListIterators[i] = tvLists.get(i).iterator(null, null);
@@ -52,7 +52,7 @@ public MergeSortTvListIterator(List<TVList> tvLists) {
         IntStream.range(0, tvListIterators.length).boxed().collect(Collectors.toList());
   }
 
-  public MergeSortTvListIterator(
+  public MergeSortTVListIterator(
       List<TVList> tvLists, Integer floatPrecision, TSEncoding encoding) {
     tvListIterators = new TVList.TVListIterator[tvLists.size()];
     for (int i = 0; i < tvLists.size(); i++) {
@@ -63,7 +63,7 @@ public MergeSortTvListIterator(
         IntStream.range(0, tvListIterators.length).boxed().collect(Collectors.toList());
   }
 
-  public MergeSortTvListIterator(TVList.TVListIterator[] tvListIterators) {
+  public MergeSortTVListIterator(TVList.TVListIterator[] tvListIterators) {
     this.tvListIterators = new TVList.TVListIterator[tvListIterators.length];
     for (int i = 0; i < tvListIterators.length; i++) {
       this.tvListIterators[i] = tvListIterators[i].clone();
@@ -156,7 +156,7 @@ public void setTVListOffsets(int[] tvListOffsets) {
   }
 
   @Override
-  public MergeSortTvListIterator clone() {
-    return new MergeSortTvListIterator(tvListIterators);
+  public MergeSortTVListIterator clone() {
+    return new MergeSortTVListIterator(tvListIterators);
   }
 }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
@@ -415,7 +415,7 @@ public int delete(long lowerBound, long upperBound) {
   }
 
   // common clone for both TVList and AlignedTVList
-  protected void cloneAs(TVList cloneList) {
+  protected synchronized void cloneAs(TVList cloneList) {
     // clone timestamps
     for (long[] timestampArray : timestamps) {
       cloneList.timestamps.add(cloneTime(timestampArray));
@@ -499,7 +499,6 @@ void updateMinMaxTimeAndSorted(long[] time, int start, int end) {
     for (int i = start; i < end; i++) {
       inPutMinTime = Math.min(inPutMinTime, time[i]);
       maxTime = Math.max(maxTime, time[i]);
-      minTime = Math.min(minTime, time[i]);
       if (inputSorted) {
         if (i < length - 1 && time[i] > time[i + 1]) {
           inputSorted = false;
@@ -508,6 +507,7 @@ void updateMinMaxTimeAndSorted(long[] time, int start, int end) {
         }
       }
     }
+    minTime = Math.min(minTime, inPutMinTime);
     if (sorted && (rowCount == 0 || time[start] >= getTime(rowCount - 1))) {
       seqRowCount += inputSeqRowCount;
     }
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkLoaderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkLoaderTest.java
@@ -21,7 +21,7 @@
 
 import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
 import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk;
-import org.apache.iotdb.db.utils.datastructure.MergeSortTvListIterator;
+import org.apache.iotdb.db.utils.datastructure.MergeSortTVListIterator;
 import org.apache.iotdb.db.utils.datastructure.TVList;
 
 import org.apache.tsfile.common.conf.TSFileConfig;
@@ -68,8 +68,8 @@ public void testBooleanMemChunkLoader() throws IOException {
     Map<TVList, Integer> booleanTvListMap = buildBooleanTvListMap();
     Mockito.when(chunk.getTvListQueryMap()).thenReturn(booleanTvListMap);
     List<TVList> booleanTvLists = new ArrayList<>(booleanTvListMap.keySet());
-    MergeSortTvListIterator timeValuePairIterator =
-        new MergeSortTvListIterator(booleanTvLists, null, null);
+    MergeSortTVListIterator timeValuePairIterator =
+        new MergeSortTVListIterator(booleanTvLists, null, null);
     Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
 
     ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
@@ -149,8 +149,8 @@ public void testInt32MemChunkLoader() throws IOException {
     Map<TVList, Integer> int32TvListMap = buildInt32TvListMap();
     Mockito.when(chunk.getTvListQueryMap()).thenReturn(int32TvListMap);
     List<TVList> int32TvLists = new ArrayList<>(int32TvListMap.keySet());
-    MergeSortTvListIterator timeValuePairIterator =
-        new MergeSortTvListIterator(int32TvLists, null, null);
+    MergeSortTVListIterator timeValuePairIterator =
+        new MergeSortTVListIterator(int32TvLists, null, null);
     Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
 
     ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
@@ -230,8 +230,8 @@ public void testInt64MemChunkLoader() throws IOException {
     Map<TVList, Integer> int64TvListMap = buildInt64TvListMap();
     Mockito.when(chunk.getTvListQueryMap()).thenReturn(int64TvListMap);
     List<TVList> int64TvLists = new ArrayList<>(int64TvListMap.keySet());
-    MergeSortTvListIterator timeValuePairIterator =
-        new MergeSortTvListIterator(int64TvLists, null, null);
+    MergeSortTVListIterator timeValuePairIterator =
+        new MergeSortTVListIterator(int64TvLists, null, null);
     Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
 
     ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
@@ -311,8 +311,8 @@ public void testFloatMemChunkLoader() throws IOException {
     Map<TVList, Integer> floatTvListMap = buildFloatTvListMap();
     Mockito.when(chunk.getTvListQueryMap()).thenReturn(floatTvListMap);
     List<TVList> floatTvLists = new ArrayList<>(floatTvListMap.keySet());
-    MergeSortTvListIterator timeValuePairIterator =
-        new MergeSortTvListIterator(floatTvLists, null, null);
+    MergeSortTVListIterator timeValuePairIterator =
+        new MergeSortTVListIterator(floatTvLists, null, null);
     Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
 
     ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
@@ -392,8 +392,8 @@ public void testDoubleMemChunkLoader() throws IOException {
     Map<TVList, Integer> doubleTvListMap = buildDoubleTvListMap();
     Mockito.when(chunk.getTvListQueryMap()).thenReturn(doubleTvListMap);
     List<TVList> doubleTvLists = new ArrayList<>(doubleTvListMap.keySet());
-    MergeSortTvListIterator timeValuePairIterator =
-        new MergeSortTvListIterator(doubleTvLists, null, null);
+    MergeSortTVListIterator timeValuePairIterator =
+        new MergeSortTVListIterator(doubleTvLists, null, null);
     Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
 
     ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
@@ -473,8 +473,8 @@ public void testTextMemChunkLoader() throws IOException {
     Map<TVList, Integer> textTvListMap = buildTextTvListMap();
     Mockito.when(chunk.getTvListQueryMap()).thenReturn(textTvListMap);
     List<TVList> textTvLists = new ArrayList<>(textTvListMap.keySet());
-    MergeSortTvListIterator timeValuePairIterator =
-        new MergeSortTvListIterator(textTvLists, null, null);
+    MergeSortTVListIterator timeValuePairIterator =
+        new MergeSortTVListIterator(textTvLists, null, null);
     Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
 
     ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
