Skip to content

[to dev/1.3] Optimize memtable region scan #16891

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JackieTien97 merged 1 commit into from
Dec 10, 2025
Merged

[to dev/1.3] Optimize memtable region scan #16891

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions ...-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2161,7 +2161,8 @@ private List<IFileScanHandle> getFileHandleListForQuery(
} else {
tsFileResource
.getProcessor()
.queryForSeriesRegionScanWithoutLock(partialPaths, context, fileScanHandles);
.queryForSeriesRegionScanWithoutLock(
partialPaths, context, fileScanHandles, globalTimeFilter);
}
}
return fileScanHandles;
Expand Down Expand Up @@ -2238,7 +2239,8 @@ private List<IFileScanHandle> getFileHandleListForQuery(
} else {
tsFileResource
.getProcessor()
.queryForDeviceRegionScanWithoutLock(devicePathsToContext, context, fileScanHandles);
.queryForDeviceRegionScanWithoutLock(
devicePathsToContext, context, fileScanHandles, globalTimeFilter);
}
}
return fileScanHandles;
Expand Down
85 changes: 57 additions & 28 deletions ...src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

Expand Down Expand Up @@ -451,7 +452,8 @@ public void queryForSeriesRegionScan(
long ttlLowerBound,
Map<String, List<IChunkMetadata>> chunkMetaDataMap,
Map<String, List<IChunkHandle>> memChunkHandleMap,
List<Pair<Modification, IMemTable>> modsToMemTabled) {
List<Pair<Modification, IMemTable>> modsToMemTabled,
Filter globalTimeFilter) {

IDeviceID deviceID = DeviceIDFactory.getInstance().getDeviceID(fullPath.getDevicePath());

Expand All @@ -469,7 +471,12 @@ public void queryForSeriesRegionScan(
(MeasurementPath) fullPath, this, modsToMemTabled, ttlLowerBound);
}
getMemChunkHandleFromMemTable(
deviceID, measurementId, chunkMetaDataMap, memChunkHandleMap, deletionList);
deviceID,
measurementId,
chunkMetaDataMap,
memChunkHandleMap,
deletionList,
globalTimeFilter);
} else {
if (!memTableMap.containsKey(deviceID)) {
return;
Expand All @@ -486,7 +493,8 @@ public void queryForSeriesRegionScan(
((AlignedPath) fullPath).getSchemaList(),
chunkMetaDataMap,
memChunkHandleMap,
deletionList);
deletionList,
globalTimeFilter);
}
}

Expand All @@ -497,7 +505,8 @@ public void queryForDeviceRegionScan(
long ttlLowerBound,
Map<String, List<IChunkMetadata>> chunkMetadataMap,
Map<String, List<IChunkHandle>> memChunkHandleMap,
List<Pair<Modification, IMemTable>> modsToMemTabled)
List<Pair<Modification, IMemTable>> modsToMemTabled,
Filter globalTimeFilter)
throws IllegalPathException {

Map<IDeviceID, IWritableMemChunkGroup> memTableMap = getMemTableMap();
Expand All @@ -515,15 +524,17 @@ public void queryForDeviceRegionScan(
chunkMetadataMap,
memChunkHandleMap,
ttlLowerBound,
modsToMemTabled);
modsToMemTabled,
globalTimeFilter);
} else {
getMemChunkHandleFromMemTable(
deviceID,
(WritableMemChunkGroup) writableMemChunkGroup,
chunkMetadataMap,
memChunkHandleMap,
ttlLowerBound,
modsToMemTabled);
modsToMemTabled,
globalTimeFilter);
}
}

Expand All @@ -532,32 +543,39 @@ private void getMemChunkHandleFromMemTable(
String measurementId,
Map<String, List<IChunkMetadata>> chunkMetadataMap,
Map<String, List<IChunkHandle>> memChunkHandleMap,
List<TimeRange> deletionList) {
List<TimeRange> deletionList,
Filter globalTimeFilter) {

WritableMemChunk memChunk =
(WritableMemChunk) memTableMap.get(deviceID).getMemChunkMap().get(measurementId);

long[] timestamps = memChunk.getFilteredTimestamp(deletionList);
if (memChunk == null) {
return;
}
Optional<Long> anySatisfiedTimestamp =
memChunk.getAnySatisfiedTimestamp(deletionList, globalTimeFilter);
if (!anySatisfiedTimestamp.isPresent()) {
return;
}
long satisfiedTimestamp = anySatisfiedTimestamp.get();

chunkMetadataMap
.computeIfAbsent(measurementId, k -> new ArrayList<>())
.add(
buildChunkMetaDataForMemoryChunk(
measurementId,
timestamps[0],
timestamps[timestamps.length - 1],
Collections.emptyList()));
buildFakeChunkMetaDataForFakeMemoryChunk(
measurementId, satisfiedTimestamp, satisfiedTimestamp, Collections.emptyList()));
memChunkHandleMap
.computeIfAbsent(measurementId, k -> new ArrayList<>())
.add(new MemChunkHandleImpl(deviceID, measurementId, timestamps));
.add(new MemChunkHandleImpl(deviceID, measurementId, new long[] {satisfiedTimestamp}));
}

