This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new aa0e230a0f5 [to dev/1.3] Fix TVList iterator state across memtable
page switches (#18652)
aa0e230a0f5 is described below
commit aa0e230a0f5342a651bd510c3a4c6ed7b3f8f703
Author: shuwenwei <[email protected]>
AuthorDate: Wed Sep 16 17:35:47 2026 +0800
[to dev/1.3] Fix TVList iterator state across memtable page switches
(#18652)
---
.../iotdb/db/utils/datastructure/TVList.java | 6 +-
.../memtable/AlignedTVListIteratorTest.java | 84 ++++++++++++++++++++++
.../memtable/NonAlignedTVListIteratorTest.java | 70 ++++++++++++++++++
3 files changed, 158 insertions(+), 2 deletions(-)
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
index 2536836069c..48260213140 100644
---
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
@@ -931,9 +931,10 @@ public abstract class TVList implements WALEntryValue {
int newIndex = getScanOrderIndex(indexInTVList);
if (newIndex > index) {
index = newIndex;
+ // If the cursor does not move, a duplicate-timestamp group prepared
for the current
+ // position remains valid. Invalidate it only after the cursor
actually advances.
+ probeNext = false;
}
-
- probeNext = false;
}
protected void prepareNext() {
@@ -1238,6 +1239,7 @@ public abstract class TVList implements WALEntryValue {
// been applied when constructing the tsBlock
TsBlock tsBlock = builder.build();
addTsBlock(tsBlock);
+ probeNext = false;
return tsBlock;
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedTVListIteratorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedTVListIteratorTest.java
index d5a8b49f726..04ea5958a2b 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedTVListIteratorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedTVListIteratorTest.java
@@ -846,6 +846,90 @@ public class AlignedTVListIteratorTest {
paginationController.getCurLimit(),
paginationController.getCurOffset());
}
+ @Test
+ public void testPageSwitchKeepsPreparedDuplicateTimestampValues() throws
IOException {
+ AlignedTVList tvList =
+ AlignedTVList.newAlignedList(
+ Arrays.asList(TSDataType.INT64, TSDataType.BOOLEAN,
TSDataType.BOOLEAN));
+ tvList.putAlignedValue(1, new Object[] {1L, true, false});
+ tvList.putAlignedValue(100, new Object[] {2L, null, false});
+ tvList.putAlignedValue(100, new Object[] {null, true, false});
+
+ Map<TVList, Integer> tvListMap = new LinkedHashMap<>();
+ tvListMap.put(tvList, tvList.rowCount());
+ AlignedReadOnlyMemChunk chunk =
+ new AlignedReadOnlyMemChunk(
+ fragmentInstanceContext,
+ Arrays.asList(0, 1, 2),
+ getMeasurementSchema(),
+ tvListMap,
+ Arrays.asList(
+ Collections.emptyList(), Collections.emptyList(),
Collections.emptyList()));
+ chunk.sortTvLists();
+ chunk.initChunkMetaFromTVListsWithFakeStatistics();
+
+ MemPointIterator iterator = chunk.createMemPointIterator(Ordering.ASC,
null);
+ List<TimeValuePair> result = new ArrayList<>();
+ // These are fake-page boundaries for one MemChunk. The middle page is
empty, but the
+ // shared iterator still receives its time range before its next page is
read.
+ for (TimeRange pageRange :
+ Arrays.asList(new TimeRange(1, 33), new TimeRange(34, 66), new
TimeRange(67, 100))) {
+ iterator.setCurrentPageTimeRange(pageRange);
+ while (iterator.hasNextTimeValuePair()) {
+ result.add(iterator.nextTimeValuePair());
+ }
+ }
+
+ Assert.assertEquals(2, result.size());
+ Assert.assertEquals(1L, result.get(0).getTimestamp());
+ Assert.assertEquals(1L, result.get(0).getValues()[0]);
+ Assert.assertEquals(100L, result.get(1).getTimestamp());
+ Assert.assertEquals(2L, result.get(1).getValues()[0]);
+ Assert.assertEquals(Boolean.TRUE, result.get(1).getValues()[1]);
+ Assert.assertEquals(Boolean.FALSE, result.get(1).getValues()[2]);
+ }
+
+ @Test
+ public void testPageSwitchKeepsPreparedDuplicateTimestampValuesDescending()
throws IOException {
+ AlignedTVList tvList =
+ AlignedTVList.newAlignedList(
+ Arrays.asList(TSDataType.INT64, TSDataType.BOOLEAN,
TSDataType.BOOLEAN));
+ tvList.putAlignedValue(1, new Object[] {null, true, false});
+ tvList.putAlignedValue(1, new Object[] {2L, null, false});
+ tvList.putAlignedValue(100, new Object[] {1L, true, false});
+
+ Map<TVList, Integer> tvListMap = new LinkedHashMap<>();
+ tvListMap.put(tvList, tvList.rowCount());
+ AlignedReadOnlyMemChunk chunk =
+ new AlignedReadOnlyMemChunk(
+ fragmentInstanceContext,
+ Arrays.asList(0, 1, 2),
+ getMeasurementSchema(),
+ tvListMap,
+ Arrays.asList(
+ Collections.emptyList(), Collections.emptyList(),
Collections.emptyList()));
+ chunk.sortTvLists();
+ chunk.initChunkMetaFromTVListsWithFakeStatistics();
+
+ MemPointIterator iterator = chunk.createMemPointIterator(Ordering.DESC,
null);
+ List<TimeValuePair> result = new ArrayList<>();
+ for (TimeRange pageRange :
+ Arrays.asList(new TimeRange(67, 100), new TimeRange(34, 66), new
TimeRange(1, 33))) {
+ iterator.setCurrentPageTimeRange(pageRange);
+ while (iterator.hasNextTimeValuePair()) {
+ result.add(iterator.nextTimeValuePair());
+ }
+ }
+
+ Assert.assertEquals(2, result.size());
+ Assert.assertEquals(100L, result.get(0).getTimestamp());
+ Assert.assertEquals(1L, result.get(0).getValues()[0]);
+ Assert.assertEquals(1L, result.get(1).getTimestamp());
+ Assert.assertEquals(2L, result.get(1).getValues()[0]);
+ Assert.assertEquals(Boolean.TRUE, result.get(1).getValues()[1]);
+ Assert.assertEquals(Boolean.FALSE, result.get(1).getValues()[2]);
+ }
+
@Test
public void testSkipTimeRange() throws QueryProcessException, IOException {
List<Map<TVList, Integer>> list =
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/NonAlignedTVListIteratorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/NonAlignedTVListIteratorTest.java
index b28979efd67..3f63812447a 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/NonAlignedTVListIteratorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/NonAlignedTVListIteratorTest.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateMachine;
import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
+import org.apache.iotdb.db.utils.datastructure.LongTVList;
import org.apache.iotdb.db.utils.datastructure.MemPointIterator;
import org.apache.iotdb.db.utils.datastructure.TVList;
@@ -717,4 +718,73 @@ public class NonAlignedTVListIteratorTest {
}
Assert.assertEquals(expectedTimestamps, resultTimestamps);
}
+
+ @Test
+ public void testBatchToPointAfterEmptyPageKeepsLatestDuplicateValue() throws
IOException {
+ LongTVList tvList = LongTVList.newList();
+ tvList.putLong(1, 1);
+ tvList.putLong(100, 2);
+ tvList.putLong(100, 3);
+
+ MemPointIterator iterator =
+ tvList.iterator(
+ Ordering.ASC,
+ tvList.rowCount(),
+ null,
+ Collections.emptyList(),
+ 0,
+ TSEncoding.PLAIN,
+ 1024);
+
+ iterator.setCurrentPageTimeRange(new TimeRange(1, 33));
+ int firstPageRows = 0;
+ while (iterator.hasNextBatch()) {
+ firstPageRows += iterator.nextBatch().getPositionCount();
+ }
+ Assert.assertEquals(1, firstPageRows);
+
+ iterator.setCurrentPageTimeRange(new TimeRange(34, 66));
+ Assert.assertFalse(iterator.hasNextBatch());
+
+ iterator.setCurrentPageTimeRange(new TimeRange(67, 100));
+ List<Long> result = new ArrayList<>();
+ while (iterator.hasNextTimeValuePair()) {
+ result.add(iterator.nextTimeValuePair().getValue().getLong());
+ }
+ Assert.assertEquals(Collections.singletonList(3L), result);
+ }
+
+ @Test
+ public void testBatchToPointAfterEmptyPageDescendingSkipsDeletedPoint()
throws IOException {
+ LongTVList tvList = LongTVList.newList();
+ tvList.putLong(10, 10);
+ tvList.putLong(100, 100);
+
+ MemPointIterator iterator =
+ tvList.iterator(
+ Ordering.DESC,
+ tvList.rowCount(),
+ null,
+ Collections.singletonList(new TimeRange(10, 10)),
+ 0,
+ TSEncoding.PLAIN,
+ 1024);
+
+ iterator.setCurrentPageTimeRange(new TimeRange(67, 100));
+ int firstPageRows = 0;
+ while (iterator.hasNextBatch()) {
+ firstPageRows += iterator.nextBatch().getPositionCount();
+ }
+ Assert.assertEquals(1, firstPageRows);
+
+ iterator.setCurrentPageTimeRange(new TimeRange(34, 66));
+ Assert.assertFalse(iterator.hasNextBatch());
+
+ iterator.setCurrentPageTimeRange(new TimeRange(1, 33));
+ List<Long> result = new ArrayList<>();
+ while (iterator.hasNextTimeValuePair()) {
+ result.add(iterator.nextTimeValuePair().getValue().getLong());
+ }
+ Assert.assertTrue(result.isEmpty());
+ }
}
