You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by hu...@apache.org on 2023/11/14 11:07:47 UTC
(iotdb) branch master updated: Implement more fine-grained time slice control for the AggregationScanOperator (#11521)
This is an automated email from the ASF dual-hosted git repository.
hui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 180e5f1fe73 Implement more fine-grained time slice control for the AggregationScanOperator (#11521)
180e5f1fe73 is described below
commit 180e5f1fe7394387d8fd6e477d29ac22099a30dc
Author: Liao Lanyu <14...@qq.com>
AuthorDate: Tue Nov 14 19:07:41 2023 +0800
Implement more fine-grained time slice control for the AggregationScanOperator (#11521)
---
.../it/env/cluster/config/MppCommonConfig.java | 7 +
.../env/cluster/config/MppSharedCommonConfig.java | 7 +
.../it/env/remote/config/RemoteCommonConfig.java | 5 +
.../org/apache/iotdb/itbase/env/CommonConfig.java | 2 +
.../db/it/aligned/IoTDBAlignedSeriesQuery5IT.java | 56 ++++++++
.../AbstractSeriesAggregationScanOperator.java | 142 ++++++++++++---------
6 files changed, 162 insertions(+), 57 deletions(-)
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
index 121a6cf99d3..4c88df36df2 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
@@ -401,6 +401,13 @@ public class MppCommonConfig extends MppBaseConfig implements CommonConfig {
return this;
}
+ @Override
+ public CommonConfig setDriverTaskExecutionTimeSliceInMs(long driverTaskExecutionTimeSliceInMs) {
+ setProperty(
+ "driver_task_execution_time_slice_in_ms", String.valueOf(driverTaskExecutionTimeSliceInMs));
+ return this;
+ }
+
// For part of the log directory
public String getClusterConfigStr() {
return fromConsensusFullNameToAbbr(properties.getProperty(CONFIG_NODE_CONSENSUS_PROTOCOL_CLASS))
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
index dc89155edb4..939f192cb0c 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
@@ -409,4 +409,11 @@ public class MppSharedCommonConfig implements CommonConfig {
cnConfig.setPipeAirGapReceiverEnabled(isPipeAirGapReceiverEnabled);
return this;
}
+
+ @Override
+ public CommonConfig setDriverTaskExecutionTimeSliceInMs(long driverTaskExecutionTimeSliceInMs) {
+ dnConfig.setDriverTaskExecutionTimeSliceInMs(driverTaskExecutionTimeSliceInMs);
+ cnConfig.setDriverTaskExecutionTimeSliceInMs(driverTaskExecutionTimeSliceInMs);
+ return this;
+ }
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java
index 667acca1265..43f759421de 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java
@@ -287,4 +287,9 @@ public class RemoteCommonConfig implements CommonConfig {
public CommonConfig setPipeAirGapReceiverEnabled(boolean isPipeAirGapReceiverEnabled) {
return this;
}
+
+ @Override
+ public CommonConfig setDriverTaskExecutionTimeSliceInMs(long driverTaskExecutionTimeSliceInMs) {
+ return this;
+ }
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java
index 41f23bdae86..874bf12eee6 100644
--- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java
@@ -129,4 +129,6 @@ public interface CommonConfig {
CommonConfig setSchemaRegionPerDataNode(double schemaRegionPerDataNode);
CommonConfig setPipeAirGapReceiverEnabled(boolean isPipeAirGapReceiverEnabled);
+
+ CommonConfig setDriverTaskExecutionTimeSliceInMs(long driverTaskExecutionTimeSliceInMs);
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQuery5IT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQuery5IT.java
new file mode 100644
index 00000000000..c2a3beefcfd
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQuery5IT.java
@@ -0,0 +1,56 @@
+/*
+ * 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.it.aligned;
+
+import org.apache.iotdb.db.it.utils.AlignedWriteUtil;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBAlignedSeriesQuery5IT extends IoTDBAlignedSeriesQueryIT {
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setEnableSeqSpaceCompaction(false)
+ .setEnableUnseqSpaceCompaction(false)
+ .setEnableCrossSpaceCompaction(false)
+ .setMaxTsBlockLineNumber(1)
+ .setMaxNumberOfPointsInPage(1)
+ .setDriverTaskExecutionTimeSliceInMs(20);
+
+ EnvFactory.getEnv().initClusterEnvironment();
+ AlignedWriteUtil.insertData();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java
index c54f42e67c1..653ee5e224e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java
@@ -66,6 +66,9 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
private final long cachedRawDataSize;
+ /** Time slice for one next call in total, shared by the inner methods of the next() method */
+ private long leftRuntimeOfOneNextCall;
+
@SuppressWarnings("squid:S107")
protected AbstractSeriesAggregationScanOperator(
PlanNodeId sourceId,
@@ -114,28 +117,33 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
@Override
public boolean hasNext() throws Exception {
- return timeRangeIterator.hasNextTimeRange();
+ return curTimeRange != null || timeRangeIterator.hasNextTimeRange();
}
@Override
public TsBlock next() throws Exception {
- // start stopwatch
- long maxRuntime = operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
+ // start stopwatch, reset leftRuntimeOfOneNextCall
long start = System.nanoTime();
+ leftRuntimeOfOneNextCall = operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
+ long maxRuntime = leftRuntimeOfOneNextCall;
while (System.nanoTime() - start < maxRuntime
- && timeRangeIterator.hasNextTimeRange()
+ && (curTimeRange != null || timeRangeIterator.hasNextTimeRange())
&& !resultTsBlockBuilder.isFull()) {
- // move to next time window
- curTimeRange = timeRangeIterator.nextTimeRange();
-
- // clear previous aggregation result
- for (Aggregator aggregator : aggregators) {
- aggregator.reset();
+ if (curTimeRange == null) {
+ // move to the next time window
+ curTimeRange = timeRangeIterator.nextTimeRange();
+ // clear previous aggregation result
+ for (Aggregator aggregator : aggregators) {
+ aggregator.reset();
+ }
}
// calculate aggregation result on current time window
- calculateNextAggregationResult();
+ // Keep curTimeRange if the calculation of this timeRange is not done
+ if (calculateAggregationResultForCurrentTimeRange()) {
+ curTimeRange = null;
+ }
}
if (resultTsBlockBuilder.getPositionCount() > 0) {
@@ -156,32 +164,42 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
}
@SuppressWarnings("squid:S112")
- protected void calculateNextAggregationResult() {
+ /** Return true if we have the result of this timeRange. */
+ protected boolean calculateAggregationResultForCurrentTimeRange() {
try {
if (calcFromCachedData()) {
updateResultTsBlock();
- return;
+ return true;
}
- // read page data firstly
if (readAndCalcFromPage()) {
updateResultTsBlock();
- return;
+ return true;
}
- // read chunk data secondly
- if (readAndCalcFromChunk()) {
+ // only when all the page data has been consumed, we need to read the chunk data
+ if (!seriesScanUtil.hasNextPage() && readAndCalcFromChunk()) {
updateResultTsBlock();
- return;
+ return true;
}
- // read from file
- if (readAndCalcFromFile()) {
+ // only when all the page and chunk data has been consumed, we need to read the file data
+ if (!seriesScanUtil.hasNextPage()
+ && !seriesScanUtil.hasNextChunk()
+ && readAndCalcFromFile()) {
updateResultTsBlock();
- return;
+ return true;
}
+ // If the TimeRange is (Long.MIN_VALUE, Long.MAX_VALUE), for Aggregators like countAggregator,
+ // we have to consume all the data before we finish the aggregation calculation.
+ if (seriesScanUtil.hasNextPage()
+ || seriesScanUtil.hasNextChunk()
+ || seriesScanUtil.hasNextFile()) {
+ return false;
+ }
updateResultTsBlock();
+ return true;
} catch (IOException e) {
throw new RuntimeException("Error while scanning the file", e);
}
@@ -214,7 +232,9 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
@SuppressWarnings({"squid:S3776", "squid:S135", "squid:S3740"})
protected boolean readAndCalcFromFile() throws IOException {
- while (seriesScanUtil.hasNextFile()) {
+ // start stopwatch
+ long start = System.nanoTime();
+ while (System.nanoTime() - start < leftRuntimeOfOneNextCall && seriesScanUtil.hasNextFile()) {
if (canUseCurrentFileStatistics()) {
Statistics fileTimeStatistics = seriesScanUtil.currentFileTimeStatistics();
if (fileTimeStatistics.getStartTime() > curTimeRange.getMax()) {
@@ -253,7 +273,9 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
@SuppressWarnings({"squid:S3776", "squid:S135", "squid:S3740"})
protected boolean readAndCalcFromChunk() throws IOException {
- while (seriesScanUtil.hasNextChunk()) {
+ // start stopwatch
+ long start = System.nanoTime();
+ while (System.nanoTime() - start < leftRuntimeOfOneNextCall && seriesScanUtil.hasNextChunk()) {
if (canUseCurrentChunkStatistics()) {
Statistics chunkTimeStatistics = seriesScanUtil.currentChunkTimeStatistics();
if (chunkTimeStatistics.getStartTime() > curTimeRange.getMax()) {
@@ -292,47 +314,53 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
@SuppressWarnings({"squid:S3776", "squid:S135", "squid:S3740"})
protected boolean readAndCalcFromPage() throws IOException {
- while (seriesScanUtil.hasNextPage()) {
- if (canUseCurrentPageStatistics()) {
- Statistics pageTimeStatistics = seriesScanUtil.currentPageTimeStatistics();
- // There is no more eligible points in current time range
- if (pageTimeStatistics.getStartTime() > curTimeRange.getMax()) {
- if (ascending) {
- return true;
- } else {
- seriesScanUtil.skipCurrentPage();
- continue;
- }
- }
- // can use pageHeader
- if (curTimeRange.contains(
- pageTimeStatistics.getStartTime(), pageTimeStatistics.getEndTime())) {
- Statistics[] statisticsList = new Statistics[subSensorSize];
- for (int i = 0; i < subSensorSize; i++) {
- statisticsList[i] = seriesScanUtil.currentPageStatistics(i);
+ // start stopwatch
+ long start = System.nanoTime();
+ try {
+ while (System.nanoTime() - start < leftRuntimeOfOneNextCall && seriesScanUtil.hasNextPage()) {
+ if (canUseCurrentPageStatistics()) {
+ Statistics pageTimeStatistics = seriesScanUtil.currentPageTimeStatistics();
+ // There is no more eligible points in current time range
+ if (pageTimeStatistics.getStartTime() > curTimeRange.getMax()) {
+ if (ascending) {
+ return true;
+ } else {
+ seriesScanUtil.skipCurrentPage();
+ continue;
+ }
}
- calcFromStatistics(pageTimeStatistics, statisticsList);
- seriesScanUtil.skipCurrentPage();
- if (isAllAggregatorsHasFinalResult(aggregators) && !isGroupByQuery) {
- return true;
- } else {
- continue;
+ // can use pageHeader
+ if (curTimeRange.contains(
+ pageTimeStatistics.getStartTime(), pageTimeStatistics.getEndTime())) {
+ Statistics[] statisticsList = new Statistics[subSensorSize];
+ for (int i = 0; i < subSensorSize; i++) {
+ statisticsList[i] = seriesScanUtil.currentPageStatistics(i);
+ }
+ calcFromStatistics(pageTimeStatistics, statisticsList);
+ seriesScanUtil.skipCurrentPage();
+ if (isAllAggregatorsHasFinalResult(aggregators) && !isGroupByQuery) {
+ return true;
+ } else {
+ continue;
+ }
}
}
- }
- // calc from page data
- TsBlock tsBlock = seriesScanUtil.nextPage();
- if (tsBlock == null || tsBlock.isEmpty()) {
- continue;
- }
+ // calc from page data
+ TsBlock tsBlock = seriesScanUtil.nextPage();
+ if (tsBlock == null || tsBlock.isEmpty()) {
+ continue;
+ }
- // calc from raw data
- if (calcFromRawData(tsBlock)) {
- return true;
+ // calc from raw data
+ if (calcFromRawData(tsBlock)) {
+ return true;
+ }
}
+ return false;
+ } finally {
+ leftRuntimeOfOneNextCall -= (System.nanoTime() - start);
}
- return false;
}
@SuppressWarnings({"squid:S3740"})