This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch rc/2.0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rc/2.0.11 by this push:
new b9a977d5542 [Fix] Reconcile actual aligned TVList memory usage (#18330)
b9a977d5542 is described below
commit b9a977d55426d7f6ea7b7b7a854fddf8f5c42102
Author: Caideyipi <[email protected]>
AuthorDate: Wed Aug 26 09:33:57 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 +-
.../dataregion/memtable/TsFileProcessorTest.java | 131 ++++++++++-
.../db/utils/datastructure/AlignedTVListTest.java | 44 ++++
5 files changed, 440 insertions(+), 80 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/TsFileProcessorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
index 040988997b2..28493c0b906 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;
@@ -45,6 +46,7 @@ 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;
+import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -62,6 +64,7 @@ import org.apache.tsfile.read.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 +539,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 +562,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());
}
@@ -585,7 +588,9 @@ public class TsFileProcessorTest {
genSingleMeasurementTablet(rowCount, true), rangeList, actualResults,
false, new long[5]);
Assert.assertEquals(
- expectedProcessor.getWorkMemTable().getTVListsRamCost(),
+ expectedProcessor.getWorkMemTable().getTVListsRamCost()
+ + 2 * AlignedTVList.bitmapReferenceRamCost()
+ + AlignedTVList.bitmapRamCost(),
actualProcessor.getWorkMemTable().getTVListsRamCost());
Assert.assertEquals(
TSStatusCode.OUT_OF_TTL.getStatusCode(),
actualResults[failedIndex].getCode());
@@ -614,7 +619,7 @@ public 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 +661,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 +690,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 +1399,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
