You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2022/11/07 06:47:21 UTC
[iotdb] branch QueryMetrics updated (37e60f727d -> 12ee362527)
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a change to branch QueryMetrics
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from 37e60f727d fix typo in logback-datanode.xml
new d15fbff6c7 Add statistics for query
new 1df99c2ec2 resolve conflicts
add f58cc6c223 Add a function quantile in UDF library (#7912)
add abbb45b891 Adjust configuration files (#7909)
add 96c860f876 [IOTDB-4807] Check nodes' status at the INVALIDATE_CACHE state (#7895)
add 95fa01f4fe Update client-go commit to fix CI
add 523e82a437 Fixed the bugs of shell (#7910)
add 24d99ef5d3 [IOTDB-4829] Let NoMoreTsBlockEvent RPC is called in async way (#7911)
add 1b1c57b6d8 [IOTDB-4857] Fix the problem when remove-datanode for ratis 1 replica (#7917)
new 12ee362527 Merge remote-tracking branch 'origin/master' into QueryMetrics
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
client-go | 2 +-
.../src/assembly/resources/conf/confignode-env.bat | 22 +-
.../resources/conf/iotdb-confignode.properties | 41 +-
.../assembly/resources/sbin/start-confignode.bat | 24 +-
.../assembly/resources/sbin/start-confignode.sh | 9 +-
.../iotdb/confignode/conf/ConfigNodeConfig.java | 167 ++-
.../confignode/conf/ConfigNodeDescriptor.java | 91 +-
.../confignode/conf/ConfigNodeStartupCheck.java | 6 +-
.../statemachine/PartitionRegionStateMachine.java | 3 +-
.../iotdb/confignode/manager/ConsensusManager.java | 25 +-
.../iotdb/confignode/manager/ProcedureManager.java | 4 +-
.../iotdb/confignode/manager/UDFManager.java | 4 +-
.../iotdb/confignode/manager/node/NodeManager.java | 5 +-
.../confignode/persistence/node/NodeInfo.java | 5 +-
.../persistence/partition/PartitionInfo.java | 9 +-
.../partition/StorageGroupPartitionTable.java | 19 +-
.../procedure/env/ConfigNodeProcedureEnv.java | 60 +-
.../procedure/env/DataNodeRemoveHandler.java | 73 +-
.../impl/node/RemoveDataNodeProcedure.java | 9 +-
.../impl/statemachine/RegionMigrateProcedure.java | 11 +-
.../procedure/store/ConfigProcedureStore.java | 2 +-
.../confignode1conf/iotdb-common.properties | 2 +-
.../confignode1conf/iotdb-confignode.properties | 2 +-
.../confignode2conf/iotdb-common.properties | 2 +-
.../confignode2conf/iotdb-confignode.properties | 2 +-
.../confignode3conf/iotdb-common.properties | 2 +-
.../confignode3conf/iotdb-confignode.properties | 2 +-
docker/src/main/Dockerfile-1c1d | 3 +
docker/src/main/Dockerfile-1c1d-influxdb | 5 +-
docs/UserGuide/Process-Data/Triggers.md | 2 +-
docs/UserGuide/Reference/Syntax-Conventions.md | 38 +-
docs/UserGuide/UDF-Library/Data-Profiling.md | 73 ++
docs/zh/UserGuide/Process-Data/Triggers.md | 2 +-
docs/zh/UserGuide/Reference/Syntax-Conventions.md | 40 +-
docs/zh/UserGuide/UDF-Library/Data-Profiling.md | 74 ++
integration-test/README.md | 2 +-
.../org/apache/iotdb/it/env/ConfigNodeWrapper.java | 2 +-
.../org/apache/iotdb/it/env/DataNodeWrapper.java | 2 +-
.../java/org/apache/iotdb/it/env/MppConfig.java | 2 +-
library-udf/src/assembly/tools/register-UDF.bat | 1 +
library-udf/src/assembly/tools/register-UDF.sh | 1 +
.../iotdb/library/dprofile/UDAFQuantile.java | 127 +++
.../library/dprofile/util/HeapLongKLLSketch.java | 356 +++++++
.../dprofile/util/KLLSketchForQuantile.java | 172 +++
.../resources/conf/iotdb-common.properties | 1126 +++++++++-----------
.../apache/iotdb/commons/conf/CommonConfig.java | 2 +-
.../apache/iotdb/commons/conf/IoTDBConstant.java | 4 +-
.../src/assembly/resources/conf/datanode-env.bat | 20 -
.../resources/conf/iotdb-datanode.properties | 47 +-
.../src/assembly/resources/sbin/start-datanode.bat | 16 -
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 6 +-
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 2 +-
.../db/mpp/execution/exchange/SinkHandle.java | 84 +-
.../fragment/FragmentInstanceContext.java | 10 +-
.../fragment/FragmentInstanceManager.java | 7 +
.../db/mpp/execution/operator/OperatorContext.java | 11 +-
.../operator/process/FilterAndProjectOperator.java | 34 +-
.../process/RawDataAggregationOperator.java | 11 +-
.../process/SingleInputAggregationOperator.java | 23 +-
.../process/SlidingWindowAggregationOperator.java | 13 +-
.../AbstractSeriesAggregationScanOperator.java | 42 +-
.../operator/source/AlignedSeriesScanOperator.java | 6 +-
.../operator/source/SeriesScanOperator.java | 6 +-
.../execution/operator/source/SeriesScanUtil.java | 3 +
.../iotdb/db/mpp/statistics/QueryStatistics.java | 15 +
.../java/org/apache/iotdb/db/service/DataNode.java | 2 +-
.../iotdb/db/service/RegionMigrateService.java | 31 +-
.../impl/DataNodeInternalRPCServiceImpl.java | 4 +-
.../apache/iotdb/db/utils/QueryDataSetUtils.java | 48 +-
.../db/mpp/execution/exchange/SinkHandleTest.java | 28 +-
.../datanode1conf/iotdb-datanode.properties | 2 +-
.../datanode2conf/iotdb-datanode.properties | 2 +-
.../datanode3conf/iotdb-datanode.properties | 2 +-
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 1 +
74 files changed, 1967 insertions(+), 1146 deletions(-)
create mode 100644 library-udf/src/main/java/org/apache/iotdb/library/dprofile/UDAFQuantile.java
create mode 100644 library-udf/src/main/java/org/apache/iotdb/library/dprofile/util/HeapLongKLLSketch.java
create mode 100644 library-udf/src/main/java/org/apache/iotdb/library/dprofile/util/KLLSketchForQuantile.java
[iotdb] 01/03: Add statistics for query
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch QueryMetrics
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d15fbff6c7475f36019a5c9dfa37a1eb3b314590
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Mon Nov 7 14:45:17 2022 +0800
Add statistics for query
---
.../fragment/FragmentInstanceContext.java | 10 ++++-
.../fragment/FragmentInstanceManager.java | 7 ++++
.../db/mpp/execution/operator/OperatorContext.java | 11 ++++-
.../operator/process/FilterAndProjectOperator.java | 34 +++++++++------
.../process/RawDataAggregationOperator.java | 11 ++++-
.../process/SingleInputAggregationOperator.java | 23 ++++++++---
.../process/SlidingWindowAggregationOperator.java | 13 +++++-
.../AbstractSeriesAggregationScanOperator.java | 42 ++++++++++---------
.../operator/source/AlignedSeriesScanOperator.java | 6 ++-
.../operator/source/SeriesScanOperator.java | 6 ++-
.../execution/operator/source/SeriesScanUtil.java | 3 ++
.../iotdb/db/mpp/statistics/QueryStatistics.java | 16 ++++++++
.../apache/iotdb/db/utils/QueryDataSetUtils.java | 48 +++++++++++++---------
13 files changed, 167 insertions(+), 63 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceContext.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceContext.java
index 286623529f..6ecb01df2f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceContext.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceContext.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.db.mpp.common.SessionInfo;
import org.apache.iotdb.db.mpp.execution.driver.DriverContext;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.statistics.QueryStatistics;
import org.apache.iotdb.db.query.context.QueryContext;
import org.slf4j.Logger;
@@ -42,6 +43,9 @@ public class FragmentInstanceContext extends QueryContext {
private static final Logger LOGGER = LoggerFactory.getLogger(FragmentInstanceContext.class);
private static final long END_TIME_INITIAL_VALUE = -1L;
+
+ private static final QueryStatistics QUERY_STATISTICS = QueryStatistics.getInstance();
+
private final FragmentInstanceId id;
// TODO if we split one fragment instance into multiple pipelines to run, we need to replace it
@@ -157,7 +161,7 @@ public class FragmentInstanceContext extends QueryContext {
}
OperatorContext operatorContext =
- new OperatorContext(operatorId, planNodeId, operatorType, this);
+ new OperatorContext(operatorId, planNodeId, operatorType, this, QUERY_STATISTICS);
operatorContexts.add(operatorContext);
return operatorContext;
}
@@ -223,4 +227,8 @@ public class FragmentInstanceContext extends QueryContext {
public SessionInfo getSessionInfo() {
return sessionInfo;
}
+
+ public void addOperationTime(String key, long costTimeInNanos) {
+ QUERY_STATISTICS.addCost(key, costTimeInNanos);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceManager.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceManager.java
index 1d7499560f..318476e43d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceManager.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.db.mpp.execution.schedule.DriverScheduler;
import org.apache.iotdb.db.mpp.execution.schedule.IDriverScheduler;
import org.apache.iotdb.db.mpp.plan.planner.LocalExecutionPlanner;
import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
+import org.apache.iotdb.db.mpp.statistics.QueryStatistics;
import org.apache.iotdb.db.utils.SetThreadName;
import io.airlift.stats.CounterStat;
@@ -47,6 +48,7 @@ import java.util.concurrent.TimeoutException;
import static java.util.Objects.requireNonNull;
import static org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
import static org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceExecution.createFragmentInstanceExecution;
+import static org.apache.iotdb.db.mpp.statistics.QueryStatistics.LOCAL_EXECUTION_PLANNER;
public class FragmentInstanceManager {
@@ -68,6 +70,8 @@ public class FragmentInstanceManager {
private static final long QUERY_TIMEOUT_MS =
IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold();
+ private static final QueryStatistics QUERY_STATISTICS = QueryStatistics.getInstance();
+
public static FragmentInstanceManager getInstance() {
return FragmentInstanceManager.InstanceHolder.INSTANCE;
}
@@ -96,6 +100,7 @@ public class FragmentInstanceManager {
FragmentInstance instance, DataRegion dataRegion) {
FragmentInstanceId instanceId = instance.getId();
+ long startTime = System.nanoTime();
try (SetThreadName fragmentInstanceName = new SetThreadName(instanceId.getFullId())) {
FragmentInstanceExecution execution =
instanceExecution.computeIfAbsent(
@@ -135,6 +140,8 @@ public class FragmentInstanceManager {
});
return execution != null ? execution.getInstanceInfo() : createFailedInstanceInfo(instanceId);
+ } finally {
+ QUERY_STATISTICS.addCost(LOCAL_EXECUTION_PLANNER, System.nanoTime() - startTime);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/OperatorContext.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/OperatorContext.java
index 447066f4c8..bc4a718422 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/OperatorContext.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/OperatorContext.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.mpp.execution.operator;
import org.apache.iotdb.db.mpp.common.SessionInfo;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.statistics.QueryStatistics;
import io.airlift.units.Duration;
@@ -38,17 +39,21 @@ public class OperatorContext {
private final String operatorType;
private final FragmentInstanceContext instanceContext;
+ private final QueryStatistics queryStatistics;
+
private Duration maxRunTime;
public OperatorContext(
int operatorId,
PlanNodeId planNodeId,
String operatorType,
- FragmentInstanceContext instanceContext) {
+ FragmentInstanceContext instanceContext,
+ QueryStatistics queryStatistics) {
this.operatorId = operatorId;
this.planNodeId = planNodeId;
this.operatorType = operatorType;
this.instanceContext = instanceContext;
+ this.queryStatistics = queryStatistics;
}
public int getOperatorId() {
@@ -75,6 +80,10 @@ public class OperatorContext {
return instanceContext.getSessionInfo();
}
+ public void addOperatorTime(String key, long costTimeInNanos) {
+ queryStatistics.addCost(key, costTimeInNanos);
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/FilterAndProjectOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/FilterAndProjectOperator.java
index eed0895a47..4c1ba15a47 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/FilterAndProjectOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/FilterAndProjectOperator.java
@@ -43,19 +43,21 @@ import com.google.common.util.concurrent.ListenableFuture;
import java.util.ArrayList;
import java.util.List;
+import static org.apache.iotdb.db.mpp.statistics.QueryStatistics.FILTER_AND_PROJECT_OPERATOR;
+
public class FilterAndProjectOperator implements ProcessOperator {
private final Operator inputOperator;
- private List<LeafColumnTransformer> filterLeafColumnTransformerList;
+ private final List<LeafColumnTransformer> filterLeafColumnTransformerList;
- private ColumnTransformer filterOutputTransformer;
+ private final ColumnTransformer filterOutputTransformer;
- private List<ColumnTransformer> commonTransformerList;
+ private final List<ColumnTransformer> commonTransformerList;
- private List<LeafColumnTransformer> projectLeafColumnTransformerList;
+ private final List<LeafColumnTransformer> projectLeafColumnTransformerList;
- private List<ColumnTransformer> projectOutputTransformerList;
+ private final List<ColumnTransformer> projectOutputTransformerList;
private final TsBlockBuilder filterTsBlockBuilder;
@@ -101,17 +103,23 @@ public class FilterAndProjectOperator implements ProcessOperator {
return null;
}
- if (!hasFilter) {
- return getTransformedTsBlock(input);
- }
+ long startTime = System.nanoTime();
- TsBlock filterResult = getFilterTsBlock(input);
+ try {
+ if (!hasFilter) {
+ return getTransformedTsBlock(input);
+ }
+
+ TsBlock filterResult = getFilterTsBlock(input);
- // contains non-mappable udf, we leave calculation for TransformOperator
- if (hasNonMappableUDF) {
- return filterResult;
+ // contains non-mappable udf, we leave calculation for TransformOperator
+ if (hasNonMappableUDF) {
+ return filterResult;
+ }
+ return getTransformedTsBlock(filterResult);
+ } finally {
+ operatorContext.addOperatorTime(FILTER_AND_PROJECT_OPERATOR, System.nanoTime() - startTime);
}
- return getTransformedTsBlock(filterResult);
}
/**
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/RawDataAggregationOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/RawDataAggregationOperator.java
index 30488c18d3..bce16f89d5 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/RawDataAggregationOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/RawDataAggregationOperator.java
@@ -68,7 +68,11 @@ public class RawDataAggregationOperator extends SingleInputAggregationOperator {
@Override
protected boolean calculateNextAggregationResult() {
+ long startTime = System.nanoTime();
+
while (!calculateFromRawData()) {
+ long endTime = System.nanoTime();
+ costTime += (endTime - startTime);
inputTsBlock = null;
// NOTE: child.next() can only be invoked once
@@ -79,19 +83,24 @@ public class RawDataAggregationOperator extends SingleInputAggregationOperator {
// if child still has next but can't be invoked now
return false;
} else {
+ startTime = System.nanoTime();
// If there are no points belong to last window, the last window will not
// initialize window and aggregators
if (!windowManager.isCurWindowInit()) {
initWindowAndAggregators();
}
+ endTime = System.nanoTime();
+ costTime += (endTime - startTime);
+ startTime = endTime;
break;
}
+ startTime = System.nanoTime();
}
updateResultTsBlock();
// Step into next window
windowManager.next();
-
+ costTime += (System.nanoTime() - startTime);
return true;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SingleInputAggregationOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SingleInputAggregationOperator.java
index 16071aea1e..894673d577 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SingleInputAggregationOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SingleInputAggregationOperator.java
@@ -33,6 +33,8 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
+import static org.apache.iotdb.db.mpp.statistics.QueryStatistics.SINGLE_INPUT_AGG_OPERATOR;
+
public abstract class SingleInputAggregationOperator implements ProcessOperator {
protected final OperatorContext operatorContext;
@@ -50,6 +52,8 @@ public abstract class SingleInputAggregationOperator implements ProcessOperator
protected final long maxRetainedSize;
protected final long maxReturnSize;
+ protected long costTime = 0L;
+
public SingleInputAggregationOperator(
OperatorContext operatorContext,
List<Aggregator> aggregators,
@@ -97,12 +101,19 @@ public abstract class SingleInputAggregationOperator implements ProcessOperator
}
}
- if (resultTsBlockBuilder.getPositionCount() > 0) {
- TsBlock resultTsBlock = resultTsBlockBuilder.build();
- resultTsBlockBuilder.reset();
- return resultTsBlock;
- } else {
- return null;
+ start = System.nanoTime();
+ try {
+ if (resultTsBlockBuilder.getPositionCount() > 0) {
+ TsBlock resultTsBlock = resultTsBlockBuilder.build();
+ resultTsBlockBuilder.reset();
+ return resultTsBlock;
+ } else {
+ return null;
+ }
+ } finally {
+ operatorContext.addOperatorTime(
+ SINGLE_INPUT_AGG_OPERATOR, costTime + (System.nanoTime() - start));
+ costTime = 0;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SlidingWindowAggregationOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SlidingWindowAggregationOperator.java
index 76499849cb..8a8ec2c813 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SlidingWindowAggregationOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SlidingWindowAggregationOperator.java
@@ -65,6 +65,7 @@ public class SlidingWindowAggregationOperator extends SingleInputAggregationOper
@Override
protected boolean calculateNextAggregationResult() {
+ long startTime = System.nanoTime();
if (curTimeRange == null && timeRangeIterator.hasNextTimeRange()) {
// move to next time window
curTimeRange = timeRangeIterator.nextTimeRange();
@@ -75,7 +76,12 @@ public class SlidingWindowAggregationOperator extends SingleInputAggregationOper
}
}
+ long endTime = System.nanoTime();
+ costTime += (endTime - startTime);
+ startTime = endTime;
+
while (!isCalculationDone()) {
+ costTime += (System.nanoTime() - startTime);
if (inputTsBlock == null) {
// NOTE: child.next() can only be invoked once
if (child.hasNext() && canCallNext) {
@@ -85,15 +91,20 @@ public class SlidingWindowAggregationOperator extends SingleInputAggregationOper
// if child still has next but can't be invoked now
return false;
} else {
+ startTime = System.nanoTime();
break;
}
}
-
+ startTime = System.nanoTime();
calculateFromCachedData();
+ endTime = System.nanoTime();
+ costTime += (endTime - startTime);
+ startTime = endTime;
}
// update result using aggregators
updateResultTsBlock();
+ costTime += (System.nanoTime() - startTime);
return true;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AbstractSeriesAggregationScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AbstractSeriesAggregationScanOperator.java
index 15ad856245..c565842d66 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AbstractSeriesAggregationScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AbstractSeriesAggregationScanOperator.java
@@ -42,6 +42,7 @@ import java.util.concurrent.TimeUnit;
import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.appendAggregationResult;
import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.calculateAggregationFromRawData;
import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.isAllAggregatorsHasFinalResult;
+import static org.apache.iotdb.db.mpp.statistics.QueryStatistics.AGG_SCAN_OPERATOR;
public abstract class AbstractSeriesAggregationScanOperator implements DataSourceOperator {
@@ -141,28 +142,31 @@ public abstract class AbstractSeriesAggregationScanOperator implements DataSourc
// start stopwatch
long maxRuntime = operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
long start = System.nanoTime();
+ try {
+ while (System.nanoTime() - start < maxRuntime
+ && timeRangeIterator.hasNextTimeRange()
+ && !resultTsBlockBuilder.isFull()) {
+ // move to next time window
+ curTimeRange = timeRangeIterator.nextTimeRange();
+
+ // clear previous aggregation result
+ for (Aggregator aggregator : aggregators) {
+ aggregator.updateTimeRange(curTimeRange);
+ }
- while (System.nanoTime() - start < maxRuntime
- && timeRangeIterator.hasNextTimeRange()
- && !resultTsBlockBuilder.isFull()) {
- // move to next time window
- curTimeRange = timeRangeIterator.nextTimeRange();
-
- // clear previous aggregation result
- for (Aggregator aggregator : aggregators) {
- aggregator.updateTimeRange(curTimeRange);
+ // calculate aggregation result on current time window
+ calculateNextAggregationResult();
}
- // calculate aggregation result on current time window
- calculateNextAggregationResult();
- }
-
- if (resultTsBlockBuilder.getPositionCount() > 0) {
- TsBlock resultTsBlock = resultTsBlockBuilder.build();
- resultTsBlockBuilder.reset();
- return resultTsBlock;
- } else {
- return null;
+ if (resultTsBlockBuilder.getPositionCount() > 0) {
+ TsBlock resultTsBlock = resultTsBlockBuilder.build();
+ resultTsBlockBuilder.reset();
+ return resultTsBlock;
+ } else {
+ return null;
+ }
+ } finally {
+ operatorContext.addOperatorTime(AGG_SCAN_OPERATOR, System.nanoTime() - start);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanOperator.java
index 8406437802..e22d7b1f65 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanOperator.java
@@ -29,6 +29,8 @@ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.io.IOException;
import java.util.HashSet;
+import static org.apache.iotdb.db.mpp.statistics.QueryStatistics.ALIGNED_SERIES_SCAN_OPERATOR;
+
public class AlignedSeriesScanOperator implements DataSourceOperator {
private final OperatorContext operatorContext;
@@ -81,7 +83,7 @@ public class AlignedSeriesScanOperator implements DataSourceOperator {
@Override
public boolean hasNext() {
-
+ long startTime = System.nanoTime();
try {
if (hasCachedTsBlock) {
return true;
@@ -115,6 +117,8 @@ public class AlignedSeriesScanOperator implements DataSourceOperator {
return hasCachedTsBlock;
} catch (IOException e) {
throw new RuntimeException("Error happened while scanning the file", e);
+ } finally {
+ operatorContext.addOperatorTime(ALIGNED_SERIES_SCAN_OPERATOR, System.nanoTime() - startTime);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesScanOperator.java
index 05685f758d..a993230378 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesScanOperator.java
@@ -30,6 +30,8 @@ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.io.IOException;
import java.util.Set;
+import static org.apache.iotdb.db.mpp.statistics.QueryStatistics.SERIES_SCAN_OPERATOR;
+
public class SeriesScanOperator implements DataSourceOperator {
private final OperatorContext operatorContext;
@@ -82,7 +84,7 @@ public class SeriesScanOperator implements DataSourceOperator {
@Override
public boolean hasNext() {
-
+ long startTime = System.nanoTime();
try {
if (hasCachedTsBlock) {
return true;
@@ -116,6 +118,8 @@ public class SeriesScanOperator implements DataSourceOperator {
return hasCachedTsBlock;
} catch (IOException e) {
throw new RuntimeException("Error happened while scanning the file", e);
+ } finally {
+ operatorContext.addOperatorTime(SERIES_SCAN_OPERATOR, System.nanoTime() - startTime);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesScanUtil.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesScanUtil.java
index 22fd2cbabc..160a794230 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesScanUtil.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesScanUtil.java
@@ -58,6 +58,7 @@ import java.util.function.ToLongFunction;
import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.iotdb.db.mpp.statistics.QueryStatistics.PAGE_READER;
public class SeriesScanUtil {
private final FragmentInstanceContext context;
@@ -1124,10 +1125,12 @@ public class SeriesScanUtil {
}
TsBlock getAllSatisfiedPageData(boolean ascending) throws IOException {
+ long startTime = System.nanoTime();
TsBlock tsBlock = data.getAllSatisfiedData();
if (!ascending) {
tsBlock.reverse();
}
+ context.addOperationTime(PAGE_READER, System.nanoTime() - startTime);
return tsBlock;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/statistics/QueryStatistics.java b/server/src/main/java/org/apache/iotdb/db/mpp/statistics/QueryStatistics.java
index e61e22f3ff..abe9b512bc 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/statistics/QueryStatistics.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/statistics/QueryStatistics.java
@@ -46,6 +46,22 @@ public class QueryStatistics {
private final Map<String, OperationStatistic> operationStatistics = new ConcurrentHashMap<>();
+ public static final String LOCAL_EXECUTION_PLANNER = "LocalExecutionPlanner";
+
+ public static final String QUERY_EXECUTION = "QueryExecution";
+
+ public static final String SERIES_SCAN_OPERATOR = "SeriesScanOperator";
+
+ public static final String ALIGNED_SERIES_SCAN_OPERATOR = "AlignedSeriesScanOperator";
+
+ public static final String AGG_SCAN_OPERATOR = "AbstractSeriesAggregationScanOperator";
+
+ public static final String FILTER_AND_PROJECT_OPERATOR = "FilterAndProjectOperator";
+
+ public static final String SINGLE_INPUT_AGG_OPERATOR = "SingleInputAggregationOperator";
+
+ public static final String PAGE_READER = "IPageReader";
+
private QueryStatistics() {
ScheduledExecutorService scheduledExecutor =
IoTDBThreadPoolFactory.newScheduledThreadPool(1, "Query-Statistics-Print");
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
index e0575880ea..a842859335 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
@@ -20,6 +20,7 @@ package org.apache.iotdb.db.utils;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.db.mpp.plan.execution.IQueryExecution;
+import org.apache.iotdb.db.mpp.statistics.QueryStatistics;
import org.apache.iotdb.db.tools.watermark.WatermarkEncoder;
import org.apache.iotdb.service.rpc.thrift.TSQueryDataSet;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
@@ -44,11 +45,15 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
+import static org.apache.iotdb.db.mpp.statistics.QueryStatistics.QUERY_EXECUTION;
+
/** TimeValuePairUtils to convert between thrift format and TsFile format. */
public class QueryDataSetUtils {
private static final int FLAG = 0x01;
+ private static final QueryStatistics QUERY_STATISTICS = QueryStatistics.getInstance();
+
private QueryDataSetUtils() {}
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
@@ -381,27 +386,32 @@ public class QueryDataSetUtils {
// To fetch required amounts of data and combine them through List
public static Pair<List<ByteBuffer>, Boolean> convertQueryResultByFetchSize(
IQueryExecution queryExecution, int fetchSize) throws IoTDBException {
- int rowCount = 0;
- List<ByteBuffer> res = new ArrayList<>();
- while (rowCount < fetchSize) {
- Optional<ByteBuffer> optionalByteBuffer = queryExecution.getByteBufferBatchResult();
- if (!optionalByteBuffer.isPresent()) {
- break;
- }
- ByteBuffer byteBuffer = optionalByteBuffer.get();
- byteBuffer.mark();
- int valueColumnCount = byteBuffer.getInt();
- for (int i = 0; i < valueColumnCount; i++) {
- byteBuffer.get();
- }
- int positionCount = byteBuffer.getInt();
- byteBuffer.reset();
- if (positionCount != 0) {
- res.add(byteBuffer);
+ long startTime = System.nanoTime();
+ try {
+ int rowCount = 0;
+ List<ByteBuffer> res = new ArrayList<>();
+ while (rowCount < fetchSize) {
+ Optional<ByteBuffer> optionalByteBuffer = queryExecution.getByteBufferBatchResult();
+ if (!optionalByteBuffer.isPresent()) {
+ break;
+ }
+ ByteBuffer byteBuffer = optionalByteBuffer.get();
+ byteBuffer.mark();
+ int valueColumnCount = byteBuffer.getInt();
+ for (int i = 0; i < valueColumnCount; i++) {
+ byteBuffer.get();
+ }
+ int positionCount = byteBuffer.getInt();
+ byteBuffer.reset();
+ if (positionCount != 0) {
+ res.add(byteBuffer);
+ }
+ rowCount += positionCount;
}
- rowCount += positionCount;
+ return new Pair<>(res, !queryExecution.hasNextResult());
+ } finally {
+ QUERY_STATISTICS.addCost(QUERY_EXECUTION, System.nanoTime() - startTime);
}
- return new Pair<>(res, !queryExecution.hasNextResult());
}
public static long[] readTimesFromBuffer(ByteBuffer buffer, int size) {
[iotdb] 03/03: Merge remote-tracking branch 'origin/master' into QueryMetrics
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch QueryMetrics
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 12ee3625275b582220e42e0b82ff63b14e3ec144
Merge: 1df99c2ec2 1b1c57b6d8
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Mon Nov 7 14:47:11 2022 +0800
Merge remote-tracking branch 'origin/master' into QueryMetrics
client-go | 2 +-
.../src/assembly/resources/conf/confignode-env.bat | 22 +-
.../resources/conf/iotdb-confignode.properties | 41 +-
.../assembly/resources/sbin/start-confignode.bat | 24 +-
.../assembly/resources/sbin/start-confignode.sh | 9 +-
.../iotdb/confignode/conf/ConfigNodeConfig.java | 167 ++-
.../confignode/conf/ConfigNodeDescriptor.java | 91 +-
.../confignode/conf/ConfigNodeStartupCheck.java | 6 +-
.../statemachine/PartitionRegionStateMachine.java | 3 +-
.../iotdb/confignode/manager/ConsensusManager.java | 25 +-
.../iotdb/confignode/manager/ProcedureManager.java | 4 +-
.../iotdb/confignode/manager/UDFManager.java | 4 +-
.../iotdb/confignode/manager/node/NodeManager.java | 5 +-
.../confignode/persistence/node/NodeInfo.java | 5 +-
.../persistence/partition/PartitionInfo.java | 9 +-
.../partition/StorageGroupPartitionTable.java | 19 +-
.../procedure/env/ConfigNodeProcedureEnv.java | 60 +-
.../procedure/env/DataNodeRemoveHandler.java | 73 +-
.../impl/node/RemoveDataNodeProcedure.java | 9 +-
.../impl/statemachine/RegionMigrateProcedure.java | 11 +-
.../procedure/store/ConfigProcedureStore.java | 2 +-
.../confignode1conf/iotdb-common.properties | 2 +-
.../confignode1conf/iotdb-confignode.properties | 2 +-
.../confignode2conf/iotdb-common.properties | 2 +-
.../confignode2conf/iotdb-confignode.properties | 2 +-
.../confignode3conf/iotdb-common.properties | 2 +-
.../confignode3conf/iotdb-confignode.properties | 2 +-
docker/src/main/Dockerfile-1c1d | 3 +
docker/src/main/Dockerfile-1c1d-influxdb | 5 +-
docs/UserGuide/Process-Data/Triggers.md | 2 +-
docs/UserGuide/Reference/Syntax-Conventions.md | 38 +-
docs/UserGuide/UDF-Library/Data-Profiling.md | 73 ++
docs/zh/UserGuide/Process-Data/Triggers.md | 2 +-
docs/zh/UserGuide/Reference/Syntax-Conventions.md | 40 +-
docs/zh/UserGuide/UDF-Library/Data-Profiling.md | 74 ++
integration-test/README.md | 2 +-
.../org/apache/iotdb/it/env/ConfigNodeWrapper.java | 2 +-
.../org/apache/iotdb/it/env/DataNodeWrapper.java | 2 +-
.../java/org/apache/iotdb/it/env/MppConfig.java | 2 +-
library-udf/src/assembly/tools/register-UDF.bat | 1 +
library-udf/src/assembly/tools/register-UDF.sh | 1 +
.../iotdb/library/dprofile/UDAFQuantile.java | 127 +++
.../library/dprofile/util/HeapLongKLLSketch.java | 356 +++++++
.../dprofile/util/KLLSketchForQuantile.java | 172 +++
.../resources/conf/iotdb-common.properties | 1126 +++++++++-----------
.../apache/iotdb/commons/conf/CommonConfig.java | 2 +-
.../apache/iotdb/commons/conf/IoTDBConstant.java | 4 +-
.../src/assembly/resources/conf/datanode-env.bat | 20 -
.../resources/conf/iotdb-datanode.properties | 47 +-
.../src/assembly/resources/sbin/start-datanode.bat | 16 -
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 6 +-
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 2 +-
.../db/mpp/execution/exchange/SinkHandle.java | 84 +-
.../java/org/apache/iotdb/db/service/DataNode.java | 2 +-
.../iotdb/db/service/RegionMigrateService.java | 31 +-
.../impl/DataNodeInternalRPCServiceImpl.java | 4 +-
.../db/mpp/execution/exchange/SinkHandleTest.java | 28 +-
.../datanode1conf/iotdb-datanode.properties | 2 +-
.../datanode2conf/iotdb-datanode.properties | 2 +-
.../datanode3conf/iotdb-datanode.properties | 2 +-
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 1 +
61 files changed, 1801 insertions(+), 1083 deletions(-)
[iotdb] 02/03: resolve conflicts
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch QueryMetrics
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1df99c2ec294942721c2ded87facba2bfb7f6105
Merge: d15fbff6c7 37e60f727d
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Mon Nov 7 14:46:43 2022 +0800
resolve conflicts
.../org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 7 ++-
.../assembly/resources/conf/logback-datanode.xml | 2 +-
.../iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java | 8 +++
.../apache/iotdb/db/mpp/plan/analyze/Analyzer.java | 10 +++-
.../db/mpp/plan/execution/QueryExecution.java | 6 +++
.../plan/execution/config/ConfigTaskVisitor.java | 7 +++
.../config/executor/ClusterConfigTaskExecutor.java | 16 ++++++
.../config/executor/IConfigTaskExecutor.java | 2 +
.../executor/StandaloneConfigTaskExecutor.java | 13 +++++
.../mpp/plan/execution/config/sys/TracingTask.java | 41 +++++++++++++++
.../iotdb/db/mpp/plan/parser/ASTVisitor.java | 7 +++
.../iotdb/db/mpp/plan/planner/LogicalPlanner.java | 7 ++-
.../planner/distribution/DistributionPlanner.java | 17 ++++--
.../db/mpp/plan/statement/StatementVisitor.java | 5 ++
.../mpp/plan/statement/sys/TracingStatement.java | 60 ++++++++++++++++++++++
.../iotdb/db/mpp/statistics/QueryStatistics.java | 11 +++-
.../service/thrift/impl/ClientRPCServiceImpl.java | 9 +++-
17 files changed, 216 insertions(+), 12 deletions(-)
diff --cc server/src/main/java/org/apache/iotdb/db/mpp/statistics/QueryStatistics.java
index abe9b512bc,1c69acedcc..55db559d19
--- a/server/src/main/java/org/apache/iotdb/db/mpp/statistics/QueryStatistics.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/statistics/QueryStatistics.java
@@@ -46,21 -46,13 +46,28 @@@ public class QueryStatistics
private final Map<String, OperationStatistic> operationStatistics = new ConcurrentHashMap<>();
+ public static final String LOCAL_EXECUTION_PLANNER = "LocalExecutionPlanner";
+
+ public static final String QUERY_EXECUTION = "QueryExecution";
+
+ public static final String SERIES_SCAN_OPERATOR = "SeriesScanOperator";
+
+ public static final String ALIGNED_SERIES_SCAN_OPERATOR = "AlignedSeriesScanOperator";
+
+ public static final String AGG_SCAN_OPERATOR = "AbstractSeriesAggregationScanOperator";
+
+ public static final String FILTER_AND_PROJECT_OPERATOR = "FilterAndProjectOperator";
+
+ public static final String SINGLE_INPUT_AGG_OPERATOR = "SingleInputAggregationOperator";
+
+ public static final String PAGE_READER = "IPageReader";
+ public static final String PARSER = "Parser";
+ public static final String ANALYZER = "Analyzer";
+ public static final String SCHEMA_FETCHER = "SchemaFetcher";
+ public static final String PARTITION_FETCHER = "PartitionFetcher";
+ public static final String LOGICAL_PLANNER = "LogicalPlanner";
+ public static final String DISTRIBUTION_PLANNER = "DistributionPlanner";
+ public static final String DISPATCHER = "Dispatcher";
private QueryStatistics() {
ScheduledExecutorService scheduledExecutor =