This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 0d94b765427 Optimize aligned memtable flush without deleted
measurements (#18406)
0d94b765427 is described below
commit 0d94b7654278520575b78e0acf26d808271034ed
Author: Caideyipi <[email protected]>
AuthorDate: Thu Aug 6 18:58:09 2026 +0800
Optimize aligned memtable flush without deleted measurements (#18406)
---
.../memtable/AlignedWritableMemChunk.java | 182 +++++++++++++++++++++
.../dataregion/memtable/MemTableFlushTaskTest.java | 64 ++++++++
2 files changed, 246 insertions(+)
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 05fd1d4073c..752e7c92b63 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
@@ -665,6 +665,188 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
BitMap allValueColDeletedMap,
int maxNumberOfPointsInPage,
List<IMeasurementSchema> activeSchemaList) {
+ // Flushing memtables are immutable, so this identity remains stable
during encoding.
+ if (activeSchemaList == schemaList) {
+ handleEncodingWithoutDeletedMeasurements(
+ ioTaskQueue,
+ chunkRange,
+ timeDuplicateInfo,
+ allValueColDeletedMap,
+ maxNumberOfPointsInPage);
+ return;
+ }
+ handleEncodingWithDeletedMeasurements(
+ ioTaskQueue,
+ chunkRange,
+ timeDuplicateInfo,
+ allValueColDeletedMap,
+ maxNumberOfPointsInPage,
+ activeSchemaList);
+ }
+
+ private void handleEncodingWithoutDeletedMeasurements(
+ BlockingQueue<Object> ioTaskQueue,
+ List<List<Integer>> chunkRange,
+ boolean[] timeDuplicateInfo,
+ BitMap allValueColDeletedMap,
+ int maxNumberOfPointsInPage) {
+ AlignedTVList alignedWorkingListForFlush = (AlignedTVList)
workingListForFlush;
+ List<TSDataType> dataTypes = alignedWorkingListForFlush.getTsDataTypes();
+ Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new
Pair[dataTypes.size()];
+ for (List<Integer> pageRange : chunkRange) {
+ AlignedChunkWriterImpl alignedChunkWriter =
+ new AlignedChunkWriterImpl(schemaList, encryptParameter);
+ for (int pageNum = 0; pageNum < pageRange.size() / 2; pageNum += 1) {
+ for (int columnIndex = 0; columnIndex < dataTypes.size();
columnIndex++) {
+ // Pair of Time and Index
+ if (Objects.nonNull(timeDuplicateInfo)
+ && lastValidPointIndexForTimeDupCheck[columnIndex] == null) {
+ lastValidPointIndexForTimeDupCheck[columnIndex] = new
Pair<>(Long.MIN_VALUE, null);
+ }
+ TSDataType tsDataType = dataTypes.get(columnIndex);
+ for (int sortedRowIndex = pageRange.get(pageNum * 2);
+ sortedRowIndex <= pageRange.get(pageNum * 2 + 1);
+ sortedRowIndex++) {
+ // skip empty row
+ if (allValueColDeletedMap != null
+ && allValueColDeletedMap.isMarked(
+ alignedWorkingListForFlush.getValueIndex(sortedRowIndex)))
{
+ continue;
+ }
+ // skip time duplicated rows
+ long time = alignedWorkingListForFlush.getTime(sortedRowIndex);
+ if (Objects.nonNull(timeDuplicateInfo)) {
+ if (!alignedWorkingListForFlush.isNullValue(
+ alignedWorkingListForFlush.getValueIndex(sortedRowIndex),
columnIndex)) {
+ lastValidPointIndexForTimeDupCheck[columnIndex].left = time;
+ lastValidPointIndexForTimeDupCheck[columnIndex].right =
+ alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
+ }
+ if (timeDuplicateInfo[sortedRowIndex]) {
+ continue;
+ }
+ }
+
+ // The part of code solves the following problem:
+ // Time: 1,2,2,3
+ // Value: 1,2,null,null
+ // When rowIndex:1, pair(min,null), timeDuplicateInfo:false,
write(T:1,V:1)
+ // When rowIndex:2, pair(2,2), timeDuplicateInfo:true, skip
writing value
+ // When rowIndex:3, pair(2,2), timeDuplicateInfo:false,
T:2==pair.left:2, write(T:2,V:2)
+ // When rowIndex:4, pair(2,2), timeDuplicateInfo:false,
T:3!=pair.left:2,
+ // write(T:3,V:null)
+
+ int originRowIndex;
+ if
(Objects.nonNull(lastValidPointIndexForTimeDupCheck[columnIndex])
+ && (time ==
lastValidPointIndexForTimeDupCheck[columnIndex].left)) {
+ originRowIndex =
lastValidPointIndexForTimeDupCheck[columnIndex].right;
+ } else {
+ originRowIndex =
alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
+ }
+
+ boolean isNull =
alignedWorkingListForFlush.isNullValue(originRowIndex, columnIndex);
+ switch (tsDataType) {
+ case BOOLEAN:
+ alignedChunkWriter.writeByColumn(
+ time,
+ !isNull
+ && alignedWorkingListForFlush.getBooleanByValueIndex(
+ originRowIndex, columnIndex),
+ isNull);
+ break;
+ case INT32:
+ case DATE:
+ alignedChunkWriter.writeByColumn(
+ time,
+ isNull
+ ? 0
+ : alignedWorkingListForFlush.getIntByValueIndex(
+ originRowIndex, columnIndex),
+ isNull);
+ break;
+ case INT64:
+ case TIMESTAMP:
+ alignedChunkWriter.writeByColumn(
+ time,
+ isNull
+ ? 0
+ : alignedWorkingListForFlush.getLongByValueIndex(
+ originRowIndex, columnIndex),
+ isNull);
+ break;
+ case FLOAT:
+ alignedChunkWriter.writeByColumn(
+ time,
+ isNull
+ ? 0
+ : alignedWorkingListForFlush.getFloatByValueIndex(
+ originRowIndex, columnIndex),
+ isNull);
+ break;
+ case DOUBLE:
+ alignedChunkWriter.writeByColumn(
+ time,
+ isNull
+ ? 0
+ : alignedWorkingListForFlush.getDoubleByValueIndex(
+ originRowIndex, columnIndex),
+ isNull);
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ case OBJECT:
+ alignedChunkWriter.writeByColumn(
+ time,
+ isNull
+ ? null
+ : alignedWorkingListForFlush.getBinaryByValueIndex(
+ originRowIndex, columnIndex),
+ isNull);
+ break;
+ default:
+ break;
+ }
+ }
+ alignedChunkWriter.nextColumn();
+ }
+
+ long[] times =
+ new long[Math.min(maxNumberOfPointsInPage,
alignedWorkingListForFlush.rowCount())];
+ int pointsInPage = 0;
+ for (int sortedRowIndex = pageRange.get(pageNum * 2);
+ sortedRowIndex <= pageRange.get(pageNum * 2 + 1);
+ sortedRowIndex++) {
+ // skip empty row
+ if (((allValueColDeletedMap != null
+ && allValueColDeletedMap.isMarked(
+
alignedWorkingListForFlush.getValueIndex(sortedRowIndex)))
+ || (alignedWorkingListForFlush.isTimeDeleted(sortedRowIndex)))) {
+ continue;
+ }
+ if (Objects.isNull(timeDuplicateInfo) ||
!timeDuplicateInfo[sortedRowIndex]) {
+ times[pointsInPage++] =
alignedWorkingListForFlush.getTime(sortedRowIndex);
+ }
+ }
+ alignedChunkWriter.write(times, pointsInPage, 0);
+ }
+ alignedChunkWriter.sealCurrentPage();
+ alignedChunkWriter.clearPageWriter();
+ try {
+ ioTaskQueue.put(alignedChunkWriter);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ private void handleEncodingWithDeletedMeasurements(
+ BlockingQueue<Object> ioTaskQueue,
+ List<List<Integer>> chunkRange,
+ boolean[] timeDuplicateInfo,
+ BitMap allValueColDeletedMap,
+ int maxNumberOfPointsInPage,
+ List<IMeasurementSchema> activeSchemaList) {
AlignedTVList alignedWorkingListForFlush = (AlignedTVList)
workingListForFlush;
List<Integer> columnIndexList = buildColumnIndexList(activeSchemaList);
Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new
Pair[activeSchemaList.size()];
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java
index 7b98a1ee99c..3c55c65517f 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java
@@ -27,15 +27,24 @@ import org.apache.iotdb.db.utils.constant.TestConstant;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.ChunkMetadata;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
import org.apache.tsfile.write.writer.RestorableTsFileIOWriter;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.LinkedBlockingQueue;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class MemTableFlushTaskTest {
@@ -161,4 +170,59 @@ public class MemTableFlushTaskTest {
assertEquals(TSDataType.BOOLEAN, chunkMetaData.getDataType());
assertEquals(endTime - startTime + 1, chunkMetaData.getNumOfPoints());
}
+
+ @Test
+ public void testAlignedFlushWithoutDeletedMeasurementsSkipsColumnMapping() {
+ TrackingAlignedWritableMemChunk memChunk = createTrackingAlignedMemChunk();
+ memChunk.putAlignedRow(1, new Object[] {1, 1L});
+ memChunk.sortTvListForFlush();
+
+ BlockingQueue<Object> ioTaskQueue = new LinkedBlockingQueue<>();
+ memChunk.encodeWorkingAlignedTVList(ioTaskQueue, 100, 100);
+
+ assertFalse(memChunk.isColumnMappingBuilt());
+ assertFalse(ioTaskQueue.isEmpty());
+ }
+
+ @Test
+ public void testAlignedFlushWithDeletedMeasurementsKeepsColumnMapping() {
+ TrackingAlignedWritableMemChunk memChunk = createTrackingAlignedMemChunk();
+ memChunk.putAlignedRow(1, new Object[] {1, 1L});
+ memChunk.removeColumn("s1");
+ memChunk.sortTvListForFlush();
+
+ BlockingQueue<Object> ioTaskQueue = new LinkedBlockingQueue<>();
+ memChunk.encodeWorkingAlignedTVList(ioTaskQueue, 100, 100);
+
+ assertTrue(memChunk.isColumnMappingBuilt());
+ assertFalse(ioTaskQueue.isEmpty());
+ }
+
+ private TrackingAlignedWritableMemChunk createTrackingAlignedMemChunk() {
+ List<IMeasurementSchema> schemas =
+ new ArrayList<>(
+ Arrays.asList(
+ new MeasurementSchema("s0", TSDataType.INT32,
TSEncoding.PLAIN),
+ new MeasurementSchema("s1", TSDataType.INT64,
TSEncoding.PLAIN)));
+ return new TrackingAlignedWritableMemChunk(schemas);
+ }
+
+ private static class TrackingAlignedWritableMemChunk extends
AlignedWritableMemChunk {
+
+ private boolean columnMappingBuilt;
+
+ private TrackingAlignedWritableMemChunk(List<IMeasurementSchema>
schemaList) {
+ super(schemaList, false);
+ }
+
+ @Override
+ public List<Integer> buildColumnIndexList(List<IMeasurementSchema>
schemaList) {
+ columnMappingBuilt = true;
+ return super.buildColumnIndexList(schemaList);
+ }
+
+ private boolean isColumnMappingBuilt() {
+ return columnMappingBuilt;
+ }
+ }
}
