Concurrently querying and writing to the memtable may cause the query results out of order (apache#16328)

Author: shuwenwei

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

@@ -51,7 +51,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 public class AlignedReadOnlyMemChunk extends ReadOnlyMemChunk {
   private final String timeChunkName;
@@ -377,23 +376,7 @@ private TsBlock buildTsBlock() {
   }
 
   private void writeValidValuesIntoTsBlock(TsBlockBuilder builder) throws IOException {
-    List<AlignedTVList> alignedTvLists =
-        alignedTvListQueryMap.keySet().stream()
-            .map(x -> (AlignedTVList) x)
-            .collect(Collectors.toList());
-    MemPointIterator timeValuePairIterator =
-        MemPointIteratorFactory.create(
-            dataTypes,
-            columnIndexList,
-            alignedTvLists,
-            Ordering.ASC,
-            null,
-            timeColumnDeletion,
-            valueColumnsDeletionList,
-            floatPrecision,
-            encodingList,
-            context.isIgnoreAllNullRows(),
-            MAX_NUMBER_OF_POINTS_IN_PAGE);
+    MemPointIterator timeValuePairIterator = createMemPointIterator(Ordering.ASC, null);
 
     while (timeValuePairIterator.hasNextTimeValuePair()) {
       TimeValuePair tvPair = timeValuePairIterator.nextTimeValuePair();
@@ -475,14 +458,17 @@ public MemPointIterator getMemPointIterator() {
 
   @Override
   public MemPointIterator createMemPointIterator(Ordering scanOrder, Filter globalTimeFilter) {
-    List<AlignedTVList> alignedTvLists =
-        alignedTvListQueryMap.keySet().stream()
-            .map(x -> (AlignedTVList) x)
-            .collect(Collectors.toList());
+    List<AlignedTVList> tvLists = new ArrayList<>(alignedTvListQueryMap.size());
+    List<Integer> tvListRowCounts = new ArrayList<>(alignedTvListQueryMap.size());
+    for (Map.Entry<TVList, Integer> entry : alignedTvListQueryMap.entrySet()) {
+      tvLists.add((AlignedTVList) entry.getKey());
+      tvListRowCounts.add(entry.getValue());
+    }
     return MemPointIteratorFactory.create(
         dataTypes,
         columnIndexList,
-        alignedTvLists,
+        tvLists,
+        tvListRowCounts,
         scanOrder,
         globalTimeFilter,
         timeColumnDeletion,

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

@@ -292,17 +292,7 @@ private TsBlock buildTsBlock() {
 
   // read all data in memory chunk and write to tsblock
   private void writeValidValuesIntoTsBlock(TsBlockBuilder builder) throws IOException {
-    List<TVList> tvLists = new ArrayList<>(tvListQueryMap.keySet());
-    MemPointIterator timeValuePairIterator =
-        MemPointIteratorFactory.create(
-            getDataType(),
-            tvLists,
-            Ordering.ASC,
-            null,
-            deletionList,
-            floatPrecision,
-            encoding,
-            MAX_NUMBER_OF_POINTS_IN_PAGE);
+    MemPointIterator timeValuePairIterator = createMemPointIterator(Ordering.ASC, null);
 
     while (timeValuePairIterator.hasNextTimeValuePair()) {
       TimeValuePair tvPair = timeValuePairIterator.nextTimeValuePair();
@@ -379,10 +369,16 @@ public MemPointIterator getMemPointIterator() {
   }
 
   public MemPointIterator createMemPointIterator(Ordering scanOrder, Filter globalTimeFilter) {
-    List<TVList> tvLists = new ArrayList<>(tvListQueryMap.keySet());
+    List<TVList> tvLists = new ArrayList<>(tvListQueryMap.size());
+    List<Integer> tvListRowCounts = new ArrayList<>(tvListQueryMap.size());
+    for (Map.Entry<TVList, Integer> entry : tvListQueryMap.entrySet()) {
+      tvLists.add(entry.getKey());
+      tvListRowCounts.add(entry.getValue());
+    }
     return MemPointIteratorFactory.create(
         dataType,
         tvLists,
+        tvListRowCounts,
         scanOrder,
         globalTimeFilter,
         deletionList,

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

@@ -1601,6 +1601,7 @@ public boolean isAllDeleted() {
 
   public AlignedTVListIterator iterator(
       Ordering scanOrder,
+      int rowCount,
       Filter globalTimeFilter,
       List<TSDataType> dataTypeList,
       List<Integer> columnIndexList,
@@ -1612,6 +1613,7 @@ public AlignedTVListIterator iterator(
       int maxNumberOfPointsInPage) {
     return new AlignedTVListIterator(
         scanOrder,
+        rowCount,
         globalTimeFilter,
         dataTypeList,
         columnIndexList,
@@ -1643,6 +1645,7 @@ public class AlignedTVListIterator extends TVListIterator {
 
     public AlignedTVListIterator(
         Ordering scanOrder,
+        int rowCount,
         Filter globalTimeFilter,
         List<TSDataType> dataTypeList,
         List<Integer> columnIndexList,
@@ -1652,7 +1655,7 @@ public AlignedTVListIterator(
         List<TSEncoding> encodingList,
         boolean ignoreAllNullRows,
         int maxNumberOfPointsInPage) {
-      super(scanOrder, globalTimeFilter, null, null, null, maxNumberOfPointsInPage);
+      super(scanOrder, rowCount, globalTimeFilter, null, null, null, maxNumberOfPointsInPage);
       this.dataTypeList = dataTypeList;
       this.columnIndexList =
           (columnIndexList == null)