This is an automated email from the ASF dual-hosted git repository. haonan pushed a commit to branch fix_alignedTVList_mem_calculate in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 4500fe59d236987f87b69090cc76cc338a27f1b0 Author: HTHou <[email protected]> AuthorDate: Thu Aug 10 16:25:53 2023 +0800 [IOTDB-6108] Fix AlignedTVList memory calculation is imprecise --- .../dataregion/memtable/TsFileProcessor.java | 147 +++++++++++++-------- .../db/utils/datastructure/AlignedTVList.java | 17 +++ .../dataregion/memtable/TsFileProcessorTest.java | 97 ++++++++++++++ 3 files changed, 203 insertions(+), 58 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..d4ff0c1d1d5 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,54 @@ 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; + 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]); + } } - // extending the column of aligned mem chunk - if (alignedMemChunk != null && !alignedMemChunk.containsMeasurement(measurements[i])) { + // here currentChunkPointNum >= 1 + if ((alignedMemChunk.alignedListSize() % PrimitiveArrayManager.ARRAY_SIZE) == 0) { + dataTypesInTVList.addAll(((AlignedTVList) alignedMemChunk.getTVList()).getTsDataTypes()); memTableIncrement += - (alignedMemChunk.alignedListSize() / PrimitiveArrayManager.ARRAY_SIZE + 1) - * dataTypes[i].getDataTypeSize(); - } - // TEXT data mem size - if (dataTypes[i] == TSDataType.TEXT && values[i] != null) { - textDataIncrement += MemUtils.getBinarySize((Binary) values[i]); + AlignedTVList.alignedTvListArrayMemCost(dataTypesInTVList.toArray(new TSDataType[0])); } } updateMemoryInfo(memTableIncrement, chunkMetadataIncrement, textDataIncrement); @@ -586,8 +598,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 +618,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 +627,62 @@ 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 + 1; + } 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.toArray(new TSDataType[0])); } } } 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..bd8db139c85 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,23 @@ public abstract class AlignedTVList extends TVList { 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; + } }
