This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch rel/1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.2 by this push:
new 05517f26edc [IOTDB-6108] Fix AlignedTVList memory calculation is
imprecise (#10827) (#10847)
05517f26edc is described below
commit 05517f26edcba72657c77c07a4dc318325fd05ee
Author: Haonan <[email protected]>
AuthorDate: Mon Aug 14 17:29:29 2023 +0800
[IOTDB-6108] Fix AlignedTVList memory calculation is imprecise (#10827)
(#10847)
---
.../dataregion/memtable/TsFileProcessor.java | 146 ++++++++++++---------
.../db/utils/datastructure/AlignedTVList.java | 46 +++++++
.../dataregion/memtable/TsFileProcessorTest.java | 97 ++++++++++++++
3 files changed, 230 insertions(+), 59 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index 52352408b19..dc7086dde70 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
@@ -464,42 +464,53 @@ public class TsFileProcessor {
long memTableIncrement = 0L;
long textDataIncrement = 0L;
long chunkMetadataIncrement = 0L;
- AlignedWritableMemChunk alignedMemChunk = null;
// get device id
IDeviceID deviceID = getDeviceID(deviceId);
-
if (workMemTable.checkIfChunkDoesNotExist(deviceID,
AlignedPath.VECTOR_PLACEHOLDER)) {
+ // 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;
+ }
+ // TEXT data mem size
+ if (dataTypes[i] == TSDataType.TEXT && values[i] != null) {
+ textDataIncrement += MemUtils.getBinarySize((Binary) values[i]);
+ }
+ }
} else {
- // here currentChunkPointNum >= 1
- long currentChunkPointNum =
- workMemTable.getCurrentTVListSize(deviceID,
AlignedPath.VECTOR_PLACEHOLDER);
- memTableIncrement +=
- (currentChunkPointNum % PrimitiveArrayManager.ARRAY_SIZE) == 0
- ? AlignedTVList.alignedTvListArrayMemCost(dataTypes)
- : 0;
- alignedMemChunk =
+ // for existed device of this mem table
+ AlignedWritableMemChunk alignedMemChunk =
((AlignedWritableMemChunkGroup)
workMemTable.getMemTableMap().get(deviceID))
.getAlignedMemChunk();
- }
- 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 != null &&
!alignedMemChunk.containsMeasurement(measurements[i])) {
- memTableIncrement +=
- (alignedMemChunk.alignedListSize() /
PrimitiveArrayManager.ARRAY_SIZE + 1)
- * dataTypes[i].getDataTypeSize();
+ List<TSDataType> dataTypesInTVList = new ArrayList<>();
+ 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] == TSDataType.TEXT && values[i] != null) {
+ textDataIncrement += MemUtils.getBinarySize((Binary) values[i]);
+ }
}
- // TEXT data mem size
- if (dataTypes[i] == TSDataType.TEXT && values[i] != null) {
- textDataIncrement += MemUtils.getBinarySize((Binary) values[i]);
+ // here currentChunkPointNum >= 1
+ if ((alignedMemChunk.alignedListSize() %
PrimitiveArrayManager.ARRAY_SIZE) == 0) {
+ dataTypesInTVList.addAll(((AlignedTVList)
alignedMemChunk.getTVList()).getTsDataTypes());
+ memTableIncrement +=
AlignedTVList.alignedTvListArrayMemCost(dataTypesInTVList);
}
}
updateMemoryInfo(memTableIncrement, chunkMetadataIncrement,
textDataIncrement);
@@ -586,8 +597,9 @@ public class TsFileProcessor {
long acquireArray =
(end - start - 1 + (currentChunkPointNum %
PrimitiveArrayManager.ARRAY_SIZE))
/ PrimitiveArrayManager.ARRAY_SIZE;
- memIncrements[0] +=
- acquireArray == 0 ? 0 : acquireArray *
TVList.tvListArrayMemCost(dataType);
+ if (acquireArray != 0) {
+ memIncrements[0] += acquireArray *
TVList.tvListArrayMemCost(dataType);
+ }
}
}
// TEXT data size
@@ -605,7 +617,6 @@ public class TsFileProcessor {
int end,
long[] memIncrements,
Object[] columns) {
- AlignedWritableMemChunk vectorMemChunk = null;
// memIncrements = [memTable, text, chunk metadata] respectively
if (workMemTable.checkIfChunkDoesNotExist(deviceId,
AlignedPath.VECTOR_PLACEHOLDER)) {
// ChunkMetadataIncrement
@@ -615,43 +626,60 @@ public class TsFileProcessor {
memIncrements[0] +=
((end - start) / PrimitiveArrayManager.ARRAY_SIZE + 1)
* AlignedTVList.alignedTvListArrayMemCost(dataTypes);
- } else {
- int currentChunkPointNum =
- (int) workMemTable.getCurrentTVListSize(deviceId,
AlignedPath.VECTOR_PLACEHOLDER);
- if (currentChunkPointNum % PrimitiveArrayManager.ARRAY_SIZE == 0) {
- memIncrements[0] +=
- ((end - start) / PrimitiveArrayManager.ARRAY_SIZE + 1)
- * AlignedTVList.alignedTvListArrayMemCost(dataTypes);
- } else {
- int acquireArray =
- (end - start - 1 + (currentChunkPointNum %
PrimitiveArrayManager.ARRAY_SIZE))
- / PrimitiveArrayManager.ARRAY_SIZE;
- memIncrements[0] +=
- acquireArray == 0
- ? 0
- : acquireArray *
AlignedTVList.alignedTvListArrayMemCost(dataTypes);
+ for (int i = 0; i < dataTypes.length; i++) {
+ TSDataType dataType = dataTypes[i];
+ String measurement = measurementIds[i];
+ Object column = columns[i];
+ if (dataType == null || column == null || measurement == null) {
+ continue;
+ }
+ // TEXT data size
+ if (dataType == TSDataType.TEXT) {
+ Binary[] binColumn = (Binary[]) columns[i];
+ memIncrements[1] += MemUtils.getBinaryColumnSize(binColumn, start,
end);
+ }
}
- vectorMemChunk =
+
+ } else {
+ AlignedWritableMemChunk alignedMemChunk =
((AlignedWritableMemChunkGroup)
workMemTable.getMemTableMap().get(deviceId))
.getAlignedMemChunk();
- }
- for (int i = 0; i < dataTypes.length; i++) {
- TSDataType dataType = dataTypes[i];
- String measurement = measurementIds[i];
- Object column = columns[i];
- if (dataType == null || column == null || measurement == null) {
- continue;
+ List<TSDataType> dataTypesInTVList = new ArrayList<>();
+ for (int i = 0; i < dataTypes.length; i++) {
+ TSDataType dataType = dataTypes[i];
+ String measurement = measurementIds[i];
+ Object column = columns[i];
+ 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 == TSDataType.TEXT) {
+ Binary[] binColumn = (Binary[]) columns[i];
+ memIncrements[1] += MemUtils.getBinaryColumnSize(binColumn, start,
end);
+ }
}
- // extending the column of aligned mem chunk
- if (vectorMemChunk != null &&
!vectorMemChunk.containsMeasurement(measurementIds[i])) {
- memIncrements[0] +=
- (vectorMemChunk.alignedListSize() /
PrimitiveArrayManager.ARRAY_SIZE + 1)
- * dataType.getDataTypeSize();
+ 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;
}
- // TEXT data size
- if (dataType == TSDataType.TEXT) {
- Binary[] binColumn = (Binary[]) columns[i];
- memIncrements[1] += MemUtils.getBinaryColumnSize(binColumn, start,
end);
+ if (acquireArray != 0) {
+ dataTypesInTVList.addAll(((AlignedTVList)
alignedMemChunk.getTVList()).getTsDataTypes());
+ memIncrements[0] +=
+ acquireArray *
AlignedTVList.alignedTvListArrayMemCost(dataTypesInTVList);
}
}
}
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 f8c1d568341..95bbd231ab9 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
@@ -854,6 +854,52 @@ public abstract class AlignedTVList extends TVList {
return size;
}
+ /**
+ * Get the single alignedTVList array mem cost by give types.
+ *
+ * @param types the types in the vector
+ * @return AlignedTvListArrayMemSize
+ */
+ public static long alignedTvListArrayMemCost(List<TSDataType> types) {
+ long size = 0;
+ // value array mem size
+ for (TSDataType type : types) {
+ if (type != null) {
+ size += (long) PrimitiveArrayManager.ARRAY_SIZE * (long)
type.getDataTypeSize();
+ }
+ }
+ // size is 0 when all types are null
+ if (size == 0) {
+ return size;
+ }
+ // time array mem size
+ size += PrimitiveArrayManager.ARRAY_SIZE * 8L;
+ // index array mem size
+ size += PrimitiveArrayManager.ARRAY_SIZE * 4L;
+ // array headers mem size
+ size += (long) NUM_BYTES_ARRAY_HEADER * (2 + types.size());
+ // Object references size in ArrayList
+ size += (long) NUM_BYTES_OBJECT_REF * (2 + types.size());
+ return size;
+ }
+
+ /**
+ * Get the single column array mem cost by give type.
+ *
+ * @param type the type of the value column
+ * @return valueListArrayMemCost
+ */
+ public static long valueListArrayMemCost(TSDataType type) {
+ long size = 0;
+ // value array mem size
+ size += (long) PrimitiveArrayManager.ARRAY_SIZE * (long)
type.getDataTypeSize();
+ // array headers mem size
+ size += NUM_BYTES_ARRAY_HEADER;
+ // Object references size in ArrayList
+ size += NUM_BYTES_OBJECT_REF;
+ return size;
+ }
+
/** Build TsBlock by column. */
public TsBlock buildTsBlock(
int floatPrecision, List<TSEncoding> encodingList, List<List<TimeRange>>
deletionList) {
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 66e325ceaba..d02507da412 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
@@ -325,6 +325,60 @@ public class TsFileProcessorTest {
Assert.assertEquals(721560, memTable.memSize());
}
+ @Test
+ public void alignedTvListRamCostTest2()
+ throws MetadataException, WriteProcessException, IOException {
+ processor =
+ new TsFileProcessor(
+ storageGroup,
+ SystemFileFactory.INSTANCE.getFile(filePath),
+ sgInfo,
+ this::closeTsFileProcessor,
+ (tsFileProcessor, updateMap, systemFlushTime) -> {},
+ true);
+ TsFileProcessorInfo tsFileProcessorInfo = new TsFileProcessorInfo(sgInfo);
+ processor.setTsFileProcessorInfo(tsFileProcessorInfo);
+ this.sgInfo.initTsFileProcessorInfo(processor);
+ SystemInfo.getInstance().reportStorageGroupStatus(sgInfo, processor);
+ // Test Tablet
+ processor.insertTablet(genInsertTableNode(0, true), 0, 10, new
TSStatus[10]);
+ IMemTable memTable = processor.getWorkMemTable();
+ Assert.assertEquals(1596808, memTable.getTVListsRamCost());
+ processor.insertTablet(genInsertTableNodeFors3000ToS6000(0, true), 0, 10,
new TSStatus[10]);
+ Assert.assertEquals(3192808, memTable.getTVListsRamCost());
+ processor.insertTablet(genInsertTableNode(100, true), 0, 10, new
TSStatus[10]);
+ Assert.assertEquals(3192808, memTable.getTVListsRamCost());
+ processor.insertTablet(genInsertTableNodeFors3000ToS6000(100, true), 0,
10, new TSStatus[10]);
+ Assert.assertEquals(3192808, memTable.getTVListsRamCost());
+ processor.insertTablet(genInsertTableNode(200, true), 0, 10, new
TSStatus[10]);
+ Assert.assertEquals(3192808, memTable.getTVListsRamCost());
+ processor.insertTablet(genInsertTableNodeFors3000ToS6000(200, true), 0,
10, new TSStatus[10]);
+ Assert.assertEquals(3192808, memTable.getTVListsRamCost());
+ processor.insertTablet(genInsertTableNode(300, true), 0, 10, new
TSStatus[10]);
+ Assert.assertEquals(6385616, memTable.getTVListsRamCost());
+ processor.insertTablet(genInsertTableNodeFors3000ToS6000(300, true), 0,
10, new TSStatus[10]);
+ Assert.assertEquals(6385616, memTable.getTVListsRamCost());
+
+ Assert.assertEquals(240000, memTable.getTotalPointsNum());
+ Assert.assertEquals(1920960, memTable.memSize());
+ // Test records
+ for (int i = 1; i <= 100; i++) {
+ TSRecord record = new TSRecord(i, deviceId);
+ record.addTuple(DataPoint.getDataPoint(dataType, measurementId,
String.valueOf(i)));
+ processor.insert(buildInsertRowNodeByTSRecord(record));
+ }
+ Assert.assertEquals(6387232, memTable.getTVListsRamCost());
+ // Test records
+ for (int i = 1; i <= 100; i++) {
+ TSRecord record = new TSRecord(i, deviceId);
+ record.addTuple(DataPoint.getDataPoint(dataType, "s1",
String.valueOf(i)));
+ processor.insert(buildInsertRowNodeByTSRecord(record));
+ }
+ Assert.assertEquals(6388848, memTable.getTVListsRamCost());
+ Assert.assertEquals(240200, memTable.getTotalPointsNum());
+ Assert.assertEquals(1923360, memTable.memSize());
+ }
+
@Test
public void nonAlignedTvListRamCostTest()
throws MetadataException, WriteProcessException, IOException {
@@ -478,4 +532,47 @@ public class TsFileProcessorTest {
return insertTabletNode;
}
+
+ private InsertTabletNode genInsertTableNodeFors3000ToS6000(long startTime,
boolean isAligned)
+ throws IllegalPathException {
+ String deviceId = "root.sg.device5";
+ String[] measurements = new String[3000];
+ TSDataType[] dataTypes = new TSDataType[3000];
+ TSEncoding[] encodings = new TSEncoding[3000];
+ MeasurementSchema[] schemas = new MeasurementSchema[3000];
+ for (int i = 0; i < 3000; i++) {
+ measurements[i] = "s" + i + 3000;
+ dataTypes[i] = TSDataType.INT64;
+ encodings[i] = TSEncoding.PLAIN;
+ schemas[i] = new MeasurementSchema(measurements[i], dataTypes[i],
encodings[i]);
+ }
+
+ long[] times = new long[10];
+ Object[] columns = new Object[3000];
+ for (int i = 0; i < 3000; i++) {
+ columns[i] = new long[10];
+ }
+
+ for (long r = 0; r < 10; r++) {
+ times[(int) r] = r + startTime;
+ for (int i = 0; i < 3000; i++) {
+ ((long[]) columns[i])[(int) r] = r;
+ }
+ }
+
+ InsertTabletNode insertTabletNode =
+ new InsertTabletNode(
+ new QueryId("test_write").genPlanNodeId(),
+ new PartialPath(deviceId),
+ isAligned,
+ measurements,
+ dataTypes,
+ times,
+ null,
+ columns,
+ times.length);
+ insertTabletNode.setMeasurementSchemas(schemas);
+
+ return insertTabletNode;
+ }
}