private void getMemAlignedChunkHandleFromMemTable(
IDeviceID deviceID,
List<IMeasurementSchema> schemaList,
Map<String, List<IChunkMetadata>> chunkMetadataList,
Map<String, List<IChunkHandle>> memChunkHandleMap,
List<List<TimeRange>> deletionList) {
List<List<TimeRange>> deletionList,
Filter globalTimeFilter) {

AlignedWritableMemChunk alignedMemChunk =
((AlignedWritableMemChunkGroup) memTableMap.get(deviceID)).getAlignedMemChunk();
Expand All @@ -574,7 +592,11 @@ private void getMemAlignedChunkHandleFromMemTable(
}

List<BitMap> bitMaps = new ArrayList<>();
long[] timestamps = alignedMemChunk.getFilteredTimestamp(deletionList, bitMaps);
long[] timestamps =
alignedMemChunk.getAnySatisfiedTimestamp(deletionList, bitMaps, globalTimeFilter);
if (timestamps.length == 0) {
return;
}

buildAlignedMemChunkHandle(
deviceID,
Expand All @@ -592,7 +614,8 @@ private void getMemAlignedChunkHandleFromMemTable(
Map<String, List<IChunkMetadata>> chunkMetadataList,
Map<String, List<IChunkHandle>> memChunkHandleMap,
long ttlLowerBound,
List<Pair<Modification, IMemTable>> modsToMemTabled)
List<Pair<Modification, IMemTable>> modsToMemTabled,
Filter globalTimeFilter)
throws IllegalPathException {

AlignedWritableMemChunk memChunk = writableMemChunkGroup.getAlignedMemChunk();
Expand All @@ -611,7 +634,10 @@ private void getMemAlignedChunkHandleFromMemTable(
}

List<BitMap> bitMaps = new ArrayList<>();
long[] timestamps = memChunk.getFilteredTimestamp(deletionList, bitMaps);
long[] timestamps = memChunk.getAnySatisfiedTimestamp(deletionList, bitMaps, globalTimeFilter);
if (timestamps.length == 0) {
return;
}
buildAlignedMemChunkHandle(
deviceID,
timestamps,
Expand All @@ -628,7 +654,8 @@ private void getMemChunkHandleFromMemTable(
Map<String, List<IChunkMetadata>> chunkMetadataMap,
Map<String, List<IChunkHandle>> memChunkHandleMap,
long ttlLowerBound,
List<Pair<Modification, IMemTable>> modsToMemTabled)
List<Pair<Modification, IMemTable>> modsToMemTabled,
Filter globalTimeFilter)
throws IllegalPathException {

for (Entry<String, IWritableMemChunk> entry :
Expand All @@ -646,18 +673,20 @@ private void getMemChunkHandleFromMemTable(
modsToMemTabled,
ttlLowerBound);
}
long[] timestamps = writableMemChunk.getFilteredTimestamp(deletionList);
Optional<Long> anySatisfiedTimestamp =
writableMemChunk.getAnySatisfiedTimestamp(deletionList, globalTimeFilter);
if (!anySatisfiedTimestamp.isPresent()) {
return;
}
long satisfiedTimestamp = anySatisfiedTimestamp.get();
chunkMetadataMap
.computeIfAbsent(measurementId, k -> new ArrayList<>())
.add(
buildChunkMetaDataForMemoryChunk(
measurementId,
timestamps[0],
timestamps[timestamps.length - 1],
Collections.emptyList()));
buildFakeChunkMetaDataForFakeMemoryChunk(
measurementId, satisfiedTimestamp, satisfiedTimestamp, Collections.emptyList()));
memChunkHandleMap
.computeIfAbsent(measurementId, k -> new ArrayList<>())
.add(new MemChunkHandleImpl(deviceID, measurementId, timestamps));
.add(new MemChunkHandleImpl(deviceID, measurementId, new long[] {satisfiedTimestamp}));
}
}

