This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 251cf340b71 Optimize aligned row and tablet memory estimation (#18426)
251cf340b71 is described below
commit 251cf340b71a7f6d45b41ad7727bf5cb90f9bffb
Author: Jiang Tian <[email protected]>
AuthorDate: Mon Aug 10 16:00:29 2026 +0800
Optimize aligned row and tablet memory estimation (#18426)
---
.../memtable/AlignedWritableMemChunk.java | 136 +++++++
.../dataregion/memtable/TsFileProcessor.java | 323 +++++++++-------
.../db/utils/datastructure/AlignedTVList.java | 421 ++++++++++++++++-----
.../AlignedRowsBranchComparisonBenchmarkTest.java | 286 ++++++++++++++
.../dataregion/memtable/TsFileProcessorTest.java | 118 +++++-
.../db/utils/datastructure/AlignedTVListTest.java | 79 +++-
6 files changed, 1129 insertions(+), 234 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
index 922153b8d68..4a0043c1559 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
@@ -21,8 +21,10 @@ package
org.apache.iotdb.db.storageengine.dataregion.memtable;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
import
org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils;
+import org.apache.iotdb.db.utils.MemUtils;
import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
import org.apache.iotdb.db.utils.datastructure.BatchEncodeInfo;
import org.apache.iotdb.db.utils.datastructure.MemPointIterator;
@@ -46,6 +48,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -58,6 +61,15 @@ import java.util.concurrent.atomic.AtomicInteger;
import static
org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE;
import static org.apache.iotdb.db.utils.ModificationUtils.isPointDeleted;
+interface AlignedRowsMemCostEstimator {
+
+ void addRow(InsertRowNode row);
+
+ long getTVListMemoryCost();
+
+ long getTextDataMemoryCost();
+}
+
public class AlignedWritableMemChunk extends AbstractWritableMemChunk {
private final Map<String, Integer> measurementIndexMap;
@@ -104,6 +116,130 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
return measurementIndexMap.containsKey(measurementId);
}
+ public long alignedWriteRowMemCost(
+ String[] measurements, TSDataType[] incomingDataTypes, Object[]
rowValues, int rowOffset) {
+ return list.alignedWriteRowMemCost(
+ mapMeasurementsToTVListColumns(measurements), incomingDataTypes,
rowValues, rowOffset);
+ }
+
+ public long alignedWriteArrayMemCost(
+ String[] measurements, TSDataType[] incomingDataTypes, Object[] columns,
int start, int end) {
+ return list.alignedWriteArrayMemCost(
+ mapMeasurementsToTVListColumns(measurements), incomingDataTypes,
columns, start, end);
+ }
+
+ public long alignedWriteRowsMemCost(List<InsertRowNode> rows) {
+ AlignedTVList workingList = list;
+ synchronized (workingList) {
+ AlignedRowsMemCostEstimator estimator =
+ new ExistingAlignedRowsMemCostEstimator(workingList, false);
+ for (InsertRowNode row : rows) {
+ estimator.addRow(row);
+ }
+ return estimator.getTVListMemoryCost();
+ }
+ }
+
+ // TsFileProcessor serializes writes for a DataRegion, so the incremental
estimator can retain a
+ // stable view of this TVList while rows from multiple devices are processed
in input order.
+ AlignedRowsMemCostEstimator newAlignedWriteRowsMemCostEstimator() {
+ return new ExistingAlignedRowsMemCostEstimator(list, true);
+ }
+
+ private final class ExistingAlignedRowsMemCostEstimator implements
AlignedRowsMemCostEstimator {
+ private final AlignedTVList.AlignedWriteRowsMemCostEstimator
tvListEstimator;
+ private final Map<String, Integer> newMeasurementIndexMap = new
HashMap<>();
+ private final boolean trackTextData;
+
+ private int nextColumnIndex = dataTypes.size();
+ private String[] previousMeasurements;
+ private int[] previousColumnIndexes = new int[0];
+ private long textDataMemoryCost;
+
+ private ExistingAlignedRowsMemCostEstimator(AlignedTVList workingList,
boolean trackTextData) {
+ tvListEstimator = workingList.newAlignedWriteRowsMemCostEstimator();
+ this.trackTextData = trackTextData;
+ }
+
+ @Override
+ public void addRow(InsertRowNode row) {
+ tvListEstimator.startRow();
+ String[] measurements = row.getMeasurements();
+ TSDataType[] incomingDataTypes = row.getDataTypes();
+ Object[] rowValues = row.getValues();
+ int[] tvListColumnIndexes = mapMeasurements(measurements);
+ int columnCount =
+ Math.min(measurements.length, Math.min(incomingDataTypes.length,
rowValues.length));
+ for (int column = 0; column < columnCount; column++) {
+ String measurement = measurements[column];
+ TSDataType dataType = incomingDataTypes[column];
+ Object value = rowValues[column];
+ tvListEstimator.addValue(tvListColumnIndexes[column], dataType, value);
+ if (trackTextData
+ && measurement != null
+ && dataType != null
+ && dataType.isBinary()
+ && value != null) {
+ textDataMemoryCost += MemUtils.getBinarySize((Binary) value);
+ }
+ }
+ }
+
+ private int[] mapMeasurements(String[] measurements) {
+ if (previousMeasurements != null
+ && (previousMeasurements == measurements
+ || Arrays.equals(previousMeasurements, measurements))) {
+ previousMeasurements = measurements;
+ return previousColumnIndexes;
+ }
+ if (previousColumnIndexes.length < measurements.length) {
+ previousColumnIndexes = new int[measurements.length];
+ }
+ Arrays.fill(previousColumnIndexes, 0, measurements.length, -1);
+ for (int i = 0; i < measurements.length; i++) {
+ String measurement = measurements[i];
+ if (measurement == null) {
+ continue;
+ }
+ Integer columnIndex = measurementIndexMap.get(measurement);
+ if (columnIndex == null) {
+ columnIndex = newMeasurementIndexMap.get(measurement);
+ if (columnIndex == null) {
+ columnIndex = nextColumnIndex++;
+ newMeasurementIndexMap.put(measurement, columnIndex);
+ }
+ }
+ previousColumnIndexes[i] = columnIndex;
+ }
+ previousMeasurements = measurements;
+ return previousColumnIndexes;
+ }
+
+ @Override
+ public long getTVListMemoryCost() {
+ return tvListEstimator.getMemoryCost();
+ }
+
+ @Override
+ public long getTextDataMemoryCost() {
+ return textDataMemoryCost;
+ }
+ }
+
+ private int[] mapMeasurementsToTVListColumns(String[] measurements) {
+ int[] columnIndexes = new int[measurements.length];
+ Arrays.fill(columnIndexes, -1);
+ for (int i = 0; i < measurements.length; i++) {
+ if (measurements[i] != null) {
+ Integer columnIndex = measurementIndexMap.get(measurements[i]);
+ if (columnIndex != null) {
+ columnIndexes[i] = columnIndex;
+ }
+ }
+ }
+ return columnIndexes;
+ }
+
@Override
public void putLong(long t, long v) {
throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
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 fff95a33d86..49e531e5915 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
@@ -97,6 +97,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -118,6 +119,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import static
org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.GET_QUERY_RESOURCE_FROM_MEM;
import static
org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.FLUSHING_MEMTABLE;
import static
org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.WORKING_MEMTABLE;
+import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF;
@SuppressWarnings("java:S1135") // ignore todos
public class TsFileProcessor {
@@ -374,15 +376,7 @@ public class TsFileProcessor {
long[] memIncrements;
long alignedMemTableIncrement = 0;
Set<IDeviceID> alignedDeviceIds = new HashSet<>();
- for (InsertRowNode insertRowNode : insertRowsNode.getInsertRowNodeList()) {
- if (insertRowNode.isAligned()) {
- alignedDeviceIds.add(insertRowNode.getDeviceID());
- }
- }
- AlignedTVListRamCostSnapshot alignedRamCostSnapshot =
- alignedDeviceIds.isEmpty()
- ? null
- : new AlignedTVListRamCostSnapshot(workMemTable, alignedDeviceIds);
+ AlignedTVListRamCostSnapshot alignedRamCostSnapshot;
long memControlStartTime = System.nanoTime();
if (insertRowsNode.isMixingAlignment()) {
@@ -395,7 +389,8 @@ public class TsFileProcessor {
nonAlignedList.add(insertRowNode);
}
}
- long[] alignedMemIncrements =
checkAlignedMemCostAndAddToTspInfoForRows(alignedList);
+ long[] alignedMemIncrements =
+ checkAlignedMemCostAndAddToTspInfoForRows(alignedList,
alignedDeviceIds);
alignedMemTableIncrement = alignedMemIncrements[0];
final long[] nonAlignedMemIncrements;
try {
@@ -411,12 +406,19 @@ public class TsFileProcessor {
} else {
if (insertRowsNode.isAligned()) {
memIncrements =
-
checkAlignedMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList());
+ checkAlignedMemCostAndAddToTspInfoForRows(
+ insertRowsNode.getInsertRowNodeList(), alignedDeviceIds);
alignedMemTableIncrement = memIncrements[0];
} else {
memIncrements =
checkMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList());
}
}
+ // Memory estimation only reserves counters. Taking the snapshot
afterwards avoids a separate
+ // row scan while still capturing TVList state before WAL and memtable
writes.
+ alignedRamCostSnapshot =
+ alignedDeviceIds.isEmpty()
+ ? null
+ : new AlignedTVListRamCostSnapshot(workMemTable, alignedDeviceIds);
// recordScheduleMemoryBlockCost
costsForMetrics[1] += System.nanoTime() - memControlStartTime;
@@ -722,7 +724,7 @@ public class TsFileProcessor {
chunkMetadataIncrement +=
ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER,
TSDataType.VECTOR)
* dataTypes.length;
- memTableIncrement += AlignedTVList.alignedTvListArrayMemCost(dataTypes);
+ memTableIncrement += estimateNewAlignedRowMemCost(measurements,
dataTypes, values);
for (int i = 0; i < dataTypes.length; i++) {
// Skip failed Measurements
if (dataTypes[i] == null || measurements[i] == null) {
@@ -736,129 +738,178 @@ public class TsFileProcessor {
} else {
// For existed device of this mem table
AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk)
memChunk;
- List<TSDataType> dataTypesInTVList = new ArrayList<>();
+ memTableIncrement +=
+ alignedMemChunk.alignedWriteRowMemCost(measurements, dataTypes,
values, 0);
for (int i = 0; i < dataTypes.length; i++) {
// Skip failed Measurements
if (dataTypes[i] == null || measurements[i] == null) {
continue;
}
- // Extending the column of aligned mem chunk
- if (!alignedMemChunk.containsMeasurement(measurements[i])) {
- memTableIncrement +=
- (alignedMemChunk.alignedListSize() /
PrimitiveArrayManager.ARRAY_SIZE + 1)
- * AlignedTVList.valueListArrayMemCost(dataTypes[i]);
- dataTypesInTVList.add(dataTypes[i]);
- }
// TEXT data mem size
if (dataTypes[i].isBinary() && values[i] != null) {
textDataIncrement += MemUtils.getBinarySize((Binary) values[i]);
}
}
- // Here currentChunkPointNum >= 1
- if ((alignedMemChunk.alignedListSize() %
PrimitiveArrayManager.ARRAY_SIZE) == 0) {
- memTableIncrement +=
alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost();
- for (TSDataType dataType : dataTypesInTVList) {
- memTableIncrement += AlignedTVList.valueListArrayMemCost(dataType);
- }
- }
}
updateMemoryInfo(memTableIncrement, chunkMetadataIncrement,
textDataIncrement);
return new long[] {memTableIncrement, textDataIncrement,
chunkMetadataIncrement};
}
@SuppressWarnings("squid:S3776") // high Cognitive Complexity
- private long[] checkAlignedMemCostAndAddToTspInfoForRows(List<InsertRowNode>
insertRowNodeList)
+ private long[] checkAlignedMemCostAndAddToTspInfoForRows(
+ List<InsertRowNode> insertRowNodeList, Set<IDeviceID> alignedDeviceIds)
throws WriteProcessException {
// Memory of increased PrimitiveArray and TEXT values, e.g., add a
long[128], add 128*8
long memTableIncrement = 0L;
long textDataIncrement = 0L;
long chunkMetadataIncrement = 0L;
- // device -> (measurements -> datatype, adding aligned TVList size)
- Map<IDeviceID, Pair<Map<String, TSDataType>, Integer>>
increasingMemTableInfo = new HashMap<>();
+ Map<IDeviceID, AlignedRowsMemCostEstimator> estimators = new HashMap<>();
for (InsertRowNode insertRowNode : insertRowNodeList) {
IDeviceID deviceId = insertRowNode.getDeviceID();
- TSDataType[] dataTypes = insertRowNode.getDataTypes();
- Object[] values = insertRowNode.getValues();
- String[] measurements = insertRowNode.getMeasurements();
-
- IWritableMemChunk memChunk =
- workMemTable.getWritableMemChunk(deviceId,
AlignedPath.VECTOR_PLACEHOLDER);
- if (memChunk == null && !increasingMemTableInfo.containsKey(deviceId)) {
- // For new device of this mem table
- // ChunkMetadataIncrement
- chunkMetadataIncrement +=
- ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER,
TSDataType.VECTOR)
- * dataTypes.length;
- memTableIncrement +=
AlignedTVList.alignedTvListArrayMemCost(dataTypes);
- for (int i = 0; i < dataTypes.length; i++) {
- // Skip failed Measurements
- if (dataTypes[i] == null || measurements[i] == null) {
- continue;
- }
- increasingMemTableInfo
- .computeIfAbsent(deviceId, k -> new Pair<>(new HashMap<>(), 1))
- .left
- .put(measurements[i], dataTypes[i]);
- // TEXT data mem size
- if (dataTypes[i].isBinary() && values[i] != null) {
- textDataIncrement += MemUtils.getBinarySize((Binary) values[i]);
- }
+ AlignedRowsMemCostEstimator estimator = estimators.get(deviceId);
+ if (estimator == null) {
+ IWritableMemChunk memChunk =
+ workMemTable.getWritableMemChunk(deviceId,
AlignedPath.VECTOR_PLACEHOLDER);
+ if (memChunk == null) {
+ estimator = new NewAlignedRowsMemCostEstimator();
+ chunkMetadataIncrement +=
+ ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER,
TSDataType.VECTOR)
+ * insertRowNode.getDataTypes().length;
+ } else {
+ estimator = ((AlignedWritableMemChunk)
memChunk).newAlignedWriteRowsMemCostEstimator();
}
+ estimators.put(deviceId, estimator);
+ alignedDeviceIds.add(deviceId);
+ }
+ estimator.addRow(insertRowNode);
+ }
- } else {
- // For existed device of this mem table
- AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk)
memChunk;
- int currentChunkPointNum = alignedMemChunk == null ? 0 :
alignedMemChunk.alignedListSize();
- List<TSDataType> dataTypesInTVList = new ArrayList<>();
- Pair<Map<String, TSDataType>, Integer> addingPointNumInfo =
- increasingMemTableInfo.computeIfAbsent(deviceId, k -> new
Pair<>(new HashMap<>(), 0));
- for (int i = 0; i < dataTypes.length; i++) {
- // Skip failed Measurements
- if (dataTypes[i] == null || measurements[i] == null) {
- continue;
- }
+ for (AlignedRowsMemCostEstimator estimator : estimators.values()) {
+ memTableIncrement += estimator.getTVListMemoryCost();
+ textDataIncrement += estimator.getTextDataMemoryCost();
+ }
+ updateMemoryInfo(memTableIncrement, chunkMetadataIncrement,
textDataIncrement);
+ return new long[] {memTableIncrement, textDataIncrement,
chunkMetadataIncrement};
+ }
- int addingPointNum = addingPointNumInfo.getRight();
- // Extending the column of aligned mem chunk
- boolean currentMemChunkContainsMeasurement =
- alignedMemChunk != null &&
alignedMemChunk.containsMeasurement(measurements[i]);
- if (!currentMemChunkContainsMeasurement
- && !addingPointNumInfo.left.containsKey(measurements[i])) {
- addingPointNumInfo.left.put(measurements[i], dataTypes[i]);
- int currentArrayNum =
- (currentChunkPointNum + addingPointNum) /
PrimitiveArrayManager.ARRAY_SIZE
- + ((currentChunkPointNum + addingPointNum) %
PrimitiveArrayManager.ARRAY_SIZE
- > 0
- ? 1
- : 0);
- memTableIncrement +=
- currentArrayNum *
AlignedTVList.valueListArrayMemCost(dataTypes[i]);
- addingPointNumInfo.left.put(measurements[i], dataTypes[i]);
- }
- // TEXT data mem size
- if (dataTypes[i].isBinary() && values[i] != null) {
- textDataIncrement += MemUtils.getBinarySize((Binary) values[i]);
+ @TestOnly private AlignedTVListRamCostSnapshot benchmarkAlignedRowsSnapshot;
+
+ @TestOnly
+ long[] benchmarkCheckAlignedMemCostAndAddToTspInfoForRows(
+ List<InsertRowNode> rows, Set<IDeviceID> alignedDeviceIds) throws
WriteProcessException {
+ long[] increments = checkAlignedMemCostAndAddToTspInfoForRows(rows,
alignedDeviceIds);
+ benchmarkAlignedRowsSnapshot =
+ alignedDeviceIds.isEmpty()
+ ? null
+ : new AlignedTVListRamCostSnapshot(workMemTable, alignedDeviceIds);
+ return increments;
+ }
+
+ /** Estimates a new aligned TVList without allocating it during the
memory-control phase. */
+ private static final class NewAlignedRowsMemCostEstimator implements
AlignedRowsMemCostEstimator {
+ private final Map<String, Integer> measurementIndexMap = new HashMap<>();
+
+ private String[] previousMeasurements;
+ private int[] previousMeasurementIndexes = new int[0];
+ private long[] primitiveArrayMemCosts = new long[0];
+ private int[] materializedMeasurementBlocks = new int[0];
+ private int rowCount;
+ private long materializedArrayMemCost;
+ private long textDataMemoryCost;
+
+ @Override
+ public void addRow(InsertRowNode row) {
+ String[] measurements = row.getMeasurements();
+ TSDataType[] dataTypes = row.getDataTypes();
+ Object[] values = row.getValues();
+ int[] measurementIndexes = mapKnownMeasurements(measurements);
+ int columnCount = Math.min(measurements.length,
Math.min(dataTypes.length, values.length));
+ int currentBlock = rowCount / PrimitiveArrayManager.ARRAY_SIZE;
+ rowCount++;
+ for (int column = 0; column < columnCount; column++) {
+ String measurement = measurements[column];
+ TSDataType dataType = dataTypes[column];
+ if (measurement == null || dataType == null) {
+ continue;
+ }
+ int measurementIndex = measurementIndexes[column];
+ if (measurementIndex < 0) {
+ Integer knownMeasurementIndex = measurementIndexMap.get(measurement);
+ if (knownMeasurementIndex == null) {
+ measurementIndex = measurementIndexMap.size();
+ measurementIndexMap.put(measurement, measurementIndex);
+ ensureMeasurementCapacity(measurementIndex + 1);
+ primitiveArrayMemCosts[measurementIndex] =
+ AlignedTVList.valueListArrayMemCost(dataType) -
NUM_BYTES_OBJECT_REF;
+ } else {
+ measurementIndex = knownMeasurementIndex;
}
+ measurementIndexes[column] = measurementIndex;
+ }
+ Object value = values[column];
+ if (value != null && materializedMeasurementBlocks[measurementIndex]
!= currentBlock) {
+ materializedMeasurementBlocks[measurementIndex] = currentBlock;
+ materializedArrayMemCost += primitiveArrayMemCosts[measurementIndex];
}
- int addingPointNum = addingPointNumInfo.right;
- // Here currentChunkPointNum + addingPointNum >= 1
- if (((currentChunkPointNum + addingPointNum) %
PrimitiveArrayManager.ARRAY_SIZE) == 0) {
- dataTypesInTVList.addAll(addingPointNumInfo.left.values());
- memTableIncrement +=
- alignedMemChunk != null
- ?
alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost()
- + dataTypesInTVList.stream()
- .mapToLong(AlignedTVList::valueListArrayMemCost)
- .sum()
- : AlignedTVList.alignedTvListArrayMemCost(
- dataTypesInTVList.toArray(new TSDataType[0]));
+ if (dataType.isBinary() && value != null) {
+ textDataMemoryCost += MemUtils.getBinarySize((Binary) value);
}
- addingPointNumInfo.setRight(addingPointNum + 1);
}
}
- updateMemoryInfo(memTableIncrement, chunkMetadataIncrement,
textDataIncrement);
- return new long[] {memTableIncrement, textDataIncrement,
chunkMetadataIncrement};
+
+ private int[] mapKnownMeasurements(String[] measurements) {
+ if (previousMeasurements != null
+ && (previousMeasurements == measurements
+ || Arrays.equals(previousMeasurements, measurements))) {
+ previousMeasurements = measurements;
+ return previousMeasurementIndexes;
+ }
+ if (previousMeasurementIndexes.length < measurements.length) {
+ previousMeasurementIndexes = new int[measurements.length];
+ }
+ Arrays.fill(previousMeasurementIndexes, 0, measurements.length, -1);
+ for (int i = 0; i < measurements.length; i++) {
+ Integer measurementIndex = measurementIndexMap.get(measurements[i]);
+ if (measurementIndex != null) {
+ previousMeasurementIndexes[i] = measurementIndex;
+ }
+ }
+ previousMeasurements = measurements;
+ return previousMeasurementIndexes;
+ }
+
+ private void ensureMeasurementCapacity(int requiredCapacity) {
+ if (requiredCapacity <= primitiveArrayMemCosts.length) {
+ return;
+ }
+ int oldCapacity = primitiveArrayMemCosts.length;
+ int newCapacity = Math.max(requiredCapacity, Math.max(4, oldCapacity <<
1));
+ primitiveArrayMemCosts = Arrays.copyOf(primitiveArrayMemCosts,
newCapacity);
+ materializedMeasurementBlocks =
Arrays.copyOf(materializedMeasurementBlocks, newCapacity);
+ Arrays.fill(materializedMeasurementBlocks, oldCapacity, newCapacity, -1);
+ }
+
+ @Override
+ public long getTVListMemoryCost() {
+ if (rowCount == 0) {
+ return 0;
+ }
+ int measurementColumnCount = measurementIndexMap.size();
+ long blockCount =
+ (rowCount + (long) PrimitiveArrayManager.ARRAY_SIZE - 1)
+ / PrimitiveArrayManager.ARRAY_SIZE;
+ return AlignedTVList.alignedTvListInitialMemCost(measurementColumnCount)
+ + blockCount
+ * AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays(
+ measurementColumnCount)
+ + materializedArrayMemCost;
+ }
+
+ @Override
+ public long getTextDataMemoryCost() {
+ return textDataMemoryCost;
+ }
}
private long[] checkMemCostAndAddToTspInfoForTablet(
@@ -965,8 +1016,7 @@ public class TsFileProcessor {
dataTypes.length
* ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER,
TSDataType.VECTOR);
memIncrements[0] +=
- ((end - start) / PrimitiveArrayManager.ARRAY_SIZE + 1)
- * AlignedTVList.alignedTvListArrayMemCost(dataTypes);
+ estimateNewAlignedArrayMemCost(measurementIds, dataTypes, columns,
start, end);
for (int i = 0; i < dataTypes.length; i++) {
TSDataType dataType = dataTypes[i];
String measurement = measurementIds[i];
@@ -983,7 +1033,6 @@ public class TsFileProcessor {
} else {
AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk)
memChunk;
- List<TSDataType> dataTypesInTVList = new ArrayList<>();
for (int i = 0; i < dataTypes.length; i++) {
TSDataType dataType = dataTypes[i];
String measurement = measurementIds[i];
@@ -991,39 +1040,53 @@ public class TsFileProcessor {
if (dataType == null || column == null || measurement == null) {
continue;
}
- // Extending the column of aligned mem chunk
- if (!alignedMemChunk.containsMeasurement(measurementIds[i])) {
- memIncrements[0] +=
- (alignedMemChunk.alignedListSize() /
PrimitiveArrayManager.ARRAY_SIZE + 1)
- * AlignedTVList.valueListArrayMemCost(dataType);
- dataTypesInTVList.add(dataType);
- }
// TEXT data size
if (dataType.isBinary()) {
Binary[] binColumn = (Binary[]) columns[i];
memIncrements[1] += MemUtils.getBinaryColumnSize(binColumn, start,
end);
}
}
- long acquireArray;
- if (alignedMemChunk.alignedListSize() % PrimitiveArrayManager.ARRAY_SIZE
== 0) {
- acquireArray = (end - start) / PrimitiveArrayManager.ARRAY_SIZE + 1L;
- } else {
- acquireArray =
- (end
- - start
- - 1
- + (alignedMemChunk.alignedListSize() %
PrimitiveArrayManager.ARRAY_SIZE))
- / PrimitiveArrayManager.ARRAY_SIZE;
+ memIncrements[0] +=
+ alignedMemChunk.alignedWriteArrayMemCost(measurementIds, dataTypes,
columns, start, end);
+ }
+ }
+
+ private long estimateNewAlignedRowMemCost(
+ String[] measurements, TSDataType[] dataTypes, Object[] values) {
+ List<TSDataType> validDataTypes = new ArrayList<>();
+ for (int i = 0; i < dataTypes.length; i++) {
+ if (measurements[i] != null && dataTypes[i] != null && values[i] !=
null) {
+ validDataTypes.add(dataTypes[i]);
}
- if (acquireArray != 0) {
- // memory of extending the TVList
- memIncrements[0] +=
- acquireArray *
alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost();
- for (TSDataType dataType : dataTypesInTVList) {
- memIncrements[0] += acquireArray *
AlignedTVList.valueListArrayMemCost(dataType);
- }
+ }
+ return estimateNewDenseAlignedMemCost(validDataTypes, 1);
+ }
+
+ private long estimateNewAlignedArrayMemCost(
+ String[] measurements, TSDataType[] dataTypes, Object[] columns, int
start, int end) {
+ List<TSDataType> validDataTypes = new ArrayList<>();
+ for (int i = 0; i < dataTypes.length; i++) {
+ if (measurements[i] != null && dataTypes[i] != null && columns[i] !=
null) {
+ validDataTypes.add(dataTypes[i]);
}
}
+ return estimateNewDenseAlignedMemCost(validDataTypes, end - start);
+ }
+
+ private long estimateNewDenseAlignedMemCost(List<TSDataType> dataTypes, int
rowCount) {
+ if (rowCount <= 0) {
+ return 0;
+ }
+ int measurementColumnCount = dataTypes.size();
+ long blockMemCost =
+
AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays(measurementColumnCount);
+ for (TSDataType dataType : dataTypes) {
+ blockMemCost += AlignedTVList.valueListArrayMemCost(dataType) -
NUM_BYTES_OBJECT_REF;
+ }
+ long blockCount =
+ (rowCount + (long) PrimitiveArrayManager.ARRAY_SIZE - 1) /
PrimitiveArrayManager.ARRAY_SIZE;
+ return AlignedTVList.alignedTvListInitialMemCost(measurementColumnCount)
+ + blockCount * blockMemCost;
}
private void reconcileAlignedTVListRamCost(
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 d319b9c9a0d..d8b16b3eb84 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
@@ -81,6 +81,8 @@ public abstract class AlignedTVList extends TVList {
private long materializedBitmapMemoryCost;
private long arrayMemCostWithoutIndex;
+ private int[] materializedValueArrayCounts;
+ private long materializedValueArrayMemCost;
/**
* A fully prepared partial clone. All allocations and validations are
completed before this plan
@@ -150,6 +152,7 @@ public abstract class AlignedTVList extends TVList {
super();
dataTypes = types;
memoryBinaryChunkSize = new long[dataTypes.size()];
+ materializedValueArrayCounts = new int[dataTypes.size()];
values = new ArrayList<>(types.size());
for (int i = 0; i < types.size(); i++) {
values.add(new ArrayList<>());
@@ -199,6 +202,8 @@ public abstract class AlignedTVList extends TVList {
alignedTvList.rowCount = this.rowCount;
alignedTvList.allValueColDeletedMap = getAllValueColDeletedMap();
alignedTvList.materializedBitmapMemoryCost =
calculateBitmapRamCost(bitMaps, null);
+ alignedTvList.refreshArrayMemCostWithoutIndex();
+ alignedTvList.refreshMaterializedValueArrayMemoryCost();
return alignedTvList;
}
@@ -210,6 +215,9 @@ public abstract class AlignedTVList extends TVList {
cloneList.values = this.values;
cloneList.bitMaps = this.bitMaps;
cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost;
+ cloneList.materializedValueArrayCounts =
+ Arrays.copyOf(materializedValueArrayCounts,
materializedValueArrayCounts.length);
+ cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost;
return cloneList;
}
@@ -304,6 +312,8 @@ public abstract class AlignedTVList extends TVList {
plan.cloneList.arrayMemCostWithoutIndex =
plan.cloneArrayMemCostWithoutIndex;
materializedBitmapMemoryCost = plan.sourceBitmapMemoryCost;
plan.cloneList.materializedBitmapMemoryCost = plan.cloneBitmapMemoryCost;
+ refreshMaterializedValueArrayMemoryCost();
+ plan.cloneList.refreshMaterializedValueArrayMemoryCost();
}
/**
@@ -331,7 +341,9 @@ public abstract class AlignedTVList extends TVList {
// Release memory for non-query columns
for (Object dataArray : columnValues) {
- PrimitiveArrayManager.release(dataArray);
+ if (dataArray != null) {
+ PrimitiveArrayManager.release(dataArray);
+ }
}
values.set(i, null);
memoryBinaryChunkSize[i] = 0;
@@ -345,6 +357,7 @@ public abstract class AlignedTVList extends TVList {
materializedBitmapMemoryCost = calculateBitmapRamCost(bitMaps,
columnsToKeep);
// Refresh per-block memory cost after releasing columns
refreshArrayMemCostWithoutIndex();
+ refreshMaterializedValueArrayMemoryCost();
}
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
@@ -360,44 +373,37 @@ public abstract class AlignedTVList extends TVList {
Object columnValue = value[i];
if (columnValue == null) {
markNullValue(i, arrayIndex, elementIndex);
+ continue;
}
List<Object> columnValues = values.get(i);
if (columnValues == null) {
throw new IllegalStateException(
String.format("Missing value arrays for aligned column index %d
during append", i));
}
+ Object valueArray = getOrCreateValueArray(i, arrayIndex);
switch (dataTypes.get(i)) {
case TEXT:
case BLOB:
case STRING:
- ((Binary[]) columnValues.get(arrayIndex))[elementIndex] =
- columnValue != null ? (Binary) columnValue : Binary.EMPTY_VALUE;
- memoryBinaryChunkSize[i] +=
- columnValue != null
- ? getBinarySize((Binary) columnValue)
- : getBinarySize(Binary.EMPTY_VALUE);
+ ((Binary[]) valueArray)[elementIndex] = (Binary) columnValue;
+ memoryBinaryChunkSize[i] += getBinarySize((Binary) columnValue);
break;
case FLOAT:
- ((float[]) columnValues.get(arrayIndex))[elementIndex] =
- columnValue != null ? (float) columnValue : Float.MIN_VALUE;
+ ((float[]) valueArray)[elementIndex] = (float) columnValue;
break;
case INT32:
case DATE:
- ((int[]) columnValues.get(arrayIndex))[elementIndex] =
- columnValue != null ? (int) columnValue : Integer.MIN_VALUE;
+ ((int[]) valueArray)[elementIndex] = (int) columnValue;
break;
case INT64:
case TIMESTAMP:
- ((long[]) columnValues.get(arrayIndex))[elementIndex] =
- columnValue != null ? (long) columnValue : Long.MIN_VALUE;
+ ((long[]) valueArray)[elementIndex] = (long) columnValue;
break;
case DOUBLE:
- ((double[]) columnValues.get(arrayIndex))[elementIndex] =
- columnValue != null ? (double) columnValue : Double.MIN_VALUE;
+ ((double[]) valueArray)[elementIndex] = (double) columnValue;
break;
case BOOLEAN:
- ((boolean[]) columnValues.get(arrayIndex))[elementIndex] =
- columnValue != null && (boolean) columnValue;
+ ((boolean[]) valueArray)[elementIndex] = (boolean) columnValue;
break;
default:
break;
@@ -517,66 +523,16 @@ public abstract class AlignedTVList extends TVList {
}
public void extendColumn(TSDataType dataType) {
- if (bitMaps == null) {
- List<List<BitMap>> localBitMaps = new ArrayList<>(values.size());
- for (int i = 0; i < values.size(); i++) {
- localBitMaps.add(null);
- }
- bitMaps = localBitMaps;
- }
List<Object> columnValue = new ArrayList<>();
- List<BitMap> columnBitMaps = new ArrayList<>();
for (int i = 0; i < timestamps.size(); i++) {
- switch (dataType) {
- case TEXT:
- case STRING:
- case BLOB:
- columnValue.add(getPrimitiveArraysByType(TSDataType.TEXT));
- break;
- case FLOAT:
- columnValue.add(getPrimitiveArraysByType(TSDataType.FLOAT));
- break;
- case INT32:
- case DATE:
- columnValue.add(getPrimitiveArraysByType(TSDataType.INT32));
- break;
- case INT64:
- case TIMESTAMP:
- columnValue.add(getPrimitiveArraysByType(TSDataType.INT64));
- break;
- case DOUBLE:
- columnValue.add(getPrimitiveArraysByType(TSDataType.DOUBLE));
- break;
- case BOOLEAN:
- columnValue.add(getPrimitiveArraysByType(TSDataType.BOOLEAN));
- break;
- default:
- break;
- }
- BitMap bitMap = new BitMap(ARRAY_SIZE);
- // The following code is for these 2 kinds of scenarios.
-
- // Eg1: If rowCount=5 and ARRAY_SIZE=2, we need to supply 3 bitmaps for
the extending column.
- // The first 2 bitmaps should mark all bits to represent 4 nulls and the
3rd bitmap should
- // mark
- // the 1st bit to represent 1 null value.
-
- // Eg2: If rowCount=4 and ARRAY_SIZE=2, we need to supply 2 bitmaps for
the extending column.
- // These 2 bitmaps should mark all bits to represent 4 nulls.
- if (i == timestamps.size() - 1 && rowCount % ARRAY_SIZE != 0) {
- for (int j = 0; j < rowCount % ARRAY_SIZE; j++) {
- bitMap.mark(j);
- }
- } else {
- bitMap.markAll();
- }
- columnBitMaps.add(bitMap);
+ columnValue.add(null);
+ }
+ if (bitMaps != null) {
+ bitMaps.add(null);
}
- this.bitMaps.add(columnBitMaps);
this.values.add(columnValue);
this.dataTypes.add(dataType);
- materializedBitmapMemoryCost +=
- (long) columnBitMaps.size() * (bitmapReferenceRamCost() +
bitmapRamCost());
+ materializedValueArrayCounts = Arrays.copyOf(materializedValueArrayCounts,
dataTypes.size());
refreshArrayMemCostWithoutIndex();
long[] tmpValueChunkRawSize = memoryBinaryChunkSize;
@@ -690,7 +646,7 @@ public abstract class AlignedTVList extends TVList {
if (bitMaps == null
|| bitMaps.get(columnIndex) == null
|| bitMaps.get(columnIndex).get(unsortedRowIndex / ARRAY_SIZE) ==
null) {
- return false;
+ return values.get(columnIndex).get(unsortedRowIndex / ARRAY_SIZE) ==
null;
}
int arrayIndex = unsortedRowIndex / ARRAY_SIZE;
int elementIndex = unsortedRowIndex % ARRAY_SIZE;
@@ -816,6 +772,9 @@ public abstract class AlignedTVList extends TVList {
}
protected Object cloneValue(TSDataType type, Object value) {
+ if (value == null) {
+ return null;
+ }
switch (type) {
case TEXT:
case BLOB:
@@ -910,6 +869,7 @@ public abstract class AlignedTVList extends TVList {
}
}
cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost;
+ cloneList.refreshMaterializedValueArrayMemoryCost();
if (hasBitMapsToMove && cloneList.bitMaps == null) {
cloneList.bitMaps = new ArrayList<>(dataTypes.size());
@@ -925,12 +885,36 @@ public abstract class AlignedTVList extends TVList {
List<Object> columnValues = values.get(i);
if (columnValues != null) {
for (Object dataArray : columnValues) {
- PrimitiveArrayManager.release(dataArray);
+ if (dataArray != null) {
+ PrimitiveArrayManager.release(dataArray);
+ }
}
columnValues.clear();
}
memoryBinaryChunkSize[i] = 0;
}
+ Arrays.fill(materializedValueArrayCounts, 0);
+ materializedValueArrayMemCost = 0;
+ }
+
+ private Object getOrCreateValueArray(int columnIndex, int arrayIndex) {
+ List<Object> columnValues = values.get(columnIndex);
+ if (columnValues == null) {
+ throw new IllegalStateException(
+ String.format("Missing value arrays for aligned column index %d",
columnIndex));
+ }
+ Object array = columnValues.get(arrayIndex);
+ if (array == null) {
+ array = getPrimitiveArraysByType(dataTypes.get(columnIndex));
+ columnValues.set(arrayIndex, array);
+ materializedValueArrayCounts[columnIndex]++;
+ materializedValueArrayMemCost +=
primitiveArrayMemCost(dataTypes.get(columnIndex));
+ int existingRows = Math.min(rowCount - arrayIndex * ARRAY_SIZE,
ARRAY_SIZE);
+ if (existingRows > 0) {
+ getBitMap(columnIndex, arrayIndex).markRange(0, existingRows);
+ }
+ }
+ return array;
}
@Override
@@ -957,7 +941,7 @@ public abstract class AlignedTVList extends TVList {
throw new IllegalStateException(
String.format("Missing value arrays for aligned column index %d
during expand", i));
}
- columnValues.add(getPrimitiveArraysByType(dataTypes.get(i)));
+ columnValues.add(null);
if (bitMaps != null && bitMaps.get(i) != null) {
bitMaps.get(i).add(null);
materializedBitmapMemoryCost += bitmapReferenceRamCost();
@@ -1078,7 +1062,7 @@ public abstract class AlignedTVList extends TVList {
}
private void arrayCopy(Object[] value, int idx, int arrayIndex, int
elementIndex, int remaining) {
- for (int i = 0; i < values.size(); i++) {
+ for (int i = 0; i < Math.min(values.size(), value.length); i++) {
if (value[i] == null) {
continue;
}
@@ -1087,11 +1071,12 @@ public abstract class AlignedTVList extends TVList {
throw new IllegalStateException(
String.format("Missing value arrays for aligned column index %d
during arrayCopy", i));
}
+ Object valueArray = getOrCreateValueArray(i, arrayIndex);
switch (dataTypes.get(i)) {
case TEXT:
case BLOB:
case STRING:
- Binary[] arrayT = ((Binary[]) columnValues.get(arrayIndex));
+ Binary[] arrayT = ((Binary[]) valueArray);
System.arraycopy(value[i], idx, arrayT, elementIndex, remaining);
// update raw size of Text chunk
@@ -1101,25 +1086,25 @@ public abstract class AlignedTVList extends TVList {
}
break;
case FLOAT:
- float[] arrayF = ((float[]) columnValues.get(arrayIndex));
+ float[] arrayF = ((float[]) valueArray);
System.arraycopy(value[i], idx, arrayF, elementIndex, remaining);
break;
case INT32:
case DATE:
- int[] arrayI = ((int[]) columnValues.get(arrayIndex));
+ int[] arrayI = ((int[]) valueArray);
System.arraycopy(value[i], idx, arrayI, elementIndex, remaining);
break;
case INT64:
case TIMESTAMP:
- long[] arrayL = ((long[]) columnValues.get(arrayIndex));
+ long[] arrayL = ((long[]) valueArray);
System.arraycopy(value[i], idx, arrayL, elementIndex, remaining);
break;
case DOUBLE:
- double[] arrayD = ((double[]) columnValues.get(arrayIndex));
+ double[] arrayD = ((double[]) valueArray);
System.arraycopy(value[i], idx, arrayD, elementIndex, remaining);
break;
case BOOLEAN:
- boolean[] arrayB = ((boolean[]) columnValues.get(arrayIndex));
+ boolean[] arrayB = ((boolean[]) valueArray);
System.arraycopy(value[i], idx, arrayB, elementIndex, remaining);
break;
default:
@@ -1198,28 +1183,64 @@ public abstract class AlignedTVList extends TVList {
return timestamps.size()
* (arrayMemCostWithoutIndex
+ (indices != null ? (long) PrimitiveArrayManager.ARRAY_SIZE *
Integer.BYTES : 0))
+ + materializedValueArrayMemCost
+ materializedBitmapMemoryCost
+ calculateContainerRamCost(null);
}
public synchronized long getRamSize(Set<Integer> columnsToClone) {
- return timestamps.size() * alignedTvListArrayMemCost(columnsToClone)
+ return timestamps.size()
+ * (calculateArrayMemCostWithoutIndex(columnsToClone)
+ + (indices != null ? (long) PrimitiveArrayManager.ARRAY_SIZE *
Integer.BYTES : 0))
+ + calculateMaterializedValueArrayMemCost(columnsToClone)
+ calculateBitmapRamCost(bitMaps, columnsToClone)
+ calculateContainerRamCost(columnsToClone);
}
private long calculateArrayMemCostWithoutIndex(Set<Integer> retainedColumns)
{
- long arrayMemCost = alignedTvListArrayMemCost(retainedColumns);
- if (indices != null) {
- arrayMemCost -= (long) PrimitiveArrayManager.ARRAY_SIZE * Integer.BYTES;
+ long arrayMemCost =
+ (long) PrimitiveArrayManager.ARRAY_SIZE * Long.BYTES
+ + 2L * NUM_BYTES_ARRAY_HEADER
+ + 2L * NUM_BYTES_OBJECT_REF;
+ for (int i = 0; i < dataTypes.size(); i++) {
+ if ((retainedColumns == null || retainedColumns.contains(i)) &&
values.get(i) != null) {
+ arrayMemCost += NUM_BYTES_OBJECT_REF;
+ }
}
return arrayMemCost;
}
+ private long calculateMaterializedValueArrayMemCost(Set<Integer>
retainedColumns) {
+ long size = 0;
+ for (int i = 0; i < dataTypes.size(); i++) {
+ if ((retainedColumns == null || retainedColumns.contains(i)) &&
values.get(i) != null) {
+ size += (long) materializedValueArrayCounts[i] *
primitiveArrayMemCost(dataTypes.get(i));
+ }
+ }
+ return size;
+ }
+
private void refreshArrayMemCostWithoutIndex() {
arrayMemCostWithoutIndex = calculateArrayMemCostWithoutIndex(null);
}
+ private void refreshMaterializedValueArrayMemoryCost() {
+ Arrays.fill(materializedValueArrayCounts, 0);
+ materializedValueArrayMemCost = 0;
+ for (int i = 0; i < values.size(); i++) {
+ List<Object> columnValues = values.get(i);
+ if (columnValues == null) {
+ continue;
+ }
+ for (Object valueArray : columnValues) {
+ if (valueArray != null) {
+ materializedValueArrayCounts[i]++;
+ materializedValueArrayMemCost +=
primitiveArrayMemCost(dataTypes.get(i));
+ }
+ }
+ }
+ }
+
/**
* Calculate the one-time container memory retained by this list.
Primitive-array references in
* the time/index/value lists and bitmap lists are already charged by the
per-block accounting, so
@@ -1324,6 +1345,228 @@ public abstract class AlignedTVList extends TVList {
return size;
}
+ /** Initial list-container memory before the first aligned row is written. */
+ public static long alignedTvListInitialMemCost(int measurementColumnCount) {
+ long arrayListShallowSize =
RamUsageEstimator.shallowSizeOfInstance(ArrayList.class);
+ long listWithReferencesSize =
+ arrayListShallowSize +
RamUsageEstimator.sizeOfObjectArray(measurementColumnCount);
+ return 2 * listWithReferencesSize
+ + RamUsageEstimator.sizeOfLongArray(measurementColumnCount)
+ + arrayListShallowSize
+ + (long) measurementColumnCount * arrayListShallowSize;
+ }
+
+ /** Memory of one aligned block excluding all lazily materialized value
arrays. */
+ public static long alignedTvListArrayMemCostWithoutPrimitiveArrays(int
measurementColumnCount) {
+ return (long) ARRAY_SIZE * Long.BYTES
+ + 2L * NUM_BYTES_ARRAY_HEADER
+ + (2L + measurementColumnCount) * NUM_BYTES_OBJECT_REF;
+ }
+
+ /**
+ * Estimate memory allocated by an aligned tablet after its measurements are
mapped to TVList
+ * columns. A negative mapped index denotes a column that will be appended
before the write.
+ */
+ public synchronized long alignedWriteArrayMemCost(
+ int[] tvListColumnIndexes,
+ TSDataType[] incomingDataTypes,
+ Object[] columns,
+ int start,
+ int end) {
+ if (start >= end) {
+ return 0;
+ }
+ int columnCount =
+ Math.min(tvListColumnIndexes.length, Math.min(columns.length,
incomingDataTypes.length));
+ int newColumnCount =
+ countNewColumns(tvListColumnIndexes, incomingDataTypes, columns,
columnCount);
+ long size = 0;
+ int block = rowCount / ARRAY_SIZE;
+ int elementIndex = rowCount % ARRAY_SIZE;
+ int inputIndex = start;
+ while (inputIndex < end) {
+ if (block >= timestamps.size()) {
+ size +=
+ arrayMemCostWithoutIndex
+ + (long) newColumnCount * NUM_BYTES_OBJECT_REF
+ + (indices != null ? (long) PrimitiveArrayManager.ARRAY_SIZE *
Integer.BYTES : 0);
+ }
+ for (int inputColumn = 0; inputColumn < columnCount; inputColumn++) {
+ TSDataType incomingDataType = incomingDataTypes[inputColumn];
+ if (incomingDataType == null || columns[inputColumn] == null) {
+ continue;
+ }
+ int tvListColumnIndex = tvListColumnIndexes[inputColumn];
+ if (tvListColumnIndex < 0 ||
!isValueArrayMaterialized(tvListColumnIndex, block)) {
+ size += primitiveArrayMemCost(incomingDataType);
+ }
+ }
+ int written = Math.min(end - inputIndex, ARRAY_SIZE - elementIndex);
+ inputIndex += written;
+ block++;
+ elementIndex = 0;
+ }
+ return size + columnExtensionMemCost(newColumnCount);
+ }
+
+ public synchronized AlignedWriteRowsMemCostEstimator
newAlignedWriteRowsMemCostEstimator() {
+ return new AlignedWriteRowsMemCostEstimator();
+ }
+
+ /** Incrementally estimates a row batch without retaining per-row input or
column mappings. */
+ public final class AlignedWriteRowsMemCostEstimator {
+ private final int initialRowCount = rowCount;
+ private final int existingBlockCount = timestamps.size();
+ private final int existingColumnCount = dataTypes.size();
+ private final long existingArrayMemCostWithoutIndex =
arrayMemCostWithoutIndex;
+ private final boolean hasIndices = indices != null;
+ private boolean[] newColumns = new boolean[Math.max(1,
existingColumnCount)];
+ private int[] materializedColumnBlocks = new int[Math.max(1,
existingColumnCount)];
+
+ private int estimatedRowCount;
+ private int currentBlock = -1;
+ private int newColumnCount;
+ private long materializedArrayMemCost;
+
+ private AlignedWriteRowsMemCostEstimator() {
+ Arrays.fill(materializedColumnBlocks, -1);
+ }
+
+ public void startRow() {
+ currentBlock = (int) ((initialRowCount + (long) estimatedRowCount) /
ARRAY_SIZE);
+ estimatedRowCount++;
+ }
+
+ public void addValue(int tvListColumnIndex, TSDataType incomingDataType,
Object value) {
+ if (tvListColumnIndex < 0 || incomingDataType == null) {
+ return;
+ }
+ ensureColumnCapacity(tvListColumnIndex + 1);
+ if (tvListColumnIndex >= existingColumnCount &&
!newColumns[tvListColumnIndex]) {
+ newColumns[tvListColumnIndex] = true;
+ newColumnCount++;
+ }
+ if (value != null
+ && !isValueArrayMaterialized(tvListColumnIndex, currentBlock)
+ && materializedColumnBlocks[tvListColumnIndex] != currentBlock) {
+ materializedColumnBlocks[tvListColumnIndex] = currentBlock;
+ materializedArrayMemCost += primitiveArrayMemCost(incomingDataType);
+ }
+ }
+
+ private void ensureColumnCapacity(int requiredCapacity) {
+ if (requiredCapacity <= newColumns.length) {
+ return;
+ }
+ int oldCapacity = newColumns.length;
+ int newCapacity = Math.max(requiredCapacity, oldCapacity << 1);
+ newColumns = Arrays.copyOf(newColumns, newCapacity);
+ materializedColumnBlocks = Arrays.copyOf(materializedColumnBlocks,
newCapacity);
+ Arrays.fill(materializedColumnBlocks, oldCapacity, newCapacity, -1);
+ }
+
+ public long getMemoryCost() {
+ if (estimatedRowCount == 0) {
+ return 0;
+ }
+ long finalBlockCount =
+ (initialRowCount + (long) estimatedRowCount + ARRAY_SIZE - 1) /
ARRAY_SIZE;
+ long newBlockCount = Math.max(0, finalBlockCount - existingBlockCount);
+ long newBlockMemCost =
+ existingArrayMemCostWithoutIndex
+ + (long) newColumnCount * NUM_BYTES_OBJECT_REF
+ + (hasIndices ? (long) ARRAY_SIZE * Integer.BYTES : 0);
+ return materializedArrayMemCost
+ + columnExtensionMemCost(newColumnCount)
+ + newBlockCount * newBlockMemCost;
+ }
+ }
+
+ /**
+ * Estimate memory allocated by one aligned row after its measurements are
mapped to TVList
+ * columns. {@code rowOffset} is the number of preceding rows in the same
pending batch.
+ */
+ public synchronized long alignedWriteRowMemCost(
+ int[] tvListColumnIndexes,
+ TSDataType[] incomingDataTypes,
+ Object[] rowValues,
+ int rowOffset) {
+ int columnCount =
+ Math.min(tvListColumnIndexes.length, Math.min(rowValues.length,
incomingDataTypes.length));
+ int newColumnCount =
+ countNewColumns(tvListColumnIndexes, incomingDataTypes, rowValues,
columnCount);
+ long size = 0;
+ int block = (rowCount + rowOffset) / ARRAY_SIZE;
+ if (block >= timestamps.size()) {
+ size +=
+ arrayMemCostWithoutIndex
+ + (long) newColumnCount * NUM_BYTES_OBJECT_REF
+ + (indices != null ? (long) PrimitiveArrayManager.ARRAY_SIZE *
Integer.BYTES : 0);
+ }
+ for (int inputColumn = 0; inputColumn < columnCount; inputColumn++) {
+ TSDataType incomingDataType = incomingDataTypes[inputColumn];
+ if (incomingDataType == null || rowValues[inputColumn] == null) {
+ continue;
+ }
+ int tvListColumnIndex = tvListColumnIndexes[inputColumn];
+ if (tvListColumnIndex < 0 ||
!isValueArrayMaterialized(tvListColumnIndex, block)) {
+ size += primitiveArrayMemCost(incomingDataType);
+ }
+ }
+ return size + columnExtensionMemCost(newColumnCount);
+ }
+
+ private static int countNewColumns(
+ int[] tvListColumnIndexes,
+ TSDataType[] incomingDataTypes,
+ Object[] inputValues,
+ int columnCount) {
+ int count = 0;
+ for (int i = 0; i < columnCount; i++) {
+ if (tvListColumnIndexes[i] < 0 && incomingDataTypes[i] != null &&
inputValues[i] != null) {
+ count++;
+ }
+ }
+ return count;
+ }
+
+ private boolean isValueArrayMaterialized(int columnIndex, int arrayIndex) {
+ if (columnIndex >= values.size()) {
+ return false;
+ }
+ List<Object> columnValues = values.get(columnIndex);
+ return columnValues != null
+ && arrayIndex < columnValues.size()
+ && columnValues.get(arrayIndex) != null;
+ }
+
+ private long columnExtensionMemCost(int newColumnCount) {
+ if (newColumnCount == 0) {
+ return 0;
+ }
+ int oldColumnCount = dataTypes.size();
+ int newTotalColumnCount = oldColumnCount + newColumnCount;
+ long size = (long) timestamps.size() * NUM_BYTES_OBJECT_REF *
newColumnCount;
+ size +=
+ RamUsageEstimator.sizeOfObjectArray(newTotalColumnCount)
+ - RamUsageEstimator.sizeOfObjectArray(oldColumnCount);
+ size +=
+ RamUsageEstimator.sizeOfLongArray(newTotalColumnCount)
+ - RamUsageEstimator.sizeOfLongArray(oldColumnCount);
+ size +=
+ RamUsageEstimator.sizeOfObjectArray(newTotalColumnCount)
+ - RamUsageEstimator.sizeOfObjectArray(oldColumnCount);
+ if (bitMaps != null) {
+ size +=
+ RamUsageEstimator.sizeOfObjectArray(newTotalColumnCount)
+ - RamUsageEstimator.sizeOfObjectArray(oldColumnCount);
+ }
+ long newColumnContainerCost =
+ RamUsageEstimator.shallowSizeOf(new ArrayList<>())
+ + (timestamps.isEmpty() ? 0 :
RamUsageEstimator.sizeOfObjectArray(0));
+ return size + newColumnCount * newColumnContainerCost;
+ }
+
/**
* Get the single alignedTVList array mem cost by give types.
*
@@ -1371,6 +1614,12 @@ public abstract class AlignedTVList extends TVList {
+ NUM_BYTES_OBJECT_REF;
}
+ /** Memory cost of one materialized primitive array, excluding its ArrayList
reference. */
+ private static long primitiveArrayMemCost(TSDataType type) {
+ return (long) PrimitiveArrayManager.ARRAY_SIZE * type.getDataTypeSize()
+ + NUM_BYTES_ARRAY_HEADER;
+ }
+
public static long bitmapRamCost() {
return BITMAP_RAM_COST;
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedRowsBranchComparisonBenchmarkTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedRowsBranchComparisonBenchmarkTest.java
new file mode 100644
index 00000000000..72d6de4990e
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedRowsBranchComparisonBenchmarkTest.java
@@ -0,0 +1,286 @@
+/*
+ * 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.commons.exception.MetadataException;
+import org.apache.iotdb.commons.file.SystemFileFactory;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
+import org.apache.iotdb.db.storageengine.dataregion.DataRegionInfo;
+import org.apache.iotdb.db.storageengine.dataregion.DataRegionTest;
+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.ManualPerformanceTestUtils;
+import org.apache.iotdb.db.utils.constant.TestConstant;
+
+import com.sun.management.ThreadMXBean;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.PlainDeviceID;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.lang.management.ManagementFactory;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+public class AlignedRowsBranchComparisonBenchmarkTest {
+
+ private static final String STORAGE_GROUP = "root.branch_benchmark";
+ private static final String DEVICE_PATH = STORAGE_GROUP + ".d0";
+ private static final IDeviceID DEVICE_ID = new PlainDeviceID(DEVICE_PATH);
+ private static final int EXISTING_COLUMNS = 48;
+ private static final int EXISTING_ROWS = PrimitiveArrayManager.ARRAY_SIZE;
+ private static final int[] ROW_COUNTS = {128, 1024, 1024, 8192};
+ private static final int[] COLUMN_COUNTS = {16, 16, 64, 64};
+ private static final int[] ITERATIONS = {976, 122, 30, 3};
+ private static final int WARMUP_ROUNDS = 3;
+ private static final int MEASUREMENT_ROUNDS = 6;
+ private static final ThreadMXBean ALLOCATION_MX_BEAN =
+ (ThreadMXBean) ManagementFactory.getThreadMXBean();
+
+ private TsFileProcessor processor;
+ private DataRegionInfo dataRegionInfo;
+ private File tsFile;
+ private IMemTable memTable;
+ private long benchmarkBlackhole;
+
+ @Before
+ public void setUp() throws Exception {
+ Assert.assertTrue(ManualPerformanceTestUtils.enableThreadMetrics());
+ EnvironmentUtils.envSetUp();
+ tsFile =
+
SystemFileFactory.INSTANCE.getFile(TestConstant.getTestTsFilePath(STORAGE_GROUP,
0, 0, 0));
+ dataRegionInfo =
+ new DataRegionInfo(
+ new DataRegionTest.DummyDataRegion(
+ TestConstant.OUTPUT_DATA_DIR + "branch-benchmark-info",
STORAGE_GROUP));
+ processor =
+ new TsFileProcessor(
+ STORAGE_GROUP,
+ tsFile,
+ dataRegionInfo,
+ ignored -> {},
+ (ignored, updateMap, systemFlushTime) -> {},
+ true);
+ processor.setTsFileProcessorInfo(new TsFileProcessorInfo(dataRegionInfo));
+ dataRegionInfo.initTsFileProcessorInfo(processor);
+ SystemInfo.getInstance().reportStorageGroupStatus(dataRegionInfo,
processor);
+
+ List<IMeasurementSchema> schemas = createSchemas(EXISTING_COLUMNS);
+ AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(new
ArrayList<>(schemas));
+ long[] times = new long[EXISTING_ROWS];
+ Object[] columns = new Object[EXISTING_COLUMNS];
+ for (int column = 0; column < EXISTING_COLUMNS; column++) {
+ int[] values = new int[EXISTING_ROWS];
+ for (int row = 0; row < EXISTING_ROWS; row++) {
+ times[row] = row;
+ values[row] = row + column;
+ }
+ columns[column] = values;
+ }
+ memChunk.putAlignedTablet(times, columns, null, 0, EXISTING_ROWS);
+ Map<IDeviceID, IWritableMemChunkGroup> map = new HashMap<>();
+ map.put(DEVICE_ID, new AlignedWritableMemChunkGroup(memChunk, new
ArrayList<>(schemas)));
+ memTable = new PrimitiveMemTable(STORAGE_GROUP, "0", map);
+ Field workMemTable =
TsFileProcessor.class.getDeclaredField("workMemTable");
+ workMemTable.setAccessible(true);
+ workMemTable.set(processor, memTable);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ try {
+ if (processor != null) {
+ processor.putMemTableBackAndClose();
+ }
+ } finally {
+ EnvironmentUtils.cleanEnv();
+ EnvironmentUtils.cleanDir(TestConstant.OUTPUT_DATA_DIR);
+ if (tsFile != null) {
+ tsFile.delete();
+ new File(tsFile.getPath() + ".resource").delete();
+ }
+ }
+ }
+
+ @Test
+ public void compareAlignedRowsMemoryEstimation() throws Exception {
+ for (int scenario = 0; scenario < ROW_COUNTS.length; scenario++) {
+ List<InsertRowNode> rows = createRows(ROW_COUNTS[scenario],
COLUMN_COUNTS[scenario]);
+ for (int round = 0; round < WARMUP_ROUNDS; round++) {
+ runUnmeasured(rows, ITERATIONS[scenario]);
+ }
+ BenchmarkResult[] results = new BenchmarkResult[MEASUREMENT_ROUNDS];
+ for (int round = 0; round < MEASUREMENT_ROUNDS; round++) {
+ results[round] = measure(rows, ITERATIONS[scenario]);
+ }
+ BenchmarkResult summary = summarize(results);
+ System.out.printf(
+ Locale.ROOT,
+ "rows=%d columns=%d iterations=%d estimate=%d time=%.1f ns/op
allocation=%.1f B/op%n",
+ ROW_COUNTS[scenario],
+ COLUMN_COUNTS[scenario],
+ ITERATIONS[scenario],
+ summary.estimate,
+ summary.nanosPerOperation,
+ summary.allocatedBytesPerOperation);
+ }
+ Assert.assertTrue(benchmarkBlackhole > 0);
+ }
+
+ private BenchmarkResult measure(List<InsertRowNode> rows, int iterations) {
+ System.gc();
+ System.runFinalization();
+ long threadId = Thread.currentThread().getId();
+ long allocatedBytesBefore =
ALLOCATION_MX_BEAN.getThreadAllocatedBytes(threadId);
+ long elapsedNanos = 0;
+ long previousCost = 0;
+ long estimate = 0;
+ for (int iteration = 0; iteration < iterations; iteration++) {
+ resetMemoryAccounting(previousCost);
+ long startNanos = System.nanoTime();
+ try {
+ long[] increments =
+ processor.benchmarkCheckAlignedMemCostAndAddToTspInfoForRows(rows,
new HashSet<>());
+ estimate = increments[0];
+ benchmarkBlackhole = estimate;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ } finally {
+ elapsedNanos += System.nanoTime() - startNanos;
+ }
+ previousCost = estimate;
+ }
+ resetMemoryAccounting(previousCost);
+ long allocatedBytes =
+ ALLOCATION_MX_BEAN.getThreadAllocatedBytes(threadId) -
allocatedBytesBefore;
+ return new BenchmarkResult(
+ (double) elapsedNanos / iterations, (double) allocatedBytes /
iterations, estimate);
+ }
+
+ private void runUnmeasured(List<InsertRowNode> rows, int iterations) {
+ long previousCost = 0;
+ for (int iteration = 0; iteration < iterations; iteration++) {
+ resetMemoryAccounting(previousCost);
+ try {
+ previousCost =
+ processor.benchmarkCheckAlignedMemCostAndAddToTspInfoForRows(rows,
new HashSet<>())[0];
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ resetMemoryAccounting(previousCost);
+ }
+
+ private void resetMemoryAccounting(long memTableIncrement) {
+ if (memTableIncrement == 0) {
+ return;
+ }
+ memTable.releaseTVListRamCost(memTableIncrement);
+ dataRegionInfo.releaseStorageGroupMemCost(memTableIncrement);
+ SystemInfo.getInstance().resetStorageGroupStatus(dataRegionInfo);
+ }
+
+ private static BenchmarkResult summarize(BenchmarkResult[] results) {
+ double[] nanos = new double[results.length];
+ double[] allocatedBytes = new double[results.length];
+ for (int i = 0; i < results.length; i++) {
+ nanos[i] = results[i].nanosPerOperation;
+ allocatedBytes[i] = results[i].allocatedBytesPerOperation;
+ }
+ return new BenchmarkResult(
+ median(nanos), median(allocatedBytes), results[results.length -
1].estimate);
+ }
+
+ private static double median(double[] values) {
+ Arrays.sort(values);
+ int middle = values.length / 2;
+ return (values.length & 1) == 1
+ ? values[middle]
+ : values[middle - 1] + (values[middle] - values[middle - 1]) / 2;
+ }
+
+ private static List<IMeasurementSchema> createSchemas(int count) {
+ List<IMeasurementSchema> schemas = new ArrayList<>(count);
+ for (int i = 0; i < count; i++) {
+ schemas.add(new MeasurementSchema("s" + i, TSDataType.INT32));
+ }
+ return schemas;
+ }
+
+ private static List<InsertRowNode> createRows(int rowCount, int columnCount)
+ throws MetadataException {
+ List<InsertRowNode> rows = new ArrayList<>(rowCount);
+ for (int row = 0; row < rowCount; row++) {
+ String[] measurements = new String[columnCount];
+ TSDataType[] dataTypes = new TSDataType[columnCount];
+ MeasurementSchema[] schemas = new MeasurementSchema[columnCount];
+ Object[] values = new Object[columnCount];
+ int offset = row % columnCount;
+ for (int column = 0; column < columnCount; column++) {
+ int sourceColumn = (column + offset) % columnCount;
+ measurements[column] = "s" + sourceColumn;
+ dataTypes[column] = TSDataType.INT32;
+ schemas[column] = new MeasurementSchema(measurements[column],
TSDataType.INT32);
+ values[column] = ((row * columnCount + column) & 3) == 0 ? null : row
+ sourceColumn;
+ }
+ rows.add(
+ new InsertRowNode(
+ new PlanNodeId("benchmark"),
+ new PartialPath(DEVICE_PATH),
+ true,
+ measurements,
+ dataTypes,
+ schemas,
+ row,
+ values,
+ false));
+ }
+ return rows;
+ }
+
+ private static final class BenchmarkResult {
+
+ private final double nanosPerOperation;
+ private final double allocatedBytesPerOperation;
+ private final long estimate;
+
+ private BenchmarkResult(
+ double nanosPerOperation, double allocatedBytesPerOperation, long
estimate) {
+ this.nanosPerOperation = nanosPerOperation;
+ this.allocatedBytesPerOperation = allocatedBytesPerOperation;
+ this.estimate = estimate;
+ }
+ }
+}
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 0e2e533cc87..c878b9109d9 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
@@ -43,6 +43,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.commons.io.FileUtils;
import org.apache.tsfile.enums.TSDataType;
@@ -74,6 +75,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -81,6 +83,7 @@ import java.util.concurrent.ExecutionException;
import static junit.framework.TestCase.assertTrue;
import static
org.apache.iotdb.db.storageengine.dataregion.DataRegionTest.buildInsertRowNodeByTSRecord;
+import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -736,7 +739,7 @@ public class TsFileProcessorTest {
processor.insertTablet(genInsertTableNodeFors3000ToS6000(200, true), 0,
10, new TSStatus[10]);
Assert.assertEquals(4152728, memTable.getTVListsRamCost());
processor.insertTablet(genInsertTableNode(300, true), 0, 10, new
TSStatus[10]);
- Assert.assertEquals(7537280, memTable.getTVListsRamCost());
+ Assert.assertEquals(5953280, memTable.getTVListsRamCost());
processor.insertTablet(genInsertTableNodeFors3000ToS6000(300, true), 0,
10, new TSStatus[10]);
Assert.assertEquals(7705280, memTable.getTVListsRamCost());
@@ -760,6 +763,7 @@ public class TsFileProcessorTest {
Assert.assertEquals(1923360, memTable.memSize());
}
+ // Bitmap accounting follows the blocks that actually contain nulls,
including a new column.
@Test
public void alignedBitmapMemoryAccountingMatchesActualAllocations()
throws MetadataException, WriteProcessException, IOException,
IllegalPathException {
@@ -803,12 +807,115 @@ public class TsFileProcessorTest {
processor.insert(extendedColumnRow, new long[4]);
int extendedColumnIndex =
alignedMemChunk.getWorkingTVList().getBitMaps().size() - 1;
- for (BitMap bitMap :
alignedMemChunk.getWorkingTVList().getBitMaps().get(extendedColumnIndex)) {
- Assert.assertNotNull(bitMap);
- }
+ List<BitMap> extendedColumnBitMaps =
+
alignedMemChunk.getWorkingTVList().getBitMaps().get(extendedColumnIndex);
+ Assert.assertNull(extendedColumnBitMaps.get(0));
+ Assert.assertNull(extendedColumnBitMaps.get(1));
+ Assert.assertNotNull(extendedColumnBitMaps.get(2));
assertAlignedTvListRamCostMatchesActual(denseTablet.getDeviceID());
}
+ // Measurement mapping and the target block determine which primitive arrays
are charged.
+ @Test
+ public void alignedWriteMemCostUsesMeasurementMapping() {
+ AlignedWritableMemChunk memChunk =
+ new AlignedWritableMemChunk(
+ Arrays.asList(
+ new MeasurementSchema("s0", TSDataType.INT64),
+ new MeasurementSchema("s1", TSDataType.INT32)));
+ memChunk.putAlignedRow(0, new Object[] {1L, null});
+
+ long estimatedMemCost =
+ memChunk.alignedWriteRowMemCost(
+ new String[] {"s1", "s0"},
+ new TSDataType[] {TSDataType.INT32, TSDataType.INT64},
+ new Object[] {1, 1L},
+ 0);
+
+ Assert.assertEquals(
+ AlignedTVList.valueListArrayMemCost(TSDataType.INT32) -
NUM_BYTES_OBJECT_REF,
+ estimatedMemCost);
+
+ InsertRowNode reorderedRow = new InsertRowNode(new PlanNodeId(""));
+ reorderedRow.setMeasurements(new String[] {"s1", "s0"});
+ reorderedRow.setDataTypes(new TSDataType[] {TSDataType.INT32,
TSDataType.INT64});
+ reorderedRow.setValues(new Object[] {1, 1L});
+ int rowsInSameBlock = 2;
+ Assert.assertEquals(
+ AlignedTVList.valueListArrayMemCost(TSDataType.INT32) -
NUM_BYTES_OBJECT_REF,
+ memChunk.alignedWriteRowsMemCost(Collections.nCopies(rowsInSameBlock,
reorderedRow)));
+
+ int remainingRowsInBlock = PrimitiveArrayManager.ARRAY_SIZE - 1;
+ memChunk.putAlignedTablet(
+ new long[remainingRowsInBlock],
+ new Object[] {new long[remainingRowsInBlock], null},
+ null,
+ 0,
+ remainingRowsInBlock);
+ Assert.assertEquals(
+ memChunk.getWorkingTVList().alignedTvListArrayMemCost(),
+ memChunk.alignedWriteArrayMemCost(
+ new String[] {"s1", "s0"},
+ new TSDataType[] {TSDataType.INT32, TSDataType.INT64},
+ new Object[] {new int[1], new long[1]},
+ 0,
+ 1));
+
+ int rowsAcrossTwoBlocks = PrimitiveArrayManager.ARRAY_SIZE + 1;
+ Assert.assertEquals(
+ 2 * memChunk.getWorkingTVList().alignedTvListArrayMemCost(),
+
memChunk.alignedWriteRowsMemCost(Collections.nCopies(rowsAcrossTwoBlocks,
reorderedRow)));
+ }
+
+ // The incremental estimator must retain new-column state when row schemas
change at a block
+ // boundary, while an all-null occurrence must not materialize the new
column in the old block.
+ @Test
+ public void alignedWriteRowsMemCostHandlesChangingMeasurementOrders() {
+ AlignedWritableMemChunk memChunk =
+ new AlignedWritableMemChunk(
+ Arrays.asList(
+ new MeasurementSchema("s0", TSDataType.INT64),
+ new MeasurementSchema("s1", TSDataType.INT32)));
+ int existingRows = PrimitiveArrayManager.ARRAY_SIZE - 1;
+ memChunk.putAlignedTablet(
+ new long[existingRows],
+ new Object[] {new long[existingRows], new int[existingRows]},
+ null,
+ 0,
+ existingRows);
+
+ InsertRowNode firstRow = new InsertRowNode(new PlanNodeId(""));
+ firstRow.setMeasurements(new String[] {"s1", "s0", "s2"});
+ firstRow.setDataTypes(new TSDataType[] {TSDataType.INT32,
TSDataType.INT64, TSDataType.DOUBLE});
+ firstRow.setValues(new Object[] {1, 1L, null});
+
+ InsertRowNode secondRow = new InsertRowNode(new PlanNodeId(""));
+ secondRow.setMeasurements(new String[] {"s2", "s0", "s1"});
+ secondRow.setDataTypes(
+ new TSDataType[] {TSDataType.DOUBLE, TSDataType.INT64,
TSDataType.INT32});
+ secondRow.setValues(new Object[] {2D, 2L, 2});
+
+ InsertRowNode thirdRow = new InsertRowNode(new PlanNodeId(""));
+ thirdRow.setMeasurements(new String[] {"s2", "s0", "s1"});
+ thirdRow.setDataTypes(new TSDataType[] {TSDataType.DOUBLE,
TSDataType.INT64, TSDataType.INT32});
+ thirdRow.setValues(new Object[] {3D, 3L, 3});
+
+ List<InsertRowNode> rows = Arrays.asList(firstRow, secondRow, thirdRow);
+ long denseArrayMemCost =
+ memChunk.alignedWriteArrayMemCost(
+ secondRow.getMeasurements(),
+ secondRow.getDataTypes(),
+ new Object[] {new double[3], new long[3], new int[3]},
+ 0,
+ 3);
+ long expectedMemCost =
+ denseArrayMemCost
+ - AlignedTVList.valueListArrayMemCost(TSDataType.DOUBLE)
+ + NUM_BYTES_OBJECT_REF;
+
+ Assert.assertEquals(expectedMemCost,
memChunk.alignedWriteRowsMemCost(rows));
+ }
+
@Test
public void nonAlignedTvListRamCostTest()
throws MetadataException, WriteProcessException, IOException {
@@ -1246,8 +1353,7 @@ public class TsFileProcessorTest {
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()) {
+ for (AlignedTVList sortedTVList : alignedMemChunk.getSortedList()) {
actualRamCost += sortedTVList.getRamSize();
}
Assert.assertEquals(actualRamCost,
processor.getWorkMemTable().getTVListsRamCost());
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 f84710bf6a0..3f8f1811997 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
@@ -32,6 +32,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Objects;
import java.util.Set;
import static
org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE;
@@ -40,6 +41,36 @@ import static
org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF;
public class AlignedTVListTest {
+ // A null-only column keeps a null array slot while a populated column
materializes its array.
+ @Test
+ public void testPrimitiveArraysAreMaterializedOnlyForNonNullColumns() {
+ AlignedTVList tvList =
+ AlignedTVList.newAlignedList(Arrays.asList(TSDataType.INT64,
TSDataType.DOUBLE));
+
+ tvList.putAlignedValue(1, new Object[] {null, 2.0D});
+ Assert.assertNull(tvList.getValues().get(0).get(0));
+ Assert.assertNotNull(tvList.getValues().get(1).get(0));
+
+ // A row containing only nulls must not allocate another value array.
+ tvList.putAlignedValue(2, new Object[] {null, null});
+ Assert.assertNull(tvList.getValues().get(0).get(0));
+ Assert.assertEquals(1,
tvList.getValues().get(1).stream().filter(Objects::nonNull).count());
+ }
+
+ // Tablet writes allocate arrays only for input columns that carry a column
vector.
+ @Test
+ public void testBatchWriteMaterializesOnlyColumnsWithValues() {
+ AlignedTVList tvList =
+ AlignedTVList.newAlignedList(Arrays.asList(TSDataType.INT64,
TSDataType.INT32));
+ long[] times = {1, 2, 3};
+ Object[] columns = {null, new int[] {1, 2, 3}};
+
+ tvList.putAlignedValues(times, columns, null, 0, times.length);
+
Assert.assertTrue(tvList.getValues().get(0).stream().allMatch(Objects::isNull));
+ Assert.assertNotNull(tvList.getValues().get(1).get(0));
+ }
+
+ // Value-array cost excludes bitmap storage because bitmaps are allocated
independently.
@Test
public void testValueListArrayMemCostExcludesBitmapReservation() {
long expected = (long) ARRAY_SIZE * Long.BYTES + NUM_BYTES_ARRAY_HEADER +
NUM_BYTES_OBJECT_REF;
@@ -52,6 +83,23 @@ public class AlignedTVListTest {
Assert.assertEquals(NUM_BYTES_OBJECT_REF,
AlignedTVList.bitmapReferenceRamCost());
}
+ @Test
+ public void testStaticNewAlignedListMemoryCosts() {
+ List<TSDataType> dataTypes = Arrays.asList(TSDataType.INT64,
TSDataType.INT32);
+ AlignedTVList tvList = AlignedTVList.newAlignedList(dataTypes);
+
+ Assert.assertEquals(
+ tvList.getRamSize(),
AlignedTVList.alignedTvListInitialMemCost(dataTypes.size()));
+ long primitiveArrayMemCost =
+ dataTypes.stream()
+ .mapToLong(
+ dataType -> AlignedTVList.valueListArrayMemCost(dataType) -
NUM_BYTES_OBJECT_REF)
+ .sum();
+ Assert.assertEquals(
+ tvList.alignedTvListArrayMemCost() - primitiveArrayMemCost,
+
AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays(dataTypes.size()));
+ }
+
@Test
public void testAlignedTVList1() {
List<TSDataType> dataTypes = new ArrayList<>();
@@ -163,6 +211,7 @@ public class AlignedTVListTest {
}
}
+ // A null first appears in the third block, so only that block receives a
compact bitmap.
@Test
public void testBitmapIsAllocatedLazilyWithCompactBackingArray() {
AlignedTVList tvList =
@@ -184,18 +233,16 @@ public class AlignedTVListTest {
ARRAY_SIZE / Byte.SIZE + 1,
firstColumnBitMaps.get(2).getByteArray().length);
Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE * 2 + 1, 0));
Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE * 2, 0));
- long primitiveArrayAndBitmapCost =
- 3L * tvList.alignedTvListArrayMemCost()
- + 3L * AlignedTVList.bitmapReferenceRamCost()
- + AlignedTVList.bitmapRamCost();
- Assert.assertTrue(tvList.getRamSize() > primitiveArrayAndBitmapCost);
+ Assert.assertEquals(3,
tvList.getValues().get(0).stream().filter(Objects::nonNull).count());
+ Assert.assertEquals(3,
tvList.getValues().get(1).stream().filter(Objects::nonNull).count());
Assert.assertEquals(tvList.getRamSize(),
tvList.calculateRamSize().getRamSize());
Assert.assertEquals(tvList.getRamSize(), tvList.clone().getRamSize());
Assert.assertEquals(tvList.getRamSize(),
tvList.cloneForFlushSort().getRamSize());
}
+ // Extending a populated TVList creates null slots but no value arrays or
bitmap structures.
@Test
- public void testExtendedColumnRamCostIncludesActualBitmaps() {
+ public void testExtendColumnDoesNotMaterializeArraysOrBitmaps() {
AlignedTVList tvList =
AlignedTVList.newAlignedList(new
ArrayList<>(Arrays.asList(TSDataType.INT64)));
for (int i = 0; i <= ARRAY_SIZE; i++) {
@@ -203,18 +250,26 @@ public class AlignedTVListTest {
}
long ramSizeBeforeExtension = tvList.getRamSize();
+ int oldColumnCount = tvList.getTsDataTypes().size();
+ long expectedExtensionCost =
+ (long) tvList.getTimestamps().size() * NUM_BYTES_OBJECT_REF
+ + 2L
+ * (RamUsageEstimator.sizeOfObjectArray(oldColumnCount + 1)
+ - RamUsageEstimator.sizeOfObjectArray(oldColumnCount))
+ + RamUsageEstimator.sizeOfLongArray(oldColumnCount + 1)
+ - RamUsageEstimator.sizeOfLongArray(oldColumnCount)
+ + RamUsageEstimator.shallowSizeOf(new ArrayList<>())
+ + RamUsageEstimator.sizeOfObjectArray(0);
tvList.extendColumn(TSDataType.INT32);
- Assert.assertTrue(
- tvList.getRamSize() - ramSizeBeforeExtension
- >= 2L
- * (AlignedTVList.valueListArrayMemCost(TSDataType.INT32)
- + AlignedTVList.bitmapReferenceRamCost()
- + AlignedTVList.bitmapRamCost()));
+ Assert.assertEquals(expectedExtensionCost, tvList.getRamSize() -
ramSizeBeforeExtension);
+
Assert.assertTrue(tvList.getValues().get(1).stream().allMatch(Objects::isNull));
+ Assert.assertNull(tvList.getBitMaps());
tvList.clear();
Assert.assertTrue(tvList.getRamSize() > 0);
}
+ // An input bitmap without marked values must not create a retained TVList
bitmap.
@Test
public void testEmptyInputBitmapsDoNotMaterializeMemTableBitmaps() {
AlignedTVList tvList =
AlignedTVList.newAlignedList(Arrays.asList(TSDataType.INT64));
