From dfac357eb57e14e978b88d07c368ea8fb1ddedbf Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 29 Jul 2026 10:04:39 +0800 Subject: [PATCH 1/3] [Fix] Reconcile actual aligned TVList memory usage (#18330) --- .../dataregion/memtable/TsFileProcessor.java | 245 +++++++++++---- .../db/utils/datastructure/AlignedTVList.java | 87 ++++-- .../iotdb/db/utils/datastructure/TVList.java | 13 +- ...BitmapMemoryAccountingPerformanceTest.java | 295 ++++++++++++++++++ .../memtable/TsFileProcessorTest.java | 126 +++++++- .../datastructure/AlignedTVListTest.java | 44 +++ 6 files changed, 731 insertions(+), 79 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index ab2066c1c6ad2..708e2b2a5f73e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -104,9 +104,11 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.CopyOnWriteArrayList; @@ -292,6 +294,10 @@ public void insert(InsertRowNode insertRowNode, long[] infoForMetrics) ensureMemTable(infoForMetrics); workMemTable.checkDataType(insertRowNode); + AlignedTVListRamCostSnapshot alignedRamCostSnapshot = + insertRowNode.isAligned() + ? new AlignedTVListRamCostSnapshot(workMemTable, insertRowNode.getDeviceID()) + : null; long[] memIncrements; long memControlStartTime = System.nanoTime(); @@ -356,11 +362,15 @@ public void insert(InsertRowNode insertRowNode, long[] infoForMetrics) insertRowNode, tsFileResource); - int pointInserted; - if (insertRowNode.isAligned()) { - pointInserted = workMemTable.insertAlignedRow(insertRowNode); - } else { - pointInserted = workMemTable.insert(insertRowNode); + int pointInserted = 0; + try { + if (insertRowNode.isAligned()) { + pointInserted = workMemTable.insertAlignedRow(insertRowNode); + } else { + pointInserted = workMemTable.insert(insertRowNode); + } + } finally { + reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements[0]); } // Update start time of this memtable @@ -385,6 +395,17 @@ public void insertRows(InsertRowsNode insertRowsNode, long[] infoForMetrics) workMemTable.checkDataType(insertRowsNode); long[] memIncrements; + long alignedMemTableIncrement = 0; + Set alignedDeviceIds = new HashSet<>(); + for (InsertRowNode insertRowNode : insertRowsNode.getInsertRowNodeList()) { + if (insertRowNode.isAligned()) { + alignedDeviceIds.add(insertRowNode.getDeviceID()); + } + } + AlignedTVListRamCostSnapshot alignedRamCostSnapshot = + alignedDeviceIds.isEmpty() + ? null + : new AlignedTVListRamCostSnapshot(workMemTable, alignedDeviceIds); long memControlStartTime = System.nanoTime(); if (insertRowsNode.isMixingAlignment()) { @@ -398,7 +419,14 @@ public void insertRows(InsertRowsNode insertRowsNode, long[] infoForMetrics) } } long[] alignedMemIncrements = checkAlignedMemCostAndAddToTspInfoForRows(alignedList); - long[] nonAlignedMemIncrements = checkMemCostAndAddToTspInfoForRows(nonAlignedList); + alignedMemTableIncrement = alignedMemIncrements[0]; + final long[] nonAlignedMemIncrements; + try { + nonAlignedMemIncrements = checkMemCostAndAddToTspInfoForRows(nonAlignedList); + } catch (final WriteProcessException e) { + rollbackMemoryInfoIfNeeded(alignedMemIncrements); + throw e; + } memIncrements = new long[3]; for (int i = 0; i < 3; i++) { memIncrements[i] = alignedMemIncrements[i] + nonAlignedMemIncrements[i]; @@ -407,6 +435,7 @@ public void insertRows(InsertRowsNode insertRowsNode, long[] infoForMetrics) if (insertRowsNode.isAligned()) { memIncrements = checkAlignedMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList()); + alignedMemTableIncrement = memIncrements[0]; } else { memIncrements = checkMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList()); } @@ -456,19 +485,24 @@ public void insertRows(InsertRowsNode insertRowsNode, long[] infoForMetrics) tsFileResource); int pointInserted = 0; - for (InsertRowNode insertRowNode : insertRowsNode.getInsertRowNodeList()) { - if (insertRowNode.isAligned()) { - pointInserted += workMemTable.insertAlignedRow(insertRowNode); - } else { - pointInserted += workMemTable.insert(insertRowNode); - } - // update start time of this memtable - tsFileResource.updateStartTime(insertRowNode.getDeviceID(), insertRowNode.getTime()); - // for sequence tsfile, we update the endTime only when the file is prepared to be closed. - // for unsequence tsfile, we have to update the endTime for each insertion. - if (!sequence) { - tsFileResource.updateEndTime(insertRowNode.getDeviceID(), insertRowNode.getTime()); + try { + for (InsertRowNode insertRowNode : insertRowsNode.getInsertRowNodeList()) { + if (insertRowNode.isAligned()) { + pointInserted += workMemTable.insertAlignedRow(insertRowNode); + } else { + pointInserted += workMemTable.insert(insertRowNode); + } + + // update start time of this memtable + tsFileResource.updateStartTime(insertRowNode.getDeviceID(), insertRowNode.getTime()); + // for sequence tsfile, we update the endTime only when the file is prepared to be closed. + // for unsequence tsfile, we have to update the endTime for each insertion. + if (!sequence) { + tsFileResource.updateEndTime(insertRowNode.getDeviceID(), insertRowNode.getTime()); + } } + } finally { + reconcileAlignedTVListRamCost(alignedRamCostSnapshot, alignedMemTableIncrement); } tsFileResource.updateProgressIndex(insertRowsNode.getProgressIndex()); @@ -583,6 +617,20 @@ public void insertTablet( ensureMemTable(infoForMetrics); workMemTable.checkDataType(insertTabletNode); + Set alignedDeviceIds = new HashSet<>(); + if (insertTabletNode.isAligned()) { + for (int[] range : rangeList) { + for (Pair deviceEndPosition : + insertTabletNode.splitByDevice(range[0], range[1])) { + alignedDeviceIds.add(deviceEndPosition.getLeft()); + } + } + } + AlignedTVListRamCostSnapshot alignedRamCostSnapshot = + alignedDeviceIds.isEmpty() + ? null + : new AlignedTVListRamCostSnapshot(workMemTable, alignedDeviceIds); + long[] memIncrements = scheduleMemoryBlock(insertTabletNode, rangeList, results, infoForMetrics); @@ -628,52 +676,63 @@ public void insertTablet( tsFileResource); int pointInserted = 0; - for (int[] rangePair : rangeList) { - int start = rangePair[0]; - int end = rangePair[1]; - try { - if (insertTabletNode.isAligned()) { - pointInserted += - workMemTable.insertAlignedTablet( - insertTabletNode, start, end, noFailure ? null : results); - } else { - pointInserted += workMemTable.insertTablet(insertTabletNode, start, end); + try { + for (int rangeIndex = 0; rangeIndex < rangeList.size(); rangeIndex++) { + final int[] rangePair = rangeList.get(rangeIndex); + int start = rangePair[0]; + int end = rangePair[1]; + try { + if (insertTabletNode.isAligned()) { + pointInserted += + workMemTable.insertAlignedTablet( + insertTabletNode, start, end, noFailure ? null : results); + } else { + pointInserted += workMemTable.insertTablet(insertTabletNode, start, end); + } + } catch (final WriteProcessException e) { + final TSStatus failureStatus = RpcUtils.getStatus(e.getErrorCode(), e.getMessage()); + for (int failedRangeIndex = rangeIndex; + failedRangeIndex < rangeList.size(); + failedRangeIndex++) { + final int[] failedRange = rangeList.get(failedRangeIndex); + for (int i = failedRange[0]; i < failedRange[1]; i++) { + results[i] = failureStatus; + } + } + throw e; } - } catch (WriteProcessException e) { for (int i = start; i < end; i++) { - results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage()); - } - throw new WriteProcessException(e); - } - for (int i = start; i < end; i++) { - if (results[i] == null - || results[i].getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - results[i] = RpcUtils.SUCCESS_STATUS; + if (results[i] == null + || results[i].getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + results[i] = RpcUtils.SUCCESS_STATUS; + } } - } - final List> deviceEndOffsetPairs = - insertTabletNode.splitByDevice(start, end); - tsFileResource.updateStartTime( - deviceEndOffsetPairs.get(0).left, insertTabletNode.getTimes()[start]); - if (!sequence) { - // For sequence tsfile, we update the endTime only when the file is prepared to be closed. - // For unsequence tsfile, we have to update the endTime for each insertion. - tsFileResource.updateEndTime( - deviceEndOffsetPairs.get(0).left, - insertTabletNode.getTimes()[deviceEndOffsetPairs.get(0).right - 1]); - } - for (int i = 1; i < deviceEndOffsetPairs.size(); i++) { - // the end offset of i - 1 is the start offset of i + final List> deviceEndOffsetPairs = + insertTabletNode.splitByDevice(start, end); tsFileResource.updateStartTime( - deviceEndOffsetPairs.get(i).left, - insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i - 1).right]); + deviceEndOffsetPairs.get(0).left, insertTabletNode.getTimes()[start]); if (!sequence) { + // For sequence tsfile, we update the endTime only when the file is prepared to be closed. + // For unsequence tsfile, we have to update the endTime for each insertion. tsFileResource.updateEndTime( + deviceEndOffsetPairs.get(0).left, + insertTabletNode.getTimes()[deviceEndOffsetPairs.get(0).right - 1]); + } + for (int i = 1; i < deviceEndOffsetPairs.size(); i++) { + // the end offset of i - 1 is the start offset of i + tsFileResource.updateStartTime( deviceEndOffsetPairs.get(i).left, - insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i).right - 1]); + insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i - 1).right]); + if (!sequence) { + tsFileResource.updateEndTime( + deviceEndOffsetPairs.get(i).left, + insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i).right - 1]); + } } } + } finally { + reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements[0]); } tsFileResource.updateProgressIndex(insertTabletNode.getProgressIndex()); @@ -1169,6 +1228,75 @@ private static boolean isFieldMeasurement( && columnCategories[index] == TsTableColumnCategory.FIELD); } + private void reconcileAlignedTVListRamCost( + AlignedTVListRamCostSnapshot snapshot, long estimatedMemTableIncrement) { + if (snapshot == null) { + return; + } + + long correction = snapshot.getMemoryCorrection(estimatedMemTableIncrement); + if (correction > 0) { + dataRegionInfo.addStorageGroupMemCost(correction); + snapshot.memTable.addTVListRamCost(correction); + } else if (correction < 0) { + long releasedMemory = -correction; + dataRegionInfo.releaseStorageGroupMemCost(releasedMemory); + snapshot.memTable.releaseTVListRamCost(releasedMemory); + SystemInfo.getInstance().resetStorageGroupStatus(dataRegionInfo); + } + } + + static final class AlignedTVListRamCostSnapshot { + + private final IMemTable memTable; + private final IDeviceID deviceId; + private final Set deviceIds; + private final long ramCostBeforeWrite; + + AlignedTVListRamCostSnapshot(IMemTable memTable, IDeviceID deviceId) { + this.memTable = memTable; + this.deviceId = deviceId; + this.deviceIds = null; + this.ramCostBeforeWrite = getRamCost(memTable, deviceId); + } + + AlignedTVListRamCostSnapshot(IMemTable memTable, Set deviceIds) { + this.memTable = memTable; + this.deviceId = null; + this.deviceIds = deviceIds; + this.ramCostBeforeWrite = getRamCost(memTable, deviceIds); + } + + long getMemoryCorrection(long estimatedMemTableIncrement) { + return (deviceId == null ? getRamCost(memTable, deviceIds) : getRamCost(memTable, deviceId)) + - ramCostBeforeWrite + - estimatedMemTableIncrement; + } + + private static long getRamCost(IMemTable memTable, Set deviceIds) { + long ramCost = 0; + for (IDeviceID currentDeviceId : deviceIds) { + ramCost += getRamCost(memTable, currentDeviceId); + } + return ramCost; + } + + private static long getRamCost(IMemTable memTable, IDeviceID deviceId) { + IWritableMemChunk memChunk = + memTable.getWritableMemChunk(deviceId, AlignedPath.VECTOR_PLACEHOLDER); + if (!(memChunk instanceof AlignedWritableMemChunk)) { + return 0; + } + + AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk) memChunk; + long ramCost = alignedMemChunk.getWorkingTVList().getRamSize(); + for (AlignedTVList sortedTVList : alignedMemChunk.getSortedList()) { + ramCost += sortedTVList.getRamSize(); + } + return ramCost; + } + } + private void updateMemoryInfo( long memTableIncrement, long chunkMetadataIncrement, long textDataIncrement) throws WriteProcessRejectException { @@ -1221,6 +1349,15 @@ private void rollbackMemoryInfo(long[] memIncrements) { workMemTable.releaseTextDataSize(textDataIncrement); } + private void rollbackMemoryInfoIfNeeded(final long[] memIncrements) { + for (final long memIncrement : memIncrements) { + if (memIncrement != 0) { + rollbackMemoryInfo(memIncrements); + return; + } + } + } + /** * Delete data which belongs to the timeseries `deviceId.measurementId` and the timestamp of which * <= 'timestamp' in the deletion.
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java index 0e44a3dbde5d3..068b4f0cc066c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java @@ -73,10 +73,9 @@ public abstract class AlignedTVList extends TVList { - private static final long BITMAP_RAM_COST_PER_BLOCK = + private static final long BITMAP_RAM_COST = RamUsageEstimator.shallowSizeOfInstance(BitMap.class) - + RamUsageEstimator.sizeOfByteArray(BitMap.getSizeOfBytes(ARRAY_SIZE)) - + NUM_BYTES_OBJECT_REF; + + RamUsageEstimator.sizeOfByteArray(BitMap.getSizeOfBytes(ARRAY_SIZE)); // Data types of this aligned tvList protected List dataTypes; @@ -84,6 +83,9 @@ public abstract class AlignedTVList extends TVList { // Record total memory size of binary column protected long[] memoryBinaryChunkSize; + private long materializedBitmapMemoryCost; + private long arrayMemCostWithoutIndex; + // Data type list -> list of TVList, add 1 when expanded -> primitive array of basic type // Index relation: columnIndex(dataTypeIndex) -> arrayIndex -> elementIndex protected List> values; @@ -106,6 +108,7 @@ public abstract class AlignedTVList extends TVList { super(); dataTypes = types; memoryBinaryChunkSize = new long[dataTypes.size()]; + refreshArrayMemCostWithoutIndex(); values = new ArrayList<>(types.size()); for (int i = 0; i < types.size(); i++) { @@ -170,7 +173,7 @@ public TVList getTvListByColumnIndex( alignedTvList.allValueColDeletedMap = ignoreAllNullRows ? getAllValueColDeletedMap() : null; alignedTvList.timeColDeletedMap = this.timeColDeletedMap; alignedTvList.timeDeletedCnt = this.timeDeletedCnt; - + alignedTvList.materializedBitmapMemoryCost = calculateBitmapRamCost(bitMaps); return alignedTvList; } @@ -183,6 +186,7 @@ public synchronized AlignedTVList cloneForFlushSort() { cloneList.values = this.values; cloneList.bitMaps = this.bitMaps; cloneList.timeColDeletedMap = this.timeColDeletedMap; + cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost; return cloneList; } @@ -218,6 +222,7 @@ public synchronized AlignedTVList clone() { } } cloneList.timeColDeletedMap = timeColDeletedMap == null ? null : timeColDeletedMap.clone(); + cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost; return cloneList; } @@ -446,6 +451,9 @@ public void extendColumn(TSDataType dataType) { this.bitMaps.add(columnBitMaps); this.values.add(columnValue); this.dataTypes.add(dataType); + materializedBitmapMemoryCost += + (long) columnBitMaps.size() * (bitmapReferenceRamCost() + bitmapRamCost()); + refreshArrayMemCostWithoutIndex(); long[] tmpValueChunkRawSize = memoryBinaryChunkSize; memoryBinaryChunkSize = new long[dataTypes.size()]; @@ -737,10 +745,13 @@ public void deleteColumn(int columnIndex) { columnBitMaps.add(new BitMap(ARRAY_SIZE)); } bitMaps.set(columnIndex, columnBitMaps); + materializedBitmapMemoryCost += + (long) columnBitMaps.size() * (bitmapReferenceRamCost() + bitmapRamCost()); } for (int i = 0; i < bitMaps.get(columnIndex).size(); i++) { if (bitMaps.get(columnIndex).get(i) == null) { bitMaps.get(columnIndex).set(i, new BitMap(ARRAY_SIZE)); + materializedBitmapMemoryCost += bitmapRamCost(); } bitMaps.get(columnIndex).get(i).markAll(); } @@ -812,6 +823,7 @@ protected void clearBitMap() { } } } + materializedBitmapMemoryCost = 0; } @Override @@ -823,6 +835,7 @@ protected void expandValues() { values.get(i).add(getPrimitiveArraysByType(dataTypes.get(i))); if (bitMaps != null && bitMaps.get(i) != null) { bitMaps.get(i).add(null); + materializedBitmapMemoryCost += bitmapReferenceRamCost(); } } } @@ -1073,11 +1086,13 @@ private BitMap getBitMap(int columnIndex, int arrayIndex) { columnBitMaps.add(null); } bitMaps.set(columnIndex, columnBitMaps); + materializedBitmapMemoryCost += (long) columnBitMaps.size() * bitmapReferenceRamCost(); } // if the bitmap in arrayIndex is null, init the bitmap if (bitMaps.get(columnIndex).get(arrayIndex) == null) { bitMaps.get(columnIndex).set(arrayIndex, new BitMap(ARRAY_SIZE)); + materializedBitmapMemoryCost += bitmapRamCost(); } return bitMaps.get(columnIndex).get(arrayIndex); @@ -1111,7 +1126,44 @@ public TSDataType getDataType() { @Override public synchronized RamInfo calculateRamSize() { return new RamInfo( - timestamps.size(), alignedTvListArrayMemCost(), rowCount, new ArrayList<>(dataTypes)); + timestamps.size(), + alignedTvListArrayMemCost(), + getRamSize(), + rowCount, + new ArrayList<>(dataTypes)); + } + + public synchronized long getRamSize() { + return (long) timestamps.size() + * (arrayMemCostWithoutIndex + + (indices != null ? (long) PrimitiveArrayManager.ARRAY_SIZE * Integer.BYTES : 0)) + + materializedBitmapMemoryCost; + } + + private void refreshArrayMemCostWithoutIndex() { + arrayMemCostWithoutIndex = alignedTvListArrayMemCost(); + if (indices != null) { + arrayMemCostWithoutIndex -= (long) PrimitiveArrayManager.ARRAY_SIZE * Integer.BYTES; + } + } + + private static long calculateBitmapRamCost(List> bitMaps) { + if (bitMaps == null) { + return 0; + } + long size = 0; + for (List columnBitMaps : bitMaps) { + if (columnBitMaps == null) { + continue; + } + size += (long) columnBitMaps.size() * bitmapReferenceRamCost(); + for (BitMap bitMap : columnBitMaps) { + if (bitMap != null) { + size += bitmapRamCost(); + } + } + } + return size; } /** @@ -1131,7 +1183,6 @@ public static long alignedTvListArrayMemCost( if (type != null && (columnCategories == null || columnCategories[i] == TsTableColumnCategory.FIELD)) { size += (long) ARRAY_SIZE * (long) type.getDataTypeSize(); - size += BITMAP_RAM_COST_PER_BLOCK; measurementColumnNum++; } } @@ -1155,12 +1206,11 @@ public static long alignedTvListArrayMemCost( */ public long alignedTvListArrayMemCost() { long size = 0; - // value & bitmap array mem size + // value array mem size for (int column = 0; column < dataTypes.size(); column++) { TSDataType type = dataTypes.get(column); if (type != null) { size += (long) PrimitiveArrayManager.ARRAY_SIZE * (long) type.getDataTypeSize(); - size += BITMAP_RAM_COST_PER_BLOCK; } } // size is 0 when all types are null @@ -1185,16 +1235,17 @@ public long alignedTvListArrayMemCost() { * @return valueListArrayMemCost */ public static long valueListArrayMemCost(TSDataType type) { - long size = 0; - // value array mem size - size += (long) PrimitiveArrayManager.ARRAY_SIZE * (long) type.getDataTypeSize(); - // bitmap object, byte array, and reference in the bitmap list - size += BITMAP_RAM_COST_PER_BLOCK; - // array headers mem size - size += NUM_BYTES_ARRAY_HEADER; - // Object references size in ArrayList - size += NUM_BYTES_OBJECT_REF; - return size; + return (long) PrimitiveArrayManager.ARRAY_SIZE * (long) type.getDataTypeSize() + + NUM_BYTES_ARRAY_HEADER + + NUM_BYTES_OBJECT_REF; + } + + public static long bitmapRamCost() { + return BITMAP_RAM_COST; + } + + public static long bitmapReferenceRamCost() { + return NUM_BYTES_OBJECT_REF; } /** Build TsBlock by column. */ 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 f96473781aea0..3141ad3c8b442 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 @@ -67,19 +67,30 @@ public abstract class TVList implements WALEntryValue { public static class RamInfo { private final int timestampsSize; private final long arrayMemCost; + private final long ramSize; private final int rowCount; private final List dataTypes; public RamInfo( int timestampCount, long arrayMemCost, int rowCount, List dataTypes) { + this(timestampCount, arrayMemCost, (long) timestampCount * arrayMemCost, rowCount, dataTypes); + } + + public RamInfo( + int timestampCount, + long arrayMemCost, + long ramSize, + int rowCount, + List dataTypes) { this.timestampsSize = timestampCount; this.rowCount = rowCount; this.arrayMemCost = arrayMemCost; + this.ramSize = ramSize; this.dataTypes = dataTypes; } public long getRamSize() { - return timestampsSize * arrayMemCost; + return ramSize; } public int getTimestampsSize() { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java new file mode 100644 index 0000000000000..1c11d9ef8ebd7 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.storageengine.dataregion.memtable; + +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils; +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Measurement; +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Summary; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.Assert; +import org.junit.Assume; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Locale; + +public class AlignedBitmapMemoryAccountingPerformanceTest { + + private static final String ENABLED_PROPERTY = "iotdb.aligned.bitmap.accounting.perf.enabled"; + private static final String COLUMNS_PROPERTY = "iotdb.aligned.bitmap.accounting.perf.columns"; + private static final String ROWS_PROPERTY = "iotdb.aligned.bitmap.accounting.perf.rows"; + private static final String WARMUP_ITERATIONS_PROPERTY = + "iotdb.aligned.bitmap.accounting.perf.warmup.iterations"; + private static final String ITERATIONS_PROPERTY = + "iotdb.aligned.bitmap.accounting.perf.iterations"; + private static final String ROUNDS_PROPERTY = "iotdb.aligned.bitmap.accounting.perf.rounds"; + private static final int RECONCILIATION_REPETITIONS = 1024; + + private static volatile long benchmarkBlackhole; + + @Test + public void alignedTabletBitmapAccountingBenchmark() { + Assume.assumeTrue( + String.format( + "Manual performance UT. Enable with -D%s=true, optionally tune -D%s, -D%s, -D%s, -D%s and -D%s.", + ENABLED_PROPERTY, + COLUMNS_PROPERTY, + ROWS_PROPERTY, + WARMUP_ITERATIONS_PROPERTY, + ITERATIONS_PROPERTY, + ROUNDS_PROPERTY), + Boolean.getBoolean(ENABLED_PROPERTY)); + Assume.assumeTrue( + "Current-thread CPU time and allocation metrics are required.", + ManualPerformanceTestUtils.enableThreadMetrics()); + + int columnCount = Integer.getInteger(COLUMNS_PROPERTY, 64); + int rowCount = Integer.getInteger(ROWS_PROPERTY, 256); + int warmupIterations = Integer.getInteger(WARMUP_ITERATIONS_PROPERTY, 200); + int iterations = Integer.getInteger(ITERATIONS_PROPERTY, 2000); + int rounds = Integer.getInteger(ROUNDS_PROPERTY, 5); + Assert.assertTrue(columnCount > 0); + Assert.assertTrue(rowCount > 0); + Assert.assertTrue(warmupIterations > 0); + Assert.assertTrue(iterations > 0); + Assert.assertTrue(rounds > 0); + + runScenario( + "dense", + createScenario(columnCount, rowCount, false), + warmupIterations, + iterations, + rounds); + runScenario( + "null-heavy", + createScenario(columnCount, rowCount, true), + warmupIterations, + iterations, + rounds); + } + + private static void runScenario( + String label, Scenario scenario, int warmupIterations, int iterations, int rounds) { + runReconciliation( + createAccountingTarget(scenario), warmupIterations * RECONCILIATION_REPETITIONS); + runWrite(scenario, createMemChunks(scenario, warmupIterations)); + + Measurement[] reconciliationMeasurements = new Measurement[rounds]; + Measurement[] writeMeasurements = new Measurement[rounds]; + for (int i = 0; i < rounds; i++) { + if ((i & 1) == 0) { + reconciliationMeasurements[i] = measureReconciliation(scenario, iterations); + writeMeasurements[i] = measureWrite(scenario, iterations); + } else { + writeMeasurements[i] = measureWrite(scenario, iterations); + reconciliationMeasurements[i] = measureReconciliation(scenario, iterations); + } + } + + Summary reconciliationSummary = + ManualPerformanceTestUtils.summarize( + reconciliationMeasurements, iterations * RECONCILIATION_REPETITIONS); + Summary writeSummary = ManualPerformanceTestUtils.summarize(writeMeasurements, iterations); + printResult( + label, scenario, warmupIterations, iterations, rounds, reconciliationSummary, writeSummary); + } + + private static Measurement measureReconciliation(Scenario scenario, int iterations) { + AccountingTarget target = createAccountingTarget(scenario); + return ManualPerformanceTestUtils.measure( + 1, () -> runReconciliation(target, iterations * RECONCILIATION_REPETITIONS)); + } + + private static Measurement measureWrite(Scenario scenario, int iterations) { + AlignedWritableMemChunk[] memChunks = createMemChunks(scenario, iterations); + return ManualPerformanceTestUtils.measure(1, () -> runWrite(scenario, memChunks)); + } + + private static void runReconciliation(AccountingTarget target, int iterations) { + long correction = 0; + for (int i = 0; i < iterations; i++) { + TsFileProcessor.AlignedTVListRamCostSnapshot snapshot = + new TsFileProcessor.AlignedTVListRamCostSnapshot(target.memTable, target.deviceId); + correction += snapshot.getMemoryCorrection(0); + } + benchmarkBlackhole = correction + iterations; + } + + private static void runWrite(Scenario scenario, AlignedWritableMemChunk[] memChunks) { + for (AlignedWritableMemChunk memChunk : memChunks) { + memChunk.writeAlignedTablet( + scenario.times, + scenario.columns, + scenario.bitMaps, + scenario.schemas, + 0, + scenario.times.length, + null); + } + benchmarkBlackhole = memChunks[memChunks.length - 1].rowCount(); + } + + private static AlignedWritableMemChunk[] createMemChunks(Scenario scenario, int count) { + AlignedWritableMemChunk[] memChunks = new AlignedWritableMemChunk[count]; + for (int i = 0; i < count; i++) { + memChunks[i] = new AlignedWritableMemChunk(new ArrayList<>(scenario.schemas), false); + } + return memChunks; + } + + private static AccountingTarget createAccountingTarget(Scenario scenario) { + AlignedWritableMemChunk memChunk = + new AlignedWritableMemChunk(new ArrayList<>(scenario.schemas), false); + memChunk.writeAlignedTablet( + scenario.times, + scenario.columns, + scenario.bitMaps, + scenario.schemas, + 0, + scenario.times.length, + null); + IDeviceID deviceId = IDeviceID.Factory.DEFAULT_FACTORY.create("root.accounting.d0"); + IMemTable memTable = + new PrimitiveMemTable( + "root.accounting", + "0", + Collections.singletonMap( + deviceId, + new AlignedWritableMemChunkGroup( + memChunk, new ArrayList<>(scenario.schemas), false))); + return new AccountingTarget(memTable, deviceId); + } + + private static Scenario createScenario(int columnCount, int rowCount, boolean nullHeavy) { + String[] measurements = new String[columnCount]; + TSDataType[] dataTypes = new TSDataType[columnCount]; + Object[] columns = new Object[columnCount]; + BitMap[] bitMaps = nullHeavy ? new BitMap[columnCount] : null; + List schemas = new ArrayList<>(columnCount); + for (int column = 0; column < columnCount; column++) { + measurements[column] = "s" + column; + dataTypes[column] = TSDataType.INT32; + schemas.add(new MeasurementSchema(measurements[column], TSDataType.INT32)); + int[] values = new int[rowCount]; + for (int row = 0; row < rowCount; row++) { + values[row] = row; + } + columns[column] = values; + if (nullHeavy) { + bitMaps[column] = new BitMap(rowCount); + for (int row = column & 1; row < rowCount; row += 2) { + bitMaps[column].mark(row); + } + } + } + long[] times = new long[rowCount]; + for (int row = 0; row < rowCount; row++) { + times[row] = row; + } + return new Scenario(measurements, dataTypes, columns, bitMaps, schemas, times); + } + + private static void printResult( + String label, + Scenario scenario, + int warmupIterations, + int iterations, + int rounds, + Summary reconciliationSummary, + Summary writeSummary) { + System.out.printf( + Locale.ROOT, + "Aligned bitmap accounting benchmark (%s): columns=%d, rows=%d, warmups=%d, iterations/round=%d, rounds=%d%n", + label, + scenario.measurements.length, + scenario.times.length, + warmupIterations, + iterations, + rounds); + printSummary("reconcile", reconciliationSummary); + printSummary("write", writeSummary); + System.out.printf( + Locale.ROOT, + " reconciliation/write CPU ratio=%.2f%%, allocation ratio=%.2f%%%n", + percentage( + reconciliationSummary.getCpuNanosPerOperation(), + writeSummary.getCpuNanosPerOperation()), + percentage( + reconciliationSummary.getAllocatedBytesPerOperation(), + writeSummary.getAllocatedBytesPerOperation())); + } + + private static void printSummary(String label, Summary summary) { + System.out.printf( + Locale.ROOT, + " %-10s CPU=%.3f us/batch, allocated=%.1f bytes/batch, peak heap delta=%.3f MiB%n", + label, + summary.getCpuNanosPerOperation() / 1_000.0, + summary.getAllocatedBytesPerOperation(), + summary.getPeakHeapDeltaBytes() / 1024.0 / 1024.0); + } + + private static double percentage(double numerator, double denominator) { + return denominator == 0 ? 0 : numerator * 100.0 / denominator; + } + + private static final class AccountingTarget { + + private final IMemTable memTable; + private final IDeviceID deviceId; + + private AccountingTarget(IMemTable memTable, IDeviceID deviceId) { + this.memTable = memTable; + this.deviceId = deviceId; + } + } + + private static final class Scenario { + + private final String[] measurements; + private final TSDataType[] dataTypes; + private final Object[] columns; + private final BitMap[] bitMaps; + private final List schemas; + private final long[] times; + + private Scenario( + String[] measurements, + TSDataType[] dataTypes, + Object[] columns, + BitMap[] bitMaps, + List schemas, + long[] times) { + this.measurements = measurements; + this.dataTypes = dataTypes; + this.columns = columns; + this.bitMaps = bitMaps; + this.schemas = schemas; + this.times = times; + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index 040988997b2a3..9593f477bf0c1 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.file.SystemFileFactory; import org.apache.iotdb.commons.path.AlignedFullPath; +import org.apache.iotdb.commons.path.AlignedPath; import org.apache.iotdb.commons.path.NonAlignedFullPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; @@ -62,6 +63,7 @@ import org.apache.tsfile.read.query.dataset.QueryDataSet; import org.apache.tsfile.read.reader.IPointReader; import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.write.record.TSRecord; import org.apache.tsfile.write.record.datapoint.DataPoint; import org.apache.tsfile.write.schema.MeasurementSchema; @@ -536,21 +538,21 @@ public void alignedTvListRamCostTest() true, new long[5]); IMemTable memTable = processor.getWorkMemTable(); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1596552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNode(100, true), Collections.singletonList(new int[] {0, 10}), new TSStatus[10], true, new long[5]); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1596552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNode(200, true), Collections.singletonList(new int[] {0, 10}), new TSStatus[10], true, new long[5]); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1596552, memTable.getTVListsRamCost()); Assert.assertEquals(90000, memTable.getTotalPointsNum()); Assert.assertEquals(720360, memTable.memSize()); // Test records @@ -559,7 +561,7 @@ public void alignedTvListRamCostTest() record.addTuple(DataPoint.getDataPoint(dataType, measurementId, String.valueOf(i))); processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); } - Assert.assertEquals(1778168, memTable.getTVListsRamCost()); + Assert.assertEquals(1598168, memTable.getTVListsRamCost()); Assert.assertEquals(90100, memTable.getTotalPointsNum()); Assert.assertEquals(721560, memTable.memSize()); } @@ -614,7 +616,7 @@ public void alignedTvListRamCostTest2() true, new long[5]); IMemTable memTable = processor.getWorkMemTable(); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1596552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNodeFors3000ToS6000(0, true), Collections.singletonList(new int[] {0, 10}), @@ -656,7 +658,7 @@ public void alignedTvListRamCostTest2() new TSStatus[10], true, new long[5]); - Assert.assertEquals(7105104, memTable.getTVListsRamCost()); + Assert.assertEquals(6937104, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNodeFors3000ToS6000(300, true), Collections.singletonList(new int[] {0, 10}), @@ -685,6 +687,65 @@ public void alignedTvListRamCostTest2() Assert.assertEquals(1923360, memTable.memSize()); } + @Test + public void alignedBitmapMemoryAccountingMatchesActualAllocations() + throws MetadataException, WriteProcessException, IOException, IllegalPathException { + processor = + new TsFileProcessor( + storageGroup, + SystemFileFactory.INSTANCE.getFile(filePath), + sgInfo, + this::closeTsFileProcessor, + (tsFileProcessor, updateMap, systemFlushTime) -> {}, + true); + TsFileProcessorInfo tsFileProcessorInfo = new TsFileProcessorInfo(sgInfo); + processor.setTsFileProcessorInfo(tsFileProcessorInfo); + this.sgInfo.initTsFileProcessorInfo(processor); + SystemInfo.getInstance().reportStorageGroupStatus(sgInfo, processor); + + int denseRowCount = PrimitiveArrayManager.ARRAY_SIZE * 2 + 1; + InsertTabletNode denseTablet = genAlignedTablet(new String[] {"s0", "s1"}, denseRowCount, 0); + processor.insertTablet( + denseTablet, + Collections.singletonList(new int[] {0, denseRowCount}), + new TSStatus[denseRowCount], + true, + new long[5]); + + AlignedWritableMemChunk alignedMemChunk = getAlignedMemChunk(denseTablet.getDeviceID()); + Assert.assertNull(alignedMemChunk.getWorkingTVList().getBitMaps()); + assertAlignedTvListRamCostMatchesActual(denseTablet.getDeviceID()); + + InsertTabletNode nullTablet = genAlignedTablet(new String[] {"s0", "s1"}, 2, denseRowCount); + BitMap secondColumnNulls = new BitMap(2); + secondColumnNulls.markAll(); + nullTablet.setBitMaps(new BitMap[] {null, secondColumnNulls}); + processor.insertTablet( + nullTablet, + Arrays.asList(new int[] {0, 1}, new int[] {1, 2}), + new TSStatus[2], + true, + new long[5]); + + List secondColumnBitMaps = alignedMemChunk.getWorkingTVList().getBitMaps().get(1); + Assert.assertNull(secondColumnBitMaps.get(0)); + Assert.assertNull(secondColumnBitMaps.get(1)); + Assert.assertNotNull(secondColumnBitMaps.get(2)); + assertAlignedTvListRamCostMatchesActual(denseTablet.getDeviceID()); + + TSRecord extendedColumnRecord = new TSRecord(deviceId, denseRowCount + 2L); + extendedColumnRecord.addTuple(DataPoint.getDataPoint(TSDataType.INT32, "s2", "1")); + InsertRowNode extendedColumnRow = buildInsertRowNodeByTSRecord(extendedColumnRecord); + extendedColumnRow.setAligned(true); + processor.insert(extendedColumnRow, new long[5]); + + int extendedColumnIndex = alignedMemChunk.getWorkingTVList().getBitMaps().size() - 1; + for (BitMap bitMap : alignedMemChunk.getWorkingTVList().getBitMaps().get(extendedColumnIndex)) { + Assert.assertNotNull(bitMap); + } + assertAlignedTvListRamCostMatchesActual(denseTablet.getDeviceID()); + } + @Test public void nonAlignedTvListRamCostTest() throws MetadataException, WriteProcessException, IOException { @@ -1335,6 +1396,59 @@ private InsertTabletNode genSingleMeasurementTablet(int rowCount, boolean isAlig rowCount); } + private AlignedWritableMemChunk getAlignedMemChunk(IDeviceID targetDeviceId) { + IWritableMemChunk memChunk = + processor + .getWorkMemTable() + .getWritableMemChunk(targetDeviceId, AlignedPath.VECTOR_PLACEHOLDER); + Assert.assertNotNull(memChunk); + return (AlignedWritableMemChunk) memChunk; + } + + private void assertAlignedTvListRamCostMatchesActual(IDeviceID targetDeviceId) { + AlignedWritableMemChunk alignedMemChunk = getAlignedMemChunk(targetDeviceId); + long actualRamCost = alignedMemChunk.getWorkingTVList().getRamSize(); + for (org.apache.iotdb.db.utils.datastructure.AlignedTVList sortedTVList : + alignedMemChunk.getSortedList()) { + actualRamCost += sortedTVList.getRamSize(); + } + Assert.assertEquals(actualRamCost, processor.getWorkMemTable().getTVListsRamCost()); + } + + private InsertTabletNode genAlignedTablet(String[] measurements, int rowCount, long startTime) + throws IllegalPathException { + TSDataType[] dataTypes = new TSDataType[measurements.length]; + MeasurementSchema[] schemas = new MeasurementSchema[measurements.length]; + Object[] columns = new Object[measurements.length]; + for (int column = 0; column < measurements.length; column++) { + dataTypes[column] = TSDataType.INT32; + schemas[column] = + new MeasurementSchema(measurements[column], TSDataType.INT32, TSEncoding.PLAIN); + columns[column] = new int[rowCount]; + for (int row = 0; row < rowCount; row++) { + ((int[]) columns[column])[row] = row; + } + } + long[] times = new long[rowCount]; + for (int row = 0; row < rowCount; row++) { + times[row] = startTime + row; + } + + InsertTabletNode insertTabletNode = + new InsertTabletNode( + new QueryId("test_write").genPlanNodeId(), + new PartialPath(deviceId), + true, + measurements, + dataTypes, + times, + null, + columns, + rowCount); + insertTabletNode.setMeasurementSchemas(schemas); + return insertTabletNode; + } + private InsertTabletNode genInsertTableNode(long startTime, boolean isAligned) throws IllegalPathException { String deviceId = "root.sg.device5"; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java index 5ab976c13104b..4cd977b7a1d3d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java @@ -26,6 +26,7 @@ import org.apache.tsfile.external.commons.lang3.ArrayUtils; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.utils.RamUsageEstimator; import org.junit.Assert; import org.junit.Test; @@ -34,9 +35,23 @@ import java.util.List; import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE; +import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_ARRAY_HEADER; +import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF; public class AlignedTVListTest { + @Test + public void testValueListArrayMemCostExcludesBitmapReservation() { + long expected = (long) ARRAY_SIZE * Long.BYTES + NUM_BYTES_ARRAY_HEADER + NUM_BYTES_OBJECT_REF; + + Assert.assertEquals(expected, AlignedTVList.valueListArrayMemCost(TSDataType.INT64)); + Assert.assertEquals( + RamUsageEstimator.shallowSizeOfInstance(BitMap.class) + + RamUsageEstimator.sizeOfByteArray(BitMap.getSizeOfBytes(ARRAY_SIZE)), + AlignedTVList.bitmapRamCost()); + Assert.assertEquals(NUM_BYTES_OBJECT_REF, AlignedTVList.bitmapReferenceRamCost()); + } + @Test public void testAlignedTVList1() { List dataTypes = new ArrayList<>(); @@ -169,6 +184,35 @@ public void testBitmapIsAllocatedLazilyWithCompactBackingArray() { BitMap.getSizeOfBytes(ARRAY_SIZE), firstColumnBitMaps.get(2).getByteArray().length); Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE * 2 + 1, 0)); Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE * 2, 0)); + Assert.assertEquals( + 3L * tvList.alignedTvListArrayMemCost() + + 3L * AlignedTVList.bitmapReferenceRamCost() + + AlignedTVList.bitmapRamCost(), + tvList.getRamSize()); + Assert.assertEquals(tvList.getRamSize(), tvList.calculateRamSize().getRamSize()); + Assert.assertEquals(tvList.getRamSize(), tvList.clone().getRamSize()); + Assert.assertEquals(tvList.getRamSize(), tvList.cloneForFlushSort().getRamSize()); + } + + @Test + public void testExtendedColumnRamCostIncludesActualBitmaps() { + AlignedTVList tvList = + AlignedTVList.newAlignedList(new ArrayList<>(Arrays.asList(TSDataType.INT64))); + for (int i = 0; i <= ARRAY_SIZE; i++) { + tvList.putAlignedValue(i, new Object[] {(long) i}); + } + + long ramSizeBeforeExtension = tvList.getRamSize(); + tvList.extendColumn(TSDataType.INT32); + + Assert.assertEquals( + 2L + * (AlignedTVList.valueListArrayMemCost(TSDataType.INT32) + + AlignedTVList.bitmapReferenceRamCost() + + AlignedTVList.bitmapRamCost()), + tvList.getRamSize() - ramSizeBeforeExtension); + tvList.clear(); + Assert.assertEquals(0, tvList.getRamSize()); } @Test From 8deb411d67f7d4fa0bd87390a0205e4041342809 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 25 Aug 2026 14:38:47 +0800 Subject: [PATCH 2/3] Delete AlignedBitmapMemoryAccountingPerformanceTest.java --- ...BitmapMemoryAccountingPerformanceTest.java | 295 ------------------ 1 file changed, 295 deletions(-) delete mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java deleted file mode 100644 index 1c11d9ef8ebd7..0000000000000 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java +++ /dev/null @@ -1,295 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.iotdb.db.storageengine.dataregion.memtable; - -import org.apache.iotdb.db.utils.ManualPerformanceTestUtils; -import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Measurement; -import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Summary; - -import org.apache.tsfile.enums.TSDataType; -import org.apache.tsfile.file.metadata.IDeviceID; -import org.apache.tsfile.utils.BitMap; -import org.apache.tsfile.write.schema.IMeasurementSchema; -import org.apache.tsfile.write.schema.MeasurementSchema; -import org.junit.Assert; -import org.junit.Assume; -import org.junit.Test; - -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Locale; - -public class AlignedBitmapMemoryAccountingPerformanceTest { - - private static final String ENABLED_PROPERTY = "iotdb.aligned.bitmap.accounting.perf.enabled"; - private static final String COLUMNS_PROPERTY = "iotdb.aligned.bitmap.accounting.perf.columns"; - private static final String ROWS_PROPERTY = "iotdb.aligned.bitmap.accounting.perf.rows"; - private static final String WARMUP_ITERATIONS_PROPERTY = - "iotdb.aligned.bitmap.accounting.perf.warmup.iterations"; - private static final String ITERATIONS_PROPERTY = - "iotdb.aligned.bitmap.accounting.perf.iterations"; - private static final String ROUNDS_PROPERTY = "iotdb.aligned.bitmap.accounting.perf.rounds"; - private static final int RECONCILIATION_REPETITIONS = 1024; - - private static volatile long benchmarkBlackhole; - - @Test - public void alignedTabletBitmapAccountingBenchmark() { - Assume.assumeTrue( - String.format( - "Manual performance UT. Enable with -D%s=true, optionally tune -D%s, -D%s, -D%s, -D%s and -D%s.", - ENABLED_PROPERTY, - COLUMNS_PROPERTY, - ROWS_PROPERTY, - WARMUP_ITERATIONS_PROPERTY, - ITERATIONS_PROPERTY, - ROUNDS_PROPERTY), - Boolean.getBoolean(ENABLED_PROPERTY)); - Assume.assumeTrue( - "Current-thread CPU time and allocation metrics are required.", - ManualPerformanceTestUtils.enableThreadMetrics()); - - int columnCount = Integer.getInteger(COLUMNS_PROPERTY, 64); - int rowCount = Integer.getInteger(ROWS_PROPERTY, 256); - int warmupIterations = Integer.getInteger(WARMUP_ITERATIONS_PROPERTY, 200); - int iterations = Integer.getInteger(ITERATIONS_PROPERTY, 2000); - int rounds = Integer.getInteger(ROUNDS_PROPERTY, 5); - Assert.assertTrue(columnCount > 0); - Assert.assertTrue(rowCount > 0); - Assert.assertTrue(warmupIterations > 0); - Assert.assertTrue(iterations > 0); - Assert.assertTrue(rounds > 0); - - runScenario( - "dense", - createScenario(columnCount, rowCount, false), - warmupIterations, - iterations, - rounds); - runScenario( - "null-heavy", - createScenario(columnCount, rowCount, true), - warmupIterations, - iterations, - rounds); - } - - private static void runScenario( - String label, Scenario scenario, int warmupIterations, int iterations, int rounds) { - runReconciliation( - createAccountingTarget(scenario), warmupIterations * RECONCILIATION_REPETITIONS); - runWrite(scenario, createMemChunks(scenario, warmupIterations)); - - Measurement[] reconciliationMeasurements = new Measurement[rounds]; - Measurement[] writeMeasurements = new Measurement[rounds]; - for (int i = 0; i < rounds; i++) { - if ((i & 1) == 0) { - reconciliationMeasurements[i] = measureReconciliation(scenario, iterations); - writeMeasurements[i] = measureWrite(scenario, iterations); - } else { - writeMeasurements[i] = measureWrite(scenario, iterations); - reconciliationMeasurements[i] = measureReconciliation(scenario, iterations); - } - } - - Summary reconciliationSummary = - ManualPerformanceTestUtils.summarize( - reconciliationMeasurements, iterations * RECONCILIATION_REPETITIONS); - Summary writeSummary = ManualPerformanceTestUtils.summarize(writeMeasurements, iterations); - printResult( - label, scenario, warmupIterations, iterations, rounds, reconciliationSummary, writeSummary); - } - - private static Measurement measureReconciliation(Scenario scenario, int iterations) { - AccountingTarget target = createAccountingTarget(scenario); - return ManualPerformanceTestUtils.measure( - 1, () -> runReconciliation(target, iterations * RECONCILIATION_REPETITIONS)); - } - - private static Measurement measureWrite(Scenario scenario, int iterations) { - AlignedWritableMemChunk[] memChunks = createMemChunks(scenario, iterations); - return ManualPerformanceTestUtils.measure(1, () -> runWrite(scenario, memChunks)); - } - - private static void runReconciliation(AccountingTarget target, int iterations) { - long correction = 0; - for (int i = 0; i < iterations; i++) { - TsFileProcessor.AlignedTVListRamCostSnapshot snapshot = - new TsFileProcessor.AlignedTVListRamCostSnapshot(target.memTable, target.deviceId); - correction += snapshot.getMemoryCorrection(0); - } - benchmarkBlackhole = correction + iterations; - } - - private static void runWrite(Scenario scenario, AlignedWritableMemChunk[] memChunks) { - for (AlignedWritableMemChunk memChunk : memChunks) { - memChunk.writeAlignedTablet( - scenario.times, - scenario.columns, - scenario.bitMaps, - scenario.schemas, - 0, - scenario.times.length, - null); - } - benchmarkBlackhole = memChunks[memChunks.length - 1].rowCount(); - } - - private static AlignedWritableMemChunk[] createMemChunks(Scenario scenario, int count) { - AlignedWritableMemChunk[] memChunks = new AlignedWritableMemChunk[count]; - for (int i = 0; i < count; i++) { - memChunks[i] = new AlignedWritableMemChunk(new ArrayList<>(scenario.schemas), false); - } - return memChunks; - } - - private static AccountingTarget createAccountingTarget(Scenario scenario) { - AlignedWritableMemChunk memChunk = - new AlignedWritableMemChunk(new ArrayList<>(scenario.schemas), false); - memChunk.writeAlignedTablet( - scenario.times, - scenario.columns, - scenario.bitMaps, - scenario.schemas, - 0, - scenario.times.length, - null); - IDeviceID deviceId = IDeviceID.Factory.DEFAULT_FACTORY.create("root.accounting.d0"); - IMemTable memTable = - new PrimitiveMemTable( - "root.accounting", - "0", - Collections.singletonMap( - deviceId, - new AlignedWritableMemChunkGroup( - memChunk, new ArrayList<>(scenario.schemas), false))); - return new AccountingTarget(memTable, deviceId); - } - - private static Scenario createScenario(int columnCount, int rowCount, boolean nullHeavy) { - String[] measurements = new String[columnCount]; - TSDataType[] dataTypes = new TSDataType[columnCount]; - Object[] columns = new Object[columnCount]; - BitMap[] bitMaps = nullHeavy ? new BitMap[columnCount] : null; - List schemas = new ArrayList<>(columnCount); - for (int column = 0; column < columnCount; column++) { - measurements[column] = "s" + column; - dataTypes[column] = TSDataType.INT32; - schemas.add(new MeasurementSchema(measurements[column], TSDataType.INT32)); - int[] values = new int[rowCount]; - for (int row = 0; row < rowCount; row++) { - values[row] = row; - } - columns[column] = values; - if (nullHeavy) { - bitMaps[column] = new BitMap(rowCount); - for (int row = column & 1; row < rowCount; row += 2) { - bitMaps[column].mark(row); - } - } - } - long[] times = new long[rowCount]; - for (int row = 0; row < rowCount; row++) { - times[row] = row; - } - return new Scenario(measurements, dataTypes, columns, bitMaps, schemas, times); - } - - private static void printResult( - String label, - Scenario scenario, - int warmupIterations, - int iterations, - int rounds, - Summary reconciliationSummary, - Summary writeSummary) { - System.out.printf( - Locale.ROOT, - "Aligned bitmap accounting benchmark (%s): columns=%d, rows=%d, warmups=%d, iterations/round=%d, rounds=%d%n", - label, - scenario.measurements.length, - scenario.times.length, - warmupIterations, - iterations, - rounds); - printSummary("reconcile", reconciliationSummary); - printSummary("write", writeSummary); - System.out.printf( - Locale.ROOT, - " reconciliation/write CPU ratio=%.2f%%, allocation ratio=%.2f%%%n", - percentage( - reconciliationSummary.getCpuNanosPerOperation(), - writeSummary.getCpuNanosPerOperation()), - percentage( - reconciliationSummary.getAllocatedBytesPerOperation(), - writeSummary.getAllocatedBytesPerOperation())); - } - - private static void printSummary(String label, Summary summary) { - System.out.printf( - Locale.ROOT, - " %-10s CPU=%.3f us/batch, allocated=%.1f bytes/batch, peak heap delta=%.3f MiB%n", - label, - summary.getCpuNanosPerOperation() / 1_000.0, - summary.getAllocatedBytesPerOperation(), - summary.getPeakHeapDeltaBytes() / 1024.0 / 1024.0); - } - - private static double percentage(double numerator, double denominator) { - return denominator == 0 ? 0 : numerator * 100.0 / denominator; - } - - private static final class AccountingTarget { - - private final IMemTable memTable; - private final IDeviceID deviceId; - - private AccountingTarget(IMemTable memTable, IDeviceID deviceId) { - this.memTable = memTable; - this.deviceId = deviceId; - } - } - - private static final class Scenario { - - private final String[] measurements; - private final TSDataType[] dataTypes; - private final Object[] columns; - private final BitMap[] bitMaps; - private final List schemas; - private final long[] times; - - private Scenario( - String[] measurements, - TSDataType[] dataTypes, - Object[] columns, - BitMap[] bitMaps, - List schemas, - long[] times) { - this.measurements = measurements; - this.dataTypes = dataTypes; - this.columns = columns; - this.bitMaps = bitMaps; - this.schemas = schemas; - this.times = times; - } - } -} From f0e35dc8da54b628efe0ebed26601eff2db28123 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 25 Aug 2026 17:50:45 +0800 Subject: [PATCH 3/3] Update TsFileProcessorTest.java --- .../dataregion/memtable/TsFileProcessorTest.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index 9593f477bf0c1..28493c0b90677 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -46,6 +46,7 @@ import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.db.utils.constant.TestConstant; +import org.apache.iotdb.db.utils.datastructure.AlignedTVList; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -587,7 +588,9 @@ public void alignedTabletKeepsFailedStatusesAndCountsWrittenRows() genSingleMeasurementTablet(rowCount, true), rangeList, actualResults, false, new long[5]); Assert.assertEquals( - expectedProcessor.getWorkMemTable().getTVListsRamCost(), + expectedProcessor.getWorkMemTable().getTVListsRamCost() + + 2 * AlignedTVList.bitmapReferenceRamCost() + + AlignedTVList.bitmapRamCost(), actualProcessor.getWorkMemTable().getTVListsRamCost()); Assert.assertEquals( TSStatusCode.OUT_OF_TTL.getStatusCode(), actualResults[failedIndex].getCode());