[To dev/1.3] Fix the AlignedTVListIterator / memtable read index problem

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

@@ -1454,8 +1454,7 @@ protected void prepareNext() {
       while (index < rows && !findValidRow) {
         // all columns values are deleted
         int convertedScanOrderValueIndex = getValueIndex(getScanOrderIndex(index));
-        if ((allValueColDeletedMap != null
-            && allValueColDeletedMap.isMarked(convertedScanOrderValueIndex))) {
+        if (isAllValueColumnsDeleted(convertedScanOrderValueIndex)) {
           index++;
           continue;
         }
@@ -1488,9 +1487,7 @@ protected void prepareNext() {
         while (index + 1 < rows
             && getTime(getScanOrderIndex(index + 1)) == getTime(getScanOrderIndex(index))) {
           index++;
-          // skip all-Null rows if allValueColDeletedMap exists
-          if (allValueColDeletedMap == null
-              || !allValueColDeletedMap.isMarked(getValueIndex(getScanOrderIndex(index)))) {
+          if (!isAllValueColumnsDeleted(getValueIndex(getScanOrderIndex(index)))) {
             for (int columnIndex = 0; columnIndex < dataTypeList.size(); columnIndex++) {
               if (!scanOrder.isAscending() && selectedIndices[columnIndex] != -1) {
                 // non -1 value means it already set the latest point index
@@ -1656,8 +1653,7 @@ public TsBlock nextBatch() {
           break;
         }
         // skip invalid row
-        if ((allValueColDeletedMap != null
-                && allValueColDeletedMap.isMarked(getValueIndex(getScanOrderIndex(index))))
+        if (isAllValueColumnsDeleted(getValueIndex(getScanOrderIndex(index)))
             || !isTimeSatisfied(time)) {
           timeInvalidInfo =
               timeInvalidInfo == null
@@ -1668,9 +1664,7 @@ public TsBlock nextBatch() {
         }
         int nextRowIndex = index + 1;
         while (nextRowIndex < rows
-            && ((allValueColDeletedMap != null
-                    && allValueColDeletedMap.isMarked(
-                        getValueIndex(getScanOrderIndex(nextRowIndex))))
+            && (isAllValueColumnsDeleted(getValueIndex(getScanOrderIndex(nextRowIndex)))
                 || !isTimeSatisfied(getTime(getScanOrderIndex(nextRowIndex))))) {
           timeInvalidInfo =
               timeInvalidInfo == null
@@ -1925,14 +1919,12 @@ public void encodeBatch(IChunkWriter chunkWriter, BatchEncodeInfo encodeInfo, lo
           break;
         }
         // skip empty row
-        if (allValueColDeletedMap != null && allValueColDeletedMap.isMarked(getValueIndex(index))) {
+        if (isAllValueColumnsDeleted(getValueIndex(index))) {
           continue;
         }
 
         int nextRowIndex = index + 1;
-        while (nextRowIndex < rows
-            && (allValueColDeletedMap != null
-                && allValueColDeletedMap.isMarked(getValueIndex(nextRowIndex)))) {
+        while (nextRowIndex < rows && isAllValueColumnsDeleted(getValueIndex(nextRowIndex))) {
           nextRowIndex++;
         }
         long time = getTime(index);
@@ -1962,8 +1954,7 @@ public void encodeBatch(IChunkWriter chunkWriter, BatchEncodeInfo encodeInfo, lo
         }
         for (int sortedRowIndex = startIndex; sortedRowIndex < index; sortedRowIndex++) {
           // skip empty row
-          if (allValueColDeletedMap != null
-              && allValueColDeletedMap.isMarked(getValueIndex(sortedRowIndex))) {
+          if (isAllValueColumnsDeleted(getValueIndex(sortedRowIndex))) {
             continue;
           }
           long time = getTime(sortedRowIndex);
@@ -2044,6 +2035,14 @@ public int[] getSelectedIndices() {
       return selectedIndices;
     }
 
+    private boolean isAllValueColumnsDeleted(int valueIndex) {
+      // A sorted row-count snapshot can point to value indices appended after the snapshot.
+      return allValueColDeletedMap != null
+          && valueIndex >= 0
+          && valueIndex < allValueColDeletedMap.getSize()
+          && allValueColDeletedMap.isMarked(valueIndex);
+    }
+
     public int getSelectedIndex(int column) {
       return selectedIndices[column];
     }

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedTVListIteratorTest.java

@@ -166,6 +166,40 @@ public void otherTest() throws IOException {
     tvListMap = buildAlignedSingleTvListMap(Collections.singletonList(new TimeRange(1, 10)));
   }
 
+  @Test
+  public void testIteratorWithStaleRowCountAfterSort() {
+    AlignedTVList tvList =
+        AlignedTVList.newAlignedList(
+            Arrays.asList(TSDataType.INT32, TSDataType.INT32, TSDataType.INT32));
+    for (int i = 100; i < 110; i++) {
+      tvList.putAlignedValue(i, new Object[] {i, i, i});
+    }
+    int staleRowCount = tvList.rowCount();
+    for (int i = 1; i <= 2; i++) {
+      tvList.putAlignedValue(i, new Object[] {i, i, i});
+    }
+    tvList.delete(1, 2);
+    tvList.sort();
+
+    AlignedTVList.AlignedTVListIterator iterator =
+        tvList.iterator(
+            Ordering.ASC,
+            staleRowCount,
+            null,
+            Arrays.asList(TSDataType.INT32, TSDataType.INT32, TSDataType.INT32),
+            Arrays.asList(0, 1, 2),
+            null,
+            null,
+            null,
+            100);
+    int count = 0;
+    while (iterator.hasNextTimeValuePair()) {
+      iterator.nextTimeValuePair();
+      count++;
+    }
+    Assert.assertEquals(staleRowCount - 2, count);
+  }
+
   @Test
   public void testAlignedWithDeletionsInTVList() throws IOException {
     Map<TVList, Integer> tvListMap =