This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new cdefafa4dfb [Fix] Reconcile actual aligned TVList memory usage
(#18330) (#18339)
cdefafa4dfb is described below
commit cdefafa4dfbf46769e6804fc22476d34dfb1729e
Author: Caideyipi <[email protected]>
AuthorDate: Thu Jul 30 10:42:54 2026 +0800
[Fix] Reconcile actual aligned TVList memory usage (#18330) (#18339)
---
.../dataregion/memtable/TsFileProcessor.java | 135 ++++++++--
.../db/utils/datastructure/AlignedTVList.java | 86 ++++--
.../iotdb/db/utils/datastructure/TVList.java | 13 +-
...ignedBitmapMemoryAccountingPerformanceTest.java | 293 +++++++++++++++++++++
.../dataregion/memtable/TsFileProcessorTest.java | 116 +++++++-
.../db/utils/datastructure/AlignedTVListTest.java | 44 ++++
6 files changed, 646 insertions(+), 41 deletions(-)
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 8c966231fb1..fff95a33d86 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.ArrayList;
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 class TsFileProcessor {
.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 class TsFileProcessor {
.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 class TsFileProcessor {
}
long[] memIncrements;
+ long alignedMemTableIncrement = 0;
+ Set<IDeviceID> 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 class TsFileProcessor {
}
}
long[] alignedMemIncrements =
checkAlignedMemCostAndAddToTspInfoForRows(alignedList);
+ alignedMemTableIncrement = alignedMemIncrements[0];
final long[] nonAlignedMemIncrements;
try {
nonAlignedMemIncrements =
checkMemCostAndAddToTspInfoForRows(nonAlignedList);
@@ -391,6 +412,7 @@ public class TsFileProcessor {
if (insertRowsNode.isAligned()) {
memIncrements =
checkAlignedMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList());
+ alignedMemTableIncrement = memIncrements[0];
} else {
memIncrements =
checkMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList());
}
@@ -435,20 +457,24 @@ public class TsFileProcessor {
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 class TsFileProcessor {
.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 class TsFileProcessor {
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 @@ public class TsFileProcessor {
}
}
+ 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<IDeviceID> 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<IDeviceID> 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<IDeviceID>
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 9e3a556710c..958b8c4e657 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 @@ import static
org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF;
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<TSDataType> 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<List<Object>> 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 abstract class AlignedTVList extends TVList {
alignedTvList.bitMaps = bitMaps;
alignedTvList.rowCount = this.rowCount;
alignedTvList.allValueColDeletedMap = getAllValueColDeletedMap();
+ alignedTvList.materializedBitmapMemoryCost =
calculateBitmapRamCost(bitMaps);
return alignedTvList;
}
@@ -150,6 +154,7 @@ public abstract class AlignedTVList extends TVList {
cloneList.memoryBinaryChunkSize = this.memoryBinaryChunkSize;
cloneList.values = this.values;
cloneList.bitMaps = this.bitMaps;
+ cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost;
return cloneList;
}
@@ -183,6 +188,7 @@ public abstract class AlignedTVList extends TVList {
}
}
}
+ cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost;
return cloneList;
}
@@ -410,6 +416,9 @@ public abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
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 @@ public abstract class AlignedTVList extends TVList {
}
}
}
+ materializedBitmapMemoryCost = 0;
}
@Override
@@ -676,6 +689,7 @@ public abstract class AlignedTVList extends TVList {
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 @@ public abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
@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<List<BitMap>> bitMaps) {
+ if (bitMaps == null) {
+ return 0;
+ }
+ long size = 0;
+ for (List<BitMap> 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 abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
*/
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 abstract class AlignedTVList extends TVList {
* @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 d21ecc83e0f..20e4a71932f 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<TSDataType> dataTypes;
public RamInfo(
int timestampCount, long arrayMemCost, int rowCount, List<TSDataType>
dataTypes) {
+ this(timestampCount, arrayMemCost, (long) timestampCount * arrayMemCost,
rowCount, dataTypes);
+ }
+
+ public RamInfo(
+ int timestampCount,
+ long arrayMemCost,
+ long ramSize,
+ int rowCount,
+ List<TSDataType> 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 00000000000..57c52727ad8
--- /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<IMeasurementSchema> 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<IMeasurementSchema> schemas;
+ private final long[] times;
+
+ private Scenario(
+ String[] measurements,
+ TSDataType[] dataTypes,
+ Object[] columns,
+ BitMap[] bitMaps,
+ List<IMeasurementSchema> 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 658faa971a3..db3b84dbf58 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.queryengine.plan.planner.plan.node.write.InsertTablet
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.common.RowRecord;
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 class TsFileProcessorTest {
// 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 class TsFileProcessorTest {
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 class TsFileProcessorTest {
// 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 class TsFileProcessorTest {
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 class TsFileProcessorTest {
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<BitMap> 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 @@ public class TsFileProcessorTest {
}
}
+ 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 6a7206d39c5..e27dc009f63 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.common.conf.TSFileConfig;
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.Arrays;
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<TSDataType> dataTypes = new ArrayList<>();
@@ -166,6 +181,35 @@ public class AlignedTVListTest {
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
