From 38a11dee52183035dae1f425ba47d4e09f660852 Mon Sep 17 00:00:00 2001 From: MiniSho <42286868+MiniSho@users.noreply.github.com> Date: Thu, 13 Jun 2024 18:01:26 +0800 Subject: [PATCH 01/11] init --- .../execution/load/LoadTsFileManager.java | 4 + .../metric/LoadTsFileCostMetricsSet.java | 92 +++++++++++++++++++ .../plan/analyze/AnalyzeVisitor.java | 5 + .../scheduler/load/LoadTsFileScheduler.java | 11 +++ .../metrics/DataNodeMetricsHelper.java | 4 + .../iotdb/db/storageengine/StorageEngine.java | 9 +- .../commons/service/metric/enums/Metric.java | 3 +- 7 files changed, 126 insertions(+), 2 deletions(-) create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index 054c664b48efe..50b7af16149fd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -32,6 +32,7 @@ import org.apache.iotdb.db.exception.DiskSpaceInsufficientException; import org.apache.iotdb.db.exception.LoadFileException; import org.apache.iotdb.db.pipe.agent.PipeAgent; +import org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode; import org.apache.iotdb.db.queryengine.plan.scheduler.load.LoadTsFileScheduler.LoadCommand; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; @@ -85,6 +86,9 @@ public class LoadTsFileManager { new AtomicReference<>(CONFIG.getLoadTsFileDirs()); private static final AtomicReference FOLDER_MANAGER = new AtomicReference<>(); + private static final LoadTsFileCostMetricsSet LOAD_TSFILE_METRICS = + LoadTsFileCostMetricsSet.getInstance(); + private final Map uuid2WriterManager = new ConcurrentHashMap<>(); private final Map uuid2CleanupTask = new ConcurrentHashMap<>(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java new file mode 100644 index 0000000000000..3ee75cc2b86a3 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java @@ -0,0 +1,92 @@ +/* + * 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.queryengine.metric; + +import org.apache.iotdb.commons.service.metric.enums.Metric; +import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.metrics.AbstractMetricService; +import org.apache.iotdb.metrics.impl.DoNothingMetricManager; +import org.apache.iotdb.metrics.metricsets.IMetricSet; +import org.apache.iotdb.metrics.type.Timer; +import org.apache.iotdb.metrics.utils.MetricLevel; +import org.apache.iotdb.metrics.utils.MetricType; + +import java.util.Arrays; + +public class LoadTsFileCostMetricsSet implements IMetricSet { + private static final LoadTsFileCostMetricsSet INSTANCE = new LoadTsFileCostMetricsSet(); + + public static final String ANALYSIS = "analysis"; + public static final String SPLIT = "split"; + public static final String WRITE = "write"; + + private LoadTsFileCostMetricsSet() { + // empty constructor + } + + private Timer analyzerTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer splitTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer writeTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + + public void recordCost(String stage, long costTimeInNanos) { + switch (stage) { + case ANALYSIS: + analyzerTimer.updateNanos(costTimeInNanos); + break; + case SPLIT: + splitTimer.updateNanos(costTimeInNanos); + break; + case WRITE: + writeTimer.updateNanos(costTimeInNanos); + break; + default: + throw new UnsupportedOperationException("Unsupported stage: " + stage); + } + } + + @Override + public void bindTo(AbstractMetricService metricService) { + analyzerTimer = + metricService.getOrCreateTimer( + Metric.LOAD_TIME_COST.toString(), MetricLevel.IMPORTANT, Tag.NAME.toString(), ANALYSIS); + splitTimer = + metricService.getOrCreateTimer( + Metric.LOAD_TIME_COST.toString(), MetricLevel.IMPORTANT, Tag.NAME.toString(), SPLIT); + writeTimer = + metricService.getOrCreateTimer( + Metric.LOAD_TIME_COST.toString(), MetricLevel.IMPORTANT, Tag.NAME.toString(), WRITE); + } + + @Override + public void unbindFrom(AbstractMetricService metricService) { + Arrays.asList(ANALYSIS, SPLIT, WRITE) + .forEach( + stage -> + metricService.remove( + MetricType.TIMER, + Metric.LOAD_TIME_COST.toString(), + Tag.NAME.toString(), + stage)); + } + + public static LoadTsFileCostMetricsSet getInstance() { + return LoadTsFileCostMetricsSet.INSTANCE; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java index 038514ee02f03..4a9effacacb49 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java @@ -58,6 +58,7 @@ import org.apache.iotdb.db.queryengine.common.schematree.IMeasurementSchemaInfo; import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree; import org.apache.iotdb.db.queryengine.execution.operator.window.WindowType; +import org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet; import org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager; import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType; @@ -181,6 +182,7 @@ import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_MATCH_PATTERN; import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.DEVICE; import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.ENDTIME; +import static org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet.ANALYSIS; import static org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet.PARTITION_FETCHER; import static org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet.SCHEMA_FETCHER; import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.bindSchemaForExpression; @@ -2827,10 +2829,13 @@ private InsertBaseStatement removeLogicalView( public Analysis visitLoadFile(LoadTsFileStatement loadTsFileStatement, MPPQueryContext context) { LoadTsfileAnalyzer loadTsfileAnalyzer = new LoadTsfileAnalyzer(loadTsFileStatement, context, partitionFetcher, schemaFetcher); + long startTime = System.nanoTime(); try { return loadTsfileAnalyzer.analyzeFileByFile(); } finally { loadTsfileAnalyzer.close(); + + LoadTsFileCostMetricsSet.getInstance().recordCost(ANALYSIS, System.nanoTime() - startTime); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java index d28681c161a8e..ceb0ef78f4f84 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java @@ -50,6 +50,7 @@ import org.apache.iotdb.db.queryengine.execution.load.TsFileSplitter; import org.apache.iotdb.db.queryengine.load.LoadTsFileDataCacheMemoryBlock; import org.apache.iotdb.db.queryengine.load.LoadTsFileMemoryManager; +import org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher; import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan; import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance; @@ -112,6 +113,8 @@ public class LoadTsFileScheduler implements IScheduler { IoTDBDescriptor.getInstance().getConfig().getThriftMaxFrameSize() >> 2; private static final int TRANSMIT_LIMIT = CommonDescriptor.getInstance().getConfig().getTTimePartitionSlotTransmitLimit(); + private static final LoadTsFileCostMetricsSet LOAD_TSFILE_COST_METRICS_SET = + LoadTsFileCostMetricsSet.getInstance(); private static final Set LOADING_FILE_SET = new HashSet<>(); @@ -230,6 +233,7 @@ public void start() { private boolean firstPhase(LoadSingleTsFileNode node) { final TsFileDataManager tsFileDataManager = new TsFileDataManager(this, node, block); + long startTime = System.nanoTime(); try { new TsFileSplitter( node.getTsFileResource().getTsFile(), tsFileDataManager::addOrSendTsFileData) @@ -253,6 +257,9 @@ private boolean firstPhase(LoadSingleTsFileNode node) { return false; } finally { tsFileDataManager.clear(); + + LOAD_TSFILE_COST_METRICS_SET.recordCost( + LoadTsFileCostMetricsSet.SPLIT, System.nanoTime() - startTime); } return true; } @@ -384,6 +391,7 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { throw new LoadReadOnlyException(); } + long startTime = System.nanoTime(); try { FragmentInstance instance = new FragmentInstance( @@ -405,6 +413,9 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { e.getFailureStatus().getMessage())); stateMachine.transitionToFailed(e.getFailureStatus()); return false; + } finally { + LOAD_TSFILE_COST_METRICS_SET.recordCost( + LoadTsFileCostMetricsSet.WRITE, System.nanoTime() - startTime); } // add metrics diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java index ade06cb875f55..e1e14939567a7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java @@ -34,6 +34,7 @@ import org.apache.iotdb.db.queryengine.metric.DataExchangeCostMetricSet; import org.apache.iotdb.db.queryengine.metric.DataExchangeCountMetricSet; import org.apache.iotdb.db.queryengine.metric.DriverSchedulerMetricSet; +import org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.metric.LoadTsFileMemMetricSet; import org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet; import org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet; @@ -93,6 +94,9 @@ public static void bind() { // bind subscription related metrics MetricService.getInstance().addMetricSet(SubscriptionMetrics.getInstance()); + + // bind load related metrics + MetricService.getInstance().addMetricSet(LoadTsFileCostMetricsSet.getInstance()); } private static void initSystemMetrics() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index d26304e113cce..3b23250014e31 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -54,6 +54,7 @@ import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException; import org.apache.iotdb.db.queryengine.execution.load.LoadTsFileManager; import org.apache.iotdb.db.queryengine.execution.load.LoadTsFileRateLimiter; +import org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode; @@ -116,6 +117,9 @@ public class StorageEngine implements IService { private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); private static final WritingMetrics WRITING_METRICS = WritingMetrics.getInstance(); + private static final LoadTsFileCostMetricsSet LOAD_TSFILE_COST_METRICS_SET = + LoadTsFileCostMetricsSet.getInstance(); + /** * a folder (system/databases/ by default) that persist system info. Each database will have a * subfolder under the systemDir. @@ -871,7 +875,7 @@ public TSStatus executeLoadCommand( boolean isGeneratedByPipe, ProgressIndex progressIndex) { TSStatus status = new TSStatus(); - + long startTime = System.currentTimeMillis(); try { switch (loadCommand) { case EXECUTE: @@ -904,6 +908,9 @@ public TSStatus executeLoadCommand( LOGGER.error("Execute load command {} error.", loadCommand, e); status.setCode(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()); status.setMessage(e.getMessage()); + } finally { + LOAD_TSFILE_COST_METRICS_SET.recordCost( + LoadTsFileCostMetricsSet.WRITE, System.nanoTime() - startTime); } return status; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java index c207435d3f191..284a13d9dc902 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java @@ -171,7 +171,8 @@ public enum Metric { SUBSCRIPTION_CURRENT_COMMIT_ID("subscription_current_commit_id"), SUBSCRIPTION_EVENT_TRANSFER("subscription_event_transfer"), // load related - LOAD_MEM("load_mem"); + LOAD_MEM("load_mem"), + LOAD_TIME_COST("load_time_cost"); final String value; From bba6831f7ed2e174e4850fd3edb3098fcfded154 Mon Sep 17 00:00:00 2001 From: MiniSho <42286868+MiniSho@users.noreply.github.com> Date: Thu, 13 Jun 2024 22:17:54 +0800 Subject: [PATCH 02/11] fix time unit --- .../iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java | 2 +- .../java/org/apache/iotdb/db/storageengine/StorageEngine.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java index cf98b8bd754f1..b3af926b67650 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java @@ -2846,7 +2846,7 @@ private InsertBaseStatement removeLogicalView( @Override public Analysis visitLoadFile(LoadTsFileStatement loadTsFileStatement, MPPQueryContext context) { - long startTime = System.currentTimeMillis(); + long startTime = System.nanoTime(); try (final LoadTsfileAnalyzer loadTsfileAnalyzer = new LoadTsfileAnalyzer(loadTsFileStatement, context, partitionFetcher, schemaFetcher)) { return loadTsfileAnalyzer.analyzeFileByFile(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index 3b23250014e31..11ba7e645ec57 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -875,7 +875,7 @@ public TSStatus executeLoadCommand( boolean isGeneratedByPipe, ProgressIndex progressIndex) { TSStatus status = new TSStatus(); - long startTime = System.currentTimeMillis(); + long startTime = System.nanoTime(); try { switch (loadCommand) { case EXECUTE: From 78410ca24071ccbba6e3f77458fbc3465b708666 Mon Sep 17 00:00:00 2001 From: MiniSho <42286868+MiniSho@users.noreply.github.com> Date: Fri, 14 Jun 2024 13:39:18 +0800 Subject: [PATCH 03/11] add type to write point metric --- .../db/queryengine/execution/load/LoadTsFileManager.java | 5 ++++- .../plan/scheduler/load/LoadTsFileScheduler.java | 8 ++++++-- .../apache/iotdb/commons/service/metric/enums/Metric.java | 1 + 3 files changed, 11 insertions(+), 3 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index 50b7af16149fd..88bec27018ab5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -416,6 +416,7 @@ private void loadAll(boolean isGeneratedByPipe, ProgressIndex progressIndex) DataRegion dataRegion = entry.getKey().getDataRegion(); dataRegion.loadNewTsFile(generateResource(writer, progressIndex), true, isGeneratedByPipe); + // Metrics dataRegion .getNonSystemDatabaseName() .ifPresent( @@ -435,7 +436,9 @@ private void loadAll(boolean isGeneratedByPipe, ProgressIndex progressIndex) Tag.DATABASE.toString(), databaseName, Tag.REGION.toString(), - dataRegion.getDataRegionId()); + dataRegion.getDataRegionId(), + Tag.TYPE.toString(), + Metric.LOAD_TSFILE.toString()); }); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java index ceb0ef78f4f84..6639c2d313327 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java @@ -444,7 +444,9 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { Tag.DATABASE.toString(), databaseName, Tag.REGION.toString(), - dataRegion.getDataRegionId()); + dataRegion.getDataRegionId(), + Tag.TYPE.toString(), + Metric.LOAD_TSFILE.toString()); if (!node.isGeneratedByRemoteConsensusLeader()) { MetricService.getInstance() .count( @@ -456,7 +458,9 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { Tag.DATABASE.toString(), databaseName, Tag.REGION.toString(), - dataRegion.getDataRegionId()); + dataRegion.getDataRegionId(), + Tag.TYPE.toString(), + Metric.LOAD_TSFILE.toString()); } }); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java index 284a13d9dc902..0989c7455361d 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java @@ -171,6 +171,7 @@ public enum Metric { SUBSCRIPTION_CURRENT_COMMIT_ID("subscription_current_commit_id"), SUBSCRIPTION_EVENT_TRANSFER("subscription_event_transfer"), // load related + LOAD_TSFILE("load_tsfile"), LOAD_MEM("load_mem"), LOAD_TIME_COST("load_time_cost"); From c1ddcb0175acebe4558d0a716c4235745ca14b4c Mon Sep 17 00:00:00 2001 From: Itami Sho <42286868+MiniSho@users.noreply.github.com> Date: Mon, 17 Jun 2024 21:33:12 +0800 Subject: [PATCH 04/11] Update LoadTsFileCostMetricsSet Co-authored-by: Steve Yurong Su --- .../iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java index 3ee75cc2b86a3..140102d4fa320 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java @@ -34,8 +34,8 @@ public class LoadTsFileCostMetricsSet implements IMetricSet { private static final LoadTsFileCostMetricsSet INSTANCE = new LoadTsFileCostMetricsSet(); public static final String ANALYSIS = "analysis"; - public static final String SPLIT = "split"; - public static final String WRITE = "write"; + public static final String FIRST_PHASE= "first_phase"; + public static final String SECOND_PHASE = "second_phase"; private LoadTsFileCostMetricsSet() { // empty constructor From a8b2f75f038522ebbde0a5cf88c11747a6fcc4e0 Mon Sep 17 00:00:00 2001 From: Itami Sho <42286868+MiniSho@users.noreply.github.com> Date: Mon, 17 Jun 2024 21:34:29 +0800 Subject: [PATCH 05/11] Update Metric Co-authored-by: Steve Yurong Su --- .../apache/iotdb/commons/service/metric/enums/Metric.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java index 0989c7455361d..7ecc3029b3787 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java @@ -171,9 +171,10 @@ public enum Metric { SUBSCRIPTION_CURRENT_COMMIT_ID("subscription_current_commit_id"), SUBSCRIPTION_EVENT_TRANSFER("subscription_event_transfer"), // load related - LOAD_TSFILE("load_tsfile"), LOAD_MEM("load_mem"), - LOAD_TIME_COST("load_time_cost"); + LOAD_TIME_COST("load_time_cost"), + LOAD_POINT_COUNT("load_point_count"), + ; final String value; From 3a40af102a2837586e4d9820f17bb631abc0c547 Mon Sep 17 00:00:00 2001 From: MiniSho <42286868+MiniSho@users.noreply.github.com> Date: Mon, 17 Jun 2024 22:03:02 +0800 Subject: [PATCH 06/11] remove unused param --- .../db/queryengine/execution/load/LoadTsFileManager.java | 4 ---- 1 file changed, 4 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index 88bec27018ab5..cd868aecb334c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -32,7 +32,6 @@ import org.apache.iotdb.db.exception.DiskSpaceInsufficientException; import org.apache.iotdb.db.exception.LoadFileException; import org.apache.iotdb.db.pipe.agent.PipeAgent; -import org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode; import org.apache.iotdb.db.queryengine.plan.scheduler.load.LoadTsFileScheduler.LoadCommand; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; @@ -86,9 +85,6 @@ public class LoadTsFileManager { new AtomicReference<>(CONFIG.getLoadTsFileDirs()); private static final AtomicReference FOLDER_MANAGER = new AtomicReference<>(); - private static final LoadTsFileCostMetricsSet LOAD_TSFILE_METRICS = - LoadTsFileCostMetricsSet.getInstance(); - private final Map uuid2WriterManager = new ConcurrentHashMap<>(); private final Map uuid2CleanupTask = new ConcurrentHashMap<>(); From d26556a425994f0e86e272ddf39155893027c84f Mon Sep 17 00:00:00 2001 From: MiniSho <42286868+MiniSho@users.noreply.github.com> Date: Mon, 17 Jun 2024 22:03:56 +0800 Subject: [PATCH 07/11] spotless --- .../iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java index 140102d4fa320..ff8d89715f22a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java @@ -34,7 +34,7 @@ public class LoadTsFileCostMetricsSet implements IMetricSet { private static final LoadTsFileCostMetricsSet INSTANCE = new LoadTsFileCostMetricsSet(); public static final String ANALYSIS = "analysis"; - public static final String FIRST_PHASE= "first_phase"; + public static final String FIRST_PHASE = "first_phase"; public static final String SECOND_PHASE = "second_phase"; private LoadTsFileCostMetricsSet() { From 99b3cd336e821e4414f439fa1018ffca810ebbf6 Mon Sep 17 00:00:00 2001 From: MiniSho <42286868+MiniSho@users.noreply.github.com> Date: Tue, 18 Jun 2024 13:38:05 +0800 Subject: [PATCH 08/11] fix --- .../execution/load/LoadTsFileManager.java | 2 +- .../metric/LoadTsFileCostMetricsSet.java | 16 +++++++++++----- .../plan/scheduler/load/LoadTsFileScheduler.java | 8 ++++---- .../iotdb/db/storageengine/StorageEngine.java | 2 +- .../commons/service/metric/enums/Metric.java | 1 + 5 files changed, 18 insertions(+), 11 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index cd868aecb334c..9052b04a2e150 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -434,7 +434,7 @@ private void loadAll(boolean isGeneratedByPipe, ProgressIndex progressIndex) Tag.REGION.toString(), dataRegion.getDataRegionId(), Tag.TYPE.toString(), - Metric.LOAD_TSFILE.toString()); + Metric.LOAD_POINT_COUNT.toString()); }); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java index ff8d89715f22a..1cea3a0b5374d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java @@ -50,10 +50,10 @@ public void recordCost(String stage, long costTimeInNanos) { case ANALYSIS: analyzerTimer.updateNanos(costTimeInNanos); break; - case SPLIT: + case FIRST_PHASE: splitTimer.updateNanos(costTimeInNanos); break; - case WRITE: + case SECOND_PHASE: writeTimer.updateNanos(costTimeInNanos); break; default: @@ -68,15 +68,21 @@ public void bindTo(AbstractMetricService metricService) { Metric.LOAD_TIME_COST.toString(), MetricLevel.IMPORTANT, Tag.NAME.toString(), ANALYSIS); splitTimer = metricService.getOrCreateTimer( - Metric.LOAD_TIME_COST.toString(), MetricLevel.IMPORTANT, Tag.NAME.toString(), SPLIT); + Metric.LOAD_TIME_COST.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + FIRST_PHASE); writeTimer = metricService.getOrCreateTimer( - Metric.LOAD_TIME_COST.toString(), MetricLevel.IMPORTANT, Tag.NAME.toString(), WRITE); + Metric.LOAD_TIME_COST.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + SECOND_PHASE); } @Override public void unbindFrom(AbstractMetricService metricService) { - Arrays.asList(ANALYSIS, SPLIT, WRITE) + Arrays.asList(ANALYSIS, FIRST_PHASE, SECOND_PHASE) .forEach( stage -> metricService.remove( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java index 6639c2d313327..116e1738d142f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java @@ -259,7 +259,7 @@ private boolean firstPhase(LoadSingleTsFileNode node) { tsFileDataManager.clear(); LOAD_TSFILE_COST_METRICS_SET.recordCost( - LoadTsFileCostMetricsSet.SPLIT, System.nanoTime() - startTime); + LoadTsFileCostMetricsSet.FIRST_PHASE, System.nanoTime() - startTime); } return true; } @@ -415,7 +415,7 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { return false; } finally { LOAD_TSFILE_COST_METRICS_SET.recordCost( - LoadTsFileCostMetricsSet.WRITE, System.nanoTime() - startTime); + LoadTsFileCostMetricsSet.SECOND_PHASE, System.nanoTime() - startTime); } // add metrics @@ -446,7 +446,7 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { Tag.REGION.toString(), dataRegion.getDataRegionId(), Tag.TYPE.toString(), - Metric.LOAD_TSFILE.toString()); + Metric.LOAD_POINT_COUNT.toString()); if (!node.isGeneratedByRemoteConsensusLeader()) { MetricService.getInstance() .count( @@ -460,7 +460,7 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { Tag.REGION.toString(), dataRegion.getDataRegionId(), Tag.TYPE.toString(), - Metric.LOAD_TSFILE.toString()); + Metric.LOAD_POINT_COUNT.toString()); } }); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index 11ba7e645ec57..1934ad47664a8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -910,7 +910,7 @@ public TSStatus executeLoadCommand( status.setMessage(e.getMessage()); } finally { LOAD_TSFILE_COST_METRICS_SET.recordCost( - LoadTsFileCostMetricsSet.WRITE, System.nanoTime() - startTime); + LoadTsFileCostMetricsSet.SECOND_PHASE, System.nanoTime() - startTime); } return status; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java index 7ecc3029b3787..6edf36e7a7476 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java @@ -172,6 +172,7 @@ public enum Metric { SUBSCRIPTION_EVENT_TRANSFER("subscription_event_transfer"), // load related LOAD_MEM("load_mem"), + LOAD_DISK_IO("load_disk_io"), LOAD_TIME_COST("load_time_cost"), LOAD_POINT_COUNT("load_point_count"), ; From c884e39b76f2f9c3fa26ec9b366bd26b1e5b3c7d Mon Sep 17 00:00:00 2001 From: MiniSho <42286868+MiniSho@users.noreply.github.com> Date: Tue, 18 Jun 2024 15:48:01 +0800 Subject: [PATCH 09/11] update --- .../LoadTsFileAnalyzeSchemaMemoryBlock.java | 2 +- .../{ => load}/LoadTsFileCostMetricsSet.java | 24 +++++++++++++++++-- .../{ => load}/LoadTsFileMemMetricSet.java | 2 +- .../plan/analyze/AnalyzeVisitor.java | 7 +++--- .../scheduler/load/LoadTsFileScheduler.java | 6 ++--- .../metrics/DataNodeMetricsHelper.java | 4 ++-- .../iotdb/db/storageengine/StorageEngine.java | 6 +++-- .../storageengine/dataregion/DataRegion.java | 4 ++++ 8 files changed, 41 insertions(+), 14 deletions(-) rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/{ => load}/LoadTsFileCostMetricsSet.java (80%) rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/{ => load}/LoadTsFileMemMetricSet.java (98%) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/load/LoadTsFileAnalyzeSchemaMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/load/LoadTsFileAnalyzeSchemaMemoryBlock.java index 17aa4150bcd31..17b52745f236c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/load/LoadTsFileAnalyzeSchemaMemoryBlock.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/load/LoadTsFileAnalyzeSchemaMemoryBlock.java @@ -22,7 +22,7 @@ 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.db.queryengine.metric.LoadTsFileMemMetricSet; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileMemMetricSet; import org.apache.iotdb.metrics.utils.MetricLevel; import org.slf4j.Logger; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileCostMetricsSet.java similarity index 80% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileCostMetricsSet.java index 1cea3a0b5374d..f057eef739a59 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileCostMetricsSet.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileCostMetricsSet.java @@ -17,13 +17,15 @@ * under the License. */ -package org.apache.iotdb.db.queryengine.metric; +package org.apache.iotdb.db.queryengine.metric.load; import org.apache.iotdb.commons.service.metric.enums.Metric; import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.metrics.AbstractMetricService; import org.apache.iotdb.metrics.impl.DoNothingMetricManager; import org.apache.iotdb.metrics.metricsets.IMetricSet; +import org.apache.iotdb.metrics.type.Rate; import org.apache.iotdb.metrics.type.Timer; import org.apache.iotdb.metrics.utils.MetricLevel; import org.apache.iotdb.metrics.utils.MetricType; @@ -44,8 +46,9 @@ private LoadTsFileCostMetricsSet() { private Timer analyzerTimer = DoNothingMetricManager.DO_NOTHING_TIMER; private Timer splitTimer = DoNothingMetricManager.DO_NOTHING_TIMER; private Timer writeTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Rate diskIORate = DoNothingMetricManager.DO_NOTHING_RATE; - public void recordCost(String stage, long costTimeInNanos) { + public void recordPhaseTimeCost(String stage, long costTimeInNanos) { switch (stage) { case ANALYSIS: analyzerTimer.updateNanos(costTimeInNanos); @@ -61,6 +64,10 @@ public void recordCost(String stage, long costTimeInNanos) { } } + public void recordDiskIO(long bytes) { + diskIORate.mark(bytes); + } + @Override public void bindTo(AbstractMetricService metricService) { analyzerTimer = @@ -78,6 +85,13 @@ public void bindTo(AbstractMetricService metricService) { MetricLevel.IMPORTANT, Tag.NAME.toString(), SECOND_PHASE); + + diskIORate = + metricService.getOrCreateRate( + Metric.LOAD_DISK_IO.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + "DataNode " + IoTDBDescriptor.getInstance().getConfig().getDataNodeId()); } @Override @@ -90,6 +104,12 @@ public void unbindFrom(AbstractMetricService metricService) { Metric.LOAD_TIME_COST.toString(), Tag.NAME.toString(), stage)); + + metricService.remove( + MetricType.RATE, + Metric.LOAD_DISK_IO.toString(), + Tag.NAME.toString(), + String.valueOf(IoTDBDescriptor.getInstance().getConfig().getDataNodeId())); } public static LoadTsFileCostMetricsSet getInstance() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileMemMetricSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileMemMetricSet.java similarity index 98% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileMemMetricSet.java rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileMemMetricSet.java index 7e822421302a7..c195381585490 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/LoadTsFileMemMetricSet.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileMemMetricSet.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.db.queryengine.metric; +package org.apache.iotdb.db.queryengine.metric.load; import org.apache.iotdb.commons.service.metric.enums.Metric; import org.apache.iotdb.commons.service.metric.enums.Tag; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java index b3af926b67650..0adf30a9b4c81 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java @@ -58,8 +58,8 @@ import org.apache.iotdb.db.queryengine.common.schematree.IMeasurementSchemaInfo; import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree; import org.apache.iotdb.db.queryengine.execution.operator.window.WindowType; -import org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager; import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType; import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher; @@ -183,9 +183,9 @@ import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_MATCH_PATTERN; import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.DEVICE; import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.ENDTIME; -import static org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet.ANALYSIS; import static org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet.PARTITION_FETCHER; import static org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet.SCHEMA_FETCHER; +import static org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet.ANALYSIS; import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.bindSchemaForExpression; import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.concatDeviceAndBindSchemaForExpression; import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer.getMeasurementExpression; @@ -2862,7 +2862,8 @@ public Analysis visitLoadFile(LoadTsFileStatement loadTsFileStatement, MPPQueryC analysis.setFailStatus(RpcUtils.getStatus(TSStatusCode.LOAD_FILE_ERROR, exceptionMessage)); return analysis; } finally { - LoadTsFileCostMetricsSet.getInstance().recordCost(ANALYSIS, System.nanoTime() - startTime); + LoadTsFileCostMetricsSet.getInstance() + .recordPhaseTimeCost(ANALYSIS, System.nanoTime() - startTime); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java index 116e1738d142f..69a36345909d1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java @@ -50,7 +50,7 @@ import org.apache.iotdb.db.queryengine.execution.load.TsFileSplitter; import org.apache.iotdb.db.queryengine.load.LoadTsFileDataCacheMemoryBlock; import org.apache.iotdb.db.queryengine.load.LoadTsFileMemoryManager; -import org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher; import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan; import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance; @@ -258,7 +258,7 @@ private boolean firstPhase(LoadSingleTsFileNode node) { } finally { tsFileDataManager.clear(); - LOAD_TSFILE_COST_METRICS_SET.recordCost( + LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( LoadTsFileCostMetricsSet.FIRST_PHASE, System.nanoTime() - startTime); } return true; @@ -414,7 +414,7 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { stateMachine.transitionToFailed(e.getFailureStatus()); return false; } finally { - LOAD_TSFILE_COST_METRICS_SET.recordCost( + LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( LoadTsFileCostMetricsSet.SECOND_PHASE, System.nanoTime() - startTime); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java index e1e14939567a7..e5e3f27f5c14e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java @@ -34,13 +34,13 @@ import org.apache.iotdb.db.queryengine.metric.DataExchangeCostMetricSet; import org.apache.iotdb.db.queryengine.metric.DataExchangeCountMetricSet; import org.apache.iotdb.db.queryengine.metric.DriverSchedulerMetricSet; -import org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet; -import org.apache.iotdb.db.queryengine.metric.LoadTsFileMemMetricSet; import org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet; import org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet; import org.apache.iotdb.db.queryengine.metric.QueryRelatedResourceMetricSet; import org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet; import org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileMemMetricSet; import org.apache.iotdb.db.subscription.metric.SubscriptionMetrics; import org.apache.iotdb.metrics.metricsets.UpTimeMetrics; import org.apache.iotdb.metrics.metricsets.disk.DiskMetrics; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index 1934ad47664a8..595e84973d668 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -54,7 +54,7 @@ import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException; import org.apache.iotdb.db.queryengine.execution.load.LoadTsFileManager; import org.apache.iotdb.db.queryengine.execution.load.LoadTsFileRateLimiter; -import org.apache.iotdb.db.queryengine.metric.LoadTsFileCostMetricsSet; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode; @@ -866,6 +866,8 @@ public TSStatus writeLoadTsFileNode( return status; } + LoadTsFileCostMetricsSet.getInstance().recordDiskIO(pieceNode.getDataSize()); + return RpcUtils.SUCCESS_STATUS; } @@ -909,7 +911,7 @@ public TSStatus executeLoadCommand( status.setCode(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()); status.setMessage(e.getMessage()); } finally { - LOAD_TSFILE_COST_METRICS_SET.recordCost( + LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( LoadTsFileCostMetricsSet.SECOND_PHASE, System.nanoTime() - startTime); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 5a58b3e1dbfce..168d959abf5ca 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -49,6 +49,7 @@ import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; import org.apache.iotdb.db.queryengine.execution.load.LoadTsFileRateLimiter; import org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache; import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; @@ -3010,6 +3011,9 @@ private boolean loadTsFileToUnSequence( tsFileManager.add(tsFileResource, false); + // metrics + LoadTsFileCostMetricsSet.getInstance().recordDiskIO(tsFileResource.getTsFile().length()); + return true; } From 327deb9a50f481b5a8bffb04efb18beb6add788f Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Thu, 20 Jun 2024 15:12:40 +0800 Subject: [PATCH 10/11] refactor --- .../execution/load/LoadTsFileRateLimiter.java | 3 ++ .../metric/load/LoadTsFileCostMetricsSet.java | 27 +++++++++--- .../scheduler/load/LoadTsFileScheduler.java | 43 +++++++++++++------ .../iotdb/db/storageengine/StorageEngine.java | 6 --- .../storageengine/dataregion/DataRegion.java | 4 -- 5 files changed, 52 insertions(+), 31 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileRateLimiter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileRateLimiter.java index ce7046bf5ee54..6d04e43eea3b1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileRateLimiter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileRateLimiter.java @@ -21,6 +21,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; import com.google.common.util.concurrent.AtomicDouble; import com.google.common.util.concurrent.RateLimiter; @@ -34,6 +35,8 @@ public class LoadTsFileRateLimiter { private final RateLimiter loadWriteRateLimiter; public void acquire(long bytes) { + LoadTsFileCostMetricsSet.getInstance().recordDiskIO(bytes); + final double throughputBytesPerSecondLimit = CONFIG.getLoadWriteThroughputBytesPerSecond(); if (throughputBytesPerSecond.get() != throughputBytesPerSecondLimit) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileCostMetricsSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileCostMetricsSet.java index f057eef739a59..b0ce2ca69a45d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileCostMetricsSet.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/load/LoadTsFileCostMetricsSet.java @@ -33,19 +33,23 @@ import java.util.Arrays; public class LoadTsFileCostMetricsSet implements IMetricSet { + private static final LoadTsFileCostMetricsSet INSTANCE = new LoadTsFileCostMetricsSet(); public static final String ANALYSIS = "analysis"; public static final String FIRST_PHASE = "first_phase"; public static final String SECOND_PHASE = "second_phase"; + public static final String LOAD_LOCALLY = "load_locally"; private LoadTsFileCostMetricsSet() { // empty constructor } private Timer analyzerTimer = DoNothingMetricManager.DO_NOTHING_TIMER; - private Timer splitTimer = DoNothingMetricManager.DO_NOTHING_TIMER; - private Timer writeTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer firstPhaseTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer secondPhaseTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer loadLocallyTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Rate diskIORate = DoNothingMetricManager.DO_NOTHING_RATE; public void recordPhaseTimeCost(String stage, long costTimeInNanos) { @@ -54,10 +58,13 @@ public void recordPhaseTimeCost(String stage, long costTimeInNanos) { analyzerTimer.updateNanos(costTimeInNanos); break; case FIRST_PHASE: - splitTimer.updateNanos(costTimeInNanos); + firstPhaseTimer.updateNanos(costTimeInNanos); break; case SECOND_PHASE: - writeTimer.updateNanos(costTimeInNanos); + secondPhaseTimer.updateNanos(costTimeInNanos); + break; + case LOAD_LOCALLY: + loadLocallyTimer.updateNanos(costTimeInNanos); break; default: throw new UnsupportedOperationException("Unsupported stage: " + stage); @@ -73,18 +80,24 @@ public void bindTo(AbstractMetricService metricService) { analyzerTimer = metricService.getOrCreateTimer( Metric.LOAD_TIME_COST.toString(), MetricLevel.IMPORTANT, Tag.NAME.toString(), ANALYSIS); - splitTimer = + firstPhaseTimer = metricService.getOrCreateTimer( Metric.LOAD_TIME_COST.toString(), MetricLevel.IMPORTANT, Tag.NAME.toString(), FIRST_PHASE); - writeTimer = + secondPhaseTimer = metricService.getOrCreateTimer( Metric.LOAD_TIME_COST.toString(), MetricLevel.IMPORTANT, Tag.NAME.toString(), SECOND_PHASE); + loadLocallyTimer = + metricService.getOrCreateTimer( + Metric.LOAD_TIME_COST.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + LOAD_LOCALLY); diskIORate = metricService.getOrCreateRate( @@ -96,7 +109,7 @@ public void bindTo(AbstractMetricService metricService) { @Override public void unbindFrom(AbstractMetricService metricService) { - Arrays.asList(ANALYSIS, FIRST_PHASE, SECOND_PHASE) + Arrays.asList(ANALYSIS, FIRST_PHASE, SECOND_PHASE, LOAD_LOCALLY) .forEach( stage -> metricService.remove( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java index 69a36345909d1..0c8ff86172c7d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java @@ -109,12 +109,13 @@ public class LoadTsFileScheduler implements IScheduler { private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); + private static final LoadTsFileCostMetricsSet LOAD_TSFILE_COST_METRICS_SET = + LoadTsFileCostMetricsSet.getInstance(); + private static final long SINGLE_SCHEDULER_MAX_MEMORY_SIZE = IoTDBDescriptor.getInstance().getConfig().getThriftMaxFrameSize() >> 2; private static final int TRANSMIT_LIMIT = CommonDescriptor.getInstance().getConfig().getTTimePartitionSlotTransmitLimit(); - private static final LoadTsFileCostMetricsSet LOAD_TSFILE_COST_METRICS_SET = - LoadTsFileCostMetricsSet.getInstance(); private static final Set LOADING_FILE_SET = new HashSet<>(); @@ -180,16 +181,38 @@ public void start() { partitionFetcher.queryDataPartition( slotList, queryContext.getSession().getUserName()))) { // do not decode, load locally - isLoadSingleTsFileSuccess = loadLocally(node); + final long startTime = System.nanoTime(); + try { + isLoadSingleTsFileSuccess = loadLocally(node); + } finally { + LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( + LoadTsFileCostMetricsSet.LOAD_LOCALLY, System.nanoTime() - startTime); + } + node.clean(); } else { // need decode, load locally or remotely, use two phases method String uuid = UUID.randomUUID().toString(); dispatcher.setUuid(uuid); allReplicaSets.clear(); - boolean isFirstPhaseSuccess = firstPhase(node); - boolean isSecondPhaseSuccess = - secondPhase(isFirstPhaseSuccess, uuid, node.getTsFileResource()); + long startTime = System.nanoTime(); + final boolean isFirstPhaseSuccess; + try { + isFirstPhaseSuccess = firstPhase(node); + } finally { + LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( + LoadTsFileCostMetricsSet.FIRST_PHASE, System.nanoTime() - startTime); + } + + startTime = System.nanoTime(); + final boolean isSecondPhaseSuccess; + try { + isSecondPhaseSuccess = + secondPhase(isFirstPhaseSuccess, uuid, node.getTsFileResource()); + } finally { + LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( + LoadTsFileCostMetricsSet.SECOND_PHASE, System.nanoTime() - startTime); + } node.clean(); if (!isFirstPhaseSuccess || !isSecondPhaseSuccess) { @@ -233,7 +256,6 @@ public void start() { private boolean firstPhase(LoadSingleTsFileNode node) { final TsFileDataManager tsFileDataManager = new TsFileDataManager(this, node, block); - long startTime = System.nanoTime(); try { new TsFileSplitter( node.getTsFileResource().getTsFile(), tsFileDataManager::addOrSendTsFileData) @@ -257,9 +279,6 @@ private boolean firstPhase(LoadSingleTsFileNode node) { return false; } finally { tsFileDataManager.clear(); - - LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( - LoadTsFileCostMetricsSet.FIRST_PHASE, System.nanoTime() - startTime); } return true; } @@ -391,7 +410,6 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { throw new LoadReadOnlyException(); } - long startTime = System.nanoTime(); try { FragmentInstance instance = new FragmentInstance( @@ -413,9 +431,6 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { e.getFailureStatus().getMessage())); stateMachine.transitionToFailed(e.getFailureStatus()); return false; - } finally { - LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( - LoadTsFileCostMetricsSet.SECOND_PHASE, System.nanoTime() - startTime); } // add metrics diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index 595e84973d668..14feefb95f5ad 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -866,8 +866,6 @@ public TSStatus writeLoadTsFileNode( return status; } - LoadTsFileCostMetricsSet.getInstance().recordDiskIO(pieceNode.getDataSize()); - return RpcUtils.SUCCESS_STATUS; } @@ -877,7 +875,6 @@ public TSStatus executeLoadCommand( boolean isGeneratedByPipe, ProgressIndex progressIndex) { TSStatus status = new TSStatus(); - long startTime = System.nanoTime(); try { switch (loadCommand) { case EXECUTE: @@ -910,9 +907,6 @@ public TSStatus executeLoadCommand( LOGGER.error("Execute load command {} error.", loadCommand, e); status.setCode(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()); status.setMessage(e.getMessage()); - } finally { - LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( - LoadTsFileCostMetricsSet.SECOND_PHASE, System.nanoTime() - startTime); } return status; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 168d959abf5ca..5a58b3e1dbfce 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -49,7 +49,6 @@ import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; import org.apache.iotdb.db.queryengine.execution.load.LoadTsFileRateLimiter; import org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet; -import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache; import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; @@ -3011,9 +3010,6 @@ private boolean loadTsFileToUnSequence( tsFileManager.add(tsFileResource, false); - // metrics - LoadTsFileCostMetricsSet.getInstance().recordDiskIO(tsFileResource.getTsFile().length()); - return true; } From 9318bcec44bd8c7ccd229c12212cf026d4e3e2b2 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Thu, 20 Jun 2024 15:13:42 +0800 Subject: [PATCH 11/11] Update StorageEngine.java --- .../org/apache/iotdb/db/storageengine/StorageEngine.java | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index 14feefb95f5ad..d26304e113cce 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -54,7 +54,6 @@ import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException; import org.apache.iotdb.db.queryengine.execution.load.LoadTsFileManager; import org.apache.iotdb.db.queryengine.execution.load.LoadTsFileRateLimiter; -import org.apache.iotdb.db.queryengine.metric.load.LoadTsFileCostMetricsSet; import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode; @@ -117,9 +116,6 @@ public class StorageEngine implements IService { private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); private static final WritingMetrics WRITING_METRICS = WritingMetrics.getInstance(); - private static final LoadTsFileCostMetricsSet LOAD_TSFILE_COST_METRICS_SET = - LoadTsFileCostMetricsSet.getInstance(); - /** * a folder (system/databases/ by default) that persist system info. Each database will have a * subfolder under the systemDir. @@ -875,6 +871,7 @@ public TSStatus executeLoadCommand( boolean isGeneratedByPipe, ProgressIndex progressIndex) { TSStatus status = new TSStatus(); + try { switch (loadCommand) { case EXECUTE: