diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java index 512879bfd3cb0..fcc7a75470401 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -757,6 +757,8 @@ private StorageEngineMessages() {} "{}: {} is closed during flush, abandon flush task"; public static final String STORAGE_LOG_THE_COMPRESSION_RATIO_OF_TSFILE_IS_TOTALMEMTABLESIZE_THE_8CE66BE3 = "The compression ratio of tsfile {} is {}, totalMemTableSize: {}, the file size: {}"; + public static final String STORAGE_LOG_THE_COMPRESSION_RATIO_OF_TSFILE_IS_TOTALMEMTABLESIZE_THE_FILE_SIZE_NULL_VALUE_RATIO_46F3B1F7 = + "The compression ratio of tsfile {} is {}, totalMemTableSize: {}, the file size: {}, null value ratio: {}"; public static final String STORAGE_LOG_STORAGE_GROUP_CLOSE_AND_REMOVE_EMPTY_FILE_72D42293 = "Storage group {} close and remove empty file {}"; public static final String STORAGE_LOG_PUT_THE_MEMTABLE_SIGNAL_OUT_OF_FLUSHINGMEMTABLES_BUT_IT_D78AF257 = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java index 96d09c1c14140..19a2e5f68f65b 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -746,6 +746,8 @@ private StorageEngineMessages() {} "{}：{} 在 flush 期间已关闭，放弃 flush 任务"; public static final String STORAGE_LOG_THE_COMPRESSION_RATIO_OF_TSFILE_IS_TOTALMEMTABLESIZE_THE_8CE66BE3 = "TsFile {} 的压缩率为 {}，totalMemTableSize：{}，文件大小：{}"; + public static final String STORAGE_LOG_THE_COMPRESSION_RATIO_OF_TSFILE_IS_TOTALMEMTABLESIZE_THE_FILE_SIZE_NULL_VALUE_RATIO_46F3B1F7 = + "TsFile {} 的压缩率为 {}，totalMemTableSize：{}，文件大小：{}，空值比例：{}"; public static final String STORAGE_LOG_STORAGE_GROUP_CLOSE_AND_REMOVE_EMPTY_FILE_72D42293 = "Storage group {} 关闭并移除空文件 {}"; public static final String STORAGE_LOG_PUT_THE_MEMTABLE_SIGNAL_OUT_OF_FLUSHINGMEMTABLES_BUT_IT_D78AF257 = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java index 1b6be1ab3db23..88197d03cd6ba 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java @@ -434,6 +434,7 @@ private void unbindWALCostMetrics(AbstractMetricService metricService) { public static final String SERIES_NUM = "series_num"; public static final String AVG_SERIES_POINT_NUM = "avg_series_points_num"; public static final String COMPRESSION_RATIO = "compression_ratio"; + public static final String NULL_VALUE_RATIO = "null_value_ratio"; public static final String EFFECTIVE_RATIO_INFO = "effective_ratio_info"; public static final String OLDEST_MEM_TABLE_RAM_WHEN_CAUSE_SNAPSHOT = "oldest_mem_table_ram_when_cause_snapshot"; @@ -598,7 +599,13 @@ public void removeWALNodeInfoMetrics(String walNodeId) { } public void createFlushingMemTableStatusMetrics(DataRegionId dataRegionId) { - Arrays.asList(MEM_TABLE_SIZE, SERIES_NUM, POINTS_NUM, COMPRESSION_RATIO, FLUSH_TSFILE_SIZE) + Arrays.asList( + MEM_TABLE_SIZE, + SERIES_NUM, + POINTS_NUM, + COMPRESSION_RATIO, + NULL_VALUE_RATIO, + FLUSH_TSFILE_SIZE) .forEach( name -> MetricService.getInstance() @@ -727,6 +734,7 @@ public void removeFlushingMemTableStatusMetrics(DataRegionId dataRegionId) { POINTS_NUM, AVG_SERIES_POINT_NUM, COMPRESSION_RATIO, + NULL_VALUE_RATIO, FLUSH_TSFILE_SIZE) .forEach( name -> @@ -790,6 +798,19 @@ public void recordTsFileCompressionRatioOfFlushingMemTable( new DataRegionId(Integer.parseInt(dataRegionId)).toString()); } + public void recordTsFileNullValueRatioOfFlushingMemTable( + String dataRegionId, double nullValueRatio) { + MetricService.getInstance() + .histogram( + (long) (nullValueRatio * 100), + Metric.FLUSHING_MEM_TABLE_STATUS.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + NULL_VALUE_RATIO, + Tag.REGION.toString(), + new DataRegionId(Integer.parseInt(dataRegionId)).toString()); + } + public void recordFlushingMemTableStatus( String storageGroup, long memSize, long seriesNum, long totalPointsNum, long avgSeriesNum) { DataRegionId dataRegionId = getDataRegionIdFromStorageGroupStr(storageGroup); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java index eea69f98a3f08..e24308cf505da 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java @@ -47,6 +47,7 @@ import org.apache.iotdb.db.utils.EncryptDBUtils; import org.apache.iotdb.db.utils.MemUtils; import org.apache.iotdb.db.utils.ModificationUtils; +import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.ChunkMetadata; @@ -104,6 +105,10 @@ public abstract class AbstractMemTable implements IMemTable { private long totalPointsNum = 0; + private long totalValueCount = 0; + + private long nullValueCount = 0; + private long totalPointsNumThreshold = 0; private long maxPlanIndex = Long.MIN_VALUE; @@ -222,13 +227,16 @@ public int insert(InsertRowNode insertRowNode) { MemUtils.getRowRecordSize(dataTypes, writableValues, insertRowNode.getColumnCategories()); write(insertRowNode.getDeviceID(), schemaList, insertRowNode.getTime(), writableValues); + int validValueCount = insertRowNode.getValidMeasurementNumber(true); int pointsInserted = - insertRowNode.getValidMeasurementNumber(true) + validValueCount - (IoTDBDescriptor.getInstance().getConfig().isIncludeNullValueInWriteThroughputMetric() ? 0 : nullPointsNumber); totalPointsNum += pointsInserted; + totalValueCount += validValueCount; + nullValueCount += nullPointsNumber; return pointsInserted; } @@ -266,12 +274,15 @@ public int insertAlignedRow(InsertRowNode insertRowNode) { dataTypes, writableValues, insertRowNode.getColumnCategories()); writeAlignedRow( insertRowNode.getDeviceID(), schemaList, insertRowNode.getTime(), writableValues); + int validValueCount = insertRowNode.getValidMeasurementNumber(true); int pointsInserted = - insertRowNode.getValidMeasurementNumber(true) + validValueCount - (IoTDBDescriptor.getInstance().getConfig().isIncludeNullValueInWriteThroughputMetric() ? 0 : nullPointsNumber); totalPointsNum += pointsInserted; + totalValueCount += validValueCount; + nullValueCount += nullPointsNumber; return pointsInserted; } @@ -279,17 +290,21 @@ public int insertAlignedRow(InsertRowNode insertRowNode) { public int insertTablet(InsertTabletNode insertTabletNode, int start, int end) throws WriteProcessException { try { - int nullPointsNumber = computeTabletNullPointsNumber(insertTabletNode, start, end, true); + int nullPointsNumber = + computeTabletNullPointsNumber(insertTabletNode, start, end, true, null); writeTabletNode(insertTabletNode, start, end); memSize += MemUtils.getTabletSize(insertTabletNode, start, end); + int validValueCount = insertTabletNode.getValidMeasurementNumber(true) * (end - start); int pointsInserted = - (insertTabletNode.getValidMeasurementNumber(true) * (end - start)) + validValueCount - (IoTDBDescriptor.getInstance() .getConfig() .isIncludeNullValueInWriteThroughputMetric() ? 0 : nullPointsNumber); totalPointsNum += pointsInserted; + totalValueCount += validValueCount; + nullValueCount += nullPointsNumber; return pointsInserted; } catch (RuntimeException e) { throw new WriteProcessException(e); @@ -301,18 +316,24 @@ public int insertAlignedTablet( InsertTabletNode insertTabletNode, int start, int end, TSStatus[] results) throws WriteProcessException { try { - int nullPointsNumber = computeTabletNullPointsNumber(insertTabletNode, start, end, true); + int nullPointsNumber = + computeTabletNullPointsNumber(insertTabletNode, start, end, true, results); writeAlignedTablet(insertTabletNode, start, end, results); // TODO-Table: what is the relation between this and TsFileProcessor.checkMemCost memSize += MemUtils.getAlignedTabletSize(insertTabletNode, start, end, results); + int validValueCount = + insertTabletNode.getValidMeasurementNumber(true) + * computeSuccessfulRowCount(results, start, end); int pointsInserted = - (insertTabletNode.getValidMeasurementNumber(true) * (end - start)) + validValueCount - (IoTDBDescriptor.getInstance() .getConfig() .isIncludeNullValueInWriteThroughputMetric() ? 0 : nullPointsNumber); totalPointsNum += pointsInserted; + totalValueCount += validValueCount; + nullValueCount += nullPointsNumber; return pointsInserted; } catch (RuntimeException e) { throw new WriteProcessException(e); @@ -320,7 +341,11 @@ public int insertAlignedTablet( } private static int computeTabletNullPointsNumber( - InsertTabletNode insertTabletNode, int start, int end, boolean countFieldOnly) { + InsertTabletNode insertTabletNode, + int start, + int end, + boolean countFieldOnly, + TSStatus[] results) { Object[] values = insertTabletNode.getBitMaps(); int nullPointsNumber = 0; if (values != null) { @@ -339,7 +364,7 @@ private static int computeTabletNullPointsNumber( BitMap bitMap = i < values.length ? (BitMap) values[i] : null; if (bitMap != null && !bitMap.isAllUnmarked()) { for (int j = start; j < end; j++) { - if (bitMap.isMarked(j)) { + if (!isFailedRow(results, j) && bitMap.isMarked(j)) { nullPointsNumber++; } } @@ -349,6 +374,22 @@ private static int computeTabletNullPointsNumber( return nullPointsNumber; } + private static int computeSuccessfulRowCount(TSStatus[] results, int start, int end) { + int successfulRowCount = 0; + for (int i = start; i < end; i++) { + if (!isFailedRow(results, i)) { + successfulRowCount++; + } + } + return successfulRowCount; + } + + private static boolean isFailedRow(TSStatus[] results, int index) { + return results != null + && results[index] != null + && results[index].getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode(); + } + @Override public void write( IDeviceID deviceId, @@ -481,6 +522,21 @@ public long getTotalPointsNum() { return totalPointsNum; } + @Override + public long getTotalValueCount() { + return totalValueCount; + } + + @Override + public long getNullValueCount() { + return nullValueCount; + } + + @Override + public double getNullValueRatio() { + return totalValueCount == 0 ? 0 : (double) nullValueCount / totalValueCount; + } + @Override public long size() { long sum = 0; @@ -501,6 +557,8 @@ public void clear() { memSize = 0; seriesNumber = 0; totalPointsNum = 0; + totalValueCount = 0; + nullValueCount = 0; totalPointsNumThreshold = 0; tvListRamCost = 0; maxPlanIndex = 0; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java index b988fc94185e4..4d38fea2c1124 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java @@ -90,6 +90,15 @@ void writeAlignedRow( long getTotalPointsNum(); + long getTotalValueCount(); + + long getNullValueCount(); + + /** + * @return the ratio of null values to all valid values written into this MemTable + */ + double getNullValueRatio(); + /** * insert into this memtable * 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 8b14459a7600e..bded5e1c62769 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 @@ -196,6 +196,10 @@ public class TsFileProcessor { /** Total memtable size for mem control. */ private long totalMemTableSize; + private long totalValueCount; + + private long nullValueCount; + private final AtomicBoolean isTotallyGeneratedByPipe = new AtomicBoolean(true); private static final String FLUSH_QUERY_WRITE_LOCKED = "{}: {} get flushQueryLock write lock"; @@ -1652,6 +1656,8 @@ private Future addAMemtableIntoFlushingList(IMemTable tobeFlushed) throws IOE if (!(tobeFlushed.isSignalMemTable() || tobeFlushed.isEmpty())) { totalMemTableSize += tobeFlushed.memSize(); + totalValueCount += tobeFlushed.getTotalValueCount(); + nullValueCount += tobeFlushed.getNullValueCount(); } WritingMetrics.getInstance() .recordMemTableLiveDuration(System.currentTimeMillis() - getWorkMemTableCreatedTime()); @@ -1942,16 +1948,20 @@ public void flushOneMemTable() { private void updateCompressionRatio() { try { double compressionRatio = ((double) totalMemTableSize) / writer.getPos(); + double nullValueRatio = totalValueCount == 0 ? 0 : (double) nullValueCount / totalValueCount; logger.info( StorageEngineMessages - .STORAGE_LOG_THE_COMPRESSION_RATIO_OF_TSFILE_IS_TOTALMEMTABLESIZE_THE_8CE66BE3, + .STORAGE_LOG_THE_COMPRESSION_RATIO_OF_TSFILE_IS_TOTALMEMTABLESIZE_THE_FILE_SIZE_NULL_VALUE_RATIO_46F3B1F7, writer.getFile().getAbsolutePath(), String.format("%.2f", compressionRatio), totalMemTableSize, - writer.getPos()); + writer.getPos(), + nullValueRatio); String dataRegionId = dataRegionInfo.getDataRegion().getDataRegionIdString(); WritingMetrics.getInstance() .recordTsFileCompressionRatioOfFlushingMemTable(dataRegionId, compressionRatio); + WritingMetrics.getInstance() + .recordTsFileNullValueRatioOfFlushingMemTable(dataRegionId, nullValueRatio); CompressionRatio.getInstance().updateRatio(totalMemTableSize, writer.getPos(), dataRegionId); } catch (IOException e) { logger.error( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTablePartialInsertTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTablePartialInsertTest.java index bdad14c20e33f..6936fd3ad51db 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTablePartialInsertTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTablePartialInsertTest.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.storageengine.dataregion.memtable; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; @@ -27,6 +28,7 @@ import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.utils.BitMap; @@ -253,6 +255,7 @@ public void testInsertTablet_oneFailedMeasurement_withBitmap_pointsInsertedNotNe assertEquals(3, points); assertEquals(3, memTable.getTotalPointsNum()); + assertEquals(0, memTable.getNullValueRatio(), 0); } @Test @@ -334,6 +337,82 @@ public void testInsertTablet_noFailure_withNullBits_pointsInsertedNotNegative() assertEquals(6, memTable.getTotalPointsNum()); } + /** + * Verifies that a non-aligned MemTable accumulates null values across row and tablet writes. The + * row contributes one null out of two values, and the tablet contributes one null out of four + * values, so the final null value ratio is 2 / 6. + */ + @Test + public void testNullValueRatioIsAccumulatedForRowAndTabletWrites() + throws IllegalPathException, WriteProcessException { + InsertRowNode rowNode = + buildInsertRowNode( + new String[] {"s0", "s1"}, new Object[] {1, null}, false, -1 /* no failure */); + memTable.insert(rowNode); + + BitMap[] bitMaps = new BitMap[2]; + bitMaps[1] = new BitMap(2); + bitMaps[1].mark(0); + InsertTabletNode tabletNode = + buildInsertTabletNode(new String[] {"s0", "s1"}, 2, bitMaps, -1 /* no failure */); + memTable.insertTablet(tabletNode, 0, 2); + + assertEquals(6, memTable.getTotalValueCount()); + assertEquals(2, memTable.getNullValueCount()); + assertEquals(1.0 / 3, memTable.getNullValueRatio(), 0.000001); + } + + /** + * Verifies that an aligned MemTable accumulates null values across row and tablet writes. The row + * contributes one null out of two values, and the tablet contributes two nulls out of four + * values, so the final null value ratio is 3 / 6. + */ + @Test + public void testNullValueRatioIsAccumulatedForAlignedWrites() + throws IllegalPathException, WriteProcessException { + InsertRowNode rowNode = + buildAlignedInsertRowNode( + new String[] {"s0", "s1"}, new Object[] {1, null}, -1 /* no failure */); + memTable.insertAlignedRow(rowNode); + + BitMap[] bitMaps = new BitMap[2]; + bitMaps[1] = new BitMap(2); + bitMaps[1].mark(0); + bitMaps[1].mark(1); + InsertTabletNode tabletNode = + buildInsertTabletNode(new String[] {"s0", "s1"}, 2, bitMaps, -1 /* no failure */); + memTable.insertAlignedTablet(tabletNode, 0, 2, null); + + assertEquals(6, memTable.getTotalValueCount()); + assertEquals(3, memTable.getNullValueCount()); + assertEquals(0.5, memTable.getNullValueRatio(), 0.000001); + } + + /** + * Verifies that failed aligned-tablet rows are excluded from null value statistics. The first row + * contributes one null out of two values, while the second row fails and contributes no values, + * so the final null value ratio is 1 / 2. + */ + @Test + public void testNullValueRatioExcludesFailedAlignedTabletRows() + throws IllegalPathException, WriteProcessException { + BitMap[] bitMaps = new BitMap[2]; + bitMaps[1] = new BitMap(2); + bitMaps[1].mark(0); + InsertTabletNode tabletNode = + buildInsertTabletNode(new String[] {"s0", "s1"}, 2, bitMaps, -1 /* no failure */); + TSStatus[] results = new TSStatus[2]; + results[1] = new TSStatus(TSStatusCode.OUT_OF_TTL.getStatusCode()); + + int points = memTable.insertAlignedTablet(tabletNode, 0, 2, results); + + assertEquals(1, points); + assertEquals(1, memTable.getTotalPointsNum()); + assertEquals(2, memTable.getTotalValueCount()); + assertEquals(1, memTable.getNullValueCount()); + assertEquals(0.5, memTable.getNullValueRatio(), 0.000001); + } + // ========================================================================= // Helpers // ========================================================================= @@ -344,6 +423,12 @@ public void testInsertTablet_noFailure_withNullBits_pointsInsertedNotNegative() */ private static InsertRowNode buildAlignedInsertRowNode( String[] measurementNames, Object[] values, int failedIndex) throws IllegalPathException { + return buildInsertRowNode(measurementNames, values, true, failedIndex); + } + + private static InsertRowNode buildInsertRowNode( + String[] measurementNames, Object[] values, boolean aligned, int failedIndex) + throws IllegalPathException { int n = measurementNames.length; TSDataType[] dataTypes = new TSDataType[n]; MeasurementSchema[] schemas = new MeasurementSchema[n]; @@ -355,7 +440,7 @@ private static InsertRowNode buildAlignedInsertRowNode( new InsertRowNode( new PlanNodeId("test"), new PartialPath("root.sg.d1"), - true /* isAligned */, + aligned, measurementNames, dataTypes, schemas,