This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/aligned-flush-fast-path in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 9fee974b726ad9fbf56ba7f72235cb7d8344ff9c Author: Caideyipi <[email protected]> AuthorDate: Wed Aug 5 17:49:22 2026 +0800 Optimize aligned memtable flush without deleted measurements --- .../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; + } + } }
