This is an automated email from the ASF dual-hosted git repository. shuwenwei pushed a commit to branch optimizeMemTableRegionScan0104 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d65c9dcc3c737e21e5a181509160ef3fae5b2d70 Author: shuwenwei <[email protected]> AuthorDate: Sun Jan 4 11:11:36 2026 +0800 Further optimize the efficiency of memtable region scan --- .../memtable/AlignedWritableMemChunk.java | 109 ++++++++++++--------- .../dataregion/memtable/WritableMemChunk.java | 44 ++++++--- .../iotdb/db/utils/datastructure/TVList.java | 4 + .../memtable/WritableMemChunkRegionScanTest.java | 87 ++++++++++++++++ 4 files changed, 184 insertions(+), 60 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java index f42867d4772..4e4e6494113 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java @@ -58,6 +58,7 @@ import java.util.TreeMap; import java.util.concurrent.BlockingQueue; import java.util.concurrent.atomic.AtomicInteger; +import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE; import static org.apache.iotdb.db.utils.ModificationUtils.isPointDeleted; public class AlignedWritableMemChunk extends AbstractWritableMemChunk { @@ -365,62 +366,74 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { // timestamp: 1 bitmap: 011 // timestamp: 2 bitmap: 101 // timestamp: 4 bitmap: 110 - for (int row = 0; row < rowCount; row++) { - // the row is deleted - if (allValueColDeletedMap != null && allValueColDeletedMap.isMarked(row)) { - continue; - } - long timestamp = alignedTVList.getTime(row); - if (globalTimeFilter != null && !globalTimeFilter.satisfy(timestamp, null)) { - continue; - } - - // Note that this method will only perform bitmap unmarking on the first occurrence of a - // non-null value in multiple timestamps for the same column. - BitMap currentRowNullValueBitmap = null; - - for (int column = 0; column < schemaList.size(); column++) { - if (alignedTVList.isNullValue(alignedTVList.getValueIndex(row), column)) { + List<long[]> timestampsList = alignedTVList.getTimestamps(); + List<int[]> indicesList = alignedTVList.getIndices(); + int row = -1; + for (int i = 0; i < timestampsList.size(); i++) { + long[] timestamps = timestampsList.get(i); + int[] indices = indicesList == null ? null : indicesList.get(i); + int limit = (i == timestampsList.size() - 1) ? rowCount - i * ARRAY_SIZE : ARRAY_SIZE; + for (int j = 0; j < limit; j++) { + row++; + // the row is deleted + if (allValueColDeletedMap != null && allValueColDeletedMap.isMarked(row)) { continue; } - - // skip deleted row - if (valueColumnsDeletionList != null - && !valueColumnsDeletionList.isEmpty() - && isPointDeleted( - timestamp, - valueColumnsDeletionList.get(column), - valueColumnDeleteCursor.get(column))) { + long timestamp = timestamps[j]; + if (globalTimeFilter != null && !globalTimeFilter.satisfy(timestamp, null)) { continue; } - if (!columnHasNonNullValue.isMarked(column)) { - hasNonNullValueColumnCount.incrementAndGet(); - columnHasNonNullValue.mark(column); - currentRowNullValueBitmap = + + // Note that this method will only perform bitmap unmarking on the first occurrence of a + // non-null value in multiple timestamps for the same column. + BitMap currentRowNullValueBitmap = null; + for (int column = 0; column < schemaList.size(); column++) { + if (alignedTVList.isNullValue(indices == null ? row : indices[j], column)) { + continue; + } + + // skip deleted row + if (valueColumnsDeletionList != null && !valueColumnsDeletionList.isEmpty()) { + List<TimeRange> columnDeletionList = valueColumnsDeletionList.get(column); + int[] deleteCursor = valueColumnDeleteCursor.get(column); + if (columnDeletionList != null && !columnDeletionList.isEmpty()) { + if (!alignedTVList.isSorted()) { + deleteCursor[0] = 0; + } + if (isPointDeleted(timestamp, columnDeletionList, deleteCursor)) { + continue; + } + } + } + if (!columnHasNonNullValue.isMarked(column)) { + hasNonNullValueColumnCount.incrementAndGet(); + columnHasNonNullValue.mark(column); + currentRowNullValueBitmap = + currentRowNullValueBitmap != null + ? currentRowNullValueBitmap + : timestampWithBitmap.computeIfAbsent( + timestamp, k -> getAllMarkedBitmap(schemaList.size())); + currentRowNullValueBitmap.unmark(column); + } + } + + if (!ignoreAllNullRows) { + timestampWithBitmap.put( + timestamp, currentRowNullValueBitmap != null ? currentRowNullValueBitmap - : timestampWithBitmap.computeIfAbsent( - timestamp, k -> getAllMarkedBitmap(schemaList.size())); - currentRowNullValueBitmap.unmark(column); + : getAllMarkedBitmap(schemaList.size())); + return; } - } - - if (!ignoreAllNullRows) { - timestampWithBitmap.put( - timestamp, - currentRowNullValueBitmap != null - ? currentRowNullValueBitmap - : getAllMarkedBitmap(schemaList.size())); - return; - } - if (currentRowNullValueBitmap == null) { - continue; - } - // found new column with non-null value - timestampWithBitmap.put(timestamp, currentRowNullValueBitmap); + if (currentRowNullValueBitmap == null) { + continue; + } + // found new column with non-null value + timestampWithBitmap.put(timestamp, currentRowNullValueBitmap); - if (hasNonNullValueColumnCount.get() == schemaList.size()) { - return; + if (hasNonNullValueColumnCount.get() == schemaList.size()) { + return; + } } } } 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 index 6499d33fcd9..0256d5b16e5 100644 --- 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 @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; +import org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager; import org.apache.iotdb.db.utils.ModificationUtils; import org.apache.iotdb.db.utils.datastructure.BatchEncodeInfo; import org.apache.iotdb.db.utils.datastructure.MemPointIterator; @@ -599,19 +600,38 @@ public class WritableMemChunk extends AbstractWritableMemChunk { && !globalTimeFilter.satisfyStartEndTime(tvlist.getMinTime(), tvlist.getMaxTime())) { return Optional.empty(); } - for (int i = 0; i < rowCount; i++) { - if (tvlist.getBitMap() != null && tvlist.isNullValue(tvlist.getValueIndex(i))) { - continue; - } - long curTime = tvlist.getTime(i); - if (deletionList != null - && ModificationUtils.isPointDeleted(curTime, deletionList, deletionCursor)) { - continue; - } - if (globalTimeFilter != null && !globalTimeFilter.satisfy(curTime, null)) { - continue; + + List<long[]> timestampsList = tvlist.getTimestamps(); + List<BitMap> bitMaps = tvlist.getBitMap(); + List<int[]> indicesList = tvlist.getIndices(); + for (int i = 0; i < timestampsList.size(); i++) { + long[] timestamps = timestampsList.get(i); + BitMap bitMap = bitMaps == null ? null : bitMaps.get(i); + int[] indices = indicesList == null ? null : indicesList.get(i); + int limit = + (i == timestampsList.size() - 1) + ? rowCount - i * PrimitiveArrayManager.ARRAY_SIZE + : PrimitiveArrayManager.ARRAY_SIZE; + for (int j = 0; j < limit; j++) { + if (bitMap != null + && (indices == null ? bitMap.isMarked(j) : tvlist.isNullValue(indices[j]))) { + continue; + } + long curTime = timestamps[j]; + if (deletionList != null && !deletionList.isEmpty()) { + if (!tvlist.isSorted()) { + deletionCursor[0] = 0; + } + if (ModificationUtils.isPointDeleted(curTime, deletionList, deletionCursor)) { + continue; + } + } + if (globalTimeFilter != null && !globalTimeFilter.satisfy(curTime, null)) { + continue; + } + + return Optional.of(curTime); } - return Optional.of(curTime); } return Optional.empty(); } 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 dd5f10f37d4..2dcd74b09ff 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 @@ -745,6 +745,10 @@ public abstract class TVList implements WALEntryValue { return timestamps; } + public List<int[]> getIndices() { + return indices; + } + public void setOwnerQuery(QueryContext queryCtx) { this.ownerQuery = queryCtx; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkRegionScanTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkRegionScanTest.java index 3967409db52..3bb0f4a8a10 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkRegionScanTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkRegionScanTest.java @@ -19,9 +19,12 @@ package org.apache.iotdb.db.storageengine.dataregion.memtable; +import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.path.AlignedFullPath; +import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.NonAlignedFullPath; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.read.filescan.IChunkHandle; import org.apache.tsfile.enums.TSDataType; @@ -46,6 +49,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Scanner; @RunWith(Parameterized.class) public class WritableMemChunkRegionScanTest { @@ -238,6 +242,53 @@ public class WritableMemChunkRegionScanTest { } } + @Test + public void testAlignedWritableMemChunkRegionScan2() throws IllegalPathException { + PrimitiveMemTable memTable = new PrimitiveMemTable("root.test", "0"); + try { + List<IMeasurementSchema> measurementSchemas = + Arrays.asList( + new MeasurementSchema("s1", TSDataType.INT32), + new MeasurementSchema("s2", TSDataType.INT32), + new MeasurementSchema("s3", TSDataType.INT32)); + AlignedWritableMemChunk writableMemChunk = null; + for (int i = 1000; i < 2000; i++) { + memTable.writeAlignedRow( + new StringArrayDeviceID("root.test.d1"), measurementSchemas, i, new Object[] {i, i, i}); + } + for (int i = 1; i < 100; i++) { + memTable.writeAlignedRow( + new StringArrayDeviceID("root.test.d1"), + measurementSchemas, + i, + new Object[] {i, null, i}); + } + + memTable.delete( + new TreeDeletionEntry( + new MeasurementPath(new StringArrayDeviceID("root.test.d1"), "s1"), + new TimeRange(1, 1500))); + writableMemChunk = + (AlignedWritableMemChunk) + memTable.getWritableMemChunk(new StringArrayDeviceID("root.test.d1"), ""); + writableMemChunk.sortTvListForFlush(); + List<BitMap> bitMaps = new ArrayList<>(); + long[] timestamps = + writableMemChunk.getAnySatisfiedTimestamp( + Arrays.asList( + Collections.emptyList(), Collections.emptyList(), Collections.emptyList()), + bitMaps, + true, + null); + Assert.assertEquals(3, timestamps.length); + Assert.assertEquals(1, timestamps[0]); + Assert.assertEquals(1000, timestamps[1]); + Assert.assertEquals(1501, timestamps[2]); + } finally { + memTable.release(); + } + } + @Test public void testTableWritableMemChunkRegionScan() { List<IMeasurementSchema> measurementSchemas = @@ -356,4 +407,40 @@ public class WritableMemChunkRegionScanTest { memTable.release(); } } + + @Test + public void testNonAlignedWritableMemChunkRegionScan2() throws IllegalPathException { + PrimitiveMemTable memTable = new PrimitiveMemTable("root.test", "0"); + try { + MeasurementSchema measurementSchema = new MeasurementSchema("s1", TSDataType.INT32); + for (int i = 1000; i < 2000; i++) { + memTable.write( + new StringArrayDeviceID("root.test.d1"), + Collections.singletonList(measurementSchema), + i, + new Object[] {i}); + } + for (int i = 1; i < 100; i++) { + memTable.write( + new StringArrayDeviceID("root.test.d1"), + Collections.singletonList(measurementSchema), + i, + new Object[] {i}); + } + memTable.delete( + new TreeDeletionEntry( + new MeasurementPath(new StringArrayDeviceID("root.test.d1"), "s1"), + new TimeRange(1, 1500))); + WritableMemChunk writableMemChunk = + (WritableMemChunk) + memTable.getWritableMemChunk(new StringArrayDeviceID("root.test.d1"), "s1"); + writableMemChunk.sortTvListForFlush(); + Optional<Long> timestamp = writableMemChunk.getAnySatisfiedTimestamp(null, null); + Assert.assertTrue(timestamp.isPresent()); + Assert.assertEquals(1501, timestamp.get().longValue()); + + } finally { + memTable.release(); + } + } }
