[To dev/1.3] Concurrently querying and writing to the memtable may cause the query results out of order

Author: shuwenwei

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

@@ -50,7 +50,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;
@@ -361,21 +360,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,
-            valueColumnsDeletionList,
-            floatPrecision,
-            encodingList,
-            MAX_NUMBER_OF_POINTS_IN_PAGE);
+    MemPointIterator timeValuePairIterator = createMemPointIterator(Ordering.ASC, null);
 
     while (timeValuePairIterator.hasNextTimeValuePair()) {
       TimeValuePair tvPair = timeValuePairIterator.nextTimeValuePair();
@@ -453,14 +438,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,
         valueColumnsDeletionList,

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

@@ -291,17 +291,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();
@@ -378,10 +368,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

@@ -1334,6 +1334,7 @@ public List<List<BitMap>> getBitMaps() {
 
   public AlignedTVListIterator iterator(
       Ordering scanOrder,
+      int rowCount,
       Filter globalTimeFilter,
       List<TSDataType> dataTypeList,
       List<Integer> columnIndexList,
@@ -1343,6 +1344,7 @@ public AlignedTVListIterator iterator(
       int maxNumberOfPointsInPage) {
     return new AlignedTVListIterator(
         scanOrder,
+        rowCount,
         globalTimeFilter,
         dataTypeList,
         columnIndexList,
@@ -1369,14 +1371,15 @@ public class AlignedTVListIterator extends TVListIterator {
 
     public AlignedTVListIterator(
         Ordering scanOrder,
+        int rowCount,
         Filter globalTimeFilter,
         List<TSDataType> dataTypeList,
         List<Integer> columnIndexList,
         List<List<TimeRange>> valueColumnsDeletionList,
         Integer floatPrecision,
         List<TSEncoding> encodingList,
         int maxNumberOfPointsInPage) {
-      super(scanOrder, globalTimeFilter, null, null, null, maxNumberOfPointsInPage);
+      super(scanOrder, rowCount, globalTimeFilter, null, null, null, maxNumberOfPointsInPage);
       this.dataTypeList = dataTypeList;
       this.columnIndexList =
           (columnIndexList == null)