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/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index 040988997b2a3..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 @@ -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; @@ -45,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; @@ -62,6 +64,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 +539,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 +562,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()); } @@ -585,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()); @@ -614,7 +619,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 +661,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 +690,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 +1399,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