From beaf2d01dfe040ab8026e0ead17c0f782652e59f 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] [Fix] Reconcile actual aligned TVList memory usage (#18330) --- .../dataregion/memtable/TsFileProcessor.java | 135 +++++++- .../db/utils/datastructure/AlignedTVList.java | 86 ++++- .../iotdb/db/utils/datastructure/TVList.java | 13 +- ...BitmapMemoryAccountingPerformanceTest.java | 293 ++++++++++++++++++ .../memtable/TsFileProcessorTest.java | 116 ++++++- .../datastructure/AlignedTVListTest.java | 44 +++ 6 files changed, 646 insertions(+), 41 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 8c966231fb1ff..fff95a33d86de 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 @@ -100,6 +100,7 @@ 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; @@ -276,6 +277,10 @@ public void insert(InsertRowNode insertRowNode, long[] costsForMetrics) .recordActiveMemTableCount(dataRegionInfo.getDataRegion().getDataRegionIdString(), 1); } + AlignedTVListRamCostSnapshot alignedRamCostSnapshot = + insertRowNode.isAligned() + ? new AlignedTVListRamCostSnapshot(workMemTable, insertRowNode.getDeviceID()) + : null; long[] memIncrements; long memControlStartTime = System.nanoTime(); @@ -329,11 +334,15 @@ public void insert(InsertRowNode insertRowNode, long[] costsForMetrics) .listenToInsertNode( dataRegionInfo.getDataRegion().getDataRegionIdString(), 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 @@ -363,6 +372,17 @@ public void insert(InsertRowsNode insertRowsNode, long[] costsForMetrics) } 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()) { @@ -376,6 +396,7 @@ public void insert(InsertRowsNode insertRowsNode, long[] costsForMetrics) } } long[] alignedMemIncrements = checkAlignedMemCostAndAddToTspInfoForRows(alignedList); + alignedMemTableIncrement = alignedMemIncrements[0]; final long[] nonAlignedMemIncrements; try { nonAlignedMemIncrements = checkMemCostAndAddToTspInfoForRows(nonAlignedList); @@ -391,6 +412,7 @@ public void insert(InsertRowsNode insertRowsNode, long[] costsForMetrics) if (insertRowsNode.isAligned()) { memIncrements = checkAlignedMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList()); + alignedMemTableIncrement = memIncrements[0]; } else { memIncrements = checkMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList()); } @@ -435,20 +457,24 @@ public void insert(InsertRowsNode insertRowsNode, long[] costsForMetrics) dataRegionInfo.getDataRegion().getDataRegionIdString(), insertRowsNode, tsFileResource); int pointInserted = 0; - for (InsertRowNode insertRowNode : insertRowsNode.getInsertRowNodeList()) { - if (insertRowNode.isAligned()) { - pointInserted += workMemTable.insertAlignedRow(insertRowNode); - } else { - pointInserted += workMemTable.insert(insertRowNode); - } + 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()); + // 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); } workMemTable.updateMemtablePointCountMetric(insertRowsNode, pointInserted); tsFileResource.updateProgressIndex(insertRowsNode.getProgressIndex()); @@ -487,6 +513,10 @@ public void insertTablet( .recordActiveMemTableCount(dataRegionInfo.getDataRegion().getDataRegionIdString(), 1); } + AlignedTVListRamCostSnapshot alignedRamCostSnapshot = + insertTabletNode.isAligned() + ? new AlignedTVListRamCostSnapshot(workMemTable, insertTabletNode.getDeviceID()) + : null; long[] memIncrements; try { long startTime = System.nanoTime(); @@ -564,6 +594,8 @@ public void insertTablet( results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage()); } throw new WriteProcessException(e); + } finally { + reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements[0]); } for (int i = start; i < end; i++) { results[i] = RpcUtils.SUCCESS_STATUS; @@ -994,6 +1026,75 @@ private void updateAlignedMemCost( } } + 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 9e3a556710cc9..958b8c4e65775 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 @@ -65,10 +65,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(ARRAY_SIZE / Byte.SIZE + 1) - + NUM_BYTES_OBJECT_REF; + + RamUsageEstimator.sizeOfByteArray(ARRAY_SIZE / Byte.SIZE + 1); // Data types of this aligned tvList protected List dataTypes; @@ -76,6 +75,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; @@ -94,6 +96,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++) { @@ -140,6 +143,7 @@ public TVList getTvListByColumnIndex(List columnIndex, List alignedTvList.bitMaps = bitMaps; alignedTvList.rowCount = this.rowCount; alignedTvList.allValueColDeletedMap = getAllValueColDeletedMap(); + alignedTvList.materializedBitmapMemoryCost = calculateBitmapRamCost(bitMaps); return alignedTvList; } @@ -150,6 +154,7 @@ public synchronized AlignedTVList cloneForFlushSort() { cloneList.memoryBinaryChunkSize = this.memoryBinaryChunkSize; cloneList.values = this.values; cloneList.bitMaps = this.bitMaps; + cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost; return cloneList; } @@ -183,6 +188,7 @@ public synchronized AlignedTVList clone() { } } } + cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost; return cloneList; } @@ -410,6 +416,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()]; @@ -591,10 +600,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(); } @@ -665,6 +677,7 @@ protected void clearBitMap() { } } } + materializedBitmapMemoryCost = 0; } @Override @@ -676,6 +689,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(); } } } @@ -856,11 +870,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); @@ -879,7 +895,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; } /** @@ -895,7 +948,6 @@ public static long alignedTvListArrayMemCost(TSDataType[] types) { for (TSDataType type : types) { if (type != null) { size += (long) ARRAY_SIZE * (long) type.getDataTypeSize(); - size += BITMAP_RAM_COST_PER_BLOCK; measurementColumnNum++; } } @@ -919,12 +971,11 @@ public static long alignedTvListArrayMemCost(TSDataType[] types) { */ 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 @@ -949,16 +1000,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 d21ecc83e0f1b..20e4a71932fa7 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 @@ -64,19 +64,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..57c52727ad883 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java @@ -0,0 +1,293 @@ +/* + * 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.file.metadata.PlainDeviceID; +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); + } + 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)); + } + return memChunks; + } + + private static AccountingTarget createAccountingTarget(Scenario scenario) { + AlignedWritableMemChunk memChunk = + new AlignedWritableMemChunk(new ArrayList<>(scenario.schemas)); + memChunk.writeAlignedTablet( + scenario.times, + scenario.columns, + scenario.bitMaps, + scenario.schemas, + 0, + scenario.times.length); + IDeviceID deviceId = new PlainDeviceID("root.accounting.d0"); + IMemTable memTable = + new PrimitiveMemTable( + "root.accounting", + "0", + Collections.singletonMap( + deviceId, + new AlignedWritableMemChunkGroup(memChunk, new ArrayList<>(scenario.schemas)))); + 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 658faa971a356..db3b84dbf583f 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 @@ -39,6 +39,7 @@ import org.apache.iotdb.db.storageengine.dataregion.DataRegionInfo; import org.apache.iotdb.db.storageengine.dataregion.DataRegionTest; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager; import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.db.utils.constant.TestConstant; @@ -58,6 +59,7 @@ import org.apache.tsfile.read.expression.QueryExpression; import org.apache.tsfile.read.query.dataset.QueryDataSet; import org.apache.tsfile.read.reader.IPointReader; +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; @@ -686,11 +688,11 @@ public void alignedTvListRamCostTest() // Test Tablet processor.insertTablet(genInsertTableNode(0, true), 0, 10, new TSStatus[10]); IMemTable memTable = processor.getWorkMemTable(); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1596552, memTable.getTVListsRamCost()); processor.insertTablet(genInsertTableNode(100, true), 0, 10, new TSStatus[10]); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1596552, memTable.getTVListsRamCost()); processor.insertTablet(genInsertTableNode(200, true), 0, 10, new TSStatus[10]); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1596552, memTable.getTVListsRamCost()); Assert.assertEquals(90000, memTable.getTotalPointsNum()); Assert.assertEquals(720360, memTable.memSize()); // Test records @@ -699,7 +701,7 @@ public void alignedTvListRamCostTest() record.addTuple(DataPoint.getDataPoint(dataType, measurementId, String.valueOf(i))); processor.insert(buildInsertRowNodeByTSRecord(record), new long[4]); } - Assert.assertEquals(1778168, memTable.getTVListsRamCost()); + Assert.assertEquals(1598168, memTable.getTVListsRamCost()); Assert.assertEquals(90100, memTable.getTotalPointsNum()); Assert.assertEquals(721560, memTable.memSize()); } @@ -722,7 +724,7 @@ public void alignedTvListRamCostTest2() // Test Tablet processor.insertTablet(genInsertTableNode(0, true), 0, 10, new TSStatus[10]); IMemTable memTable = processor.getWorkMemTable(); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1596552, memTable.getTVListsRamCost()); processor.insertTablet(genInsertTableNodeFors3000ToS6000(0, true), 0, 10, new TSStatus[10]); Assert.assertEquals(3552552, memTable.getTVListsRamCost()); processor.insertTablet(genInsertTableNode(100, true), 0, 10, new TSStatus[10]); @@ -734,7 +736,7 @@ public void alignedTvListRamCostTest2() processor.insertTablet(genInsertTableNodeFors3000ToS6000(200, true), 0, 10, new TSStatus[10]); Assert.assertEquals(3552552, memTable.getTVListsRamCost()); processor.insertTablet(genInsertTableNode(300, true), 0, 10, new TSStatus[10]); - Assert.assertEquals(7105104, memTable.getTVListsRamCost()); + Assert.assertEquals(6937104, memTable.getTVListsRamCost()); processor.insertTablet(genInsertTableNodeFors3000ToS6000(300, true), 0, 10, new TSStatus[10]); Assert.assertEquals(7105104, memTable.getTVListsRamCost()); @@ -758,6 +760,55 @@ 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, 0, denseRowCount, new TSStatus[denseRowCount]); + + 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, 0, 2, new TSStatus[2]); + + 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(denseRowCount + 2L, deviceId); + extendedColumnRecord.addTuple(DataPoint.getDataPoint(TSDataType.INT32, "s2", "1")); + InsertRowNode extendedColumnRow = buildInsertRowNodeByTSRecord(extendedColumnRecord); + extendedColumnRow.setAligned(true); + processor.insert(extendedColumnRow, new long[4]); + + 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 { @@ -1183,6 +1234,59 @@ private void closeTsFileProcessor(TsFileProcessor unsealedTsFileProcessor) } } + 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 6a7206d39c584..e27dc009f637e 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 @@ -23,6 +23,7 @@ import org.apache.tsfile.enums.TSDataType; 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; @@ -31,9 +32,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(ARRAY_SIZE / Byte.SIZE + 1), + AlignedTVList.bitmapRamCost()); + Assert.assertEquals(NUM_BYTES_OBJECT_REF, AlignedTVList.bitmapReferenceRamCost()); + } + @Test public void testAlignedTVList1() { List dataTypes = new ArrayList<>(); @@ -166,6 +181,35 @@ public void testBitmapIsAllocatedLazilyWithCompactBackingArray() { ARRAY_SIZE / Byte.SIZE + 1, 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