diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java index 375f560c9ee4..314a098211c2 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java @@ -83,10 +83,7 @@ public void testAddingManyColumnsAfterManyRowsDoesNotExhaustWriteMemory() throws * NEW_COLUMN_COUNT * AlignedTVList.valueListArrayMemCost(TSDataType.INT64); long lazyAllocationCost = - (long) historicalBlockCount - * NEW_COLUMN_COUNT - * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray() - + (long) NEW_COLUMN_COUNT * AlignedTVList.primitiveArrayMemCost(TSDataType.INT64); + (long) NEW_COLUMN_COUNT * AlignedTVList.valueListArrayMemCost(TSDataType.INT64); long dataNodeMaxHeapSize = DATANODE_MAX_HEAP_SIZE_IN_MB * 1024L * 1024L; Assert.assertTrue( 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 285e29b40650..ee9b39d2ffe4 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 @@ -313,6 +313,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; @@ -379,11 +383,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 @@ -408,6 +416,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()) { @@ -421,6 +440,7 @@ public void insertRows(InsertRowsNode insertRowsNode, long[] infoForMetrics) } } long[] alignedMemIncrements = checkAlignedMemCostAndAddToTspInfoForRows(alignedList); + alignedMemTableIncrement = alignedMemIncrements[0]; final long[] nonAlignedMemIncrements; try { nonAlignedMemIncrements = checkMemCostAndAddToTspInfoForRows(nonAlignedList); @@ -436,6 +456,7 @@ public void insertRows(InsertRowsNode insertRowsNode, long[] infoForMetrics) if (insertRowsNode.isAligned()) { memIncrements = checkAlignedMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList()); + alignedMemTableIncrement = memIncrements[0]; } else { memIncrements = checkMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList()); } @@ -486,19 +507,23 @@ 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()); @@ -623,6 +648,19 @@ 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); @@ -670,59 +708,63 @@ public void insertTablet( tsFileResource); int pointInserted = 0; - 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; + 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; } - throw e; - } - for (int i = start; i < end; i++) { - if (results[i] == null - || results[i].getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - results[i] = RpcUtils.SUCCESS_STATUS; + for (int i = start; i < end; i++) { + 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()); @@ -853,52 +895,33 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRow( ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER, TSDataType.VECTOR) * writableFieldDataTypes.length; // The first row creates the first aligned TVList block. All writable fields have a non-null - // value, so this includes the timestamp array, value primitive arrays, bitmap reservations, - // array headers, and ArrayList references for that block. + // value, so this includes the timestamp array, value primitive arrays, array headers, and + // ArrayList references for that block. memTableIncrement += AlignedTVList.alignedTvListArrayMemCost(writableFieldDataTypes, null); } else { // For existed device of this mem table AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk) memChunk; int currentPointNum = alignedMemChunk.alignedListSize(); - int currentArrayNum = - currentPointNum / PrimitiveArrayManager.ARRAY_SIZE - + (currentPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0); int targetArrayIndex = currentPointNum / PrimitiveArrayManager.ARRAY_SIZE; - int newColumnCount = 0; for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { // Skip failed Measurements if (!isWritableFieldMeasurement(measurements, dataTypes, values, columnCategories, i)) { continue; } - if (!alignedMemChunk.containsMeasurement(measurements[i])) { - // Extending a column adds one null value-list placeholder and one conservatively reserved - // bitmap to every historical block. No value primitive array is allocated there. - memTableIncrement += - currentArrayNum * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); - newColumnCount++; - } if (!isValueArrayMaterialized(alignedMemChunk, measurements[i], targetArrayIndex)) { // The non-null value materializes this column's primitive array in the target block. - // This cost contains the primitive-array payload and its array header. - memTableIncrement += AlignedTVList.primitiveArrayMemCost(dataTypes[i]); + // Null historical placeholders are not charged. + memTableIncrement += AlignedTVList.valueListArrayMemCost(dataTypes[i]); } } if ((currentPointNum % PrimitiveArrayManager.ARRAY_SIZE) == 0) { - // Starting a block allocates its timestamp array, optional sort-index array, value-list - // null - // placeholders, bitmap reservations, array headers, and ArrayList references for all - // existing columns. Value primitive arrays are excluded because they were charged above - // only for written columns. + // Starting a block allocates its timestamp array and optional sort-index array. Value + // arrays and their references are charged above only for written columns. memTableIncrement += alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCostWithoutPrimitiveArrays(); - // The working TVList does not contain newly extended columns yet, so add their value-list - // placeholder and bitmap reservation for this new block separately. - memTableIncrement += - (long) newColumnCount * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); } } - for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { if (isWritableFieldMeasurement(measurements, dataTypes, values, columnCategories, i) && dataTypes[i].isBinary()) { @@ -942,7 +965,7 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRows(List ins ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER, TSDataType.VECTOR) * writableFieldDataTypes.length; // The first row creates the first complete aligned TVList block: timestamp array, value - // primitive arrays for the writable fields, bitmap reservations, headers, and references. + // primitive arrays for the writable fields, headers, and references. memTableIncrement += AlignedTVList.alignedTvListArrayMemCost(writableFieldDataTypes, null); for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { // Skip failed Measurements @@ -980,13 +1003,6 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRows(List ins if (!currentMemChunkContainsMeasurement && !addingPointNumInfo.left.containsKey(measurements[i])) { addingPointNumInfo.left.put(measurements[i], dataTypes[i]); - int currentArrayNum = - pointNumBeforeCurrentRow / PrimitiveArrayManager.ARRAY_SIZE - + (pointNumBeforeCurrentRow % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0); - // A column first seen in this batch adds a null value-list placeholder and a bitmap - // reservation to every block that already exists before the current row. - memTableIncrement += - currentArrayNum * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); } Set materializedArrayIndexes = materializedArraysInCurrentBatch @@ -994,32 +1010,25 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRows(List ins .computeIfAbsent(measurements[i], key -> new HashSet<>()); if (!isValueArrayMaterialized(alignedMemChunk, measurements[i], targetArrayIndex) && materializedArrayIndexes.add(targetArrayIndex)) { - // Charge the payload and header of a value primitive array exactly once when this - // batch first writes a non-null value to the column in the target block. - memTableIncrement += AlignedTVList.primitiveArrayMemCost(dataTypes[i]); + // Charge a value array and its reference exactly once when this batch first writes a + // non-null value to the column in the target block. + memTableIncrement += AlignedTVList.valueListArrayMemCost(dataTypes[i]); } } if ((pointNumBeforeCurrentRow % PrimitiveArrayManager.ARRAY_SIZE) == 0) { if (alignedMemChunk == null) { - // A device created earlier in this batch starts another block. Charge the timestamp - // array, value-list placeholders, bitmap reservations, headers, and references for all - // columns discovered in the batch, excluding their value primitive arrays. + // A device created earlier in this batch starts another timestamp block. Value arrays + // and their references are charged only when materialized. memTableIncrement += AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays( addingPointNumInfo.left.values().toArray(new TSDataType[0]), null); } else { - // An existing aligned TVList starts another block. Charge its timestamp array, optional - // sort-index array, value-list placeholders, bitmap reservations, headers, and - // references for columns that were already present before this batch. + // An existing aligned TVList starts another timestamp block. Value arrays and their + // references are charged only when materialized. memTableIncrement += alignedMemChunk .getWorkingTVList() .alignedTvListArrayMemCostWithoutPrimitiveArrays(); - // Columns first introduced by this batch are absent from the working TVList's type - // list, so add one value-list placeholder and bitmap reservation for each of them. - memTableIncrement += - (long) addingPointNumInfo.left.size() - * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); } } addingPointNumInfo.setRight(addingPointNum + 1); @@ -1172,9 +1181,8 @@ private void updateAlignedMemCost( int numArraysToAdd = incomingPointNum / PrimitiveArrayManager.ARRAY_SIZE + (incomingPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0); - // Each new block allocates its timestamp array, value-list null placeholders, bitmap - // reservations, array headers, and ArrayList references. Value primitive arrays are excluded - // here because all-null and failed-only column segments leave them unmaterialized. + // Each new block allocates its timestamp array. Value arrays and their references are + // excluded here because all-null and failed-only column segments leave them unmaterialized. memIncrements[0] += numArraysToAdd * AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays( @@ -1182,7 +1190,7 @@ private void updateAlignedMemCost( // Add the payload and header of a value primitive array only for a column/block segment that // contains at least one successful non-null value. memIncrements[0] += - calculateTabletPrimitiveArrayMemCost( + calculateTabletValueArrayMemCost( null, measurementIds, dataTypes, @@ -1195,28 +1203,11 @@ private void updateAlignedMemCost( 0); } else { AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk) memChunk; - List newDataTypes = new ArrayList<>(); int currentPointNum = alignedMemChunk.alignedListSize(); int newPointNum = currentPointNum + incomingPointNum; int currentArrayCnt = currentPointNum / PrimitiveArrayManager.ARRAY_SIZE + (currentPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0); - for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { - TSDataType dataType = dataTypes[i]; - if (!isWritableFieldMeasurement(measurementIds, dataTypes, columns, columnCategories, i)) { - continue; - } - - if (!alignedMemChunk.containsMeasurement(measurementIds[i])) { - // A new column adds one null value-list placeholder and one conservatively reserved - // bitmap to every historical block. It does not materialize value primitive arrays in - // those blocks. - memIncrements[0] += - currentArrayCnt * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); - newDataTypes.add(dataType); - } - } - // calculate how many new arrays will be added after this insertion int newArrayCnt = newPointNum / PrimitiveArrayManager.ARRAY_SIZE @@ -1224,26 +1215,18 @@ private void updateAlignedMemCost( long acquireArray = newArrayCnt - currentArrayCnt; if (acquireArray != 0) { - // Each acquired block adds a timestamp array, optional sort-index array, value-list null - // placeholders, bitmap reservations, array headers, and ArrayList references for columns - // already present in the working TVList. Value primitive arrays are charged separately - // below. + // Each acquired block adds a timestamp array and optional sort-index array. Value arrays + // and their references are charged separately below. memIncrements[0] += acquireArray * alignedMemChunk .getWorkingTVList() .alignedTvListArrayMemCostWithoutPrimitiveArrays(); - for (TSDataType ignored : newDataTypes) { - // Newly extended columns are not included in the working TVList's per-block cost yet, so - // add their value-list placeholder and bitmap reservation to every acquired block. - memIncrements[0] += - acquireArray * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); - } } // Charge the payload and header of each value primitive array that this tablet actually // materializes. Arrays already present in the working TVList are not charged again. memIncrements[0] += - calculateTabletPrimitiveArrayMemCost( + calculateTabletValueArrayMemCost( alignedMemChunk, measurementIds, dataTypes, @@ -1255,7 +1238,6 @@ private void updateAlignedMemCost( end, currentPointNum); } - // flexible-length data size for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { TSDataType dataType = dataTypes[i]; @@ -1273,13 +1255,12 @@ private void updateAlignedMemCost( } /** - * Calculates only the value primitive arrays that an aligned tablet will materialize. The - * per-block timestamp arrays, value-list placeholders, bitmap reservations, headers, and - * references are charged by the caller. A column/block pair is charged only when the tablet has - * at least one successful non-null value in that block and the working TVList has not already - * allocated its value array. + * Calculates only the value arrays and list references that an aligned tablet will materialize. + * The per-block timestamp arrays are charged by the caller. A column/block pair is charged only + * when the tablet has at least one successful non-null value in that block and the working TVList + * has not already allocated its value array. */ - private static long calculateTabletPrimitiveArrayMemCost( + private static long calculateTabletValueArrayMemCost( AlignedWritableMemChunk alignedMemChunk, String[] measurementIds, TSDataType[] dataTypes, @@ -1308,9 +1289,9 @@ private static long calculateTabletPrimitiveArrayMemCost( if (containsNonNullValue(bitMap, results, inputIndex, length) && !isValueArrayMaterialized( alignedMemChunk, measurementIds[column], targetArrayIndex)) { - // One successful non-null write materializes the whole fixed-size value array, whose - // memory consists of the primitive payload and the array header. - size += AlignedTVList.primitiveArrayMemCost(dataTypes[column]); + // One successful non-null write materializes the fixed-size value array and its list + // reference. + size += AlignedTVList.valueListArrayMemCost(dataTypes[column]); } inputIndex += length; targetArrayIndex++; @@ -1395,6 +1376,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 { 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 644890d0bed3..a589e1ea97a6 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 @@ -72,8 +72,8 @@ public abstract class AlignedTVList extends TVList { - private static final long BITMAP_RAM_COST_PER_BLOCK = - BitMap.createBitMapDynamically(ARRAY_SIZE).ramBytesUsed() + NUM_BYTES_OBJECT_REF; + private static final long BITMAP_RAM_COST = + BitMap.createBitMapDynamically(ARRAY_SIZE).ramBytesUsed(); // Data types of this aligned tvList protected List dataTypes; @@ -84,6 +84,8 @@ public abstract class AlignedTVList extends TVList { // Number and memory cost of primitive arrays that have been allocated for each value column private int[] materializedValueArrayCounts; private long materializedValueArrayMemCost; + private long materializedBitmapMemoryCost; + private long arrayMemCostWithoutPrimitiveArraysAndIndex; // Data type list -> list of TVList, add 1 when expanded -> primitive array of basic type // Index relation: columnIndex(dataTypeIndex) -> arrayIndex -> elementIndex @@ -108,6 +110,7 @@ public abstract class AlignedTVList extends TVList { dataTypes = types; memoryBinaryChunkSize = new long[dataTypes.size()]; materializedValueArrayCounts = new int[dataTypes.size()]; + refreshArrayMemCostWithoutPrimitiveArrays(); values = new ArrayList<>(types.size()); for (int i = 0; i < types.size(); i++) { @@ -174,13 +177,14 @@ public TVList getTvListByColumnIndex( alignedTvList.allValueColDeletedMap = ignoreAllNullRows ? getAllValueColDeletedMap() : null; alignedTvList.timeColDeletedMap = this.timeColDeletedMap; alignedTvList.timeDeletedCnt = this.timeDeletedCnt; + alignedTvList.materializedBitmapMemoryCost = calculateBitmapRamCost(bitMaps); for (int i = 0; i < columnIndexList.size(); i++) { int columnIndex = columnIndexList.get(i); if (columnIndex != -1 && values.get(i) != null) { int materializedArrayCount = materializedValueArrayCounts[columnIndex]; alignedTvList.materializedValueArrayCounts[i] = materializedArrayCount; alignedTvList.materializedValueArrayMemCost += - (long) materializedArrayCount * primitiveArrayMemCost(dataTypeList.get(i)); + (long) materializedArrayCount * valueListArrayMemCost(dataTypeList.get(i)); } } @@ -199,6 +203,7 @@ public synchronized AlignedTVList cloneForFlushSort() { cloneList.materializedValueArrayCounts = Arrays.copyOf(materializedValueArrayCounts, materializedValueArrayCounts.length); cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost; + cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost; return cloneList; } @@ -237,6 +242,7 @@ public synchronized AlignedTVList clone() { cloneList.materializedValueArrayCounts = Arrays.copyOf(materializedValueArrayCounts, materializedValueArrayCounts.length); cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost; + cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost; return cloneList; } @@ -428,9 +434,12 @@ public void extendColumn(TSDataType dataType) { } columnBitMaps.add(bitMap); } + materializedBitmapMemoryCost += + (long) timestamps.size() * (bitmapReferenceRamCost() + bitmapRamCost()); this.bitMaps.add(columnBitMaps); this.values.add(columnValue); this.dataTypes.add(dataType); + refreshArrayMemCostWithoutPrimitiveArrays(); long[] tmpValueChunkRawSize = memoryBinaryChunkSize; memoryBinaryChunkSize = new long[dataTypes.size()]; @@ -726,10 +735,13 @@ public void deleteColumn(int columnIndex) { columnBitMaps.add(BitMap.createBitMapDynamically(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, BitMap.createBitMapDynamically(ARRAY_SIZE)); + materializedBitmapMemoryCost += bitmapRamCost(); } bitMaps.get(columnIndex).get(i).markAll(); } @@ -808,6 +820,7 @@ protected void clearBitMap() { } } } + materializedBitmapMemoryCost = 0; } @Override @@ -819,6 +832,7 @@ protected void expandValues() { values.get(i).add(null); if (bitMaps != null && bitMaps.get(i) != null) { bitMaps.get(i).add(null); + materializedBitmapMemoryCost += bitmapReferenceRamCost(); } } } @@ -1089,7 +1103,7 @@ private Object getOrCreateValueArray(int columnIndex, int arrayIndex) { valueArray = getPrimitiveArraysByType(dataTypes.get(columnIndex)); columnValues.set(arrayIndex, valueArray); materializedValueArrayCounts[columnIndex]++; - materializedValueArrayMemCost += primitiveArrayMemCost(dataTypes.get(columnIndex)); + materializedValueArrayMemCost += valueListArrayMemCost(dataTypes.get(columnIndex)); } return valueArray; } @@ -1111,11 +1125,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, BitMap.createBitMapDynamically(ARRAY_SIZE)); + materializedBitmapMemoryCost += bitmapRamCost(); } return bitMaps.get(columnIndex).get(arrayIndex); @@ -1142,12 +1158,36 @@ public synchronized RamInfo calculateRamSize() { return new RamInfo( timestamps.size(), alignedTvListArrayMemCost(), - (long) timestamps.size() * alignedTvListArrayMemCostWithoutPrimitiveArrays() - + materializedValueArrayMemCost, + getRamSize(), rowCount, new ArrayList<>(dataTypes)); } + public synchronized long getRamSize() { + return (long) timestamps.size() * alignedTvListArrayMemCostWithoutPrimitiveArrays() + + materializedValueArrayMemCost + + materializedBitmapMemoryCost; + } + + 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 += bitMap.ramBytesUsed(); + } + } + } + return size; + } + /** * Get the single alignedTVList array mem cost by give types. * @@ -1165,7 +1205,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++; } } @@ -1194,7 +1233,6 @@ public long alignedTvListArrayMemCost() { 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 @@ -1213,13 +1251,21 @@ public long alignedTvListArrayMemCost() { } public long alignedTvListArrayMemCostWithoutPrimitiveArrays() { + return arrayMemCostWithoutPrimitiveArraysAndIndex + + (indices != null ? (long) PrimitiveArrayManager.ARRAY_SIZE * Integer.BYTES : 0); + } + + private void refreshArrayMemCostWithoutPrimitiveArrays() { long size = alignedTvListArrayMemCost(); + if (indices != null) { + size -= (long) PrimitiveArrayManager.ARRAY_SIZE * Integer.BYTES; + } for (TSDataType dataType : dataTypes) { if (dataType != null) { - size -= primitiveArrayMemCost(dataType); + size -= valueListArrayMemCost(dataType); } } - return size; + arrayMemCostWithoutPrimitiveArraysAndIndex = size; } public static long alignedTvListArrayMemCostWithoutPrimitiveArrays( @@ -1229,7 +1275,7 @@ public static long alignedTvListArrayMemCostWithoutPrimitiveArrays( TSDataType dataType = types[i]; if (dataType != null && (columnCategories == null || columnCategories[i] == TsTableColumnCategory.FIELD)) { - size -= primitiveArrayMemCost(dataType); + size -= valueListArrayMemCost(dataType); } } return size; @@ -1242,7 +1288,9 @@ public static long alignedTvListArrayMemCostWithoutPrimitiveArrays( * @return valueListArrayMemCost */ public static long valueListArrayMemCost(TSDataType type) { - return primitiveArrayMemCost(type) + valueListArrayMemCostWithoutPrimitiveArray(); + // Charge the reference only after the value array is materialized. Null historical + // placeholders do not contribute to write-memory accounting. + return primitiveArrayMemCost(type) + NUM_BYTES_OBJECT_REF; } public static long primitiveArrayMemCost(TSDataType type) { @@ -1251,13 +1299,12 @@ public static long primitiveArrayMemCost(TSDataType type) { + NUM_BYTES_ARRAY_HEADER; } - public static long valueListArrayMemCostWithoutPrimitiveArray() { - long size = 0; - // bitmap object, byte array, and reference in the bitmap list - size += BITMAP_RAM_COST_PER_BLOCK; - // null placeholder reference in the value ArrayList - size += NUM_BYTES_OBJECT_REF; - return size; + 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/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 000000000000..8a872f6af61e --- /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] = BitMap.createBitMapDynamically(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 3a85754a71c7..a8193a9c0b28 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; @@ -63,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; @@ -537,21 +539,21 @@ public void alignedTvListRamCostTest() true, new long[5]); IMemTable memTable = processor.getWorkMemTable(); - Assert.assertEquals(1752552, 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(1752552, 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(1752552, memTable.getTVListsRamCost()); + Assert.assertEquals(1596552, memTable.getTVListsRamCost()); Assert.assertEquals(90000, memTable.getTotalPointsNum()); Assert.assertEquals(720360, memTable.memSize()); // Test records @@ -560,7 +562,7 @@ public void alignedTvListRamCostTest() record.addTuple(DataPoint.getDataPoint(dataType, measurementId, String.valueOf(i))); processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); } - Assert.assertEquals(1754168, memTable.getTVListsRamCost()); + Assert.assertEquals(1598168, memTable.getTVListsRamCost()); Assert.assertEquals(90100, memTable.getTotalPointsNum()); Assert.assertEquals(721560, memTable.memSize()); } @@ -587,7 +589,9 @@ public void alignedTabletDoesNotChargePrimitiveArrayForFailedOnlyBlock() Assert.assertEquals( expectedProcessor.getWorkMemTable().getTVListsRamCost() - - AlignedTVList.primitiveArrayMemCost(dataType), + - AlignedTVList.valueListArrayMemCost(dataType) + + 2 * AlignedTVList.bitmapReferenceRamCost() + + AlignedTVList.bitmapRamCost(), actualProcessor.getWorkMemTable().getTVListsRamCost()); Assert.assertEquals( TSStatusCode.OUT_OF_TTL.getStatusCode(), actualResults[failedIndex].getCode()); @@ -616,7 +620,10 @@ public void alignedTabletOnlyChargesMaterializedPrimitiveArrays() AlignedTVList.alignedTvListArrayMemCost( new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null); Assert.assertEquals( - denseBlockCost - AlignedTVList.primitiveArrayMemCost(TSDataType.INT32), + denseBlockCost + - AlignedTVList.valueListArrayMemCost(TSDataType.INT32) + + 2 * AlignedTVList.bitmapReferenceRamCost() + + AlignedTVList.bitmapRamCost(), processor.getWorkMemTable().getTVListsRamCost() - ramCostBeforeNewBlock); } @@ -642,10 +649,63 @@ public void alignedRowOnlyChargesMaterializedPrimitiveArrays() AlignedTVList.alignedTvListArrayMemCost( new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null); Assert.assertEquals( - denseBlockCost - AlignedTVList.primitiveArrayMemCost(TSDataType.INT32), + denseBlockCost + - AlignedTVList.valueListArrayMemCost(TSDataType.INT32) + + 2 * AlignedTVList.bitmapReferenceRamCost() + + AlignedTVList.bitmapRamCost(), processor.getWorkMemTable().getTVListsRamCost() - ramCostBeforeNewBlock); } + @Test + public void alignedBitmapMemoryAccountingMatchesActualAllocations() + throws MetadataException, WriteProcessException, IOException, IllegalPathException { + processor = newTestProcessor(filePath + ".bitmap-accounting"); + int denseRowCount = PrimitiveArrayManager.ARRAY_SIZE * 2 + 1; + processor.insertTablet( + genAlignedTablet(new String[] {"s0", "s1"}, denseRowCount, 0), + Collections.singletonList(new int[] {0, denseRowCount}), + new TSStatus[denseRowCount], + true, + new long[5]); + + AlignedWritableMemChunk alignedMemChunk = getAlignedMemChunk(deviceId); + Assert.assertNull(alignedMemChunk.getWorkingTVList().getBitMaps()); + assertAlignedTvListRamCostMatchesActual(deviceId); + + 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(deviceId); + + 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.getMeasurementIndex("s2"); + Assert.assertNull( + alignedMemChunk.getWorkingTVList().getValues().get(extendedColumnIndex).get(0)); + Assert.assertNull( + alignedMemChunk.getWorkingTVList().getValues().get(extendedColumnIndex).get(1)); + for (BitMap bitMap : alignedMemChunk.getWorkingTVList().getBitMaps().get(extendedColumnIndex)) { + Assert.assertNotNull(bitMap); + } + assertAlignedTvListRamCostMatchesActual(deviceId); + } + @Test public void alignedTvListRamCostTest2() throws MetadataException, WriteProcessException, IOException { @@ -669,7 +729,7 @@ public void alignedTvListRamCostTest2() true, new long[5]); IMemTable memTable = processor.getWorkMemTable(); - Assert.assertEquals(1752552, memTable.getTVListsRamCost()); + Assert.assertEquals(1596552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNodeFors3000ToS6000(0, true), Collections.singletonList(new int[] {0, 10}), @@ -711,7 +771,7 @@ public void alignedTvListRamCostTest2() new TSStatus[10], true, new long[5]); - Assert.assertEquals(5425104, memTable.getTVListsRamCost()); + Assert.assertEquals(5269104, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNodeFors3000ToS6000(300, true), Collections.singletonList(new int[] {0, 10}), @@ -1009,7 +1069,9 @@ public void testAlignedSparseRowDoesNotChargeUnallocatedPrimitiveArrayOnNewBlock long denseRowRamIncrement = memTable.getTVListsRamCost() - ramCostBeforeDenseRow; Assert.assertEquals( - AlignedTVList.primitiveArrayMemCost(dataType), + AlignedTVList.valueListArrayMemCost(dataType) + - 2 * AlignedTVList.bitmapReferenceRamCost() + - AlignedTVList.bitmapRamCost(), denseRowRamIncrement - sparseRowRamIncrement); } @@ -1404,6 +1466,23 @@ private void insertAlignedRow( targetProcessor.insert(node, new long[5]); } + private AlignedWritableMemChunk getAlignedMemChunk(String targetDevice) { + IWritableMemChunk memChunk = + processor + .getWorkMemTable() + .getWritableMemChunk( + IDeviceID.Factory.DEFAULT_FACTORY.create(targetDevice), + AlignedPath.VECTOR_PLACEHOLDER); + Assert.assertNotNull(memChunk); + return (AlignedWritableMemChunk) memChunk; + } + + private void assertAlignedTvListRamCostMatchesActual(String targetDevice) { + Assert.assertEquals( + getAlignedMemChunk(targetDevice).getWorkingTVList().calculateRamSize().getRamSize(), + processor.getWorkMemTable().getTVListsRamCost()); + } + private InsertTabletNode genSingleMeasurementTablet(int rowCount, boolean isAligned) throws IllegalPathException { String[] measurements = new String[] {measurementId}; 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 06e422ad776a..381f733bf4dd 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 @@ -45,14 +45,13 @@ public class AlignedTVListTest { @Test - public void testValueListArrayMemCostIncludesBitmapImplementation() { - long expected = - (long) ARRAY_SIZE * Long.BYTES - + BitMap.createBitMapDynamically(ARRAY_SIZE).ramBytesUsed() - + NUM_BYTES_ARRAY_HEADER - + 2L * NUM_BYTES_OBJECT_REF; + public void testValueListArrayMemCostExcludesLazyBitmapAndNullPlaceholder() { + long expected = (long) ARRAY_SIZE * Long.BYTES + NUM_BYTES_ARRAY_HEADER + NUM_BYTES_OBJECT_REF; Assert.assertEquals(expected, AlignedTVList.valueListArrayMemCost(TSDataType.INT64)); + Assert.assertEquals( + BitMap.createBitMapDynamically(ARRAY_SIZE).ramBytesUsed(), AlignedTVList.bitmapRamCost()); + Assert.assertEquals(NUM_BYTES_OBJECT_REF, AlignedTVList.bitmapReferenceRamCost()); } @Test @@ -189,6 +188,9 @@ public void testBitmapIsAllocatedLazilyWithCompactBackingArray() { firstColumnBitMaps.get(2).ramBytesUsed() < new BitMap(ARRAY_SIZE).ramBytesUsed()); Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE * 2 + 1, 0)); Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE * 2, 0)); + Assert.assertEquals( + 3L * AlignedTVList.bitmapReferenceRamCost() + AlignedTVList.bitmapRamCost(), + tvList.calculateRamSize().getRamSize() - 3L * tvList.alignedTvListArrayMemCost()); } @Test @@ -263,10 +265,14 @@ public void testPrimitiveArraysAreAllocatedOnFirstWrite() { Assert.assertTrue(tvList.isNullValue(0, 1)); Assert.assertEquals(1, tvList.getLongByValueIndex(ARRAY_SIZE + 1, 1)); + long ramSizeBeforeExtension = tvList.calculateRamSize().getRamSize(); tvList.extendColumn(TSDataType.INT32); Assert.assertNull(tvList.getValues().get(2).get(0)); Assert.assertNull(tvList.getValues().get(2).get(1)); + Assert.assertEquals( + 2L * (AlignedTVList.bitmapReferenceRamCost() + AlignedTVList.bitmapRamCost()), + tvList.calculateRamSize().getRamSize() - ramSizeBeforeExtension); long ramSizeBeforeExtendedColumnMaterialization = tvList.calculateRamSize().getRamSize(); tvList.putAlignedValue(ARRAY_SIZE + 2L, new Object[] {null, null, 2}); @@ -277,7 +283,7 @@ public void testPrimitiveArraysAreAllocatedOnFirstWrite() { Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE + 2, 2)); Assert.assertEquals(2, tvList.getIntByValueIndex(ARRAY_SIZE + 2, 2)); Assert.assertEquals( - AlignedTVList.primitiveArrayMemCost(TSDataType.INT32), + AlignedTVList.valueListArrayMemCost(TSDataType.INT32), tvList.calculateRamSize().getRamSize() - ramSizeBeforeExtendedColumnMaterialization); } @@ -293,7 +299,7 @@ public void testCalculateRamSizeCountsMaterializedPrimitiveArrays() { tvList.putAlignedValue(ARRAY_SIZE + 1L, new Object[] {1L, 1L}); Assert.assertEquals( - AlignedTVList.primitiveArrayMemCost(TSDataType.INT64), + AlignedTVList.valueListArrayMemCost(TSDataType.INT64), tvList.calculateRamSize().getRamSize() - ramSizeBeforeMaterialization); Assert.assertEquals( @@ -307,7 +313,9 @@ public void testCalculateRamSizeCountsMaterializedPrimitiveArrays() { Assert.assertEquals( (long) projectedTvList.getValues().get(0).size() * projectedTvList.alignedTvListArrayMemCostWithoutPrimitiveArrays() - + AlignedTVList.primitiveArrayMemCost(TSDataType.INT64), + + AlignedTVList.valueListArrayMemCost(TSDataType.INT64) + + (long) projectedTvList.getBitMaps().get(0).size() + * (AlignedTVList.bitmapReferenceRamCost() + AlignedTVList.bitmapRamCost()), projectedTvList.calculateRamSize().getRamSize()); tvList.clear(); @@ -325,7 +333,10 @@ public void testCalculateRamSizeExcludesUnallocatedPrimitiveArrays() { int blockCount = tvList.getValues().get(0).size(); long denseRamSize = blockCount * tvList.alignedTvListArrayMemCost(); long expectedRamSize = - denseRamSize - blockCount * AlignedTVList.primitiveArrayMemCost(TSDataType.INT64); + denseRamSize + - blockCount * AlignedTVList.valueListArrayMemCost(TSDataType.INT64) + + (long) blockCount + * (AlignedTVList.bitmapReferenceRamCost() + AlignedTVList.bitmapRamCost()); Assert.assertEquals(expectedRamSize, tvList.calculateRamSize().getRamSize()); }