You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/08/25 06:27:04 UTC
(iotdb) branch to-20111-dd created (now dfac357eb57)
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a change to branch to-20111-dd
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at dfac357eb57 [Fix] Reconcile actual aligned TVList memory usage (#18330)
This branch includes the following new commits:
new dfac357eb57 [Fix] Reconcile actual aligned TVList memory usage (#18330)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: [Fix] Reconcile actual aligned TVList memory usage (#18330)
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch to-20111-dd
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit dfac357eb57e14e978b88d07c368ea8fb1ddedbf
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Wed Jul 29 10:04:39 2026 +0800
[Fix] Reconcile actual aligned TVList memory usage (#18330)
---
.../dataregion/memtable/TsFileProcessor.java | 245 +++++++++++++----
.../db/utils/datastructure/AlignedTVList.java | 87 ++++--
.../iotdb/db/utils/datastructure/TVList.java | 13 +-
...ignedBitmapMemoryAccountingPerformanceTest.java | 295 +++++++++++++++++++++
.../dataregion/memtable/TsFileProcessorTest.java | 126 ++++++++-
.../db/utils/datastructure/AlignedTVListTest.java | 44 +++
6 files changed, 731 insertions(+), 79 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 ab2066c1c6a..708e2b2a5f7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -104,9 +104,11 @@ import java.util.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;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -292,6 +294,10 @@ public class TsFileProcessor {
ensureMemTable(infoForMetrics);
workMemTable.checkDataType(insertRowNode);
+ AlignedTVListRamCostSnapshot alignedRamCostSnapshot =
+ insertRowNode.isAligned()
+ ? new AlignedTVListRamCostSnapshot(workMemTable, insertRowNode.getDeviceID())
+ : null;
long[] memIncrements;
long memControlStartTime = System.nanoTime();
@@ -356,11 +362,15 @@ public class TsFileProcessor {
insertRowNode,
tsFileResource);
- int pointInserted;
- if (insertRowNode.isAligned()) {
- pointInserted = workMemTable.insertAlignedRow(insertRowNode);
- } else {
- pointInserted = workMemTable.insert(insertRowNode);
+ int pointInserted = 0;
+ try {
+ if (insertRowNode.isAligned()) {
+ pointInserted = workMemTable.insertAlignedRow(insertRowNode);
+ } else {
+ pointInserted = workMemTable.insert(insertRowNode);
+ }
+ } finally {
+ reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements[0]);
}
// Update start time of this memtable
@@ -385,6 +395,17 @@ public class TsFileProcessor {
workMemTable.checkDataType(insertRowsNode);
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()) {
@@ -398,7 +419,14 @@ public class TsFileProcessor {
}
}
long[] alignedMemIncrements = checkAlignedMemCostAndAddToTspInfoForRows(alignedList);
- long[] nonAlignedMemIncrements = checkMemCostAndAddToTspInfoForRows(nonAlignedList);
+ alignedMemTableIncrement = alignedMemIncrements[0];
+ final long[] nonAlignedMemIncrements;
+ try {
+ nonAlignedMemIncrements = checkMemCostAndAddToTspInfoForRows(nonAlignedList);
+ } catch (final WriteProcessException e) {
+ rollbackMemoryInfoIfNeeded(alignedMemIncrements);
+ throw e;
+ }
memIncrements = new long[3];
for (int i = 0; i < 3; i++) {
memIncrements[i] = alignedMemIncrements[i] + nonAlignedMemIncrements[i];
@@ -407,6 +435,7 @@ public class TsFileProcessor {
if (insertRowsNode.isAligned()) {
memIncrements =
checkAlignedMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList());
+ alignedMemTableIncrement = memIncrements[0];
} else {
memIncrements = checkMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList());
}
@@ -456,19 +485,24 @@ public class TsFileProcessor {
tsFileResource);
int pointInserted = 0;
- for (InsertRowNode insertRowNode : insertRowsNode.getInsertRowNodeList()) {
- if (insertRowNode.isAligned()) {
- pointInserted += workMemTable.insertAlignedRow(insertRowNode);
- } else {
- pointInserted += workMemTable.insert(insertRowNode);
- }
- // update start time of this memtable
- tsFileResource.updateStartTime(insertRowNode.getDeviceID(), insertRowNode.getTime());
- // for sequence tsfile, we update the endTime only when the file is prepared to be closed.
- // for unsequence tsfile, we have to update the endTime for each insertion.
- if (!sequence) {
- tsFileResource.updateEndTime(insertRowNode.getDeviceID(), insertRowNode.getTime());
+ try {
+ for (InsertRowNode insertRowNode : insertRowsNode.getInsertRowNodeList()) {
+ if (insertRowNode.isAligned()) {
+ pointInserted += workMemTable.insertAlignedRow(insertRowNode);
+ } else {
+ pointInserted += workMemTable.insert(insertRowNode);
+ }
+
+ // update start time of this memtable
+ tsFileResource.updateStartTime(insertRowNode.getDeviceID(), insertRowNode.getTime());
+ // for sequence tsfile, we update the endTime only when the file is prepared to be closed.
+ // for unsequence tsfile, we have to update the endTime for each insertion.
+ if (!sequence) {
+ tsFileResource.updateEndTime(insertRowNode.getDeviceID(), insertRowNode.getTime());
+ }
}
+ } finally {
+ reconcileAlignedTVListRamCost(alignedRamCostSnapshot, alignedMemTableIncrement);
}
tsFileResource.updateProgressIndex(insertRowsNode.getProgressIndex());
@@ -583,6 +617,20 @@ public class TsFileProcessor {
ensureMemTable(infoForMetrics);
workMemTable.checkDataType(insertTabletNode);
+ Set<IDeviceID> alignedDeviceIds = new HashSet<>();
+ if (insertTabletNode.isAligned()) {
+ for (int[] range : rangeList) {
+ for (Pair<IDeviceID, Integer> deviceEndPosition :
+ insertTabletNode.splitByDevice(range[0], range[1])) {
+ alignedDeviceIds.add(deviceEndPosition.getLeft());
+ }
+ }
+ }
+ AlignedTVListRamCostSnapshot alignedRamCostSnapshot =
+ alignedDeviceIds.isEmpty()
+ ? null
+ : new AlignedTVListRamCostSnapshot(workMemTable, alignedDeviceIds);
+
long[] memIncrements =
scheduleMemoryBlock(insertTabletNode, rangeList, results, infoForMetrics);
@@ -628,52 +676,63 @@ public class TsFileProcessor {
tsFileResource);
int pointInserted = 0;
- for (int[] rangePair : rangeList) {
- int start = rangePair[0];
- int end = rangePair[1];
- try {
- if (insertTabletNode.isAligned()) {
- pointInserted +=
- workMemTable.insertAlignedTablet(
- insertTabletNode, start, end, noFailure ? null : results);
- } else {
- pointInserted += workMemTable.insertTablet(insertTabletNode, start, end);
+ try {
+ for (int rangeIndex = 0; rangeIndex < rangeList.size(); rangeIndex++) {
+ final int[] rangePair = rangeList.get(rangeIndex);
+ int start = rangePair[0];
+ int end = rangePair[1];
+ try {
+ if (insertTabletNode.isAligned()) {
+ pointInserted +=
+ workMemTable.insertAlignedTablet(
+ insertTabletNode, start, end, noFailure ? null : results);
+ } else {
+ pointInserted += workMemTable.insertTablet(insertTabletNode, start, end);
+ }
+ } catch (final WriteProcessException e) {
+ final TSStatus failureStatus = RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ for (int failedRangeIndex = rangeIndex;
+ failedRangeIndex < rangeList.size();
+ failedRangeIndex++) {
+ final int[] failedRange = rangeList.get(failedRangeIndex);
+ for (int i = failedRange[0]; i < failedRange[1]; i++) {
+ results[i] = failureStatus;
+ }
+ }
+ throw e;
}
- } catch (WriteProcessException e) {
for (int i = start; i < end; i++) {
- results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage());
- }
- throw new WriteProcessException(e);
- }
- for (int i = start; i < end; i++) {
- if (results[i] == null
- || results[i].getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- results[i] = RpcUtils.SUCCESS_STATUS;
+ if (results[i] == null
+ || results[i].getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ results[i] = RpcUtils.SUCCESS_STATUS;
+ }
}
- }
- final List<Pair<IDeviceID, Integer>> 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<Pair<IDeviceID, Integer>> deviceEndOffsetPairs =
+ insertTabletNode.splitByDevice(start, end);
tsFileResource.updateStartTime(
- deviceEndOffsetPairs.get(i).left,
- insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i - 1).right]);
+ deviceEndOffsetPairs.get(0).left, insertTabletNode.getTimes()[start]);
if (!sequence) {
+ // For sequence tsfile, we update the endTime only when the file is prepared to be closed.
+ // For unsequence tsfile, we have to update the endTime for each insertion.
tsFileResource.updateEndTime(
+ deviceEndOffsetPairs.get(0).left,
+ insertTabletNode.getTimes()[deviceEndOffsetPairs.get(0).right - 1]);
+ }
+ for (int i = 1; i < deviceEndOffsetPairs.size(); i++) {
+ // the end offset of i - 1 is the start offset of i
+ tsFileResource.updateStartTime(
deviceEndOffsetPairs.get(i).left,
- insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i).right - 1]);
+ insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i - 1).right]);
+ if (!sequence) {
+ tsFileResource.updateEndTime(
+ deviceEndOffsetPairs.get(i).left,
+ insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i).right - 1]);
+ }
}
}
+ } finally {
+ reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements[0]);
}
tsFileResource.updateProgressIndex(insertTabletNode.getProgressIndex());
@@ -1169,6 +1228,75 @@ public class TsFileProcessor {
&& 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<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 {
@@ -1221,6 +1349,15 @@ public class TsFileProcessor {
workMemTable.releaseTextDataSize(textDataIncrement);
}
+ private void rollbackMemoryInfoIfNeeded(final long[] memIncrements) {
+ for (final long memIncrement : memIncrements) {
+ if (memIncrement != 0) {
+ rollbackMemoryInfo(memIncrements);
+ return;
+ }
+ }
+ }
+
/**
* Delete data which belongs to the timeseries `deviceId.measurementId` and the timestamp of which
* <= 'timestamp' in the deletion. <br>
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 0e44a3dbde5..068b4f0cc06 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
@@ -73,10 +73,9 @@ 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(BitMap.getSizeOfBytes(ARRAY_SIZE))
- + NUM_BYTES_OBJECT_REF;
+ + RamUsageEstimator.sizeOfByteArray(BitMap.getSizeOfBytes(ARRAY_SIZE));
// Data types of this aligned tvList
protected List<TSDataType> dataTypes;
@@ -84,6 +83,9 @@ public abstract class AlignedTVList extends TVList {
// Record total memory size of binary column
protected long[] memoryBinaryChunkSize;
+ private long materializedBitmapMemoryCost;
+ private long arrayMemCostWithoutIndex;
+
// Data type list -> list of TVList, add 1 when expanded -> primitive array of basic type
// Index relation: columnIndex(dataTypeIndex) -> arrayIndex -> elementIndex
protected List<List<Object>> values;
@@ -106,6 +108,7 @@ public abstract class AlignedTVList extends TVList {
super();
dataTypes = types;
memoryBinaryChunkSize = new long[dataTypes.size()];
+ refreshArrayMemCostWithoutIndex();
values = new ArrayList<>(types.size());
for (int i = 0; i < types.size(); i++) {
@@ -170,7 +173,7 @@ public abstract class AlignedTVList extends TVList {
alignedTvList.allValueColDeletedMap = ignoreAllNullRows ? getAllValueColDeletedMap() : null;
alignedTvList.timeColDeletedMap = this.timeColDeletedMap;
alignedTvList.timeDeletedCnt = this.timeDeletedCnt;
-
+ alignedTvList.materializedBitmapMemoryCost = calculateBitmapRamCost(bitMaps);
return alignedTvList;
}
@@ -183,6 +186,7 @@ public abstract class AlignedTVList extends TVList {
cloneList.values = this.values;
cloneList.bitMaps = this.bitMaps;
cloneList.timeColDeletedMap = this.timeColDeletedMap;
+ cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost;
return cloneList;
}
@@ -218,6 +222,7 @@ public abstract class AlignedTVList extends TVList {
}
}
cloneList.timeColDeletedMap = timeColDeletedMap == null ? null : timeColDeletedMap.clone();
+ cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost;
return cloneList;
}
@@ -446,6 +451,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()];
@@ -737,10 +745,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();
}
@@ -812,6 +823,7 @@ public abstract class AlignedTVList extends TVList {
}
}
}
+ materializedBitmapMemoryCost = 0;
}
@Override
@@ -823,6 +835,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();
}
}
}
@@ -1073,11 +1086,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);
@@ -1111,7 +1126,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;
}
/**
@@ -1131,7 +1183,6 @@ public abstract class AlignedTVList extends TVList {
if (type != null
&& (columnCategories == null || columnCategories[i] == TsTableColumnCategory.FIELD)) {
size += (long) ARRAY_SIZE * (long) type.getDataTypeSize();
- size += BITMAP_RAM_COST_PER_BLOCK;
measurementColumnNum++;
}
}
@@ -1155,12 +1206,11 @@ public 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
@@ -1185,16 +1235,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 f96473781ae..3141ad3c8b4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
@@ -67,19 +67,30 @@ public abstract class TVList implements WALEntryValue {
public static class RamInfo {
private final int timestampsSize;
private final long arrayMemCost;
+ private final long ramSize;
private final int rowCount;
private final List<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..1c11d9ef8eb
--- /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<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 040988997b2..9593f477bf0 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.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.file.SystemFileFactory;
import org.apache.iotdb.commons.path.AlignedFullPath;
+import org.apache.iotdb.commons.path.AlignedPath;
import org.apache.iotdb.commons.path.NonAlignedFullPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
@@ -62,6 +63,7 @@ import org.apache.tsfile.read.expression.QueryExpression;
import org.apache.tsfile.read.query.dataset.QueryDataSet;
import org.apache.tsfile.read.reader.IPointReader;
import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.write.record.TSRecord;
import org.apache.tsfile.write.record.datapoint.DataPoint;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -536,21 +538,21 @@ public class TsFileProcessorTest {
true,
new long[5]);
IMemTable memTable = processor.getWorkMemTable();
- Assert.assertEquals(1776552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1596552, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNode(100, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(1776552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1596552, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNode(200, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(1776552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1596552, memTable.getTVListsRamCost());
Assert.assertEquals(90000, memTable.getTotalPointsNum());
Assert.assertEquals(720360, memTable.memSize());
// Test records
@@ -559,7 +561,7 @@ public class TsFileProcessorTest {
record.addTuple(DataPoint.getDataPoint(dataType, measurementId, String.valueOf(i)));
processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]);
}
- Assert.assertEquals(1778168, memTable.getTVListsRamCost());
+ Assert.assertEquals(1598168, memTable.getTVListsRamCost());
Assert.assertEquals(90100, memTable.getTotalPointsNum());
Assert.assertEquals(721560, memTable.memSize());
}
@@ -614,7 +616,7 @@ public class TsFileProcessorTest {
true,
new long[5]);
IMemTable memTable = processor.getWorkMemTable();
- Assert.assertEquals(1776552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1596552, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNodeFors3000ToS6000(0, true),
Collections.singletonList(new int[] {0, 10}),
@@ -656,7 +658,7 @@ public class TsFileProcessorTest {
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(7105104, memTable.getTVListsRamCost());
+ Assert.assertEquals(6937104, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNodeFors3000ToS6000(300, true),
Collections.singletonList(new int[] {0, 10}),
@@ -685,6 +687,65 @@ public 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,
+ Collections.singletonList(new int[] {0, denseRowCount}),
+ new TSStatus[denseRowCount],
+ true,
+ new long[5]);
+
+ AlignedWritableMemChunk alignedMemChunk = getAlignedMemChunk(denseTablet.getDeviceID());
+ Assert.assertNull(alignedMemChunk.getWorkingTVList().getBitMaps());
+ assertAlignedTvListRamCostMatchesActual(denseTablet.getDeviceID());
+
+ InsertTabletNode nullTablet = genAlignedTablet(new String[] {"s0", "s1"}, 2, denseRowCount);
+ BitMap secondColumnNulls = new BitMap(2);
+ secondColumnNulls.markAll();
+ nullTablet.setBitMaps(new BitMap[] {null, secondColumnNulls});
+ processor.insertTablet(
+ nullTablet,
+ Arrays.asList(new int[] {0, 1}, new int[] {1, 2}),
+ new TSStatus[2],
+ true,
+ new long[5]);
+
+ List<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(deviceId, denseRowCount + 2L);
+ extendedColumnRecord.addTuple(DataPoint.getDataPoint(TSDataType.INT32, "s2", "1"));
+ InsertRowNode extendedColumnRow = buildInsertRowNodeByTSRecord(extendedColumnRecord);
+ extendedColumnRow.setAligned(true);
+ processor.insert(extendedColumnRow, new long[5]);
+
+ int extendedColumnIndex = alignedMemChunk.getWorkingTVList().getBitMaps().size() - 1;
+ for (BitMap bitMap : alignedMemChunk.getWorkingTVList().getBitMaps().get(extendedColumnIndex)) {
+ Assert.assertNotNull(bitMap);
+ }
+ assertAlignedTvListRamCostMatchesActual(denseTablet.getDeviceID());
+ }
+
@Test
public void nonAlignedTvListRamCostTest()
throws MetadataException, WriteProcessException, IOException {
@@ -1335,6 +1396,59 @@ public class TsFileProcessorTest {
rowCount);
}
+ private AlignedWritableMemChunk getAlignedMemChunk(IDeviceID targetDeviceId) {
+ IWritableMemChunk memChunk =
+ processor
+ .getWorkMemTable()
+ .getWritableMemChunk(targetDeviceId, AlignedPath.VECTOR_PLACEHOLDER);
+ Assert.assertNotNull(memChunk);
+ return (AlignedWritableMemChunk) memChunk;
+ }
+
+ private void assertAlignedTvListRamCostMatchesActual(IDeviceID targetDeviceId) {
+ AlignedWritableMemChunk alignedMemChunk = getAlignedMemChunk(targetDeviceId);
+ long actualRamCost = alignedMemChunk.getWorkingTVList().getRamSize();
+ for (org.apache.iotdb.db.utils.datastructure.AlignedTVList sortedTVList :
+ alignedMemChunk.getSortedList()) {
+ actualRamCost += sortedTVList.getRamSize();
+ }
+ Assert.assertEquals(actualRamCost, processor.getWorkMemTable().getTVListsRamCost());
+ }
+
+ private InsertTabletNode genAlignedTablet(String[] measurements, int rowCount, long startTime)
+ throws IllegalPathException {
+ TSDataType[] dataTypes = new TSDataType[measurements.length];
+ MeasurementSchema[] schemas = new MeasurementSchema[measurements.length];
+ Object[] columns = new Object[measurements.length];
+ for (int column = 0; column < measurements.length; column++) {
+ dataTypes[column] = TSDataType.INT32;
+ schemas[column] =
+ new MeasurementSchema(measurements[column], TSDataType.INT32, TSEncoding.PLAIN);
+ columns[column] = new int[rowCount];
+ for (int row = 0; row < rowCount; row++) {
+ ((int[]) columns[column])[row] = row;
+ }
+ }
+ long[] times = new long[rowCount];
+ for (int row = 0; row < rowCount; row++) {
+ times[row] = startTime + row;
+ }
+
+ InsertTabletNode insertTabletNode =
+ new InsertTabletNode(
+ new QueryId("test_write").genPlanNodeId(),
+ new PartialPath(deviceId),
+ true,
+ measurements,
+ dataTypes,
+ times,
+ null,
+ columns,
+ rowCount);
+ insertTabletNode.setMeasurementSchemas(schemas);
+ return insertTabletNode;
+ }
+
private InsertTabletNode genInsertTableNode(long startTime, boolean isAligned)
throws IllegalPathException {
String deviceId = "root.sg.device5";
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
index 5ab976c1310..4cd977b7a1d 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
@@ -26,6 +26,7 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.external.commons.lang3.ArrayUtils;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.utils.RamUsageEstimator;
import org.junit.Assert;
import org.junit.Test;
@@ -34,9 +35,23 @@ import java.util.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(BitMap.getSizeOfBytes(ARRAY_SIZE)),
+ AlignedTVList.bitmapRamCost());
+ Assert.assertEquals(NUM_BYTES_OBJECT_REF, AlignedTVList.bitmapReferenceRamCost());
+ }
+
@Test
public void testAlignedTVList1() {
List<TSDataType> dataTypes = new ArrayList<>();
@@ -169,6 +184,35 @@ public class AlignedTVListTest {
BitMap.getSizeOfBytes(ARRAY_SIZE), firstColumnBitMaps.get(2).getByteArray().length);
Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE * 2 + 1, 0));
Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE * 2, 0));
+ Assert.assertEquals(
+ 3L * tvList.alignedTvListArrayMemCost()
+ + 3L * AlignedTVList.bitmapReferenceRamCost()
+ + AlignedTVList.bitmapRamCost(),
+ tvList.getRamSize());
+ Assert.assertEquals(tvList.getRamSize(), tvList.calculateRamSize().getRamSize());
+ Assert.assertEquals(tvList.getRamSize(), tvList.clone().getRamSize());
+ Assert.assertEquals(tvList.getRamSize(), tvList.cloneForFlushSort().getRamSize());
+ }
+
+ @Test
+ public void testExtendedColumnRamCostIncludesActualBitmaps() {
+ AlignedTVList tvList =
+ AlignedTVList.newAlignedList(new ArrayList<>(Arrays.asList(TSDataType.INT64)));
+ for (int i = 0; i <= ARRAY_SIZE; i++) {
+ tvList.putAlignedValue(i, new Object[] {(long) i});
+ }
+
+ long ramSizeBeforeExtension = tvList.getRamSize();
+ tvList.extendColumn(TSDataType.INT32);
+
+ Assert.assertEquals(
+ 2L
+ * (AlignedTVList.valueListArrayMemCost(TSDataType.INT32)
+ + AlignedTVList.bitmapReferenceRamCost()
+ + AlignedTVList.bitmapRamCost()),
+ tvList.getRamSize() - ramSizeBeforeExtension);
+ tvList.clear();
+ Assert.assertEquals(0, tvList.getRamSize());
}
@Test