Expand All @@ -681,7 +710,7 @@ private void buildAlignedMemChunkHandle(
chunkMetadataList
.computeIfAbsent(measurement, k -> new ArrayList<>())
.add(
buildChunkMetaDataForMemoryChunk(
buildFakeChunkMetaDataForFakeMemoryChunk(
measurement, startEndTime[0], startEndTime[1], deletion));
chunkHandleMap
.computeIfAbsent(measurement, k -> new ArrayList<>())
Expand Down Expand Up @@ -712,7 +741,7 @@ private long[] calculateStartEndTime(long[] timestamps, List<BitMap> bitMaps, in
return new long[] {startTime, endTime};
}

private IChunkMetadata buildChunkMetaDataForMemoryChunk(
private IChunkMetadata buildFakeChunkMetaDataForFakeMemoryChunk(
String measurement, long startTime, long endTime, List<TimeRange> deletionList) {
TimeStatistics timeStatistics = new TimeStatistics();
timeStatistics.setStartTime(startTime);
Expand Down
114 changes: 90 additions & 24 deletions ...n/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@

import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.read.filter.basic.Filter;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.utils.Pair;
Expand All @@ -52,6 +53,7 @@
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import static org.apache.iotdb.db.utils.ModificationUtils.isPointDeleted;

Expand Down Expand Up @@ -712,29 +714,89 @@ private Pair<Object[], BitMap[]> checkAndReorderColumnValuesInInsertPlan(
return new Pair<>(reorderedColumnValues, reorderedBitMaps);
}

private void filterDeletedTimeStamp(
public long[] getAnySatisfiedTimestamp(
List<List<TimeRange>> deletionList, List<BitMap> bitMaps, Filter globalTimeFilter) {
BitMap columnHasNonNullValue = new BitMap(schemaList.size());
AtomicInteger hasNonNullValueColumnCount = new AtomicInteger(0);
Map<Long, BitMap> timestampWithBitmap = new TreeMap<>();

getAnySatisfiedTimestamp(
list,
deletionList,
timestampWithBitmap,
globalTimeFilter,
columnHasNonNullValue,
hasNonNullValueColumnCount);
for (int i = 0;
i < sortedList.size() && hasNonNullValueColumnCount.get() < schemaList.size();
i++) {
getAnySatisfiedTimestamp(
sortedList.get(i),
deletionList,
timestampWithBitmap,
globalTimeFilter,
columnHasNonNullValue,
hasNonNullValueColumnCount);
}

long[] timestamps = new long[timestampWithBitmap.size()];
int idx = 0;
for (Map.Entry<Long, BitMap> entry : timestampWithBitmap.entrySet()) {
timestamps[idx++] = entry.getKey();
bitMaps.add(entry.getValue());
}
return timestamps;
}

private void getAnySatisfiedTimestamp(
AlignedTVList alignedTVList,
List<List<TimeRange>> valueColumnsDeletionList,
Map<Long, BitMap> timestampWithBitmap) {
Map<Long, BitMap> timestampWithBitmap,
Filter globalTimeFilter,
BitMap columnHasNonNullValue,
AtomicInteger hasNonNullValueColumnCount) {
if (globalTimeFilter != null
&& !globalTimeFilter.satisfyStartEndTime(
alignedTVList.getMinTime(), alignedTVList.getMaxTime())) {
return;
}
BitMap allValueColDeletedMap = alignedTVList.getAllValueColDeletedMap();

int rowCount = alignedTVList.rowCount();
List<int[]> valueColumnDeleteCursor = new ArrayList<>();
if (valueColumnsDeletionList != null) {
valueColumnsDeletionList.forEach(x -> valueColumnDeleteCursor.add(new int[] {0}));
}

// example:
// globalTimeFilter:null, ignoreAllNullRows: true
// tvList:
// time s1 s2 s3
// 1 1 null null
// 2 null 1 null
// 2 1 1 null
// 3 1 null null
// 4 1 null 1
// timestampWithBitmap:
// 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;

BitMap bitMap = new BitMap(schemaList.size());
for (int column = 0; column < schemaList.size(); column++) {
if (alignedTVList.isNullValue(alignedTVList.getValueIndex(row), column)) {
bitMap.mark(column);
continue;
}

// skip deleted row
Expand All @@ -744,32 +806,36 @@ && isPointDeleted(
timestamp,
valueColumnsDeletionList.get(column),
valueColumnDeleteCursor.get(column))) {
bitMap.mark(column);
}

// skip all-null row
if (bitMap.isAllMarked()) {
continue;
}
timestampWithBitmap.put(timestamp, bitMap);
if (!columnHasNonNullValue.isMarked(column)) {
hasNonNullValueColumnCount.incrementAndGet();
columnHasNonNullValue.mark(column);
currentRowNullValueBitmap =
currentRowNullValueBitmap != null
? currentRowNullValueBitmap
: timestampWithBitmap.computeIfAbsent(
timestamp, k -> getAllMarkedBitmap(schemaList.size()));
currentRowNullValueBitmap.unmark(column);
}
}
}
}

public long[] getFilteredTimestamp(List<List<TimeRange>> deletionList, List<BitMap> bitMaps) {
Map<Long, BitMap> timestampWithBitmap = new TreeMap<>();
if (currentRowNullValueBitmap == null) {
continue;
}
// found new column with non-null value
timestampWithBitmap.put(timestamp, currentRowNullValueBitmap);

filterDeletedTimeStamp(list, deletionList, timestampWithBitmap);
for (AlignedTVList alignedTVList : sortedList) {
filterDeletedTimeStamp(alignedTVList, deletionList, timestampWithBitmap);
if (hasNonNullValueColumnCount.get() == schemaList.size()) {
return;
}
}
}

List<Long> filteredTimestamps = new ArrayList<>();
for (Map.Entry<Long, BitMap> entry : timestampWithBitmap.entrySet()) {
filteredTimestamps.add(entry.getKey());
bitMaps.add(entry.getValue());
}
return filteredTimestamps.stream().mapToLong(Long::valueOf).toArray();
private BitMap getAllMarkedBitmap(int size) {
BitMap bitMap = new BitMap(size);
bitMap.markAll();
return bitMap;
}

// Choose maximum avgPointSizeOfLargestColumn among working and sorted AlignedTVList as
Expand Down
Loading
Loading