This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch fix_flushed_flush_average_points in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 06bb2f373b6239dff86555a8bde23594871b9bb0 Author: Tian Jiang <[email protected]> AuthorDate: Thu Aug 13 11:15:17 2026 +0800 fix flush average points num may be mixed in one region --- .../iotdb/db/service/metrics/WritingMetrics.java | 34 +++++---- .../iotdb/db/utils/datastructure/TVList.java | 3 +- .../db/service/metrics/WritingMetricsTest.java | 87 ++++++++++++++++++++++ 3 files changed, 108 insertions(+), 16 deletions(-) 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 88197d03cd6..bd2f124e060 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 @@ -45,6 +45,8 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Objects; +import java.util.concurrent.atomic.DoubleAdder; +import java.util.concurrent.atomic.LongAdder; public class WritingMetrics implements IMetricSet { private static final WritingMetrics INSTANCE = new WritingMetrics(); @@ -461,7 +463,9 @@ public class WritingMetrics implements IMetricSet { private Counter manualFlushMemtableCounter = DoNothingMetricManager.DO_NOTHING_COUNTER; private Counter memControlFlushMemtableCounter = DoNothingMetricManager.DO_NOTHING_COUNTER; - private Histogram avgPointHistogram = DoNothingMetricManager.DO_NOTHING_HISTOGRAM; + // TVList preallocation needs a process-lifetime, node-wide mean independent of metric lifecycle. + private final DoubleAdder avgSeriesPointNumSum = new DoubleAdder(); + private final LongAdder flushedMemTableCount = new LongAdder(); private AutoGauge tableDiskUsageCacheBlockedRequestNumGauge = DoNothingMetricManager.DO_NOTHING_AUTO_GAUGE; @@ -603,6 +607,7 @@ public class WritingMetrics implements IMetricSet { MEM_TABLE_SIZE, SERIES_NUM, POINTS_NUM, + AVG_SERIES_POINT_NUM, COMPRESSION_RATIO, NULL_VALUE_RATIO, FLUSH_TSFILE_SIZE) @@ -616,15 +621,6 @@ public class WritingMetrics implements IMetricSet { name, Tag.REGION.toString(), dataRegionId.toString())); - avgPointHistogram = - MetricService.getInstance() - .getOrCreateHistogram( - Metric.FLUSHING_MEM_TABLE_STATUS.toString(), - MetricLevel.IMPORTANT, - Tag.NAME.toString(), - AVG_SERIES_POINT_NUM, - Tag.REGION.toString(), - dataRegionId.toString()); } public Counter createWalFlushMemTableCounterMetrics() { @@ -746,7 +742,6 @@ public class WritingMetrics implements IMetricSet { name, Tag.REGION.toString(), dataRegionId.toString())); - avgPointHistogram = DoNothingMetricManager.DO_NOTHING_HISTOGRAM; } public void recordWALNodeEffectiveInfoRatio(String walNodeId, double ratio) { @@ -845,7 +840,17 @@ public class WritingMetrics implements IMetricSet { POINTS_NUM, Tag.REGION.toString(), dataRegionId.toString()); - avgPointHistogram.update(avgSeriesNum); + MetricService.getInstance() + .histogram( + avgSeriesNum, + Metric.FLUSHING_MEM_TABLE_STATUS.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + AVG_SERIES_POINT_NUM, + Tag.REGION.toString(), + dataRegionId.toString()); + avgSeriesPointNumSum.add(avgSeriesNum); + flushedMemTableCount.increment(); } public void recordFlushTsFileSize(String storageGroup, long size) { @@ -1046,7 +1051,8 @@ public class WritingMetrics implements IMetricSet { return INSTANCE; } - public Histogram getAvgPointHistogram() { - return avgPointHistogram; + public double getGlobalAvgSeriesPointNum() { + long count = flushedMemTableCount.sum(); + return count == 0 ? 0 : avgSeriesPointNumSum.sum() / count; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java index a7aa2ff16e1..506f82d9e10 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java @@ -1462,8 +1462,7 @@ public abstract class TVList implements WALEntryValue { if (System.currentTimeMillis() - defaultArrayNumLastUpdatedTimeMs > 10_000) { defaultArrayNumLastUpdatedTimeMs = System.currentTimeMillis(); defaultArrayNum = - ((int) WritingMetrics.getInstance().getAvgPointHistogram().takeSnapshot().getMean() - / ARRAY_SIZE); + ((int) WritingMetrics.getInstance().getGlobalAvgSeriesPointNum() / ARRAY_SIZE); } return defaultArrayNum; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/WritingMetricsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/WritingMetricsTest.java new file mode 100644 index 00000000000..4e889c50421 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/WritingMetricsTest.java @@ -0,0 +1,87 @@ +/* + * 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.service.metrics; + +import org.apache.iotdb.commons.consensus.DataRegionId; +import org.apache.iotdb.commons.service.metric.MetricService; +import org.apache.iotdb.commons.service.metric.enums.Metric; +import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.metrics.config.MetricConfigDescriptor; +import org.apache.iotdb.metrics.type.Histogram; +import org.apache.iotdb.metrics.utils.MetricLevel; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import static org.apache.iotdb.db.service.metrics.WritingMetrics.AVG_SERIES_POINT_NUM; +import static org.junit.Assert.assertEquals; + +public class WritingMetricsTest { + + private static final DataRegionId FIRST_REGION = new DataRegionId(1); + private static final DataRegionId SECOND_REGION = new DataRegionId(2); + private static final MetricService METRIC_SERVICE = MetricService.getInstance(); + private static final WritingMetrics WRITING_METRICS = WritingMetrics.getInstance(); + + @BeforeClass + public static void setUp() { + MetricConfigDescriptor.getInstance().getMetricConfig().setMetricLevel(MetricLevel.IMPORTANT); + METRIC_SERVICE.startService(); + WRITING_METRICS.createFlushingMemTableStatusMetrics(FIRST_REGION); + WRITING_METRICS.createFlushingMemTableStatusMetrics(SECOND_REGION); + } + + @AfterClass + public static void tearDown() { + WRITING_METRICS.removeFlushingMemTableStatusMetrics(FIRST_REGION); + WRITING_METRICS.removeFlushingMemTableStatusMetrics(SECOND_REGION); + METRIC_SERVICE.stopService(); + } + + /** + * Verifies that flushing two DataRegions records each average series point sample in the + * histogram carrying that Region's tag, instead of routing every sample to the last-created + * histogram. + */ + @Test + public void testRecordAverageSeriesPointNumByDataRegion() { + WRITING_METRICS.recordFlushingMemTableStatus("root.db-1", 100, 2, 20, 10); + WRITING_METRICS.recordFlushingMemTableStatus("root.db-2", 200, 4, 80, 20); + + Histogram firstRegionHistogram = getAverageSeriesPointHistogram(FIRST_REGION); + Histogram secondRegionHistogram = getAverageSeriesPointHistogram(SECOND_REGION); + + assertEquals(1, firstRegionHistogram.getCount()); + assertEquals(10, firstRegionHistogram.takeSnapshot().getSum(), 0.001); + assertEquals(1, secondRegionHistogram.getCount()); + assertEquals(20, secondRegionHistogram.takeSnapshot().getSum(), 0.001); + } + + private Histogram getAverageSeriesPointHistogram(DataRegionId dataRegionId) { + return METRIC_SERVICE.getOrCreateHistogram( + Metric.FLUSHING_MEM_TABLE_STATUS.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + AVG_SERIES_POINT_NUM, + Tag.REGION.toString(), + dataRegionId.toString()); + } +}
