This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 9e909d3ce26 [Fix] Reconcile actual aligned TVList memory usage (#18330)
9e909d3ce26 is described below
commit 9e909d3ce26002507efd7a72cc53e7bf40c145bc
Author: Caideyipi <[email protected]>
AuthorDate: Wed Jul 29 10:04:39 2026 +0800
[Fix] Reconcile actual aligned TVList memory usage (#18330)
---
.../db/it/IoTDBAlignedTVListLazyAllocationIT.java | 5 +-
.../dataregion/memtable/TsFileProcessor.java | 348 ++++++++++++---------
.../db/utils/datastructure/AlignedTVList.java | 85 +++--
...ignedBitmapMemoryAccountingPerformanceTest.java | 295 +++++++++++++++++
.../dataregion/memtable/TsFileProcessorTest.java | 99 +++++-
.../db/utils/datastructure/AlignedTVListTest.java | 31 +-
6 files changed, 671 insertions(+), 192 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java
index 375f560c9ee..314a098211c 100644
---
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java
@@ -83,10 +83,7 @@ public class IoTDBAlignedTVListLazyAllocationIT {
* NEW_COLUMN_COUNT
* AlignedTVList.valueListArrayMemCost(TSDataType.INT64);
long lazyAllocationCost =
- (long) historicalBlockCount
- * NEW_COLUMN_COUNT
- * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray()
- + (long) NEW_COLUMN_COUNT *
AlignedTVList.primitiveArrayMemCost(TSDataType.INT64);
+ (long) NEW_COLUMN_COUNT *
AlignedTVList.valueListArrayMemCost(TSDataType.INT64);
long dataNodeMaxHeapSize = DATANODE_MAX_HEAP_SIZE_IN_MB * 1024L * 1024L;
Assert.assertTrue(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index 285e29b4065..ee9b39d2ffe 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -313,6 +313,10 @@ public class TsFileProcessor {
ensureMemTable(infoForMetrics);
workMemTable.checkDataType(insertRowNode);
+ AlignedTVListRamCostSnapshot alignedRamCostSnapshot =
+ insertRowNode.isAligned()
+ ? new AlignedTVListRamCostSnapshot(workMemTable,
insertRowNode.getDeviceID())
+ : null;
long[] memIncrements;
@@ -379,11 +383,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
@@ -408,6 +416,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()) {
@@ -421,6 +440,7 @@ public class TsFileProcessor {
}
}
long[] alignedMemIncrements =
checkAlignedMemCostAndAddToTspInfoForRows(alignedList);
+ alignedMemTableIncrement = alignedMemIncrements[0];
final long[] nonAlignedMemIncrements;
try {
nonAlignedMemIncrements =
checkMemCostAndAddToTspInfoForRows(nonAlignedList);
@@ -436,6 +456,7 @@ public class TsFileProcessor {
if (insertRowsNode.isAligned()) {
memIncrements =
checkAlignedMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList());
+ alignedMemTableIncrement = memIncrements[0];
} else {
memIncrements =
checkMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList());
}
@@ -486,19 +507,23 @@ 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());
@@ -623,6 +648,19 @@ 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);
@@ -670,59 +708,63 @@ public class TsFileProcessor {
tsFileResource);
int pointInserted = 0;
- for (int rangeIndex = 0; rangeIndex < rangeList.size(); rangeIndex++) {
- final int[] rangePair = rangeList.get(rangeIndex);
- int start = rangePair[0];
- int end = rangePair[1];
- try {
- if (insertTabletNode.isAligned()) {
- pointInserted +=
- workMemTable.insertAlignedTablet(
- insertTabletNode, start, end, noFailure ? null : results);
- } else {
- pointInserted += workMemTable.insertTablet(insertTabletNode, start,
end);
- }
- } catch (final WriteProcessException e) {
- final TSStatus failureStatus = RpcUtils.getStatus(e.getErrorCode(),
e.getMessage());
- for (int failedRangeIndex = rangeIndex;
- failedRangeIndex < rangeList.size();
- failedRangeIndex++) {
- final int[] failedRange = rangeList.get(failedRangeIndex);
- for (int i = failedRange[0]; i < failedRange[1]; i++) {
- results[i] = failureStatus;
+ try {
+ for (int rangeIndex = 0; rangeIndex < rangeList.size(); rangeIndex++) {
+ final int[] rangePair = rangeList.get(rangeIndex);
+ int start = rangePair[0];
+ int end = rangePair[1];
+ try {
+ if (insertTabletNode.isAligned()) {
+ pointInserted +=
+ workMemTable.insertAlignedTablet(
+ insertTabletNode, start, end, noFailure ? null : results);
+ } else {
+ pointInserted += workMemTable.insertTablet(insertTabletNode,
start, end);
+ }
+ } catch (final WriteProcessException e) {
+ final TSStatus failureStatus = RpcUtils.getStatus(e.getErrorCode(),
e.getMessage());
+ for (int failedRangeIndex = rangeIndex;
+ failedRangeIndex < rangeList.size();
+ failedRangeIndex++) {
+ final int[] failedRange = rangeList.get(failedRangeIndex);
+ for (int i = failedRange[0]; i < failedRange[1]; i++) {
+ results[i] = failureStatus;
+ }
}
+ throw e;
}
- throw e;
- }
- for (int i = start; i < end; i++) {
- if (results[i] == null
- || results[i].getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- results[i] = RpcUtils.SUCCESS_STATUS;
+ for (int i = start; i < end; i++) {
+ if (results[i] == null
+ || results[i].getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ results[i] = RpcUtils.SUCCESS_STATUS;
+ }
}
- }
- final List<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());
@@ -853,52 +895,33 @@ public class TsFileProcessor {
ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER,
TSDataType.VECTOR)
* writableFieldDataTypes.length;
// The first row creates the first aligned TVList block. All writable
fields have a non-null
- // value, so this includes the timestamp array, value primitive arrays,
bitmap reservations,
- // array headers, and ArrayList references for that block.
+ // value, so this includes the timestamp array, value primitive arrays,
array headers, and
+ // ArrayList references for that block.
memTableIncrement +=
AlignedTVList.alignedTvListArrayMemCost(writableFieldDataTypes, null);
} else {
// For existed device of this mem table
AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk)
memChunk;
int currentPointNum = alignedMemChunk.alignedListSize();
- int currentArrayNum =
- currentPointNum / PrimitiveArrayManager.ARRAY_SIZE
- + (currentPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 :
0);
int targetArrayIndex = currentPointNum /
PrimitiveArrayManager.ARRAY_SIZE;
- int newColumnCount = 0;
for (int i = 0; dataTypes != null && i < dataTypes.length; i++) {
// Skip failed Measurements
if (!isWritableFieldMeasurement(measurements, dataTypes, values,
columnCategories, i)) {
continue;
}
- if (!alignedMemChunk.containsMeasurement(measurements[i])) {
- // Extending a column adds one null value-list placeholder and one
conservatively reserved
- // bitmap to every historical block. No value primitive array is
allocated there.
- memTableIncrement +=
- currentArrayNum *
AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray();
- newColumnCount++;
- }
if (!isValueArrayMaterialized(alignedMemChunk, measurements[i],
targetArrayIndex)) {
// The non-null value materializes this column's primitive array in
the target block.
- // This cost contains the primitive-array payload and its array
header.
- memTableIncrement +=
AlignedTVList.primitiveArrayMemCost(dataTypes[i]);
+ // Null historical placeholders are not charged.
+ memTableIncrement +=
AlignedTVList.valueListArrayMemCost(dataTypes[i]);
}
}
if ((currentPointNum % PrimitiveArrayManager.ARRAY_SIZE) == 0) {
- // Starting a block allocates its timestamp array, optional sort-index
array, value-list
- // null
- // placeholders, bitmap reservations, array headers, and ArrayList
references for all
- // existing columns. Value primitive arrays are excluded because they
were charged above
- // only for written columns.
+ // Starting a block allocates its timestamp array and optional
sort-index array. Value
+ // arrays and their references are charged above only for written
columns.
memTableIncrement +=
alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCostWithoutPrimitiveArrays();
- // The working TVList does not contain newly extended columns yet, so
add their value-list
- // placeholder and bitmap reservation for this new block separately.
- memTableIncrement +=
- (long) newColumnCount *
AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray();
}
}
-
for (int i = 0; dataTypes != null && i < dataTypes.length; i++) {
if (isWritableFieldMeasurement(measurements, dataTypes, values,
columnCategories, i)
&& dataTypes[i].isBinary()) {
@@ -942,7 +965,7 @@ public class TsFileProcessor {
ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER,
TSDataType.VECTOR)
* writableFieldDataTypes.length;
// The first row creates the first complete aligned TVList block:
timestamp array, value
- // primitive arrays for the writable fields, bitmap reservations,
headers, and references.
+ // primitive arrays for the writable fields, headers, and references.
memTableIncrement +=
AlignedTVList.alignedTvListArrayMemCost(writableFieldDataTypes, null);
for (int i = 0; dataTypes != null && i < dataTypes.length; i++) {
// Skip failed Measurements
@@ -980,13 +1003,6 @@ public class TsFileProcessor {
if (!currentMemChunkContainsMeasurement
&& !addingPointNumInfo.left.containsKey(measurements[i])) {
addingPointNumInfo.left.put(measurements[i], dataTypes[i]);
- int currentArrayNum =
- pointNumBeforeCurrentRow / PrimitiveArrayManager.ARRAY_SIZE
- + (pointNumBeforeCurrentRow %
PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0);
- // A column first seen in this batch adds a null value-list
placeholder and a bitmap
- // reservation to every block that already exists before the
current row.
- memTableIncrement +=
- currentArrayNum *
AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray();
}
Set<Integer> materializedArrayIndexes =
materializedArraysInCurrentBatch
@@ -994,32 +1010,25 @@ public class TsFileProcessor {
.computeIfAbsent(measurements[i], key -> new HashSet<>());
if (!isValueArrayMaterialized(alignedMemChunk, measurements[i],
targetArrayIndex)
&& materializedArrayIndexes.add(targetArrayIndex)) {
- // Charge the payload and header of a value primitive array
exactly once when this
- // batch first writes a non-null value to the column in the target
block.
- memTableIncrement +=
AlignedTVList.primitiveArrayMemCost(dataTypes[i]);
+ // Charge a value array and its reference exactly once when this
batch first writes a
+ // non-null value to the column in the target block.
+ memTableIncrement +=
AlignedTVList.valueListArrayMemCost(dataTypes[i]);
}
}
if ((pointNumBeforeCurrentRow % PrimitiveArrayManager.ARRAY_SIZE) ==
0) {
if (alignedMemChunk == null) {
- // A device created earlier in this batch starts another block.
Charge the timestamp
- // array, value-list placeholders, bitmap reservations, headers,
and references for all
- // columns discovered in the batch, excluding their value
primitive arrays.
+ // A device created earlier in this batch starts another timestamp
block. Value arrays
+ // and their references are charged only when materialized.
memTableIncrement +=
AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays(
addingPointNumInfo.left.values().toArray(new
TSDataType[0]), null);
} else {
- // An existing aligned TVList starts another block. Charge its
timestamp array, optional
- // sort-index array, value-list placeholders, bitmap reservations,
headers, and
- // references for columns that were already present before this
batch.
+ // An existing aligned TVList starts another timestamp block.
Value arrays and their
+ // references are charged only when materialized.
memTableIncrement +=
alignedMemChunk
.getWorkingTVList()
.alignedTvListArrayMemCostWithoutPrimitiveArrays();
- // Columns first introduced by this batch are absent from the
working TVList's type
- // list, so add one value-list placeholder and bitmap reservation
for each of them.
- memTableIncrement +=
- (long) addingPointNumInfo.left.size()
- *
AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray();
}
}
addingPointNumInfo.setRight(addingPointNum + 1);
@@ -1172,9 +1181,8 @@ public class TsFileProcessor {
int numArraysToAdd =
incomingPointNum / PrimitiveArrayManager.ARRAY_SIZE
+ (incomingPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 :
0);
- // Each new block allocates its timestamp array, value-list null
placeholders, bitmap
- // reservations, array headers, and ArrayList references. Value
primitive arrays are excluded
- // here because all-null and failed-only column segments leave them
unmaterialized.
+ // Each new block allocates its timestamp array. Value arrays and their
references are
+ // excluded here because all-null and failed-only column segments leave
them unmaterialized.
memIncrements[0] +=
numArraysToAdd
* AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays(
@@ -1182,7 +1190,7 @@ public class TsFileProcessor {
// Add the payload and header of a value primitive array only for a
column/block segment that
// contains at least one successful non-null value.
memIncrements[0] +=
- calculateTabletPrimitiveArrayMemCost(
+ calculateTabletValueArrayMemCost(
null,
measurementIds,
dataTypes,
@@ -1195,28 +1203,11 @@ public class TsFileProcessor {
0);
} else {
AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk)
memChunk;
- List<TSDataType> newDataTypes = new ArrayList<>();
int currentPointNum = alignedMemChunk.alignedListSize();
int newPointNum = currentPointNum + incomingPointNum;
int currentArrayCnt =
currentPointNum / PrimitiveArrayManager.ARRAY_SIZE
+ (currentPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 :
0);
- for (int i = 0; dataTypes != null && i < dataTypes.length; i++) {
- TSDataType dataType = dataTypes[i];
- if (!isWritableFieldMeasurement(measurementIds, dataTypes, columns,
columnCategories, i)) {
- continue;
- }
-
- if (!alignedMemChunk.containsMeasurement(measurementIds[i])) {
- // A new column adds one null value-list placeholder and one
conservatively reserved
- // bitmap to every historical block. It does not materialize value
primitive arrays in
- // those blocks.
- memIncrements[0] +=
- currentArrayCnt *
AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray();
- newDataTypes.add(dataType);
- }
- }
-
// calculate how many new arrays will be added after this insertion
int newArrayCnt =
newPointNum / PrimitiveArrayManager.ARRAY_SIZE
@@ -1224,26 +1215,18 @@ public class TsFileProcessor {
long acquireArray = newArrayCnt - currentArrayCnt;
if (acquireArray != 0) {
- // Each acquired block adds a timestamp array, optional sort-index
array, value-list null
- // placeholders, bitmap reservations, array headers, and ArrayList
references for columns
- // already present in the working TVList. Value primitive arrays are
charged separately
- // below.
+ // Each acquired block adds a timestamp array and optional sort-index
array. Value arrays
+ // and their references are charged separately below.
memIncrements[0] +=
acquireArray
* alignedMemChunk
.getWorkingTVList()
.alignedTvListArrayMemCostWithoutPrimitiveArrays();
- for (TSDataType ignored : newDataTypes) {
- // Newly extended columns are not included in the working TVList's
per-block cost yet, so
- // add their value-list placeholder and bitmap reservation to every
acquired block.
- memIncrements[0] +=
- acquireArray *
AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray();
- }
}
// Charge the payload and header of each value primitive array that this
tablet actually
// materializes. Arrays already present in the working TVList are not
charged again.
memIncrements[0] +=
- calculateTabletPrimitiveArrayMemCost(
+ calculateTabletValueArrayMemCost(
alignedMemChunk,
measurementIds,
dataTypes,
@@ -1255,7 +1238,6 @@ public class TsFileProcessor {
end,
currentPointNum);
}
-
// flexible-length data size
for (int i = 0; dataTypes != null && i < dataTypes.length; i++) {
TSDataType dataType = dataTypes[i];
@@ -1273,13 +1255,12 @@ public class TsFileProcessor {
}
/**
- * Calculates only the value primitive arrays that an aligned tablet will
materialize. The
- * per-block timestamp arrays, value-list placeholders, bitmap reservations,
headers, and
- * references are charged by the caller. A column/block pair is charged only
when the tablet has
- * at least one successful non-null value in that block and the working
TVList has not already
- * allocated its value array.
+ * Calculates only the value arrays and list references that an aligned
tablet will materialize.
+ * The per-block timestamp arrays are charged by the caller. A column/block
pair is charged only
+ * when the tablet has at least one successful non-null value in that block
and the working TVList
+ * has not already allocated its value array.
*/
- private static long calculateTabletPrimitiveArrayMemCost(
+ private static long calculateTabletValueArrayMemCost(
AlignedWritableMemChunk alignedMemChunk,
String[] measurementIds,
TSDataType[] dataTypes,
@@ -1308,9 +1289,9 @@ public class TsFileProcessor {
if (containsNonNullValue(bitMap, results, inputIndex, length)
&& !isValueArrayMaterialized(
alignedMemChunk, measurementIds[column], targetArrayIndex)) {
- // One successful non-null write materializes the whole fixed-size
value array, whose
- // memory consists of the primitive payload and the array header.
- size += AlignedTVList.primitiveArrayMemCost(dataTypes[column]);
+ // One successful non-null write materializes the fixed-size value
array and its list
+ // reference.
+ size += AlignedTVList.valueListArrayMemCost(dataTypes[column]);
}
inputIndex += length;
targetArrayIndex++;
@@ -1395,6 +1376,75 @@ 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 {
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 644890d0bed..a589e1ea97a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
@@ -72,8 +72,8 @@ 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 =
- BitMap.createBitMapDynamically(ARRAY_SIZE).ramBytesUsed() +
NUM_BYTES_OBJECT_REF;
+ private static final long BITMAP_RAM_COST =
+ BitMap.createBitMapDynamically(ARRAY_SIZE).ramBytesUsed();
// Data types of this aligned tvList
protected List<TSDataType> dataTypes;
@@ -84,6 +84,8 @@ public abstract class AlignedTVList extends TVList {
// Number and memory cost of primitive arrays that have been allocated for
each value column
private int[] materializedValueArrayCounts;
private long materializedValueArrayMemCost;
+ private long materializedBitmapMemoryCost;
+ private long arrayMemCostWithoutPrimitiveArraysAndIndex;
// Data type list -> list of TVList, add 1 when expanded -> primitive array
of basic type
// Index relation: columnIndex(dataTypeIndex) -> arrayIndex -> elementIndex
@@ -108,6 +110,7 @@ public abstract class AlignedTVList extends TVList {
dataTypes = types;
memoryBinaryChunkSize = new long[dataTypes.size()];
materializedValueArrayCounts = new int[dataTypes.size()];
+ refreshArrayMemCostWithoutPrimitiveArrays();
values = new ArrayList<>(types.size());
for (int i = 0; i < types.size(); i++) {
@@ -174,13 +177,14 @@ public abstract class AlignedTVList extends TVList {
alignedTvList.allValueColDeletedMap = ignoreAllNullRows ?
getAllValueColDeletedMap() : null;
alignedTvList.timeColDeletedMap = this.timeColDeletedMap;
alignedTvList.timeDeletedCnt = this.timeDeletedCnt;
+ alignedTvList.materializedBitmapMemoryCost =
calculateBitmapRamCost(bitMaps);
for (int i = 0; i < columnIndexList.size(); i++) {
int columnIndex = columnIndexList.get(i);
if (columnIndex != -1 && values.get(i) != null) {
int materializedArrayCount = materializedValueArrayCounts[columnIndex];
alignedTvList.materializedValueArrayCounts[i] = materializedArrayCount;
alignedTvList.materializedValueArrayMemCost +=
- (long) materializedArrayCount *
primitiveArrayMemCost(dataTypeList.get(i));
+ (long) materializedArrayCount *
valueListArrayMemCost(dataTypeList.get(i));
}
}
@@ -199,6 +203,7 @@ public abstract class AlignedTVList extends TVList {
cloneList.materializedValueArrayCounts =
Arrays.copyOf(materializedValueArrayCounts,
materializedValueArrayCounts.length);
cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost;
+ cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost;
return cloneList;
}
@@ -237,6 +242,7 @@ public abstract class AlignedTVList extends TVList {
cloneList.materializedValueArrayCounts =
Arrays.copyOf(materializedValueArrayCounts,
materializedValueArrayCounts.length);
cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost;
+ cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost;
return cloneList;
}
@@ -428,9 +434,12 @@ public abstract class AlignedTVList extends TVList {
}
columnBitMaps.add(bitMap);
}
+ materializedBitmapMemoryCost +=
+ (long) timestamps.size() * (bitmapReferenceRamCost() +
bitmapRamCost());
this.bitMaps.add(columnBitMaps);
this.values.add(columnValue);
this.dataTypes.add(dataType);
+ refreshArrayMemCostWithoutPrimitiveArrays();
long[] tmpValueChunkRawSize = memoryBinaryChunkSize;
memoryBinaryChunkSize = new long[dataTypes.size()];
@@ -726,10 +735,13 @@ public abstract class AlignedTVList extends TVList {
columnBitMaps.add(BitMap.createBitMapDynamically(ARRAY_SIZE));
}
bitMaps.set(columnIndex, columnBitMaps);
+ materializedBitmapMemoryCost +=
+ (long) columnBitMaps.size() * (bitmapReferenceRamCost() +
bitmapRamCost());
}
for (int i = 0; i < bitMaps.get(columnIndex).size(); i++) {
if (bitMaps.get(columnIndex).get(i) == null) {
bitMaps.get(columnIndex).set(i,
BitMap.createBitMapDynamically(ARRAY_SIZE));
+ materializedBitmapMemoryCost += bitmapRamCost();
}
bitMaps.get(columnIndex).get(i).markAll();
}
@@ -808,6 +820,7 @@ public abstract class AlignedTVList extends TVList {
}
}
}
+ materializedBitmapMemoryCost = 0;
}
@Override
@@ -819,6 +832,7 @@ public abstract class AlignedTVList extends TVList {
values.get(i).add(null);
if (bitMaps != null && bitMaps.get(i) != null) {
bitMaps.get(i).add(null);
+ materializedBitmapMemoryCost += bitmapReferenceRamCost();
}
}
}
@@ -1089,7 +1103,7 @@ public abstract class AlignedTVList extends TVList {
valueArray = getPrimitiveArraysByType(dataTypes.get(columnIndex));
columnValues.set(arrayIndex, valueArray);
materializedValueArrayCounts[columnIndex]++;
- materializedValueArrayMemCost +=
primitiveArrayMemCost(dataTypes.get(columnIndex));
+ materializedValueArrayMemCost +=
valueListArrayMemCost(dataTypes.get(columnIndex));
}
return valueArray;
}
@@ -1111,11 +1125,13 @@ 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,
BitMap.createBitMapDynamically(ARRAY_SIZE));
+ materializedBitmapMemoryCost += bitmapRamCost();
}
return bitMaps.get(columnIndex).get(arrayIndex);
@@ -1142,12 +1158,36 @@ public abstract class AlignedTVList extends TVList {
return new RamInfo(
timestamps.size(),
alignedTvListArrayMemCost(),
- (long) timestamps.size() *
alignedTvListArrayMemCostWithoutPrimitiveArrays()
- + materializedValueArrayMemCost,
+ getRamSize(),
rowCount,
new ArrayList<>(dataTypes));
}
+ public synchronized long getRamSize() {
+ return (long) timestamps.size() *
alignedTvListArrayMemCostWithoutPrimitiveArrays()
+ + materializedValueArrayMemCost
+ + materializedBitmapMemoryCost;
+ }
+
+ private static long calculateBitmapRamCost(List<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 += bitMap.ramBytesUsed();
+ }
+ }
+ }
+ return size;
+ }
+
/**
* Get the single alignedTVList array mem cost by give types.
*
@@ -1165,7 +1205,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++;
}
}
@@ -1194,7 +1233,6 @@ public abstract class AlignedTVList extends TVList {
TSDataType type = dataTypes.get(column);
if (type != null) {
size += (long) PrimitiveArrayManager.ARRAY_SIZE * (long)
type.getDataTypeSize();
- size += BITMAP_RAM_COST_PER_BLOCK;
}
}
// size is 0 when all types are null
@@ -1213,13 +1251,21 @@ public abstract class AlignedTVList extends TVList {
}
public long alignedTvListArrayMemCostWithoutPrimitiveArrays() {
+ return arrayMemCostWithoutPrimitiveArraysAndIndex
+ + (indices != null ? (long) PrimitiveArrayManager.ARRAY_SIZE *
Integer.BYTES : 0);
+ }
+
+ private void refreshArrayMemCostWithoutPrimitiveArrays() {
long size = alignedTvListArrayMemCost();
+ if (indices != null) {
+ size -= (long) PrimitiveArrayManager.ARRAY_SIZE * Integer.BYTES;
+ }
for (TSDataType dataType : dataTypes) {
if (dataType != null) {
- size -= primitiveArrayMemCost(dataType);
+ size -= valueListArrayMemCost(dataType);
}
}
- return size;
+ arrayMemCostWithoutPrimitiveArraysAndIndex = size;
}
public static long alignedTvListArrayMemCostWithoutPrimitiveArrays(
@@ -1229,7 +1275,7 @@ public abstract class AlignedTVList extends TVList {
TSDataType dataType = types[i];
if (dataType != null
&& (columnCategories == null || columnCategories[i] ==
TsTableColumnCategory.FIELD)) {
- size -= primitiveArrayMemCost(dataType);
+ size -= valueListArrayMemCost(dataType);
}
}
return size;
@@ -1242,7 +1288,9 @@ public abstract class AlignedTVList extends TVList {
* @return valueListArrayMemCost
*/
public static long valueListArrayMemCost(TSDataType type) {
- return primitiveArrayMemCost(type) +
valueListArrayMemCostWithoutPrimitiveArray();
+ // Charge the reference only after the value array is materialized. Null
historical
+ // placeholders do not contribute to write-memory accounting.
+ return primitiveArrayMemCost(type) + NUM_BYTES_OBJECT_REF;
}
public static long primitiveArrayMemCost(TSDataType type) {
@@ -1251,13 +1299,12 @@ public abstract class AlignedTVList extends TVList {
+ NUM_BYTES_ARRAY_HEADER;
}
- public static long valueListArrayMemCostWithoutPrimitiveArray() {
- long size = 0;
- // bitmap object, byte array, and reference in the bitmap list
- size += BITMAP_RAM_COST_PER_BLOCK;
- // null placeholder reference in the value ArrayList
- size += NUM_BYTES_OBJECT_REF;
- return size;
+ public static long bitmapRamCost() {
+ return BITMAP_RAM_COST;
+ }
+
+ public static long bitmapReferenceRamCost() {
+ return NUM_BYTES_OBJECT_REF;
}
/** Build TsBlock by column. */
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java
new file mode 100644
index 00000000000..8a872f6af61
--- /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] = BitMap.createBitMapDynamically(rowCount);
+ for (int row = column & 1; row < rowCount; row += 2) {
+ bitMaps[column].mark(row);
+ }
+ }
+ }
+ long[] times = new long[rowCount];
+ for (int row = 0; row < rowCount; row++) {
+ times[row] = row;
+ }
+ return new Scenario(measurements, dataTypes, columns, bitMaps, schemas,
times);
+ }
+
+ private static void printResult(
+ String label,
+ Scenario scenario,
+ int warmupIterations,
+ int iterations,
+ int rounds,
+ Summary reconciliationSummary,
+ Summary writeSummary) {
+ System.out.printf(
+ Locale.ROOT,
+ "Aligned bitmap accounting benchmark (%s): columns=%d, rows=%d,
warmups=%d, iterations/round=%d, rounds=%d%n",
+ label,
+ scenario.measurements.length,
+ scenario.times.length,
+ warmupIterations,
+ iterations,
+ rounds);
+ printSummary("reconcile", reconciliationSummary);
+ printSummary("write", writeSummary);
+ System.out.printf(
+ Locale.ROOT,
+ " reconciliation/write CPU ratio=%.2f%%, allocation ratio=%.2f%%%n",
+ percentage(
+ reconciliationSummary.getCpuNanosPerOperation(),
+ writeSummary.getCpuNanosPerOperation()),
+ percentage(
+ reconciliationSummary.getAllocatedBytesPerOperation(),
+ writeSummary.getAllocatedBytesPerOperation()));
+ }
+
+ private static void printSummary(String label, Summary summary) {
+ System.out.printf(
+ Locale.ROOT,
+ " %-10s CPU=%.3f us/batch, allocated=%.1f bytes/batch, peak heap
delta=%.3f MiB%n",
+ label,
+ summary.getCpuNanosPerOperation() / 1_000.0,
+ summary.getAllocatedBytesPerOperation(),
+ summary.getPeakHeapDeltaBytes() / 1024.0 / 1024.0);
+ }
+
+ private static double percentage(double numerator, double denominator) {
+ return denominator == 0 ? 0 : numerator * 100.0 / denominator;
+ }
+
+ private static final class AccountingTarget {
+
+ private final IMemTable memTable;
+ private final IDeviceID deviceId;
+
+ private AccountingTarget(IMemTable memTable, IDeviceID deviceId) {
+ this.memTable = memTable;
+ this.deviceId = deviceId;
+ }
+ }
+
+ private static final class Scenario {
+
+ private final String[] measurements;
+ private final TSDataType[] dataTypes;
+ private final Object[] columns;
+ private final BitMap[] bitMaps;
+ private final List<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 3a85754a71c..a8193a9c0b2 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;
@@ -63,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;
@@ -537,21 +539,21 @@ public class TsFileProcessorTest {
true,
new long[5]);
IMemTable memTable = processor.getWorkMemTable();
- Assert.assertEquals(1752552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1596552, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNode(100, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(1752552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1596552, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNode(200, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(1752552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1596552, memTable.getTVListsRamCost());
Assert.assertEquals(90000, memTable.getTotalPointsNum());
Assert.assertEquals(720360, memTable.memSize());
// Test records
@@ -560,7 +562,7 @@ public class TsFileProcessorTest {
record.addTuple(DataPoint.getDataPoint(dataType, measurementId,
String.valueOf(i)));
processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]);
}
- Assert.assertEquals(1754168, memTable.getTVListsRamCost());
+ Assert.assertEquals(1598168, memTable.getTVListsRamCost());
Assert.assertEquals(90100, memTable.getTotalPointsNum());
Assert.assertEquals(721560, memTable.memSize());
}
@@ -587,7 +589,9 @@ public class TsFileProcessorTest {
Assert.assertEquals(
expectedProcessor.getWorkMemTable().getTVListsRamCost()
- - AlignedTVList.primitiveArrayMemCost(dataType),
+ - AlignedTVList.valueListArrayMemCost(dataType)
+ + 2 * AlignedTVList.bitmapReferenceRamCost()
+ + AlignedTVList.bitmapRamCost(),
actualProcessor.getWorkMemTable().getTVListsRamCost());
Assert.assertEquals(
TSStatusCode.OUT_OF_TTL.getStatusCode(),
actualResults[failedIndex].getCode());
@@ -616,7 +620,10 @@ public class TsFileProcessorTest {
AlignedTVList.alignedTvListArrayMemCost(
new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null);
Assert.assertEquals(
- denseBlockCost - AlignedTVList.primitiveArrayMemCost(TSDataType.INT32),
+ denseBlockCost
+ - AlignedTVList.valueListArrayMemCost(TSDataType.INT32)
+ + 2 * AlignedTVList.bitmapReferenceRamCost()
+ + AlignedTVList.bitmapRamCost(),
processor.getWorkMemTable().getTVListsRamCost() -
ramCostBeforeNewBlock);
}
@@ -642,10 +649,63 @@ public class TsFileProcessorTest {
AlignedTVList.alignedTvListArrayMemCost(
new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null);
Assert.assertEquals(
- denseBlockCost - AlignedTVList.primitiveArrayMemCost(TSDataType.INT32),
+ denseBlockCost
+ - AlignedTVList.valueListArrayMemCost(TSDataType.INT32)
+ + 2 * AlignedTVList.bitmapReferenceRamCost()
+ + AlignedTVList.bitmapRamCost(),
processor.getWorkMemTable().getTVListsRamCost() -
ramCostBeforeNewBlock);
}
+ @Test
+ public void alignedBitmapMemoryAccountingMatchesActualAllocations()
+ throws MetadataException, WriteProcessException, IOException,
IllegalPathException {
+ processor = newTestProcessor(filePath + ".bitmap-accounting");
+ int denseRowCount = PrimitiveArrayManager.ARRAY_SIZE * 2 + 1;
+ processor.insertTablet(
+ genAlignedTablet(new String[] {"s0", "s1"}, denseRowCount, 0),
+ Collections.singletonList(new int[] {0, denseRowCount}),
+ new TSStatus[denseRowCount],
+ true,
+ new long[5]);
+
+ AlignedWritableMemChunk alignedMemChunk = getAlignedMemChunk(deviceId);
+ Assert.assertNull(alignedMemChunk.getWorkingTVList().getBitMaps());
+ assertAlignedTvListRamCostMatchesActual(deviceId);
+
+ InsertTabletNode nullTablet = genAlignedTablet(new String[] {"s0", "s1"},
2, denseRowCount);
+ BitMap secondColumnNulls = new BitMap(2);
+ secondColumnNulls.markAll();
+ nullTablet.setBitMaps(new BitMap[] {null, secondColumnNulls});
+ processor.insertTablet(
+ nullTablet,
+ Arrays.asList(new int[] {0, 1}, new int[] {1, 2}),
+ new TSStatus[2],
+ true,
+ new long[5]);
+
+ List<BitMap> secondColumnBitMaps =
alignedMemChunk.getWorkingTVList().getBitMaps().get(1);
+ Assert.assertNull(secondColumnBitMaps.get(0));
+ Assert.assertNull(secondColumnBitMaps.get(1));
+ Assert.assertNotNull(secondColumnBitMaps.get(2));
+ assertAlignedTvListRamCostMatchesActual(deviceId);
+
+ TSRecord extendedColumnRecord = new TSRecord(deviceId, denseRowCount + 2L);
+ extendedColumnRecord.addTuple(DataPoint.getDataPoint(TSDataType.INT32,
"s2", "1"));
+ InsertRowNode extendedColumnRow =
buildInsertRowNodeByTSRecord(extendedColumnRecord);
+ extendedColumnRow.setAligned(true);
+ processor.insert(extendedColumnRow, new long[5]);
+
+ int extendedColumnIndex = alignedMemChunk.getMeasurementIndex("s2");
+ Assert.assertNull(
+
alignedMemChunk.getWorkingTVList().getValues().get(extendedColumnIndex).get(0));
+ Assert.assertNull(
+
alignedMemChunk.getWorkingTVList().getValues().get(extendedColumnIndex).get(1));
+ for (BitMap bitMap :
alignedMemChunk.getWorkingTVList().getBitMaps().get(extendedColumnIndex)) {
+ Assert.assertNotNull(bitMap);
+ }
+ assertAlignedTvListRamCostMatchesActual(deviceId);
+ }
+
@Test
public void alignedTvListRamCostTest2()
throws MetadataException, WriteProcessException, IOException {
@@ -669,7 +729,7 @@ public class TsFileProcessorTest {
true,
new long[5]);
IMemTable memTable = processor.getWorkMemTable();
- Assert.assertEquals(1752552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1596552, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNodeFors3000ToS6000(0, true),
Collections.singletonList(new int[] {0, 10}),
@@ -711,7 +771,7 @@ public class TsFileProcessorTest {
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(5425104, memTable.getTVListsRamCost());
+ Assert.assertEquals(5269104, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNodeFors3000ToS6000(300, true),
Collections.singletonList(new int[] {0, 10}),
@@ -1009,7 +1069,9 @@ public class TsFileProcessorTest {
long denseRowRamIncrement = memTable.getTVListsRamCost() -
ramCostBeforeDenseRow;
Assert.assertEquals(
- AlignedTVList.primitiveArrayMemCost(dataType),
+ AlignedTVList.valueListArrayMemCost(dataType)
+ - 2 * AlignedTVList.bitmapReferenceRamCost()
+ - AlignedTVList.bitmapRamCost(),
denseRowRamIncrement - sparseRowRamIncrement);
}
@@ -1404,6 +1466,23 @@ public class TsFileProcessorTest {
targetProcessor.insert(node, new long[5]);
}
+ private AlignedWritableMemChunk getAlignedMemChunk(String targetDevice) {
+ IWritableMemChunk memChunk =
+ processor
+ .getWorkMemTable()
+ .getWritableMemChunk(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(targetDevice),
+ AlignedPath.VECTOR_PLACEHOLDER);
+ Assert.assertNotNull(memChunk);
+ return (AlignedWritableMemChunk) memChunk;
+ }
+
+ private void assertAlignedTvListRamCostMatchesActual(String targetDevice) {
+ Assert.assertEquals(
+
getAlignedMemChunk(targetDevice).getWorkingTVList().calculateRamSize().getRamSize(),
+ processor.getWorkMemTable().getTVListsRamCost());
+ }
+
private InsertTabletNode genSingleMeasurementTablet(int rowCount, boolean
isAligned)
throws IllegalPathException {
String[] measurements = new String[] {measurementId};
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
index 06e422ad776..381f733bf4d 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
@@ -45,14 +45,13 @@ import static
org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF;
public class AlignedTVListTest {
@Test
- public void testValueListArrayMemCostIncludesBitmapImplementation() {
- long expected =
- (long) ARRAY_SIZE * Long.BYTES
- + BitMap.createBitMapDynamically(ARRAY_SIZE).ramBytesUsed()
- + NUM_BYTES_ARRAY_HEADER
- + 2L * NUM_BYTES_OBJECT_REF;
+ public void testValueListArrayMemCostExcludesLazyBitmapAndNullPlaceholder() {
+ long expected = (long) ARRAY_SIZE * Long.BYTES + NUM_BYTES_ARRAY_HEADER +
NUM_BYTES_OBJECT_REF;
Assert.assertEquals(expected,
AlignedTVList.valueListArrayMemCost(TSDataType.INT64));
+ Assert.assertEquals(
+ BitMap.createBitMapDynamically(ARRAY_SIZE).ramBytesUsed(),
AlignedTVList.bitmapRamCost());
+ Assert.assertEquals(NUM_BYTES_OBJECT_REF,
AlignedTVList.bitmapReferenceRamCost());
}
@Test
@@ -189,6 +188,9 @@ public class AlignedTVListTest {
firstColumnBitMaps.get(2).ramBytesUsed() < new
BitMap(ARRAY_SIZE).ramBytesUsed());
Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE * 2 + 1, 0));
Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE * 2, 0));
+ Assert.assertEquals(
+ 3L * AlignedTVList.bitmapReferenceRamCost() +
AlignedTVList.bitmapRamCost(),
+ tvList.calculateRamSize().getRamSize() - 3L *
tvList.alignedTvListArrayMemCost());
}
@Test
@@ -263,10 +265,14 @@ public class AlignedTVListTest {
Assert.assertTrue(tvList.isNullValue(0, 1));
Assert.assertEquals(1, tvList.getLongByValueIndex(ARRAY_SIZE + 1, 1));
+ long ramSizeBeforeExtension = tvList.calculateRamSize().getRamSize();
tvList.extendColumn(TSDataType.INT32);
Assert.assertNull(tvList.getValues().get(2).get(0));
Assert.assertNull(tvList.getValues().get(2).get(1));
+ Assert.assertEquals(
+ 2L * (AlignedTVList.bitmapReferenceRamCost() +
AlignedTVList.bitmapRamCost()),
+ tvList.calculateRamSize().getRamSize() - ramSizeBeforeExtension);
long ramSizeBeforeExtendedColumnMaterialization =
tvList.calculateRamSize().getRamSize();
tvList.putAlignedValue(ARRAY_SIZE + 2L, new Object[] {null, null, 2});
@@ -277,7 +283,7 @@ public class AlignedTVListTest {
Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE + 2, 2));
Assert.assertEquals(2, tvList.getIntByValueIndex(ARRAY_SIZE + 2, 2));
Assert.assertEquals(
- AlignedTVList.primitiveArrayMemCost(TSDataType.INT32),
+ AlignedTVList.valueListArrayMemCost(TSDataType.INT32),
tvList.calculateRamSize().getRamSize() -
ramSizeBeforeExtendedColumnMaterialization);
}
@@ -293,7 +299,7 @@ public class AlignedTVListTest {
tvList.putAlignedValue(ARRAY_SIZE + 1L, new Object[] {1L, 1L});
Assert.assertEquals(
- AlignedTVList.primitiveArrayMemCost(TSDataType.INT64),
+ AlignedTVList.valueListArrayMemCost(TSDataType.INT64),
tvList.calculateRamSize().getRamSize() - ramSizeBeforeMaterialization);
Assert.assertEquals(
@@ -307,7 +313,9 @@ public class AlignedTVListTest {
Assert.assertEquals(
(long) projectedTvList.getValues().get(0).size()
*
projectedTvList.alignedTvListArrayMemCostWithoutPrimitiveArrays()
- + AlignedTVList.primitiveArrayMemCost(TSDataType.INT64),
+ + AlignedTVList.valueListArrayMemCost(TSDataType.INT64)
+ + (long) projectedTvList.getBitMaps().get(0).size()
+ * (AlignedTVList.bitmapReferenceRamCost() +
AlignedTVList.bitmapRamCost()),
projectedTvList.calculateRamSize().getRamSize());
tvList.clear();
@@ -325,7 +333,10 @@ public class AlignedTVListTest {
int blockCount = tvList.getValues().get(0).size();
long denseRamSize = blockCount * tvList.alignedTvListArrayMemCost();
long expectedRamSize =
- denseRamSize - blockCount *
AlignedTVList.primitiveArrayMemCost(TSDataType.INT64);
+ denseRamSize
+ - blockCount *
AlignedTVList.valueListArrayMemCost(TSDataType.INT64)
+ + (long) blockCount
+ * (AlignedTVList.bitmapReferenceRamCost() +
AlignedTVList.bitmapRamCost());
Assert.assertEquals(expectedRamSize,
tvList.calculateRamSize().getRamSize());
}
