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/07/15 01:04:24 UTC
[iotdb] branch master updated: [IOTDB-3714] Refactor operators related to aggregate query to support batch processing (#6622)
This is an automated email from the ASF dual-hosted git repository.
jackietien 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 8cd42fa076 [IOTDB-3714] Refactor operators related to aggregate query to support batch processing (#6622)
8cd42fa076 is described below
commit 8cd42fa07612378b983ce33321a123ee0796a071
Author: liuminghui233 <36...@users.noreply.github.com>
AuthorDate: Fri Jul 15 09:04:20 2022 +0800
[IOTDB-3714] Refactor operators related to aggregate query to support batch processing (#6622)
---
docs/UserGuide/Reference/DataNode-Config-Manual.md | 28 +-
.../UserGuide/Reference/DataNode-Config-Manual.md | 28 +-
.../util/TSFileConfigUtilCompletenessTest.java | 4 +-
.../java/org/apache/iotdb/it/env/MppConfig.java | 6 +
.../org/apache/iotdb/itbase/env/BaseConfig.java | 8 +
.../aligned/IoTDBAggregationWithDeletion2IT.java | 3 +
.../it/aligned/IoTDBAggregationWithDeletionIT.java | 4 +
.../db/it/aligned/IoTDBAlignedSeriesQuery2IT.java | 3 +
.../db/it/aligned/IoTDBAlignedSeriesQuery3IT.java | 4 +-
.../db/it/aligned/IoTDBAlignedSeriesQueryIT.java | 4 +
.../db/it/aligned/IoTDBGroupByLevelQueryIT.java | 5 +
...GroupByQueryWithValueFilterWithDeletion2IT.java | 3 +
...BGroupByQueryWithValueFilterWithDeletionIT.java | 4 +
...upByQueryWithoutValueFilterWithDeletion2IT.java | 3 +
...oupByQueryWithoutValueFilterWithDeletionIT.java | 4 +
.../iotdb/db/it/env/StandaloneEnvConfig.java | 11 +
.../resources/conf/iotdb-datanode.properties | 12 +
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 12 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 22 ++
.../iotdb/db/mpp/aggregation/Aggregator.java | 13 +-
.../EmptyQueueSlidingWindowAggregator.java | 1 -
.../MonotonicQueueSlidingWindowAggregator.java | 5 +-
.../NormalQueueSlidingWindowAggregator.java | 1 -
.../slidingwindow/SlidingWindowAggregator.java | 2 +-
.../db/mpp/execution/operator/AggregationUtil.java | 161 +++++++++
.../db/mpp/execution/operator/OperatorContext.java | 35 ++
.../operator/process/AggregationOperator.java | 144 +++++---
.../process/RawDataAggregationOperator.java | 187 ++--------
.../process/SingleInputAggregationOperator.java | 149 ++++++++
.../process/SlidingWindowAggregationOperator.java | 155 +++------
.../AbstractSeriesAggregationScanOperator.java | 350 +++++++++++++++++++
.../AlignedSeriesAggregationScanOperator.java | 377 +-------------------
.../operator/source/AlignedSeriesScanUtil.java | 38 ++
.../source/SeriesAggregationScanOperator.java | 386 +--------------------
.../execution/operator/source/SeriesScanUtil.java | 57 +--
.../mpp/execution/schedule/DriverTaskThread.java | 6 +-
.../mpp/execution/timer/ITimeSliceAllocator.java | 29 ++
.../timer/RuleBasedTimeSliceAllocator.java | 66 ++++
.../db/mpp/plan/planner/LocalExecutionPlanner.java | 184 ++++++----
.../operator/AggregationOperatorTest.java | 42 ++-
.../AlignedSeriesAggregationScanOperatorTest.java | 92 +++--
.../operator/LastQueryMergeOperatorTest.java | 15 +
.../operator/RawDataAggregationOperatorTest.java | 8 +
.../SeriesAggregationScanOperatorTest.java | 88 +++--
.../SlidingWindowAggregationOperatorTest.java | 46 ++-
.../operator/UpdateLastCacheOperatorTest.java | 8 +
.../iotdb/tsfile/common/conf/TSFileConfig.java | 22 ++
.../file/metadata/statistics/Statistics.java | 5 +
.../iotdb/tsfile/read/common/block/TsBlock.java | 4 +
.../tsfile/read/common/block/TsBlockBuilder.java | 6 +-
.../read/common/block/TsBlockBuilderStatus.java | 4 +-
.../tsfile/read/common/block/TsBlockUtil.java | 59 ++++
52 files changed, 1627 insertions(+), 1286 deletions(-)
diff --git a/docs/UserGuide/Reference/DataNode-Config-Manual.md b/docs/UserGuide/Reference/DataNode-Config-Manual.md
index 5e5c5ba473..3199ede743 100644
--- a/docs/UserGuide/Reference/DataNode-Config-Manual.md
+++ b/docs/UserGuide/Reference/DataNode-Config-Manual.md
@@ -199,7 +199,7 @@ The permission definitions are in ${IOTDB\_CONF}/conf/jmx.access.
|Default| 5 |
|Effective|After restarting system|
-* mpp\_data\_exchange\_core\_pool\_size=1
+* mpp\_data\_exchange\_keep\_alive\_time\_in\_ms
|Name| mpp\_data\_exchange\_keep\_alive\_time\_in\_ms |
|:---:|:---|
@@ -208,6 +208,32 @@ The permission definitions are in ${IOTDB\_CONF}/conf/jmx.access.
|Default| 1000 |
|Effective|After restarting system|
+* driver\_task\_execution\_time\_slice\_in\_ms
+
+|Name| driver\_task\_execution\_time\_slice\_in\_ms |
+|:---:|:---|
+|Description| Maximum execution time of a DriverTask |
+|Type| int |
+|Default| 100 |
+|Effective|After restarting system|
+
+* max\_tsblock\_size\_in\_bytes
+
+|Name| max\_tsblock\_size\_in\_bytes |
+|:---:|:---|
+|Description| Maximum capacity of a TsBlock |
+|Type| int |
+|Default| 1024 * 1024 (1 MB) |
+|Effective|After restarting system|
+
+* max\_tsblock\_line\_numbers
+
+|Name| max\_tsblock\_line\_numbers |
+|:---:|:---|
+|Description| Maximum number of lines in a single TsBlock |
+|Type| int |
+|Default| 1000 |
+|Effective|After restarting system|
### DataNode Internal Service
diff --git a/docs/zh/UserGuide/Reference/DataNode-Config-Manual.md b/docs/zh/UserGuide/Reference/DataNode-Config-Manual.md
index c6f8f36671..7b03b9ef36 100644
--- a/docs/zh/UserGuide/Reference/DataNode-Config-Manual.md
+++ b/docs/zh/UserGuide/Reference/DataNode-Config-Manual.md
@@ -181,7 +181,7 @@ IoTDB DataNode 与 Standalone 模式共用一套配置文件，均位于 IoTDB
|默认值| 5 |
|改后生效方式|重启服务生效|
-* mpp\_data\_exchange\_core\_pool\_size
+* mpp\_data\_exchange\_keep\_alive\_time\_in\_ms
|名字| mpp\_data\_exchange\_keep\_alive\_time\_in\_ms |
|:---:|:---|
@@ -190,6 +190,32 @@ IoTDB DataNode 与 Standalone 模式共用一套配置文件，均位于 IoTDB
|默认值| 1000 |
|改后生效方式|重启服务生效|
+* driver\_task\_execution\_time\_slice\_in\_ms
+
+|名字| driver\_task\_execution\_time\_slice\_in\_ms |
+|:---:|:---|
+|描述| 单个 DriverTask 最长执行时间 |
+|类型| int |
+|默认值| 100 |
+|改后生效方式|重启服务生效|
+
+* max\_tsblock\_size\_in\_bytes
+
+|名字| max\_tsblock\_size\_in\_bytes |
+|:---:|:---|
+|描述| 单个 TsBlock 的最大容量 |
+|类型| int |
+|默认值| 1024 * 1024 (1 MB) |
+|改后生效方式|重启服务生效|
+
+* max\_tsblock\_line\_numbers
+
+|名字| max\_tsblock\_line\_numbers |
+|:---:|:---|
+|描述| 单个 TsBlock 的最大行数 |
+|类型| int |
+|默认值| 1000 |
+|改后生效方式|重启服务生效|
### DataNode 内部服务参数
diff --git a/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/util/TSFileConfigUtilCompletenessTest.java b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/util/TSFileConfigUtilCompletenessTest.java
index 156baa4e0c..5ed14b4051 100644
--- a/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/util/TSFileConfigUtilCompletenessTest.java
+++ b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/util/TSFileConfigUtilCompletenessTest.java
@@ -72,7 +72,9 @@ public class TSFileConfigUtilCompletenessTest {
"setUseKerberos",
"setValueEncoder",
"setFreqEncodingSNR",
- "setFreqEncodingBlockSize"
+ "setFreqEncodingBlockSize",
+ "setMaxTsBlockLineNumber",
+ "setMaxTsBlockSizeInBytes"
};
Set<String> newSetters =
Arrays.stream(TSFileConfig.class.getMethods())
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/MppConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/MppConfig.java
index 18b5dd942e..89a2611fcc 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/MppConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/MppConfig.java
@@ -198,4 +198,10 @@ public class MppConfig implements BaseConfig {
"avg_series_point_number_threshold", String.valueOf(avgSeriesPointNumberThreshold));
return this;
}
+
+ @Override
+ public BaseConfig setMaxTsBlockLineNumber(int maxTsBlockLineNumber) {
+ engineProperties.setProperty("max_tsblock_line_number", String.valueOf(maxTsBlockLineNumber));
+ return this;
+ }
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseConfig.java
index e538c319a4..8066de123d 100644
--- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseConfig.java
@@ -185,4 +185,12 @@ public interface BaseConfig {
default BaseConfig setAvgSeriesPointNumberThreshold(int avgSeriesPointNumberThreshold) {
return this;
}
+
+ default int getMaxTsBlockLineNumber() {
+ return 1000;
+ }
+
+ default BaseConfig setMaxTsBlockLineNumber(int maxTsBlockLineNumber) {
+ return this;
+ }
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAggregationWithDeletion2IT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAggregationWithDeletion2IT.java
index 3decbd8bd4..23b894fae7 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAggregationWithDeletion2IT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAggregationWithDeletion2IT.java
@@ -47,11 +47,13 @@ public class IoTDBAggregationWithDeletion2IT extends IoTDBAggregationWithDeletio
enableUnseqSpaceCompaction = ConfigFactory.getConfig().isEnableUnseqSpaceCompaction();
enableCrossSpaceCompaction = ConfigFactory.getConfig().isEnableCrossSpaceCompaction();
numOfPointsPerPage = ConfigFactory.getConfig().getMaxNumberOfPointsInPage();
+ maxTsBlockLineNumber = ConfigFactory.getConfig().getMaxTsBlockLineNumber();
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(false);
ConfigFactory.getConfig().setMaxNumberOfPointsInPage(3);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(3);
EnvFactory.getEnv().initBeforeClass();
AlignedWriteUtil.insertData();
@@ -74,5 +76,6 @@ public class IoTDBAggregationWithDeletion2IT extends IoTDBAggregationWithDeletio
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
ConfigFactory.getConfig().setMaxNumberOfPointsInPage(numOfPointsPerPage);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(maxTsBlockLineNumber);
}
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAggregationWithDeletionIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAggregationWithDeletionIT.java
index da4c147e69..aca4adc514 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAggregationWithDeletionIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAggregationWithDeletionIT.java
@@ -50,15 +50,18 @@ public class IoTDBAggregationWithDeletionIT {
protected static boolean enableSeqSpaceCompaction;
protected static boolean enableUnseqSpaceCompaction;
protected static boolean enableCrossSpaceCompaction;
+ protected static int maxTsBlockLineNumber;
@BeforeClass
public static void setUp() throws Exception {
enableSeqSpaceCompaction = ConfigFactory.getConfig().isEnableSeqSpaceCompaction();
enableUnseqSpaceCompaction = ConfigFactory.getConfig().isEnableUnseqSpaceCompaction();
enableCrossSpaceCompaction = ConfigFactory.getConfig().isEnableCrossSpaceCompaction();
+ maxTsBlockLineNumber = ConfigFactory.getConfig().getMaxTsBlockLineNumber();
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(false);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(3);
EnvFactory.getEnv().initBeforeClass();
AlignedWriteUtil.insertData();
@@ -80,6 +83,7 @@ public class IoTDBAggregationWithDeletionIT {
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(enableSeqSpaceCompaction);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(maxTsBlockLineNumber);
}
@Test
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQuery2IT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQuery2IT.java
index 9842ddfeb7..f0fadcbe96 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQuery2IT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQuery2IT.java
@@ -41,11 +41,13 @@ public class IoTDBAlignedSeriesQuery2IT extends IoTDBAlignedSeriesQueryIT {
enableUnseqSpaceCompaction = ConfigFactory.getConfig().isEnableUnseqSpaceCompaction();
enableCrossSpaceCompaction = ConfigFactory.getConfig().isEnableCrossSpaceCompaction();
numOfPointsPerPage = ConfigFactory.getConfig().getMaxNumberOfPointsInPage();
+ maxTsBlockLineNumber = ConfigFactory.getConfig().getMaxTsBlockLineNumber();
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(false);
ConfigFactory.getConfig().setMaxNumberOfPointsInPage(2);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(3);
EnvFactory.getEnv().initBeforeClass();
AlignedWriteUtil.insertData();
}
@@ -57,5 +59,6 @@ public class IoTDBAlignedSeriesQuery2IT extends IoTDBAlignedSeriesQueryIT {
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
ConfigFactory.getConfig().setMaxNumberOfPointsInPage(numOfPointsPerPage);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(maxTsBlockLineNumber);
}
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQuery3IT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQuery3IT.java
index 697658379e..fb01fa9a83 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQuery3IT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQuery3IT.java
@@ -41,11 +41,12 @@ public class IoTDBAlignedSeriesQuery3IT extends IoTDBAlignedSeriesQueryIT {
enableUnseqSpaceCompaction = ConfigFactory.getConfig().isEnableUnseqSpaceCompaction();
enableCrossSpaceCompaction = ConfigFactory.getConfig().isEnableCrossSpaceCompaction();
numOfPointsPerPage = ConfigFactory.getConfig().getMaxNumberOfPointsInPage();
-
+ maxTsBlockLineNumber = ConfigFactory.getConfig().getMaxTsBlockLineNumber();
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(false);
ConfigFactory.getConfig().setMaxNumberOfPointsInPage(3);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(3);
EnvFactory.getEnv().initBeforeClass();
AlignedWriteUtil.insertData();
}
@@ -57,5 +58,6 @@ public class IoTDBAlignedSeriesQuery3IT extends IoTDBAlignedSeriesQueryIT {
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
ConfigFactory.getConfig().setMaxNumberOfPointsInPage(numOfPointsPerPage);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(maxTsBlockLineNumber);
}
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQueryIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQueryIT.java
index 93eaf497d7..e13639c32d 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQueryIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBAlignedSeriesQueryIT.java
@@ -70,15 +70,18 @@ public class IoTDBAlignedSeriesQueryIT {
protected static boolean enableSeqSpaceCompaction;
protected static boolean enableUnseqSpaceCompaction;
protected static boolean enableCrossSpaceCompaction;
+ protected static int maxTsBlockLineNumber;
@BeforeClass
public static void setUp() throws Exception {
enableSeqSpaceCompaction = ConfigFactory.getConfig().isEnableSeqSpaceCompaction();
enableUnseqSpaceCompaction = ConfigFactory.getConfig().isEnableUnseqSpaceCompaction();
enableCrossSpaceCompaction = ConfigFactory.getConfig().isEnableCrossSpaceCompaction();
+ maxTsBlockLineNumber = ConfigFactory.getConfig().getMaxTsBlockLineNumber();
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(false);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(3);
EnvFactory.getEnv().initBeforeClass();
AlignedWriteUtil.insertData();
}
@@ -89,6 +92,7 @@ public class IoTDBAlignedSeriesQueryIT {
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(enableSeqSpaceCompaction);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(maxTsBlockLineNumber);
}
// ----------------------------------------Last Query-----------------------------------------
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByLevelQueryIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByLevelQueryIT.java
index 5909f6d3c6..65300100e1 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByLevelQueryIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByLevelQueryIT.java
@@ -43,15 +43,19 @@ public class IoTDBGroupByLevelQueryIT {
protected static boolean enableSeqSpaceCompaction;
protected static boolean enableUnseqSpaceCompaction;
protected static boolean enableCrossSpaceCompaction;
+ protected static int maxTsBlockLineNumber;
@BeforeClass
public static void setUp() throws Exception {
enableSeqSpaceCompaction = ConfigFactory.getConfig().isEnableSeqSpaceCompaction();
enableUnseqSpaceCompaction = ConfigFactory.getConfig().isEnableUnseqSpaceCompaction();
enableCrossSpaceCompaction = ConfigFactory.getConfig().isEnableCrossSpaceCompaction();
+ maxTsBlockLineNumber = ConfigFactory.getConfig().getMaxTsBlockLineNumber();
+
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(false);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(3);
EnvFactory.getEnv().initBeforeClass();
AlignedWriteUtil.insertData();
@@ -90,6 +94,7 @@ public class IoTDBGroupByLevelQueryIT {
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(enableSeqSpaceCompaction);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(maxTsBlockLineNumber);
}
@Test
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithValueFilterWithDeletion2IT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithValueFilterWithDeletion2IT.java
index 2c029ed5ac..8888370399 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithValueFilterWithDeletion2IT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithValueFilterWithDeletion2IT.java
@@ -45,10 +45,12 @@ public class IoTDBGroupByQueryWithValueFilterWithDeletion2IT
enableUnseqSpaceCompaction = ConfigFactory.getConfig().isEnableUnseqSpaceCompaction();
enableCrossSpaceCompaction = ConfigFactory.getConfig().isEnableCrossSpaceCompaction();
numOfPointsPerPage = ConfigFactory.getConfig().getMaxNumberOfPointsInPage();
+ maxTsBlockLineNumber = ConfigFactory.getConfig().getMaxTsBlockLineNumber();
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(false);
ConfigFactory.getConfig().setMaxNumberOfPointsInPage(3);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(3);
EnvFactory.getEnv().initBeforeClass();
@@ -70,6 +72,7 @@ public class IoTDBGroupByQueryWithValueFilterWithDeletion2IT
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
ConfigFactory.getConfig().setMaxNumberOfPointsInPage(numOfPointsPerPage);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(maxTsBlockLineNumber);
EnvFactory.getEnv().cleanAfterClass();
}
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithValueFilterWithDeletionIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithValueFilterWithDeletionIT.java
index d7053ff7a9..a1112d33fd 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithValueFilterWithDeletionIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithValueFilterWithDeletionIT.java
@@ -52,15 +52,18 @@ public class IoTDBGroupByQueryWithValueFilterWithDeletionIT {
protected static boolean enableSeqSpaceCompaction;
protected static boolean enableUnseqSpaceCompaction;
protected static boolean enableCrossSpaceCompaction;
+ protected static int maxTsBlockLineNumber;
@BeforeClass
public static void setUp() throws Exception {
enableSeqSpaceCompaction = ConfigFactory.getConfig().isEnableSeqSpaceCompaction();
enableUnseqSpaceCompaction = ConfigFactory.getConfig().isEnableUnseqSpaceCompaction();
enableCrossSpaceCompaction = ConfigFactory.getConfig().isEnableCrossSpaceCompaction();
+ maxTsBlockLineNumber = ConfigFactory.getConfig().getMaxTsBlockLineNumber();
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(false);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(3);
EnvFactory.getEnv().initBeforeClass();
@@ -81,6 +84,7 @@ public class IoTDBGroupByQueryWithValueFilterWithDeletionIT {
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(enableSeqSpaceCompaction);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(maxTsBlockLineNumber);
EnvFactory.getEnv().cleanAfterClass();
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithoutValueFilterWithDeletion2IT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithoutValueFilterWithDeletion2IT.java
index 528ec2ac24..f4e8ea19c0 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithoutValueFilterWithDeletion2IT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithoutValueFilterWithDeletion2IT.java
@@ -45,11 +45,13 @@ public class IoTDBGroupByQueryWithoutValueFilterWithDeletion2IT
enableUnseqSpaceCompaction = ConfigFactory.getConfig().isEnableUnseqSpaceCompaction();
enableCrossSpaceCompaction = ConfigFactory.getConfig().isEnableCrossSpaceCompaction();
numOfPointsPerPage = ConfigFactory.getConfig().getMaxNumberOfPointsInPage();
+ maxTsBlockLineNumber = ConfigFactory.getConfig().getMaxTsBlockLineNumber();
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(false);
ConfigFactory.getConfig().setMaxNumberOfPointsInPage(3);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(3);
EnvFactory.getEnv().initBeforeClass();
AlignedWriteUtil.insertData();
@@ -72,5 +74,6 @@ public class IoTDBGroupByQueryWithoutValueFilterWithDeletion2IT
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
ConfigFactory.getConfig().setMaxNumberOfPointsInPage(numOfPointsPerPage);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(maxTsBlockLineNumber);
}
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithoutValueFilterWithDeletionIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithoutValueFilterWithDeletionIT.java
index d6b76d3a39..bb60bd1c25 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithoutValueFilterWithDeletionIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aligned/IoTDBGroupByQueryWithoutValueFilterWithDeletionIT.java
@@ -52,6 +52,7 @@ public class IoTDBGroupByQueryWithoutValueFilterWithDeletionIT {
protected static boolean enableUnseqSpaceCompaction;
protected static boolean enableCrossSpaceCompaction;
protected static long prevPartitionInterval;
+ protected static int maxTsBlockLineNumber;
private static final String TIMESTAMP_STR = "Time";
@@ -60,9 +61,11 @@ public class IoTDBGroupByQueryWithoutValueFilterWithDeletionIT {
enableSeqSpaceCompaction = ConfigFactory.getConfig().isEnableSeqSpaceCompaction();
enableUnseqSpaceCompaction = ConfigFactory.getConfig().isEnableUnseqSpaceCompaction();
enableCrossSpaceCompaction = ConfigFactory.getConfig().isEnableCrossSpaceCompaction();
+ maxTsBlockLineNumber = ConfigFactory.getConfig().getMaxTsBlockLineNumber();
ConfigFactory.getConfig().setEnableSeqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(false);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(false);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(3);
EnvFactory.getEnv().initBeforeClass();
AlignedWriteUtil.insertData();
@@ -84,6 +87,7 @@ public class IoTDBGroupByQueryWithoutValueFilterWithDeletionIT {
ConfigFactory.getConfig().setEnableUnseqSpaceCompaction(enableUnseqSpaceCompaction);
ConfigFactory.getConfig().setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
ConfigFactory.getConfig().setPartitionInterval(prevPartitionInterval);
+ ConfigFactory.getConfig().setMaxTsBlockLineNumber(maxTsBlockLineNumber);
EnvFactory.getEnv().cleanAfterClass();
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/env/StandaloneEnvConfig.java b/integration-test/src/test/java/org/apache/iotdb/db/it/env/StandaloneEnvConfig.java
index 55e65be26d..30d21973a5 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/env/StandaloneEnvConfig.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/env/StandaloneEnvConfig.java
@@ -231,4 +231,15 @@ public class StandaloneEnvConfig implements BaseConfig {
.setAvgSeriesPointNumberThreshold(avgSeriesPointNumberThreshold);
return this;
}
+
+ @Override
+ public int getMaxTsBlockLineNumber() {
+ return TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber();
+ }
+
+ @Override
+ public BaseConfig setMaxTsBlockLineNumber(int maxTsBlockLineNumber) {
+ TSFileDescriptor.getInstance().getConfig().setMaxTsBlockLineNumber(maxTsBlockLineNumber);
+ return this;
+ }
}
diff --git a/server/src/assembly/resources/conf/iotdb-datanode.properties b/server/src/assembly/resources/conf/iotdb-datanode.properties
index 0bbdb3c64e..66c8460c5c 100644
--- a/server/src/assembly/resources/conf/iotdb-datanode.properties
+++ b/server/src/assembly/resources/conf/iotdb-datanode.properties
@@ -493,6 +493,18 @@ timestamp_precision=ms
# Datatype: int, Unit: ms
# default_fill_interval=-1
+# the max execution time of a DriverTask
+# Datatype: int, Unit: ms
+# driver_task_execution_time_slice_in_ms=100
+
+# the max capacity of a TsBlock
+# Datatype: int, Unit: byte
+# max_tsblock_size_in_bytes=1048576
+
+# the max number of lines in a single TsBlock
+# Datatype: int
+# max_tsblock_line_numbers=1000
+
####################
### Compaction Configurations
####################
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 36560c79e3..66d48dfa21 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -950,6 +950,10 @@ public class IoTDBConfig {
private String readConsistencyLevel = "strong";
+ /** Maximum execution time of a DriverTask */
+ /** Maximum execution time of a DriverTask */
+ private int driverTaskExecutionTimeSliceInMs = 100;
+
IoTDBConfig() {}
public float getUdfMemoryBudgetInMB() {
@@ -3019,4 +3023,12 @@ public class IoTDBConfig {
public void setReadConsistencyLevel(String readConsistencyLevel) {
this.readConsistencyLevel = readConsistencyLevel;
}
+
+ public int getDriverTaskExecutionTimeSliceInMs() {
+ return driverTaskExecutionTimeSliceInMs;
+ }
+
+ public void setDriverTaskExecutionTimeSliceInMs(int driverTaskExecutionTimeSliceInMs) {
+ this.driverTaskExecutionTimeSliceInMs = driverTaskExecutionTimeSliceInMs;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index a25d6674d7..2c3a2a6ed0 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -1198,6 +1198,22 @@ public class IoTDBDescriptor {
"max_degree_of_index_node",
Integer.toString(
TSFileDescriptor.getInstance().getConfig().getMaxDegreeOfIndexNode()))));
+ TSFileDescriptor.getInstance()
+ .getConfig()
+ .setMaxTsBlockSizeInBytes(
+ Integer.parseInt(
+ properties.getProperty(
+ "max_tsblock_size_in_bytes",
+ Integer.toString(
+ TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes()))));
+ TSFileDescriptor.getInstance()
+ .getConfig()
+ .setMaxTsBlockLineNumber(
+ Integer.parseInt(
+ properties.getProperty(
+ "max_tsblock_line_number",
+ Integer.toString(
+ TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber()))));
}
// Mqtt related
@@ -1723,6 +1739,12 @@ public class IoTDBDescriptor {
Integer.parseInt(
properties.getProperty(
"partition_cache_size", Integer.toString(conf.getPartitionCacheSize()))));
+
+ conf.setDriverTaskExecutionTimeSliceInMs(
+ Integer.parseInt(
+ properties.getProperty(
+ "driver_task_execution_time_slice_in_ms",
+ Integer.toString(conf.getDriverTaskExecutionTimeSliceInMs()))));
}
/** Get default encode algorithm by data type */
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/Aggregator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/Aggregator.java
index b1bd9bfb63..a1f73ef627 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/Aggregator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/Aggregator.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.read.common.block.column.Column;
import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
+import java.util.Collections;
import java.util.List;
import static com.google.common.base.Preconditions.checkArgument;
@@ -45,6 +46,8 @@ public class Aggregator {
public Aggregator(Accumulator accumulator, AggregationStep step) {
this.accumulator = accumulator;
this.step = step;
+ this.inputLocationList =
+ Collections.singletonList(new InputLocation[] {new InputLocation(0, 0)});
}
// Used for aggregateOperator
@@ -108,16 +111,9 @@ public class Aggregator {
}
}
- public void processStatistics(Statistics statistics) {
- accumulator.addStatistics(statistics);
- }
-
- /** Used for AlignedSeriesAggregateScanOperator. */
+ /** Used for SeriesAggregateScanOperator. */
public void processStatistics(Statistics[] statistics) {
for (InputLocation[] inputLocations : inputLocationList) {
- checkArgument(
- inputLocations[0].getTsBlockIndex() == 0,
- "AlignedSeriesAggregateScanOperator can only process one tsBlock input.");
int valueIndex = inputLocations[0].getValueColumnIndex();
accumulator.addStatistics(statistics[valueIndex]);
}
@@ -141,6 +137,7 @@ public class Aggregator {
}
public void updateTimeRange(TimeRange curTimeRange) {
+ reset();
this.curTimeRange = curTimeRange;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/EmptyQueueSlidingWindowAggregator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/EmptyQueueSlidingWindowAggregator.java
index 52b009e922..a3a0a1b735 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/EmptyQueueSlidingWindowAggregator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/EmptyQueueSlidingWindowAggregator.java
@@ -50,7 +50,6 @@ public class EmptyQueueSlidingWindowAggregator extends SlidingWindowAggregator {
@Override
public void processPartialResult(PartialAggregationResult partialResult) {
if (!partialResult.isNull()) {
- this.accumulator.reset();
this.accumulator.addIntermediate(partialResult.getPartialResult());
lastTime = partialResult.getTime();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/MonotonicQueueSlidingWindowAggregator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/MonotonicQueueSlidingWindowAggregator.java
index 7706edfb2c..35eeb8f2e1 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/MonotonicQueueSlidingWindowAggregator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/MonotonicQueueSlidingWindowAggregator.java
@@ -68,9 +68,6 @@ public class MonotonicQueueSlidingWindowAggregator extends SlidingWindowAggregat
deque.removeLast();
}
deque.addLast(partialResult);
- this.accumulator.reset();
- if (!deque.isEmpty()) {
- this.accumulator.addIntermediate(deque.getFirst().getPartialResult());
- }
+ this.accumulator.addIntermediate(deque.getFirst().getPartialResult());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/NormalQueueSlidingWindowAggregator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/NormalQueueSlidingWindowAggregator.java
index eecaa8cffe..d89981a831 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/NormalQueueSlidingWindowAggregator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/NormalQueueSlidingWindowAggregator.java
@@ -53,7 +53,6 @@ public class NormalQueueSlidingWindowAggregator extends SlidingWindowAggregator
if (!partialResult.isNull()) {
deque.addLast(partialResult);
}
- this.accumulator.reset();
if (!deque.isEmpty()) {
this.accumulator.addIntermediate(deque.getFirst().getPartialResult());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/SlidingWindowAggregator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/SlidingWindowAggregator.java
index aed468c8d8..ec848b354d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/SlidingWindowAggregator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/slidingwindow/SlidingWindowAggregator.java
@@ -66,7 +66,7 @@ public abstract class SlidingWindowAggregator extends Aggregator {
valueColumn[i] = tsBlock.getColumn(inputLocation.getValueColumnIndex());
}
processPartialResult(new PartialAggregationResult(timeColumn, valueColumn));
- return tsBlock.getPositionCount();
+ return 1;
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/AggregationUtil.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/AggregationUtil.java
new file mode 100644
index 0000000000..5465800553
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/AggregationUtil.java
@@ -0,0 +1,161 @@
+/*
+ * 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.mpp.execution.operator;
+
+import org.apache.iotdb.db.mpp.aggregation.Aggregator;
+import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.ITimeRangeIterator;
+import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.SingleTimeWindowIterator;
+import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.TimeRangeIteratorFactory;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
+import org.apache.iotdb.tsfile.read.common.TimeRange;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import java.util.List;
+
+import static org.apache.iotdb.tsfile.read.common.block.TsBlockUtil.skipPointsOutOfTimeRange;
+
+public class AggregationUtil {
+
+ private AggregationUtil() {
+ // forbidding instantiation
+ }
+
+ /**
+ * If groupByTimeParameter is null, which means it's an aggregation query without down sampling.
+ * Aggregation query has only one time window and the result set of it does not contain a
+ * timestamp, so it doesn't matter what the time range returns.
+ */
+ public static ITimeRangeIterator initTimeRangeIterator(
+ GroupByTimeParameter groupByTimeParameter,
+ boolean ascending,
+ boolean outputPartialTimeWindow) {
+ if (groupByTimeParameter == null) {
+ return new SingleTimeWindowIterator(0, Long.MAX_VALUE);
+ } else {
+ return TimeRangeIteratorFactory.getTimeRangeIterator(
+ groupByTimeParameter.getStartTime(),
+ groupByTimeParameter.getEndTime(),
+ groupByTimeParameter.getInterval(),
+ groupByTimeParameter.getSlidingStep(),
+ ascending,
+ groupByTimeParameter.isIntervalByMonth(),
+ groupByTimeParameter.isSlidingStepByMonth(),
+ groupByTimeParameter.isLeftCRightO(),
+ outputPartialTimeWindow);
+ }
+ }
+
+ /**
+ * Calculate aggregation value on the time range from the tsBlock containing raw data.
+ *
+ * @return left - whether the aggregation calculation of the current time range has done; right -
+ * remaining tsBlock
+ */
+ public static Pair<Boolean, TsBlock> calculateAggregationFromRawData(
+ TsBlock inputTsBlock,
+ List<Aggregator> aggregators,
+ TimeRange curTimeRange,
+ boolean ascending) {
+ if (inputTsBlock == null || inputTsBlock.isEmpty()) {
+ return new Pair<>(false, inputTsBlock);
+ }
+
+ // check if the tsBlock does not contain points in current interval
+ if (satisfiedTimeRange(inputTsBlock, curTimeRange, ascending)) {
+ // skip points that cannot be calculated
+ if ((ascending && inputTsBlock.getStartTime() < curTimeRange.getMin())
+ || (!ascending && inputTsBlock.getStartTime() > curTimeRange.getMax())) {
+ inputTsBlock = skipPointsOutOfTimeRange(inputTsBlock, curTimeRange, ascending);
+ }
+
+ int lastReadRowIndex = 0;
+ for (Aggregator aggregator : aggregators) {
+ // current agg method has been calculated
+ if (aggregator.hasFinalResult()) {
+ continue;
+ }
+
+ lastReadRowIndex = Math.max(lastReadRowIndex, aggregator.processTsBlock(inputTsBlock));
+ }
+ if (lastReadRowIndex >= inputTsBlock.getPositionCount()) {
+ inputTsBlock = null;
+ } else {
+ inputTsBlock = inputTsBlock.subTsBlock(lastReadRowIndex);
+ }
+ }
+
+ // judge whether the calculation finished
+ boolean isTsBlockOutOfBound =
+ inputTsBlock != null
+ && (ascending
+ ? inputTsBlock.getEndTime() > curTimeRange.getMax()
+ : inputTsBlock.getEndTime() < curTimeRange.getMin());
+ return new Pair<>(
+ isAllAggregatorsHasFinalResult(aggregators) || isTsBlockOutOfBound, inputTsBlock);
+ }
+
+ /** Append a row of aggregation results to the result tsBlock. */
+ public static void appendAggregationResult(
+ TsBlockBuilder tsBlockBuilder,
+ List<? extends Aggregator> aggregators,
+ ITimeRangeIterator timeRangeIterator) {
+ TimeColumnBuilder timeColumnBuilder = tsBlockBuilder.getTimeColumnBuilder();
+ // Use start time of current time range as time column
+ timeColumnBuilder.writeLong(timeRangeIterator.currentOutputTime());
+ ColumnBuilder[] columnBuilders = tsBlockBuilder.getValueColumnBuilders();
+ int columnIndex = 0;
+ for (Aggregator aggregator : aggregators) {
+ ColumnBuilder[] columnBuilder = new ColumnBuilder[aggregator.getOutputType().length];
+ columnBuilder[0] = columnBuilders[columnIndex++];
+ if (columnBuilder.length > 1) {
+ columnBuilder[1] = columnBuilders[columnIndex++];
+ }
+ aggregator.outputResult(columnBuilder);
+ }
+ tsBlockBuilder.declarePosition();
+ }
+
+ /** @return whether the tsBlock contains the data of the current time window */
+ public static boolean satisfiedTimeRange(
+ TsBlock tsBlock, TimeRange curTimeRange, boolean ascending) {
+ if (tsBlock == null || tsBlock.isEmpty()) {
+ return false;
+ }
+
+ return ascending
+ ? (tsBlock.getEndTime() >= curTimeRange.getMin()
+ && tsBlock.getStartTime() <= curTimeRange.getMax())
+ : (tsBlock.getEndTime() <= curTimeRange.getMax()
+ && tsBlock.getStartTime() >= curTimeRange.getMin());
+ }
+
+ public static boolean isAllAggregatorsHasFinalResult(List<Aggregator> aggregators) {
+ for (Aggregator aggregator : aggregators) {
+ if (!aggregator.hasFinalResult()) {
+ return false;
+ }
+ }
+ return true;
+ }
+}
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 0fb63897a3..356e0c4eed 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,10 @@ package org.apache.iotdb.db.mpp.execution.operator;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import io.airlift.units.Duration;
+
+import java.util.Objects;
+
/**
* Contains information about {@link Operator} execution.
*
@@ -33,6 +37,8 @@ public class OperatorContext {
private final String operatorType;
private final FragmentInstanceContext instanceContext;
+ private Duration maxRunTime;
+
public OperatorContext(
int operatorId,
PlanNodeId planNodeId,
@@ -48,7 +54,36 @@ public class OperatorContext {
return operatorId;
}
+ public String getOperatorType() {
+ return operatorType;
+ }
+
public FragmentInstanceContext getInstanceContext() {
return instanceContext;
}
+
+ public Duration getMaxRunTime() {
+ return maxRunTime;
+ }
+
+ public void setMaxRunTime(Duration maxRunTime) {
+ this.maxRunTime = maxRunTime;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ OperatorContext that = (OperatorContext) o;
+ return operatorId == that.operatorId;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(operatorId);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java
index e4de130719..76b81c2873 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java
@@ -27,36 +27,40 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.TimeRange;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
-import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
-import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
import com.google.common.util.concurrent.ListenableFuture;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.TimeUnit;
-import static org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregationScanOperator.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.appendAggregationResult;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.initTimeRangeIterator;
/**
* AggregationOperator can process the situation: aggregation of intermediate aggregate result, it
- * will output one result based on time interval. One intermediate tsBlock input will only contain
- * the result of one time interval exactly.
+ * will output one tsBlock contain many results on aggregation time intervals. One intermediate
+ * tsBlock input will contain the result of many time intervals.
*/
public class AggregationOperator implements ProcessOperator {
private final OperatorContext operatorContext;
- private final List<Aggregator> aggregators;
- private final List<Operator> children;
+ private final List<Operator> children;
private final int inputOperatorsCount;
private final TsBlock[] inputTsBlocks;
- private final TsBlockBuilder tsBlockBuilder;
+ private final boolean[] canCallNext;
private final ITimeRangeIterator timeRangeIterator;
// current interval of aggregation window [curStartTime, curEndTime)
private TimeRange curTimeRange;
+ private final List<Aggregator> aggregators;
+
+ // using for building result tsBlock
+ private final TsBlockBuilder resultTsBlockBuilder;
+
public AggregationOperator(
OperatorContext operatorContext,
List<Aggregator> aggregators,
@@ -65,18 +69,21 @@ public class AggregationOperator implements ProcessOperator {
GroupByTimeParameter groupByTimeParameter,
boolean outputPartialTimeWindow) {
this.operatorContext = operatorContext;
- this.aggregators = aggregators;
this.children = children;
+ this.aggregators = aggregators;
this.inputOperatorsCount = children.size();
this.inputTsBlocks = new TsBlock[inputOperatorsCount];
+ this.canCallNext = new boolean[inputOperatorsCount];
+
+ this.timeRangeIterator =
+ initTimeRangeIterator(groupByTimeParameter, ascending, outputPartialTimeWindow);
+
List<TSDataType> dataTypes = new ArrayList<>();
for (Aggregator aggregator : aggregators) {
dataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
}
- tsBlockBuilder = new TsBlockBuilder(dataTypes);
- this.timeRangeIterator =
- initTimeRangeIterator(groupByTimeParameter, ascending, outputPartialTimeWindow);
+ this.resultTsBlockBuilder = new TsBlockBuilder(dataTypes);
}
@Override
@@ -95,37 +102,55 @@ public class AggregationOperator implements ProcessOperator {
return NOT_BLOCKED;
}
+ @Override
+ public boolean hasNext() {
+ return curTimeRange != null || timeRangeIterator.hasNextTimeRange();
+ }
+
@Override
public TsBlock next() {
- // update input tsBlock
- if (curTimeRange == null && timeRangeIterator.hasNextTimeRange()) {
- curTimeRange = timeRangeIterator.nextTimeRange();
- }
+ // start stopwatch
+ long maxRuntime = operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
+ long start = System.nanoTime();
+
+ // reset operator state
+ resultTsBlockBuilder.reset();
for (int i = 0; i < inputOperatorsCount; i++) {
- if (inputTsBlocks[i] != null) {
- continue;
+ canCallNext[i] = true;
+ }
+
+ while (System.nanoTime() - start < maxRuntime
+ && (curTimeRange != null || timeRangeIterator.hasNextTimeRange())
+ && !resultTsBlockBuilder.isFull()) {
+ boolean hasCachedData = prepareInput();
+ if (!hasCachedData) {
+ break;
}
- inputTsBlocks[i] = children.get(i).next();
- if (inputTsBlocks[i] == null) {
- return null;
+
+ if (curTimeRange == null && timeRangeIterator.hasNextTimeRange()) {
+ // 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();
}
- // consume current input tsBlocks
- for (Aggregator aggregator : aggregators) {
- aggregator.reset();
- aggregator.processTsBlocks(inputTsBlocks);
- }
- // output result from aggregator
- curTimeRange = null;
- for (int i = 0; i < inputOperatorsCount; i++) {
- inputTsBlocks[i] = null;
+
+ if (resultTsBlockBuilder.getPositionCount() > 0) {
+ return resultTsBlockBuilder.build();
+ } else {
+ return null;
}
- return updateResultTsBlockFromAggregators(tsBlockBuilder, aggregators, timeRangeIterator);
}
@Override
- public boolean hasNext() {
- return curTimeRange != null || timeRangeIterator.hasNextTimeRange();
+ public boolean isFinished() {
+ return !this.hasNext();
}
@Override
@@ -135,30 +160,43 @@ public class AggregationOperator implements ProcessOperator {
}
}
- @Override
- public boolean isFinished() {
- return !this.hasNext();
+ private boolean prepareInput() {
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ if (inputTsBlocks[i] != null) {
+ continue;
+ }
+ if (!canCallNext[i]) {
+ return false;
+ }
+
+ inputTsBlocks[i] = children.get(i).next();
+ canCallNext[i] = false;
+ if (inputTsBlocks[i] == null) {
+ return false;
+ }
+ }
+ return true;
}
- public static TsBlock updateResultTsBlockFromAggregators(
- TsBlockBuilder tsBlockBuilder,
- List<? extends Aggregator> aggregators,
- ITimeRangeIterator timeRangeIterator) {
- tsBlockBuilder.reset();
- TimeColumnBuilder timeColumnBuilder = tsBlockBuilder.getTimeColumnBuilder();
- // Use start time of current time range as time column
- timeColumnBuilder.writeLong(timeRangeIterator.currentOutputTime());
- ColumnBuilder[] columnBuilders = tsBlockBuilder.getValueColumnBuilders();
- int columnIndex = 0;
+ private void calculateNextAggregationResult() {
+ // consume current input tsBlocks
for (Aggregator aggregator : aggregators) {
- ColumnBuilder[] columnBuilder = new ColumnBuilder[aggregator.getOutputType().length];
- columnBuilder[0] = columnBuilders[columnIndex++];
- if (columnBuilder.length > 1) {
- columnBuilder[1] = columnBuilders[columnIndex++];
+ aggregator.processTsBlocks(inputTsBlocks);
+ }
+
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ inputTsBlocks[i] = inputTsBlocks[i].skipFirst();
+ if (inputTsBlocks[i].isEmpty()) {
+ inputTsBlocks[i] = null;
}
- aggregator.outputResult(columnBuilder);
}
- tsBlockBuilder.declarePosition();
- return tsBlockBuilder.build();
+
+ // update result using aggregators
+ updateResultTsBlock();
+ }
+
+ private void updateResultTsBlock() {
+ curTimeRange = null;
+ appendAggregationResult(resultTsBlockBuilder, aggregators, timeRangeIterator);
}
}
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 01b0625c60..88fb5fe20a 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
@@ -20,24 +20,15 @@
package org.apache.iotdb.db.mpp.execution.operator.process;
import org.apache.iotdb.db.mpp.aggregation.Aggregator;
-import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.ITimeRangeIterator;
import org.apache.iotdb.db.mpp.execution.operator.Operator;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.read.common.TimeRange;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
-import org.apache.iotdb.tsfile.read.common.block.TsBlock.TsBlockSingleColumnIterator;
-import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
-import org.apache.iotdb.tsfile.read.common.block.column.TimeColumn;
+import org.apache.iotdb.tsfile.utils.Pair;
-import com.google.common.util.concurrent.ListenableFuture;
-
-import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
-import static org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregationScanOperator.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.calculateAggregationFromRawData;
/**
* RawDataAggregationOperator is used to process raw data tsBlock input calculating using value
@@ -47,22 +38,9 @@ import static org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregatio
* <p>Since raw data query with value filter is processed by FilterOperator above TimeJoinOperator,
* there we can see RawDataAggregateOperator as a one-to-one(one input, ont output) operator.
*
- * <p>Return aggregation result in one time interval once.
+ * <p>Return aggregation result in many time intervals once.
*/
-public class RawDataAggregationOperator implements ProcessOperator {
-
- private final OperatorContext operatorContext;
- private final List<Aggregator> aggregators;
- private final Operator child;
- private final boolean ascending;
- private ITimeRangeIterator timeRangeIterator;
- // current interval of aggregation window [curStartTime, curEndTime)
- private TimeRange curTimeRange;
-
- private TsBlock preCachedData;
-
- // Using for building result tsBlock
- private final TsBlockBuilder tsBlockBuilder;
+public class RawDataAggregationOperator extends SingleInputAggregationOperator {
public RawDataAggregationOperator(
OperatorContext operatorContext,
@@ -70,161 +48,36 @@ public class RawDataAggregationOperator implements ProcessOperator {
Operator child,
boolean ascending,
GroupByTimeParameter groupByTimeParameter) {
- this.operatorContext = operatorContext;
- this.aggregators = aggregators;
- this.child = child;
- this.ascending = ascending;
-
- List<TSDataType> dataTypes = new ArrayList<>();
- for (Aggregator aggregator : aggregators) {
- dataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
- }
- tsBlockBuilder = new TsBlockBuilder(dataTypes);
- this.timeRangeIterator = initTimeRangeIterator(groupByTimeParameter, ascending, true);
- }
-
- @Override
- public OperatorContext getOperatorContext() {
- return operatorContext;
- }
-
- @Override
- public ListenableFuture<?> isBlocked() {
- return child.isBlocked();
+ super(operatorContext, aggregators, child, ascending, groupByTimeParameter, true);
}
@Override
- public TsBlock next() {
- // Move to next timeRange
- if (curTimeRange == null && timeRangeIterator.hasNextTimeRange()) {
- curTimeRange = timeRangeIterator.nextTimeRange();
- for (Aggregator aggregator : aggregators) {
- aggregator.reset();
- aggregator.updateTimeRange(curTimeRange);
- }
- }
+ protected boolean calculateNextAggregationResult() {
+ while (!calcFromRawData()) {
+ inputTsBlock = null;
- // 1. Calculate aggregation result based on current time window
- boolean canCallNext = true;
- while (!calcFromCacheData(curTimeRange)) {
- preCachedData = null;
- // child.next can only be invoked once
+ // NOTE: child.next() can only be invoked once
if (child.hasNext() && canCallNext) {
- preCachedData = child.next();
+ inputTsBlock = child.next();
canCallNext = false;
- // if child still has next but can't be invoked now
} else if (child.hasNext()) {
- return null;
+ // if child still has next but can't be invoked now
+ return false;
} else {
break;
}
}
- // 2. Update result using aggregators
- curTimeRange = null;
- return AggregationOperator.updateResultTsBlockFromAggregators(
- tsBlockBuilder, aggregators, timeRangeIterator);
- }
-
- @Override
- public boolean hasNext() {
- return curTimeRange != null || timeRangeIterator.hasNextTimeRange();
- }
-
- @Override
- public void close() throws Exception {
- child.close();
- }
-
- @Override
- public boolean isFinished() {
- return !this.hasNext();
- }
-
- /** @return if already get the result */
- private boolean calcFromCacheData(TimeRange curTimeRange) {
- if (preCachedData == null || preCachedData.isEmpty()) {
- return false;
- }
- // check if the batchData does not contain points in current interval
- if (satisfied(preCachedData, curTimeRange, ascending)) {
- // skip points that cannot be calculated
- if ((ascending && preCachedData.getStartTime() < curTimeRange.getMin())
- || (!ascending && preCachedData.getStartTime() > curTimeRange.getMax())) {
- preCachedData = skipOutOfTimeRangePoints(preCachedData, curTimeRange, ascending);
- }
-
- int lastReadRowIndex = 0;
- for (Aggregator aggregator : aggregators) {
- // current agg method has been calculated
- if (aggregator.hasFinalResult()) {
- continue;
- }
-
- lastReadRowIndex = Math.max(lastReadRowIndex, aggregator.processTsBlock(preCachedData));
- }
- if (lastReadRowIndex >= preCachedData.getPositionCount()) {
- preCachedData = null;
- } else {
- preCachedData = preCachedData.subTsBlock(lastReadRowIndex);
- }
- }
- // The result is calculated from the cache
- return (preCachedData != null
- && (ascending
- ? preCachedData.getEndTime() > curTimeRange.getMax()
- : preCachedData.getEndTime() < curTimeRange.getMin()))
- || isEndCalc(aggregators);
- }
-
- // skip points that cannot be calculated
- public static TsBlock skipOutOfTimeRangePoints(
- TsBlock tsBlock, TimeRange curTimeRange, boolean ascending) {
- TimeColumn timeColumn = tsBlock.getTimeColumn();
- long targetTime = ascending ? curTimeRange.getMin() : curTimeRange.getMax();
- int left = 0, right = timeColumn.getPositionCount() - 1, mid;
- // if ascending, find the first greater than or equal to targetTime
- // else, find the first less than or equal to targetTime
- while (left < right) {
- mid = (left + right) >> 1;
- if (timeColumn.getLongWithoutCheck(mid) < targetTime) {
- if (ascending) {
- left = mid + 1;
- } else {
- right = mid;
- }
- } else if (timeColumn.getLongWithoutCheck(mid) > targetTime) {
- if (ascending) {
- right = mid;
- } else {
- left = mid + 1;
- }
- } else if (timeColumn.getLongWithoutCheck(mid) == targetTime) {
- return tsBlock.subTsBlock(mid);
- }
- }
- return tsBlock.subTsBlock(left);
- }
-
- public static boolean satisfied(TsBlock tsBlock, TimeRange timeRange, boolean ascending) {
- TsBlockSingleColumnIterator tsBlockIterator = tsBlock.getTsBlockSingleColumnIterator();
- if (tsBlockIterator == null || !tsBlockIterator.hasNext()) {
- return false;
- }
+ // update result using aggregators
+ updateResultTsBlock();
- return ascending
- ? (tsBlockIterator.getEndTime() >= timeRange.getMin()
- && tsBlockIterator.currentTime() <= timeRange.getMax())
- : (tsBlockIterator.getEndTime() <= timeRange.getMax()
- && tsBlockIterator.currentTime() >= timeRange.getMin());
+ return true;
}
- public static boolean isEndCalc(List<Aggregator> aggregators) {
- for (Aggregator aggregator : aggregators) {
- if (!aggregator.hasFinalResult()) {
- return false;
- }
- }
- return true;
+ private boolean calcFromRawData() {
+ Pair<Boolean, TsBlock> calcResult =
+ calculateAggregationFromRawData(inputTsBlock, aggregators, curTimeRange, ascending);
+ inputTsBlock = calcResult.getRight();
+ return calcResult.getLeft();
}
}
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
new file mode 100644
index 0000000000..c59de13988
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SingleInputAggregationOperator.java
@@ -0,0 +1,149 @@
+/*
+ * 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.mpp.execution.operator.process;
+
+import org.apache.iotdb.db.mpp.aggregation.Aggregator;
+import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.ITimeRangeIterator;
+import org.apache.iotdb.db.mpp.execution.operator.Operator;
+import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.TimeRange;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+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.initTimeRangeIterator;
+
+public abstract class SingleInputAggregationOperator implements ProcessOperator {
+
+ protected final OperatorContext operatorContext;
+ protected final boolean ascending;
+
+ protected final Operator child;
+ protected TsBlock inputTsBlock;
+ protected boolean canCallNext;
+
+ protected final ITimeRangeIterator timeRangeIterator;
+ // current interval of aggregation window [curStartTime, curEndTime)
+ protected TimeRange curTimeRange;
+
+ protected final List<Aggregator> aggregators;
+
+ // using for building result tsBlock
+ protected final TsBlockBuilder resultTsBlockBuilder;
+
+ public SingleInputAggregationOperator(
+ OperatorContext operatorContext,
+ List<Aggregator> aggregators,
+ Operator child,
+ boolean ascending,
+ GroupByTimeParameter groupByTimeParameter,
+ boolean outputPartialTimeWindow) {
+ this.operatorContext = operatorContext;
+ this.ascending = ascending;
+ this.child = child;
+ this.aggregators = aggregators;
+
+ this.timeRangeIterator =
+ initTimeRangeIterator(groupByTimeParameter, ascending, outputPartialTimeWindow);
+
+ List<TSDataType> dataTypes = new ArrayList<>();
+ for (Aggregator aggregator : aggregators) {
+ dataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
+ }
+ this.resultTsBlockBuilder = new TsBlockBuilder(dataTypes);
+ }
+
+ @Override
+ public OperatorContext getOperatorContext() {
+ return operatorContext;
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return curTimeRange != null || timeRangeIterator.hasNextTimeRange();
+ }
+
+ @Override
+ public TsBlock next() {
+ // start stopwatch
+ long maxRuntime = operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
+ long start = System.nanoTime();
+
+ // reset operator state
+ resultTsBlockBuilder.reset();
+ canCallNext = true;
+
+ while (System.nanoTime() - start < maxRuntime
+ && (curTimeRange != null || timeRangeIterator.hasNextTimeRange())
+ && !resultTsBlockBuilder.isFull()) {
+ if (curTimeRange == null && timeRangeIterator.hasNextTimeRange()) {
+ // 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
+ if (!calculateNextAggregationResult()) {
+ break;
+ }
+ }
+
+ if (resultTsBlockBuilder.getPositionCount() > 0) {
+ return resultTsBlockBuilder.build();
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public boolean isFinished() {
+ return !this.hasNext();
+ }
+
+ @Override
+ public void close() throws Exception {
+ child.close();
+ }
+
+ protected abstract boolean calculateNextAggregationResult();
+
+ protected void updateResultTsBlock() {
+ curTimeRange = null;
+ appendAggregationResult(resultTsBlockBuilder, aggregators, timeRangeIterator);
+ }
+}
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 2685e3e465..10303016c5 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
@@ -20,149 +20,88 @@
package org.apache.iotdb.db.mpp.execution.operator.process;
import org.apache.iotdb.db.mpp.aggregation.Aggregator;
-import org.apache.iotdb.db.mpp.aggregation.slidingwindow.SlidingWindowAggregator;
import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.ITimeRangeIterator;
import org.apache.iotdb.db.mpp.execution.operator.Operator;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.TimeRange;
-import org.apache.iotdb.tsfile.read.common.block.TsBlock;
-import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
-import com.google.common.util.concurrent.ListenableFuture;
-
-import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import static com.google.common.base.Preconditions.checkArgument;
-import static org.apache.iotdb.db.mpp.execution.operator.process.AggregationOperator.updateResultTsBlockFromAggregators;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.satisfied;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.skipOutOfTimeRangePoints;
-import static org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregationScanOperator.initTimeRangeIterator;
-
-public class SlidingWindowAggregationOperator implements ProcessOperator {
-
- private final OperatorContext operatorContext;
- private final Operator child;
-
- private TsBlock cachedTsBlock;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.initTimeRangeIterator;
- private final List<SlidingWindowAggregator> aggregators;
+public class SlidingWindowAggregationOperator extends SingleInputAggregationOperator {
- private final ITimeRangeIterator timeRangeIterator;
- // current interval of aggregation window [curStartTime, curEndTime)
- private TimeRange curTimeRange;
-
- private final boolean ascending;
-
- private final TsBlockBuilder tsBlockBuilder;
+ private final ITimeRangeIterator subTimeRangeIterator;
+ // current interval of pre-aggregation window [curStartTime, curEndTime)
+ private TimeRange curSubTimeRange;
public SlidingWindowAggregationOperator(
OperatorContext operatorContext,
- List<SlidingWindowAggregator> aggregators,
+ List<Aggregator> aggregators,
Operator child,
boolean ascending,
GroupByTimeParameter groupByTimeParameter) {
+ super(operatorContext, aggregators, child, ascending, groupByTimeParameter, false);
checkArgument(
groupByTimeParameter != null,
"GroupByTimeParameter cannot be null in SlidingWindowAggregationOperator");
-
- this.operatorContext = operatorContext;
- this.aggregators = aggregators;
- this.child = child;
- List<TSDataType> outputDataTypes = new ArrayList<>();
- for (Aggregator aggregator : aggregators) {
- outputDataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
- }
- this.tsBlockBuilder = new TsBlockBuilder(outputDataTypes);
- this.timeRangeIterator = initTimeRangeIterator(groupByTimeParameter, ascending, false);
- this.ascending = ascending;
- }
-
- @Override
- public boolean hasNext() {
- return curTimeRange != null || timeRangeIterator.hasNextTimeRange();
+ this.subTimeRangeIterator = initTimeRangeIterator(groupByTimeParameter, ascending, true);
}
@Override
- public TsBlock next() {
- // Move to next timeRange
- if (curTimeRange == null && timeRangeIterator.hasNextTimeRange()) {
- curTimeRange = timeRangeIterator.nextTimeRange();
- for (Aggregator aggregator : aggregators) {
- aggregator.updateTimeRange(curTimeRange);
+ protected boolean calculateNextAggregationResult() {
+ while (!isCalculationDone()) {
+ if (inputTsBlock == null) {
+ // NOTE: child.next() can only be invoked once
+ if (child.hasNext() && canCallNext) {
+ inputTsBlock = child.next();
+ canCallNext = false;
+ } else if (child.hasNext()) {
+ // if child still has next but can't be invoked now
+ return false;
+ } else {
+ break;
+ }
}
- }
- // 1. Calculate aggregation result based on current time window
- boolean canCallNext = true;
- while (!calcFromTsBlock(cachedTsBlock, curTimeRange)) {
- cachedTsBlock = null;
- // child.next can only be invoked once
- if (child.hasNext() && canCallNext) {
- cachedTsBlock = child.next();
- canCallNext = false;
- // if child still has next but can't be invoked now
- } else if (child.hasNext()) {
- return null;
- } else {
- break;
- }
+ calculateFromCachedData();
}
- // 2. Update result using aggregators
- curTimeRange = null;
- return updateResultTsBlockFromAggregators(tsBlockBuilder, aggregators, timeRangeIterator);
+ // update result using aggregators
+ updateResultTsBlock();
+
+ return true;
}
- private boolean calcFromTsBlock(TsBlock inputTsBlock, TimeRange timeRange) {
- if (inputTsBlock == null || inputTsBlock.isEmpty()) {
- return false;
+ /** @return if already get the result */
+ private boolean isCalculationDone() {
+ if (curSubTimeRange == null && !subTimeRangeIterator.hasNextTimeRange()) {
+ return true;
}
- // check if the batchData does not contain points in current interval
- if (satisfied(inputTsBlock, timeRange, ascending)) {
- // skip points that cannot be calculated
- if ((ascending && inputTsBlock.getStartTime() < timeRange.getMin())
- || (!ascending && inputTsBlock.getStartTime() > timeRange.getMax())) {
- inputTsBlock = skipOutOfTimeRangePoints(inputTsBlock, timeRange, ascending);
- }
- int lastReadRowIndex = 0;
- for (SlidingWindowAggregator aggregator : aggregators) {
- lastReadRowIndex = Math.max(lastReadRowIndex, aggregator.processTsBlock(inputTsBlock));
- }
- if (lastReadRowIndex >= inputTsBlock.getPositionCount()) {
- inputTsBlock = null;
- } else {
- inputTsBlock = inputTsBlock.subTsBlock(lastReadRowIndex);
- }
+ if (curSubTimeRange == null && subTimeRangeIterator.hasNextTimeRange()) {
+ curSubTimeRange = subTimeRangeIterator.nextTimeRange();
}
- // The result is calculated from the cache
- return inputTsBlock != null
- && (ascending
- ? inputTsBlock.getEndTime() > timeRange.getMax()
- : inputTsBlock.getEndTime() < timeRange.getMin());
- }
-
- @Override
- public OperatorContext getOperatorContext() {
- return operatorContext;
+ return ascending
+ ? curSubTimeRange.getMin() > curTimeRange.getMax()
+ : curSubTimeRange.getMax() < curTimeRange.getMin();
}
- @Override
- public ListenableFuture<?> isBlocked() {
- return child.isBlocked();
- }
+ private void calculateFromCachedData() {
+ if (inputTsBlock == null || inputTsBlock.isEmpty()) {
+ return;
+ }
- @Override
- public boolean isFinished() {
- return !this.hasNext();
- }
+ for (Aggregator aggregator : aggregators) {
+ aggregator.processTsBlock(inputTsBlock);
+ }
- @Override
- public void close() throws Exception {
- child.close();
+ inputTsBlock = inputTsBlock.skipFirst();
+ if (inputTsBlock.isEmpty()) {
+ inputTsBlock = null;
+ }
+ curSubTimeRange = null;
}
}
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
new file mode 100644
index 0000000000..2d7d671be9
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AbstractSeriesAggregationScanOperator.java
@@ -0,0 +1,350 @@
+/*
+ * 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.mpp.execution.operator.source;
+
+import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
+import org.apache.iotdb.db.mpp.aggregation.Aggregator;
+import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.ITimeRangeIterator;
+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.plan.planner.plan.parameter.GroupByTimeParameter;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
+import org.apache.iotdb.tsfile.read.common.TimeRange;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+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.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.isAllAggregatorsHasFinalResult;
+
+public abstract class AbstractSeriesAggregationScanOperator implements DataSourceOperator {
+
+ protected final PlanNodeId sourceId;
+ protected final OperatorContext operatorContext;
+ protected final boolean ascending;
+ protected final boolean isGroupByQuery;
+
+ protected SeriesScanUtil seriesScanUtil;
+ protected int subSensorSize;
+
+ protected TsBlock inputTsBlock;
+
+ protected final ITimeRangeIterator timeRangeIterator;
+ // current interval of aggregation window [curStartTime, curEndTime)
+ protected TimeRange curTimeRange;
+
+ // We still think aggregator in SeriesAggregateScanOperator is a inputRaw step.
+ // But in facing of statistics, it will invoke another method processStatistics()
+ protected final List<Aggregator> aggregators;
+
+ // using for building result tsBlock
+ protected final TsBlockBuilder resultTsBlockBuilder;
+
+ protected boolean finished = false;
+
+ public AbstractSeriesAggregationScanOperator(
+ PlanNodeId sourceId,
+ OperatorContext context,
+ SeriesScanUtil seriesScanUtil,
+ int subSensorSize,
+ List<Aggregator> aggregators,
+ boolean ascending,
+ GroupByTimeParameter groupByTimeParameter) {
+ this.sourceId = sourceId;
+ this.operatorContext = context;
+ this.ascending = ascending;
+ this.isGroupByQuery = groupByTimeParameter != null;
+ this.seriesScanUtil = seriesScanUtil;
+ this.subSensorSize = subSensorSize;
+ this.aggregators = aggregators;
+
+ this.timeRangeIterator = initTimeRangeIterator(groupByTimeParameter, ascending, true);
+
+ List<TSDataType> dataTypes = new ArrayList<>();
+ for (Aggregator aggregator : aggregators) {
+ dataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
+ }
+ this.resultTsBlockBuilder = new TsBlockBuilder(dataTypes);
+ }
+
+ @Override
+ public PlanNodeId getSourceId() {
+ return sourceId;
+ }
+
+ @Override
+ public OperatorContext getOperatorContext() {
+ return operatorContext;
+ }
+
+ @Override
+ public void initQueryDataSource(QueryDataSource dataSource) {
+ seriesScanUtil.initQueryDataSource(dataSource);
+ }
+
+ @Override
+ public boolean hasNext() {
+ return timeRangeIterator.hasNextTimeRange();
+ }
+
+ @Override
+ public TsBlock next() {
+ // start stopwatch
+ long maxRuntime = operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
+ long start = System.nanoTime();
+
+ // reset operator state
+ resultTsBlockBuilder.reset();
+
+ 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();
+ }
+
+ if (resultTsBlockBuilder.getPositionCount() > 0) {
+ return resultTsBlockBuilder.build();
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public boolean isFinished() {
+ return finished || (finished = !hasNext());
+ }
+
+ protected void calculateNextAggregationResult() {
+ try {
+ if (calcFromCachedData()) {
+ updateResultTsBlock();
+ return;
+ }
+
+ // read page data firstly
+ if (readAndCalcFromPage()) {
+ updateResultTsBlock();
+ return;
+ }
+
+ // read chunk data secondly
+ if (readAndCalcFromChunk()) {
+ updateResultTsBlock();
+ return;
+ }
+
+ // read from file
+ if (readAndCalcFromFile()) {
+ updateResultTsBlock();
+ return;
+ }
+
+ updateResultTsBlock();
+ } catch (IOException e) {
+ throw new RuntimeException("Error while scanning the file", e);
+ }
+ }
+
+ protected void updateResultTsBlock() {
+ appendAggregationResult(resultTsBlockBuilder, aggregators, timeRangeIterator);
+ }
+
+ protected boolean calcFromCachedData() {
+ return calcFromRawData(inputTsBlock);
+ }
+
+ private boolean calcFromRawData(TsBlock tsBlock) {
+ Pair<Boolean, TsBlock> calcResult =
+ calculateAggregationFromRawData(tsBlock, aggregators, curTimeRange, ascending);
+ inputTsBlock = calcResult.getRight();
+ return calcResult.getLeft();
+ }
+
+ protected void calcFromStatistics(Statistics[] statistics) {
+ for (Aggregator aggregator : aggregators) {
+ if (aggregator.hasFinalResult()) {
+ continue;
+ }
+ aggregator.processStatistics(statistics);
+ }
+ }
+
+ protected boolean readAndCalcFromFile() throws IOException {
+ while (seriesScanUtil.hasNextFile()) {
+ if (canUseCurrentFileStatistics()) {
+ Statistics fileTimeStatistics = seriesScanUtil.currentFileTimeStatistics();
+ if (fileTimeStatistics.getStartTime() > curTimeRange.getMax()) {
+ if (ascending) {
+ return true;
+ } else {
+ seriesScanUtil.skipCurrentFile();
+ continue;
+ }
+ }
+ // calc from fileMetaData
+ if (curTimeRange.contains(
+ fileTimeStatistics.getStartTime(), fileTimeStatistics.getEndTime())) {
+ Statistics[] statisticsList = new Statistics[subSensorSize];
+ for (int i = 0; i < subSensorSize; i++) {
+ statisticsList[i] = seriesScanUtil.currentFileStatistics(i);
+ }
+ calcFromStatistics(statisticsList);
+ seriesScanUtil.skipCurrentFile();
+ if (isAllAggregatorsHasFinalResult(aggregators) && !isGroupByQuery) {
+ return true;
+ } else {
+ continue;
+ }
+ }
+ }
+
+ // read chunk
+ if (readAndCalcFromChunk()) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ protected boolean readAndCalcFromChunk() throws IOException {
+ while (seriesScanUtil.hasNextChunk()) {
+ if (canUseCurrentChunkStatistics()) {
+ Statistics chunkTimeStatistics = seriesScanUtil.currentChunkTimeStatistics();
+ if (chunkTimeStatistics.getStartTime() > curTimeRange.getMax()) {
+ if (ascending) {
+ return true;
+ } else {
+ seriesScanUtil.skipCurrentChunk();
+ continue;
+ }
+ }
+ // calc from chunkMetaData
+ if (curTimeRange.contains(
+ chunkTimeStatistics.getStartTime(), chunkTimeStatistics.getEndTime())) {
+ // calc from chunkMetaData
+ Statistics[] statisticsList = new Statistics[subSensorSize];
+ for (int i = 0; i < subSensorSize; i++) {
+ statisticsList[i] = seriesScanUtil.currentChunkStatistics(i);
+ }
+ calcFromStatistics(statisticsList);
+ seriesScanUtil.skipCurrentChunk();
+ if (isAllAggregatorsHasFinalResult(aggregators) && !isGroupByQuery) {
+ return true;
+ } else {
+ continue;
+ }
+ }
+ }
+
+ // read page
+ if (readAndCalcFromPage()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ 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);
+ }
+ calcFromStatistics(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 raw data
+ if (calcFromRawData(tsBlock)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ protected boolean canUseCurrentFileStatistics() throws IOException {
+ Statistics fileStatistics = seriesScanUtil.currentFileTimeStatistics();
+ return !seriesScanUtil.isFileOverlapped()
+ && fileStatistics.containedByTimeFilter(seriesScanUtil.getTimeFilter())
+ && !seriesScanUtil.currentFileModified();
+ }
+
+ protected boolean canUseCurrentChunkStatistics() throws IOException {
+ Statistics chunkStatistics = seriesScanUtil.currentChunkTimeStatistics();
+ return !seriesScanUtil.isChunkOverlapped()
+ && chunkStatistics.containedByTimeFilter(seriesScanUtil.getTimeFilter())
+ && !seriesScanUtil.currentChunkModified();
+ }
+
+ protected boolean canUseCurrentPageStatistics() throws IOException {
+ Statistics currentPageStatistics = seriesScanUtil.currentPageTimeStatistics();
+ if (currentPageStatistics == null) {
+ return false;
+ }
+ return !seriesScanUtil.isPageOverlapped()
+ && currentPageStatistics.containedByTimeFilter(seriesScanUtil.getTimeFilter())
+ && !seriesScanUtil.currentPageModified();
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesAggregationScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesAggregationScanOperator.java
index 375b203976..f9fa1f8d52 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesAggregationScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesAggregationScanOperator.java
@@ -19,55 +19,18 @@
package org.apache.iotdb.db.mpp.execution.operator.source;
-import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.metadata.path.AlignedPath;
import org.apache.iotdb.db.mpp.aggregation.Aggregator;
-import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.ITimeRangeIterator;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
-import org.apache.iotdb.db.mpp.execution.operator.process.AggregationOperator;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
-import org.apache.iotdb.tsfile.read.common.TimeRange;
-import org.apache.iotdb.tsfile.read.common.block.TsBlock;
-import org.apache.iotdb.tsfile.read.common.block.TsBlock.TsBlockSingleColumnIterator;
-import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.isEndCalc;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.skipOutOfTimeRangePoints;
-import static org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregationScanOperator.initTimeRangeIterator;
-
/** This operator is responsible to do the aggregation calculation especially for aligned series. */
-public class AlignedSeriesAggregationScanOperator implements DataSourceOperator {
-
- private final OperatorContext operatorContext;
- private final PlanNodeId sourceId;
- private final AlignedSeriesScanUtil alignedSeriesScanUtil;
- private final int subSensorSize;
- private final boolean ascending;
- // We still think aggregator in AlignedSeriesAggregateScanOperator is a inputRaw step.
- // But in facing of statistics, it will invoke another method processStatistics()
- private List<Aggregator> aggregators;
-
- private ITimeRangeIterator timeRangeIterator;
- // current interval of aggregation window [curStartTime, curEndTime)
- private TimeRange curTimeRange;
- private boolean isGroupByQuery;
-
- private TsBlock preCachedData;
-
- private TsBlockBuilder tsBlockBuilder;
- private TsBlock resultTsBlock;
- private boolean hasCachedTsBlock = false;
- private boolean finished = false;
+public class AlignedSeriesAggregationScanOperator extends AbstractSeriesAggregationScanOperator {
public AlignedSeriesAggregationScanOperator(
PlanNodeId sourceId,
@@ -77,341 +40,19 @@ public class AlignedSeriesAggregationScanOperator implements DataSourceOperator
Filter timeFilter,
boolean ascending,
GroupByTimeParameter groupByTimeParameter) {
- this.sourceId = sourceId;
- this.operatorContext = context;
- this.ascending = ascending;
- this.alignedSeriesScanUtil =
+ super(
+ sourceId,
+ context,
new AlignedSeriesScanUtil(
seriesPath,
new HashSet<>(seriesPath.getMeasurementList()),
context.getInstanceContext(),
timeFilter,
null,
- ascending);
- this.subSensorSize = seriesPath.getMeasurementList().size();
- this.aggregators = aggregators;
- List<TSDataType> dataTypes = new ArrayList<>();
- for (Aggregator aggregator : aggregators) {
- dataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
- }
- tsBlockBuilder = new TsBlockBuilder(dataTypes);
- this.timeRangeIterator = initTimeRangeIterator(groupByTimeParameter, ascending, true);
- this.isGroupByQuery = groupByTimeParameter != null;
- }
-
- @Override
- public OperatorContext getOperatorContext() {
- return operatorContext;
- }
-
- @Override
- public TsBlock next() {
- if (hasCachedTsBlock || hasNext()) {
- hasCachedTsBlock = false;
- return resultTsBlock;
- }
- return null;
- }
-
- @Override
- public boolean hasNext() {
- if (hasCachedTsBlock) {
- return true;
- }
- try {
- if (!timeRangeIterator.hasNextTimeRange()) {
- return false;
- }
- curTimeRange = timeRangeIterator.nextTimeRange();
-
- // 1. Clear previous aggregation result
- for (Aggregator aggregator : aggregators) {
- aggregator.reset();
- aggregator.updateTimeRange(curTimeRange);
- }
-
- // 2. Calculate aggregation result based on current time window
- if (calcFromCacheData(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
-
- // read page data firstly
- if (readAndCalcFromPage(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
-
- // read chunk data secondly
- if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
-
- // read from file first
- while (alignedSeriesScanUtil.hasNextFile()) {
- if (canUseCurrentFileStatistics()) {
- Statistics fileTimeStatistics = alignedSeriesScanUtil.currentFileTimeStatistics();
- if (fileTimeStatistics.getStartTime() > curTimeRange.getMax()) {
- if (ascending) {
- updateResultTsBlockFromAggregators();
- return true;
- } else {
- alignedSeriesScanUtil.skipCurrentFile();
- continue;
- }
- }
- // calc from fileMetaData
- if (curTimeRange.contains(
- fileTimeStatistics.getStartTime(), fileTimeStatistics.getEndTime())) {
- Statistics[] statisticsList = new Statistics[subSensorSize];
- for (int i = 0; i < subSensorSize; i++) {
- statisticsList[i] = alignedSeriesScanUtil.currentFileStatistics(i);
- }
- calcFromStatistics(statisticsList);
- alignedSeriesScanUtil.skipCurrentFile();
- if (isEndCalc(aggregators) && !isGroupByQuery) {
- break;
- } else {
- continue;
- }
- }
- }
-
- // read chunk
- if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
- }
-
- updateResultTsBlockFromAggregators();
- return true;
- } catch (IOException e) {
- throw new RuntimeException("Error while scanning the file", e);
- }
- }
-
- @Override
- public boolean isFinished() {
- return finished || (finished = !hasNext());
- }
-
- @Override
- public void initQueryDataSource(QueryDataSource dataSource) {
- alignedSeriesScanUtil.initQueryDataSource(dataSource);
- }
-
- @Override
- public PlanNodeId getSourceId() {
- return sourceId;
- }
-
- private void updateResultTsBlockFromAggregators() {
- resultTsBlock =
- AggregationOperator.updateResultTsBlockFromAggregators(
- tsBlockBuilder, aggregators, timeRangeIterator);
- hasCachedTsBlock = true;
- }
-
- /** @return if already get the result */
- private boolean calcFromCacheData(TimeRange curTimeRange) throws IOException {
- calcFromBatch(preCachedData, curTimeRange);
- // The result is calculated from the cache
- return (preCachedData != null
- && (ascending
- ? preCachedData.getEndTime() > curTimeRange.getMax()
- : preCachedData.getEndTime() < curTimeRange.getMin()))
- || isEndCalc(aggregators);
- }
-
- @SuppressWarnings("squid:S3776")
- private void calcFromBatch(TsBlock tsBlock, TimeRange curTimeRange) {
- // check if the batchData does not contain points in current interval
- if (tsBlock != null && satisfied(tsBlock, curTimeRange, ascending)) {
- // skip points that cannot be calculated
- if ((ascending && tsBlock.getStartTime() < curTimeRange.getMin())
- || (!ascending && tsBlock.getStartTime() > curTimeRange.getMax())) {
- tsBlock = skipOutOfTimeRangePoints(tsBlock, curTimeRange, ascending);
- }
-
- int lastReadRowIndex = 0;
- for (Aggregator aggregator : aggregators) {
- // current agg method has been calculated
- if (aggregator.hasFinalResult()) {
- continue;
- }
-
- lastReadRowIndex = Math.max(lastReadRowIndex, aggregator.processTsBlock(tsBlock));
- }
- if (lastReadRowIndex >= tsBlock.getPositionCount()) {
- tsBlock = null;
- } else {
- tsBlock = tsBlock.subTsBlock(lastReadRowIndex);
- }
-
- // can calc for next interval
- if (tsBlock != null && tsBlock.getTsBlockSingleColumnIterator().hasNext()) {
- preCachedData = tsBlock;
- }
- }
- }
-
- private boolean satisfied(TsBlock tsBlock, TimeRange timeRange, boolean ascending) {
- TsBlockSingleColumnIterator tsBlockIterator = tsBlock.getTsBlockSingleColumnIterator();
- if (tsBlockIterator == null || !tsBlockIterator.hasNext()) {
- return false;
- }
-
- if (ascending
- && (tsBlockIterator.getEndTime() < timeRange.getMin()
- || tsBlockIterator.currentTime() > timeRange.getMax())) {
- return false;
- }
- if (!ascending
- && (tsBlockIterator.getEndTime() > timeRange.getMax()
- || tsBlockIterator.currentTime() < timeRange.getMin())) {
- preCachedData = tsBlock;
- return false;
- }
- return true;
- }
-
- @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
- private boolean readAndCalcFromPage(TimeRange curTimeRange) throws IOException {
- while (alignedSeriesScanUtil.hasNextPage()) {
- if (canUseCurrentPageStatistics()) {
- Statistics pageTimeStatistics = alignedSeriesScanUtil.currentPageTimeStatistics();
- // There is no more eligible points in current time range
- if (pageTimeStatistics.getStartTime() > curTimeRange.getMax()) {
- if (ascending) {
- return true;
- } else {
- alignedSeriesScanUtil.skipCurrentPage();
- continue;
- }
- }
- // can use pageHeader
- if (canUseCurrentPageStatistics()
- && curTimeRange.contains(
- pageTimeStatistics.getStartTime(), pageTimeStatistics.getEndTime())) {
- Statistics[] statisticsList = new Statistics[subSensorSize];
- for (int i = 0; i < subSensorSize; i++) {
- statisticsList[i] = alignedSeriesScanUtil.currentPageStatistics(i);
- }
- calcFromStatistics(statisticsList);
- alignedSeriesScanUtil.skipCurrentPage();
- if (isEndCalc(aggregators) && !isGroupByQuery) {
- return true;
- } else {
- continue;
- }
- }
- }
-
- // calc from page data
- TsBlock tsBlock = alignedSeriesScanUtil.nextPage();
- TsBlockSingleColumnIterator tsBlockIterator = tsBlock.getTsBlockSingleColumnIterator();
- if (tsBlockIterator == null || !tsBlockIterator.hasNext()) {
- continue;
- }
-
- // stop calc and cached current batchData
- if (ascending && tsBlockIterator.currentTime() > curTimeRange.getMax()) {
- preCachedData = tsBlock;
- return true;
- }
-
- // calc from batch data
- calcFromBatch(tsBlock, curTimeRange);
-
- // judge whether the calculation finished
- boolean isTsBlockOutOfBound =
- ascending
- ? tsBlock.getEndTime() > curTimeRange.getMax()
- : tsBlock.getEndTime() < curTimeRange.getMin();
- if (isEndCalc(aggregators) || isTsBlockOutOfBound) {
- return true;
- }
- }
- return false;
- }
-
- private boolean readAndCalcFromChunk(TimeRange curTimeRange) throws IOException {
- while (alignedSeriesScanUtil.hasNextChunk()) {
- if (canUseCurrentChunkStatistics()) {
- Statistics chunkTimeStatistics = alignedSeriesScanUtil.currentChunkTimeStatistics();
- if (chunkTimeStatistics.getStartTime() > curTimeRange.getMax()) {
- if (ascending) {
- return true;
- } else {
- alignedSeriesScanUtil.skipCurrentChunk();
- continue;
- }
- }
- // calc from chunkMetaData
- if (curTimeRange.contains(
- chunkTimeStatistics.getStartTime(), chunkTimeStatistics.getEndTime())) {
- // calc from chunkMetaData
- Statistics[] statisticsList = new Statistics[subSensorSize];
- for (int i = 0; i < subSensorSize; i++) {
- statisticsList[i] = alignedSeriesScanUtil.currentChunkStatistics(i);
- }
- calcFromStatistics(statisticsList);
- alignedSeriesScanUtil.skipCurrentChunk();
- if (isEndCalc(aggregators) && !isGroupByQuery) {
- return true;
- } else {
- continue;
- }
- }
- }
-
- // read page
- if (readAndCalcFromPage(curTimeRange)) {
- return true;
- }
- }
- return false;
- }
-
- private void calcFromStatistics(Statistics[] statistics) {
- for (int i = 0; i < aggregators.size(); i++) {
- Aggregator aggregator = aggregators.get(i);
- if (aggregator.hasFinalResult()) {
- continue;
- }
- aggregator.processStatistics(statistics);
- }
- }
-
- public boolean canUseCurrentFileStatistics() throws IOException {
- Statistics fileStatistics = alignedSeriesScanUtil.currentFileTimeStatistics();
- return !alignedSeriesScanUtil.isFileOverlapped()
- && containedByTimeFilter(fileStatistics)
- && !alignedSeriesScanUtil.currentFileModified();
- }
-
- public boolean canUseCurrentChunkStatistics() throws IOException {
- Statistics chunkStatistics = alignedSeriesScanUtil.currentChunkTimeStatistics();
- return !alignedSeriesScanUtil.isChunkOverlapped()
- && containedByTimeFilter(chunkStatistics)
- && !alignedSeriesScanUtil.currentChunkModified();
- }
-
- public boolean canUseCurrentPageStatistics() throws IOException {
- Statistics currentPageStatistics = alignedSeriesScanUtil.currentPageTimeStatistics();
- if (currentPageStatistics == null) {
- return false;
- }
- return !alignedSeriesScanUtil.isPageOverlapped()
- && containedByTimeFilter(currentPageStatistics)
- && !alignedSeriesScanUtil.currentPageModified();
- }
-
- private boolean containedByTimeFilter(Statistics statistics) {
- Filter timeFilter = alignedSeriesScanUtil.getTimeFilter();
- return timeFilter == null
- || timeFilter.containStartEndTime(statistics.getStartTime(), statistics.getEndTime());
+ ascending),
+ seriesPath.getMeasurementList().size(),
+ aggregators,
+ ascending,
+ groupByTimeParameter);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanUtil.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanUtil.java
index 5b9b3079e4..27d8ab3d9c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanUtil.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanUtil.java
@@ -28,8 +28,10 @@ import org.apache.iotdb.db.query.reader.universal.AlignedPriorityMergeReader;
import org.apache.iotdb.db.query.reader.universal.DescPriorityMergeReader;
import org.apache.iotdb.db.query.reader.universal.PriorityMergeReader;
import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.file.metadata.AlignedChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.AlignedTimeSeriesMetadata;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.reader.IPointReader;
@@ -57,6 +59,42 @@ public class AlignedSeriesScanUtil extends SeriesScanUtil {
.getSchemaList().stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
}
+ @Override
+ protected Statistics currentFileStatistics(int index) throws IOException {
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentFileTimeStatistics() throws IOException {
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getTimeStatistics();
+ }
+
+ @Override
+ protected Statistics currentChunkStatistics(int index) throws IOException {
+ return ((AlignedChunkMetadata) firstChunkMetadata).getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentChunkTimeStatistics() {
+ return ((AlignedChunkMetadata) firstChunkMetadata).getTimeStatistics();
+ }
+
+ @Override
+ protected Statistics currentPageStatistics(int index) throws IOException {
+ if (firstPageReader == null) {
+ return null;
+ }
+ return firstPageReader.getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentPageTimeStatistics() throws IOException {
+ if (firstPageReader == null) {
+ return null;
+ }
+ return firstPageReader.getTimeStatistics();
+ }
+
@Override
protected PriorityMergeReader getPriorityMergeReader() {
return new AlignedPriorityMergeReader();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregationScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregationScanOperator.java
index 56b1a33b4a..157f51f2fa 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregationScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregationScanOperator.java
@@ -19,61 +19,24 @@
package org.apache.iotdb.db.mpp.execution.operator.source;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.mpp.aggregation.Aggregator;
-import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.ITimeRangeIterator;
-import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.SingleTimeWindowIterator;
-import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.TimeRangeIteratorFactory;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
-import org.apache.iotdb.db.mpp.execution.operator.process.AggregationOperator;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
-import org.apache.iotdb.tsfile.read.common.TimeRange;
-import org.apache.iotdb.tsfile.read.common.block.TsBlock;
-import org.apache.iotdb.tsfile.read.common.block.TsBlock.TsBlockSingleColumnIterator;
-import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.Set;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.isEndCalc;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.skipOutOfTimeRangePoints;
-
/**
* This operator is responsible to do the aggregation calculation for one series based on global
* time range and time split parameter.
*
- * <p>Every time next() is invoked, one tsBlock which contains current time window will be returned.
+ * <p>Every time next() is invoked, one tsBlock which contains many time windows will be returned.
* In sliding window situation, current time window is a pre-aggregation window. If there is no time
* split parameter, i.e. aggregation without groupBy, just one tsBlock will be returned.
*/
-public class SeriesAggregationScanOperator implements DataSourceOperator {
-
- private final OperatorContext operatorContext;
- private final PlanNodeId sourceId;
- private final SeriesScanUtil seriesScanUtil;
- private final boolean ascending;
- // We still think aggregator in SeriesAggregateScanOperator is a inputRaw step.
- // But in facing of statistics, it will invoke another method processStatistics()
- private List<Aggregator> aggregators;
-
- private ITimeRangeIterator timeRangeIterator;
- // current interval of aggregation window [curStartTime, curEndTime)
- private TimeRange curTimeRange;
- private boolean isGroupByQuery;
-
- private TsBlock preCachedData;
-
- private TsBlockBuilder tsBlockBuilder;
- private TsBlock resultTsBlock;
- private boolean hasCachedTsBlock = false;
- private boolean finished = false;
+public class SeriesAggregationScanOperator extends AbstractSeriesAggregationScanOperator {
public SeriesAggregationScanOperator(
PlanNodeId sourceId,
@@ -84,10 +47,9 @@ public class SeriesAggregationScanOperator implements DataSourceOperator {
Filter timeFilter,
boolean ascending,
GroupByTimeParameter groupByTimeParameter) {
- this.sourceId = sourceId;
- this.operatorContext = context;
- this.ascending = ascending;
- this.seriesScanUtil =
+ super(
+ sourceId,
+ context,
new SeriesScanUtil(
seriesPath,
allSensors,
@@ -95,338 +57,10 @@ public class SeriesAggregationScanOperator implements DataSourceOperator {
context.getInstanceContext(),
timeFilter,
null,
- ascending);
- this.aggregators = aggregators;
- List<TSDataType> dataTypes = new ArrayList<>();
- for (Aggregator aggregator : aggregators) {
- dataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
- }
- tsBlockBuilder = new TsBlockBuilder(dataTypes);
- this.timeRangeIterator = initTimeRangeIterator(groupByTimeParameter, ascending, true);
- this.isGroupByQuery = groupByTimeParameter != null;
- }
-
- /**
- * If groupByTimeParameter is null, which means it's an aggregation query without down sampling.
- * Aggregation query has only one time window and the result set of it does not contain a
- * timestamp, so it doesn't matter what the time range returns.
- */
- public static ITimeRangeIterator initTimeRangeIterator(
- GroupByTimeParameter groupByTimeParameter,
- boolean ascending,
- boolean outputPartialTimeWindow) {
- if (groupByTimeParameter == null) {
- return new SingleTimeWindowIterator(0, Long.MAX_VALUE);
- } else {
- return TimeRangeIteratorFactory.getTimeRangeIterator(
- groupByTimeParameter.getStartTime(),
- groupByTimeParameter.getEndTime(),
- groupByTimeParameter.getInterval(),
- groupByTimeParameter.getSlidingStep(),
- ascending,
- groupByTimeParameter.isIntervalByMonth(),
- groupByTimeParameter.isSlidingStepByMonth(),
- groupByTimeParameter.isLeftCRightO(),
- outputPartialTimeWindow);
- }
- }
-
- @Override
- public OperatorContext getOperatorContext() {
- return operatorContext;
- }
-
- @Override
- public TsBlock next() {
- if (hasCachedTsBlock || hasNext()) {
- hasCachedTsBlock = false;
- return resultTsBlock;
- }
- return null;
- }
-
- @Override
- public boolean hasNext() {
- if (hasCachedTsBlock) {
- return true;
- }
- try {
- if (!timeRangeIterator.hasNextTimeRange()) {
- return false;
- }
- curTimeRange = timeRangeIterator.nextTimeRange();
-
- // 1. Clear previous aggregation result
- for (Aggregator aggregator : aggregators) {
- aggregator.reset();
- aggregator.updateTimeRange(curTimeRange);
- }
-
- // 2. Calculate aggregation result based on current time window
- if (calcFromCacheData(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
-
- // read page data firstly
- if (readAndCalcFromPage(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
-
- // read chunk data secondly
- if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
-
- // read from file first
- while (seriesScanUtil.hasNextFile()) {
- if (canUseCurrentFileStatistics()) {
- Statistics fileStatistics = seriesScanUtil.currentFileStatistics();
- if (fileStatistics.getStartTime() > curTimeRange.getMax()) {
- if (ascending) {
- updateResultTsBlockFromAggregators();
- return true;
- } else {
- seriesScanUtil.skipCurrentFile();
- continue;
- }
- }
- // calc from fileMetaData
- if (curTimeRange.contains(fileStatistics.getStartTime(), fileStatistics.getEndTime())) {
- calcFromStatistics(fileStatistics);
- seriesScanUtil.skipCurrentFile();
- if (isEndCalc(aggregators) && !isGroupByQuery) {
- break;
- } else {
- continue;
- }
- }
- }
-
- // read chunk
- if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
- }
-
- updateResultTsBlockFromAggregators();
- return true;
- } catch (IOException e) {
- throw new RuntimeException("Error while scanning the file", e);
- }
- }
-
- private void updateResultTsBlockFromAggregators() {
- resultTsBlock =
- AggregationOperator.updateResultTsBlockFromAggregators(
- tsBlockBuilder, aggregators, timeRangeIterator);
- hasCachedTsBlock = true;
- }
-
- @Override
- public boolean isFinished() {
- return finished || (finished = !hasNext());
- }
-
- @Override
- public PlanNodeId getSourceId() {
- return sourceId;
- }
-
- @Override
- public void initQueryDataSource(QueryDataSource dataSource) {
- seriesScanUtil.initQueryDataSource(dataSource);
- }
-
- /** @return if already get the result */
- private boolean calcFromCacheData(TimeRange curTimeRange) throws IOException {
- calcFromBatch(preCachedData, curTimeRange);
- // The result is calculated from the cache
- return (preCachedData != null
- && (ascending
- ? preCachedData.getEndTime() > curTimeRange.getMax()
- : preCachedData.getEndTime() < curTimeRange.getMin()))
- || isEndCalc(aggregators);
- }
-
- @SuppressWarnings("squid:S3776")
- private void calcFromBatch(TsBlock tsBlock, TimeRange curTimeRange) {
- // check if the batchData does not contain points in current interval
- if (tsBlock != null && satisfied(tsBlock, curTimeRange, ascending)) {
- // skip points that cannot be calculated
- if ((ascending && tsBlock.getStartTime() < curTimeRange.getMin())
- || (!ascending && tsBlock.getStartTime() > curTimeRange.getMax())) {
- tsBlock = skipOutOfTimeRangePoints(tsBlock, curTimeRange, ascending);
- }
-
- int lastReadRowIndex = 0;
- for (Aggregator aggregator : aggregators) {
- // current agg method has been calculated
- if (aggregator.hasFinalResult()) {
- continue;
- }
-
- lastReadRowIndex = Math.max(lastReadRowIndex, aggregator.processTsBlock(tsBlock));
- }
- if (lastReadRowIndex >= tsBlock.getPositionCount()) {
- tsBlock = null;
- } else {
- tsBlock = tsBlock.subTsBlock(lastReadRowIndex);
- }
-
- // can calc for next interval
- if (tsBlock != null && tsBlock.getTsBlockSingleColumnIterator().hasNext()) {
- preCachedData = tsBlock;
- }
- }
- }
-
- private boolean satisfied(TsBlock tsBlock, TimeRange timeRange, boolean ascending) {
- TsBlockSingleColumnIterator tsBlockIterator = tsBlock.getTsBlockSingleColumnIterator();
- if (tsBlockIterator == null || !tsBlockIterator.hasNext()) {
- return false;
- }
-
- if (ascending
- && (tsBlockIterator.getEndTime() < timeRange.getMin()
- || tsBlockIterator.currentTime() > timeRange.getMax())) {
- return false;
- }
- if (!ascending
- && (tsBlockIterator.getEndTime() > timeRange.getMax()
- || tsBlockIterator.currentTime() < timeRange.getMin())) {
- preCachedData = tsBlock;
- return false;
- }
- return true;
- }
-
- @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
- private boolean readAndCalcFromPage(TimeRange curTimeRange) throws IOException {
- while (seriesScanUtil.hasNextPage()) {
- if (canUseCurrentPageStatistics()) {
- Statistics pageStatistics = seriesScanUtil.currentPageStatistics();
- // There is no more eligible points in current time range
- if (pageStatistics.getStartTime() > curTimeRange.getMax()) {
- if (ascending) {
- return true;
- } else {
- seriesScanUtil.skipCurrentPage();
- continue;
- }
- }
- // can use pageHeader
- if (curTimeRange.contains(pageStatistics.getStartTime(), pageStatistics.getEndTime())) {
- calcFromStatistics(pageStatistics);
- seriesScanUtil.skipCurrentPage();
- if (isEndCalc(aggregators) && !isGroupByQuery) {
- return true;
- } else {
- continue;
- }
- }
- }
-
- // calc from page data
- TsBlock tsBlock = seriesScanUtil.nextPage();
- TsBlockSingleColumnIterator tsBlockIterator = tsBlock.getTsBlockSingleColumnIterator();
- if (tsBlockIterator == null || !tsBlockIterator.hasNext()) {
- continue;
- }
-
- // stop calc and cached current batchData
- if (ascending && tsBlockIterator.currentTime() > curTimeRange.getMax()) {
- preCachedData = tsBlock;
- return true;
- }
-
- // calc from batch data
- calcFromBatch(tsBlock, curTimeRange);
-
- // judge whether the calculation finished
- boolean isTsBlockOutOfBound =
- ascending
- ? tsBlock.getEndTime() > curTimeRange.getMax()
- : tsBlock.getEndTime() < curTimeRange.getMin();
- if (isEndCalc(aggregators) || isTsBlockOutOfBound) {
- return true;
- }
- }
- return false;
- }
-
- private boolean readAndCalcFromChunk(TimeRange curTimeRange) throws IOException {
- while (seriesScanUtil.hasNextChunk()) {
- if (canUseCurrentChunkStatistics()) {
- Statistics chunkStatistics = seriesScanUtil.currentChunkStatistics();
- if (chunkStatistics.getStartTime() > curTimeRange.getMax()) {
- if (ascending) {
- return true;
- } else {
- seriesScanUtil.skipCurrentChunk();
- continue;
- }
- }
- // calc from chunkMetaData
- if (curTimeRange.contains(chunkStatistics.getStartTime(), chunkStatistics.getEndTime())) {
- calcFromStatistics(chunkStatistics);
- seriesScanUtil.skipCurrentChunk();
- if (isEndCalc(aggregators) && !isGroupByQuery) {
- return true;
- } else {
- continue;
- }
- }
- }
-
- // read page
- if (readAndCalcFromPage(curTimeRange)) {
- return true;
- }
- }
- return false;
- }
-
- private void calcFromStatistics(Statistics statistics) {
- for (int i = 0; i < aggregators.size(); i++) {
- Aggregator aggregator = aggregators.get(i);
- if (aggregator.hasFinalResult()) {
- continue;
- }
- aggregator.processStatistics(statistics);
- }
- }
-
- public boolean canUseCurrentFileStatistics() throws IOException {
- Statistics fileStatistics = seriesScanUtil.currentFileStatistics();
- return !seriesScanUtil.isFileOverlapped()
- && containedByTimeFilter(fileStatistics)
- && !seriesScanUtil.currentFileModified();
- }
-
- public boolean canUseCurrentChunkStatistics() throws IOException {
- Statistics chunkStatistics = seriesScanUtil.currentChunkStatistics();
- return !seriesScanUtil.isChunkOverlapped()
- && containedByTimeFilter(chunkStatistics)
- && !seriesScanUtil.currentChunkModified();
- }
-
- public boolean canUseCurrentPageStatistics() throws IOException {
- Statistics currentPageStatistics = seriesScanUtil.currentPageStatistics();
- if (currentPageStatistics == null) {
- return false;
- }
- return !seriesScanUtil.isPageOverlapped()
- && containedByTimeFilter(currentPageStatistics)
- && !seriesScanUtil.currentPageModified();
- }
-
- private boolean containedByTimeFilter(Statistics statistics) {
- Filter timeFilter = seriesScanUtil.getTimeFilter();
- return timeFilter == null
- || timeFilter.containStartEndTime(statistics.getStartTime(), statistics.getEndTime());
+ ascending),
+ 1,
+ aggregators,
+ ascending,
+ groupByTimeParameter);
}
}
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 279f6be932..483ef03afa 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
@@ -29,8 +29,6 @@ import org.apache.iotdb.db.query.reader.universal.PriorityMergeReader;
import org.apache.iotdb.db.utils.FileLoaderUtils;
import org.apache.iotdb.db.utils.QueryUtils;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
-import org.apache.iotdb.tsfile.file.metadata.AlignedChunkMetadata;
-import org.apache.iotdb.tsfile.file.metadata.AlignedTimeSeriesMetadata;
import org.apache.iotdb.tsfile.file.metadata.IChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.ITimeSeriesMetadata;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -58,6 +56,8 @@ import java.util.Set;
import java.util.function.ToLongFunction;
import java.util.stream.Collectors;
+import static com.google.common.base.Preconditions.checkArgument;
+
public class SeriesScanUtil {
private final FragmentInstanceContext context;
@@ -226,18 +226,13 @@ public class SeriesScanUtil {
return firstTimeSeriesMetadata.getStatistics();
}
- Statistics currentFileStatistics(int index) throws IOException {
- if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
- throw new IOException("Can only get statistics by index from alignedTimeSeriesMetaData");
- }
- return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getStatistics(index);
+ protected Statistics currentFileStatistics(int index) throws IOException {
+ checkArgument(index == 0, "Only one sensor in non-aligned SeriesScanUtil.");
+ return currentFileStatistics();
}
- public Statistics currentFileTimeStatistics() throws IOException {
- if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
- throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
- }
- return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getTimeStatistics();
+ protected Statistics currentFileTimeStatistics() throws IOException {
+ return currentFileStatistics();
}
boolean currentFileModified() throws IOException {
@@ -362,18 +357,13 @@ public class SeriesScanUtil {
return firstChunkMetadata.getStatistics();
}
- Statistics currentChunkStatistics(int index) throws IOException {
- if (!(firstChunkMetadata instanceof AlignedChunkMetadata)) {
- throw new IOException("Can only get statistics by index from vectorChunkMetaData");
- }
- return ((AlignedChunkMetadata) firstChunkMetadata).getStatistics(index);
+ protected Statistics currentChunkStatistics(int index) throws IOException {
+ checkArgument(index == 0, "Only one sensor in non-aligned SeriesScanUtil.");
+ return currentChunkStatistics();
}
- Statistics currentChunkTimeStatistics() throws IOException {
- if (!(firstChunkMetadata instanceof AlignedChunkMetadata)) {
- throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
- }
- return ((AlignedChunkMetadata) firstChunkMetadata).getTimeStatistics();
+ protected Statistics currentChunkTimeStatistics() throws IOException {
+ return currentChunkStatistics();
}
boolean currentChunkModified() throws IOException {
@@ -590,24 +580,13 @@ public class SeriesScanUtil {
return firstPageReader.getStatistics();
}
- Statistics currentPageStatistics(int index) throws IOException {
- if (firstPageReader == null) {
- return null;
- }
- if (!(firstPageReader.isAlignedPageReader())) {
- throw new IOException("Can only get statistics by index from AlignedPageReader");
- }
- return firstPageReader.getStatistics(index);
+ protected Statistics currentPageStatistics(int index) throws IOException {
+ checkArgument(index == 0, "Only one sensor in non-aligned SeriesScanUtil.");
+ return currentPageStatistics();
}
- Statistics currentPageTimeStatistics() throws IOException {
- if (firstPageReader == null) {
- return null;
- }
- if (!(firstPageReader.isAlignedPageReader())) {
- throw new IOException("Can only get statistics of time column from AlignedPageReader");
- }
- return firstPageReader.getTimeStatistics();
+ protected Statistics currentPageTimeStatistics() throws IOException {
+ return currentPageStatistics();
}
boolean currentPageModified() throws IOException {
@@ -1094,7 +1073,7 @@ public class SeriesScanUtil {
return orderUtils;
}
- private class VersionPageReader {
+ protected class VersionPageReader {
protected PriorityMergeReader.MergeReaderPriority version;
protected IPageReader data;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverTaskThread.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverTaskThread.java
index 02a795cff3..f49806ac31 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverTaskThread.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverTaskThread.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.mpp.execution.schedule;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.execution.driver.IDriver;
import org.apache.iotdb.db.mpp.execution.schedule.queue.IndexedBlockingQueue;
import org.apache.iotdb.db.mpp.execution.schedule.task.DriverTask;
@@ -34,7 +35,10 @@ import java.util.concurrent.TimeUnit;
/** the worker thread of {@link DriverTask} */
public class DriverTaskThread extends AbstractDriverThread {
- public static final Duration EXECUTION_TIME_SLICE = new Duration(100, TimeUnit.MILLISECONDS);
+ public static final Duration EXECUTION_TIME_SLICE =
+ new Duration(
+ IoTDBDescriptor.getInstance().getConfig().getDriverTaskExecutionTimeSliceInMs(),
+ TimeUnit.MILLISECONDS);
// As the callback is lightweight enough, there's no need to use another one thread to execute.
private static final Executor listeningExecutor = MoreExecutors.directExecutor();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/timer/ITimeSliceAllocator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/timer/ITimeSliceAllocator.java
new file mode 100644
index 0000000000..d23da3fbc7
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/timer/ITimeSliceAllocator.java
@@ -0,0 +1,29 @@
+/*
+ * 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.mpp.execution.timer;
+
+import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
+
+import io.airlift.units.Duration;
+
+public interface ITimeSliceAllocator {
+
+ Duration getMaxRunTime(OperatorContext operatorContext);
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/timer/RuleBasedTimeSliceAllocator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/timer/RuleBasedTimeSliceAllocator.java
new file mode 100644
index 0000000000..910b2295bc
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/timer/RuleBasedTimeSliceAllocator.java
@@ -0,0 +1,66 @@
+/*
+ * 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.mpp.execution.timer;
+
+import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
+import org.apache.iotdb.db.mpp.execution.schedule.DriverTaskThread;
+
+import io.airlift.units.Duration;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public class RuleBasedTimeSliceAllocator implements ITimeSliceAllocator {
+
+ private final long EXECUTION_TIME_SLICE_IN_MS =
+ DriverTaskThread.EXECUTION_TIME_SLICE.roundTo(TimeUnit.MILLISECONDS);
+
+ private final Map<OperatorContext, Integer> operatorToWeightMap;
+
+ private int totalWeight;
+
+ public RuleBasedTimeSliceAllocator() {
+ this.operatorToWeightMap = new HashMap<>();
+ this.totalWeight = 0;
+ }
+
+ public void recordExecutionWeight(OperatorContext operatorContext, int weight) {
+ checkState(
+ !operatorToWeightMap.containsKey(operatorContext), "Same operator has been weighted");
+ operatorToWeightMap.put(operatorContext, weight);
+ totalWeight += weight;
+ }
+
+ private int getWeight(OperatorContext operatorContext) {
+ checkState(
+ operatorToWeightMap.containsKey(operatorContext), "This operator has not been weighted");
+ return operatorToWeightMap.get(operatorContext);
+ }
+
+ @Override
+ public Duration getMaxRunTime(OperatorContext operatorContext) {
+ return new Duration(
+ (double) EXECUTION_TIME_SLICE_IN_MS * getWeight(operatorContext) / totalWeight,
+ TimeUnit.MILLISECONDS);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanner.java
index 343c328eea..fd2aebef52 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanner.java
@@ -28,7 +28,6 @@ import org.apache.iotdb.db.metadata.path.MeasurementPath;
import org.apache.iotdb.db.metadata.schemaregion.ISchemaRegion;
import org.apache.iotdb.db.mpp.aggregation.AccumulatorFactory;
import org.apache.iotdb.db.mpp.aggregation.Aggregator;
-import org.apache.iotdb.db.mpp.aggregation.slidingwindow.SlidingWindowAggregator;
import org.apache.iotdb.db.mpp.aggregation.slidingwindow.SlidingWindowAggregatorFactory;
import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
import org.apache.iotdb.db.mpp.execution.driver.DataDriver;
@@ -106,6 +105,8 @@ import org.apache.iotdb.db.mpp.execution.operator.source.ExchangeOperator;
import org.apache.iotdb.db.mpp.execution.operator.source.LastCacheScanOperator;
import org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregationScanOperator;
import org.apache.iotdb.db.mpp.execution.operator.source.SeriesScanOperator;
+import org.apache.iotdb.db.mpp.execution.timer.ITimeSliceAllocator;
+import org.apache.iotdb.db.mpp.execution.timer.RuleBasedTimeSliceAllocator;
import org.apache.iotdb.db.mpp.plan.analyze.TypeProvider;
import org.apache.iotdb.db.mpp.plan.expression.leaf.TimeSeriesOperand;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
@@ -182,7 +183,6 @@ import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
import static org.apache.iotdb.db.mpp.execution.operator.LastQueryUtil.satisfyFilter;
import static org.apache.iotdb.db.mpp.plan.constant.DataNodeEndPoints.isSameNode;
-import static org.apache.iotdb.tsfile.file.metadata.enums.TSDataType.BOOLEAN;
/**
* Used to plan a fragment instance. Currently, we simply change it from PlanNode to executable
@@ -218,6 +218,13 @@ public class LocalExecutionPlanner {
Operator root = plan.accept(new Visitor(), context);
+ ITimeSliceAllocator timeSliceAllocator = context.getTimeSliceAllocator();
+ instanceContext
+ .getOperatorContexts()
+ .forEach(
+ operatorContext ->
+ operatorContext.setMaxRunTime(timeSliceAllocator.getMaxRunTime(operatorContext)));
+
DataDriverContext dataDriverContext =
new DataDriverContext(
instanceContext,
@@ -240,6 +247,13 @@ public class LocalExecutionPlanner {
Operator root = plan.accept(new Visitor(), context);
+ ITimeSliceAllocator timeSliceAllocator = context.getTimeSliceAllocator();
+ instanceContext
+ .getOperatorContexts()
+ .forEach(
+ operatorContext ->
+ operatorContext.setMaxRunTime(timeSliceAllocator.getMaxRunTime(operatorContext)));
+
return new SchemaDriver(root, context.getSinkHandle(), schemaDriverContext);
}
@@ -274,7 +288,7 @@ public class LocalExecutionPlanner {
context.addSourceOperator(seriesScanOperator);
context.addPath(seriesPath);
-
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return seriesScanOperator;
}
@@ -300,7 +314,9 @@ public class LocalExecutionPlanner {
context.addSourceOperator(seriesScanOperator);
context.addPath(seriesPath);
-
+ context
+ .getTimeSliceAllocator()
+ .recordExecutionWeight(operatorContext, seriesPath.getColumnNum());
return seriesScanOperator;
}
@@ -350,7 +366,7 @@ public class LocalExecutionPlanner {
context.addSourceOperator(seriesAggregationScanOperator);
context.addPath(seriesPath);
-
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, aggregators.size());
return seriesAggregationScanOperator;
}
@@ -368,6 +384,7 @@ public class LocalExecutionPlanner {
node.getPlanNodeId(),
SchemaQueryOrderByHeatOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new SchemaQueryOrderByHeatOperator(operatorContext, children);
}
@@ -398,6 +415,7 @@ public class LocalExecutionPlanner {
context.getNextOperatorId(),
node.getPlanNodeId(),
TimeSeriesSchemaScanOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new TimeSeriesSchemaScanOperator(
node.getPlanNodeId(),
operatorContext,
@@ -419,6 +437,7 @@ public class LocalExecutionPlanner {
context.getNextOperatorId(),
node.getPlanNodeId(),
DevicesSchemaScanOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new DevicesSchemaScanOperator(
node.getPlanNodeId(),
operatorContext,
@@ -441,6 +460,7 @@ public class LocalExecutionPlanner {
context.getNextOperatorId(),
node.getPlanNodeId(),
SchemaQueryMergeOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new SchemaQueryMergeOperator(node.getPlanNodeId(), operatorContext, children);
}
@@ -454,7 +474,8 @@ public class LocalExecutionPlanner {
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- CountSchemaMergeNode.class.getSimpleName());
+ CountMergeOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new CountMergeOperator(node.getPlanNodeId(), operatorContext, children);
}
@@ -464,7 +485,8 @@ public class LocalExecutionPlanner {
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- CountSchemaMergeNode.class.getSimpleName());
+ DevicesCountOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new DevicesCountOperator(
node.getPlanNodeId(), operatorContext, node.getPath(), node.isPrefixPath());
}
@@ -476,7 +498,8 @@ public class LocalExecutionPlanner {
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- TimeSeriesCountNode.class.getSimpleName());
+ TimeSeriesCountOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new TimeSeriesCountOperator(
node.getPlanNodeId(), operatorContext, node.getPath(), node.isPrefixPath());
}
@@ -488,7 +511,8 @@ public class LocalExecutionPlanner {
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- LevelTimeSeriesCountNode.class.getSimpleName());
+ LevelTimeSeriesCountOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new LevelTimeSeriesCountOperator(
node.getPlanNodeId(),
operatorContext,
@@ -504,7 +528,8 @@ public class LocalExecutionPlanner {
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- NodePathsSchemaScanNode.class.getSimpleName());
+ NodePathsSchemaScanOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new NodePathsSchemaScanOperator(
node.getPlanNodeId(), operatorContext, node.getPrefixPath(), node.getLevel());
}
@@ -513,37 +538,39 @@ public class LocalExecutionPlanner {
public Operator visitNodeManagementMemoryMerge(
NodeManagementMemoryMergeNode node, LocalExecutionPlanContext context) {
Operator child = node.getChild().accept(this, context);
- return new NodeManageMemoryMergeOperator(
+ OperatorContext operatorContext =
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- NodeManageMemoryMergeOperator.class.getSimpleName()),
- node.getData(),
- child);
+ NodeManageMemoryMergeOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+ return new NodeManageMemoryMergeOperator(operatorContext, node.getData(), child);
}
@Override
public Operator visitNodePathConvert(
NodePathsConvertNode node, LocalExecutionPlanContext context) {
Operator child = node.getChild().accept(this, context);
- return new NodePathsConvertOperator(
+ OperatorContext operatorContext =
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- NodeManageMemoryMergeOperator.class.getSimpleName()),
- child);
+ NodePathsConvertOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+ return new NodePathsConvertOperator(operatorContext, child);
}
@Override
public Operator visitNodePathsCount(
NodePathsCountNode node, LocalExecutionPlanContext context) {
Operator child = node.getChild().accept(this, context);
- return new NodePathsCountOperator(
+ OperatorContext operatorContext =
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- NodeManageMemoryMergeOperator.class.getSimpleName()),
- child);
+ NodePathsCountOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+ return new NodePathsCountOperator(operatorContext, child);
}
@Override
@@ -581,7 +608,7 @@ public class LocalExecutionPlanner {
context.addSourceOperator(aggregateScanOperator);
context.addPath(seriesPath);
-
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, aggregators.size());
return aggregateScanOperator;
}
@@ -601,6 +628,8 @@ public class LocalExecutionPlanner {
.map(deviceName -> node.getDeviceToMeasurementIndexesMap().get(deviceName))
.collect(Collectors.toList());
List<TSDataType> outputColumnTypes = getOutputColumnTypes(node, context.getTypeProvider());
+
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new DeviceViewOperator(
operatorContext, node.getDevices(), children, deviceColumnIndex, outputColumnTypes);
}
@@ -631,6 +660,8 @@ public class LocalExecutionPlanner {
break;
}
}
+
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new DeviceMergeOperator(
operatorContext, node.getDevices(), children, dataTypes, selector, timeComparator);
}
@@ -647,32 +678,25 @@ public class LocalExecutionPlanner {
List<TSDataType> inputDataTypes = getOutputColumnTypes(node.getChild(), context.typeProvider);
int inputColumns = inputDataTypes.size();
FillPolicy fillPolicy = descriptor.getFillPolicy();
+ OperatorContext operatorContext =
+ context.instanceContext.addOperatorContext(
+ context.getNextOperatorId(),
+ node.getPlanNodeId(),
+ FillOperator.class.getSimpleName());
switch (fillPolicy) {
case VALUE:
Literal literal = descriptor.getFillValue();
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new FillOperator(
- context.instanceContext.addOperatorContext(
- context.getNextOperatorId(),
- node.getPlanNodeId(),
- FillOperator.class.getSimpleName()),
- getConstantFill(inputColumns, inputDataTypes, literal),
- child);
+ operatorContext, getConstantFill(inputColumns, inputDataTypes, literal), child);
case PREVIOUS:
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new FillOperator(
- context.instanceContext.addOperatorContext(
- context.getNextOperatorId(),
- node.getPlanNodeId(),
- FillOperator.class.getSimpleName()),
- getPreviousFill(inputColumns, inputDataTypes),
- child);
+ operatorContext, getPreviousFill(inputColumns, inputDataTypes), child);
case LINEAR:
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new LinearFillOperator(
- context.instanceContext.addOperatorContext(
- context.getNextOperatorId(),
- node.getPlanNodeId(),
- LinearFillOperator.class.getSimpleName()),
- getLinearFill(inputColumns, inputDataTypes),
- child);
+ operatorContext, getLinearFill(inputColumns, inputDataTypes), child);
default:
throw new IllegalArgumentException("Unknown fill policy: " + fillPolicy);
}
@@ -779,6 +803,8 @@ public class LocalExecutionPlanner {
final List<TSDataType> inputDataTypes = getInputColumnTypes(node, context.getTypeProvider());
final Map<String, List<InputLocation>> inputLocations = makeLayout(node);
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+
try {
return new TransformOperator(
operatorContext,
@@ -806,6 +832,8 @@ public class LocalExecutionPlanner {
final List<TSDataType> inputDataTypes = getInputColumnTypes(node, context.getTypeProvider());
final Map<String, List<InputLocation>> inputLocations = makeLayout(node);
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+
try {
return new FilterOperator(
operatorContext,
@@ -859,6 +887,8 @@ public class LocalExecutionPlanner {
context.getNextOperatorId(),
node.getPlanNodeId(),
AggregationOperator.class.getSimpleName());
+
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, aggregators.size());
return new AggregationOperator(
operatorContext, aggregators, children, ascending, node.getGroupByTimeParameter(), false);
}
@@ -876,7 +906,7 @@ public class LocalExecutionPlanner {
SlidingWindowAggregationOperator.class.getSimpleName());
Operator child = node.getChild().accept(this, context);
boolean ascending = node.getScanOrder() == OrderBy.TIMESTAMP_ASC;
- List<SlidingWindowAggregator> aggregators = new ArrayList<>();
+ List<Aggregator> aggregators = new ArrayList<>();
Map<String, List<InputLocation>> layout = makeLayout(node);
for (AggregationDescriptor descriptor : node.getAggregationDescriptorList()) {
List<InputLocation[]> inputLocationList = calcInputLocationList(descriptor, layout);
@@ -891,6 +921,8 @@ public class LocalExecutionPlanner {
inputLocationList,
descriptor.getStep()));
}
+
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, aggregators.size());
return new SlidingWindowAggregationOperator(
operatorContext, aggregators, child, ascending, node.getGroupByTimeParameter());
}
@@ -898,25 +930,27 @@ public class LocalExecutionPlanner {
@Override
public Operator visitLimit(LimitNode node, LocalExecutionPlanContext context) {
Operator child = node.getChild().accept(this, context);
- return new LimitOperator(
+ OperatorContext operatorContext =
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- LimitOperator.class.getSimpleName()),
- node.getLimit(),
- child);
+ LimitOperator.class.getSimpleName());
+
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+ return new LimitOperator(operatorContext, node.getLimit(), child);
}
@Override
public Operator visitOffset(OffsetNode node, LocalExecutionPlanContext context) {
Operator child = node.getChild().accept(this, context);
- return new OffsetOperator(
+ OperatorContext operatorContext =
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- OffsetOperator.class.getSimpleName()),
- node.getOffset(),
- child);
+ OffsetOperator.class.getSimpleName());
+
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+ return new OffsetOperator(operatorContext, node.getOffset(), child);
}
@Override
@@ -953,6 +987,7 @@ public class LocalExecutionPlanner {
context.getNextOperatorId(),
node.getPlanNodeId(),
RawDataAggregationOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, aggregators.size());
return new RawDataAggregationOperator(
operatorContext,
aggregators,
@@ -965,6 +1000,7 @@ public class LocalExecutionPlanner {
context.getNextOperatorId(),
node.getPlanNodeId(),
AggregationOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, aggregators.size());
return new AggregationOperator(
operatorContext,
aggregators,
@@ -1021,6 +1057,8 @@ public class LocalExecutionPlanner {
List<OutputColumn> outputColumns = generateOutputColumns(node);
List<ColumnMerger> mergers = createColumnMergers(outputColumns, timeComparator);
List<TSDataType> outputColumnTypes = getOutputColumnTypes(node, context.getTypeProvider());
+
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new TimeJoinOperator(
operatorContext,
children,
@@ -1065,6 +1103,8 @@ public class LocalExecutionPlanner {
context.getNextOperatorId(),
node.getPlanNodeId(),
ExchangeOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 0);
+
FragmentInstanceId localInstanceId = context.instanceContext.getId();
FragmentInstanceId remoteInstanceId = node.getUpstreamInstanceId();
@@ -1125,6 +1165,7 @@ public class LocalExecutionPlanner {
context.getNextOperatorId(),
node.getPlanNodeId(),
SchemaFetchMergeOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new SchemaFetchMergeOperator(operatorContext, children);
}
@@ -1136,6 +1177,7 @@ public class LocalExecutionPlanner {
context.getNextOperatorId(),
node.getPlanNodeId(),
SchemaFetchScanOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
return new SchemaFetchScanOperator(
node.getPlanNodeId(),
operatorContext,
@@ -1171,11 +1213,14 @@ public class LocalExecutionPlanner {
LastQueryScanNode node, LocalExecutionPlanContext context, PartialPath fullPath) {
SeriesAggregationScanOperator lastQueryScan = createLastQueryScanOperator(node, context);
- return new UpdateLastCacheOperator(
+ OperatorContext operatorContext =
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- UpdateLastCacheOperator.class.getSimpleName()),
+ UpdateLastCacheOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+ return new UpdateLastCacheOperator(
+ operatorContext,
lastQueryScan,
fullPath,
node.getSeriesPath().getSeriesType(),
@@ -1207,6 +1252,7 @@ public class LocalExecutionPlanner {
null);
context.addSourceOperator(seriesAggregationScanOperator);
context.addPath(seriesPath);
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, aggregators.size());
return seriesAggregationScanOperator;
}
@@ -1240,11 +1286,14 @@ public class LocalExecutionPlanner {
AlignedSeriesAggregationScanOperator lastQueryScan =
createLastQueryScanOperator(node, context);
- return new UpdateLastCacheOperator(
+ OperatorContext operatorContext =
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- UpdateLastCacheOperator.class.getSimpleName()),
+ UpdateLastCacheOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+ return new UpdateLastCacheOperator(
+ operatorContext,
lastQueryScan,
fullPath,
node.getSeriesPath().getSchemaList().get(0).getType(),
@@ -1275,6 +1324,7 @@ public class LocalExecutionPlanner {
null);
context.addSourceOperator(seriesAggregationScanOperator);
context.addPath(seriesPath);
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, aggregators.size());
return seriesAggregationScanOperator;
}
@@ -1306,23 +1356,25 @@ public class LocalExecutionPlanner {
timeValuePair.getValue().getDataType().name());
}
+ OperatorContext operatorContext =
+ context.instanceContext.addOperatorContext(
+ context.getNextOperatorId(),
+ context.firstCachedPlanNodeId,
+ LastCacheScanOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
LastCacheScanOperator operator =
new LastCacheScanOperator(
- context.instanceContext.addOperatorContext(
- context.getNextOperatorId(),
- context.firstCachedPlanNodeId,
- LastCacheScanOperator.class.getSimpleName()),
- context.firstCachedPlanNodeId,
- builder.build());
+ operatorContext, context.firstCachedPlanNodeId, builder.build());
operatorList.add(operator);
}
- return new LastQueryMergeOperator(
+ OperatorContext operatorContext =
context.instanceContext.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- LastQueryMergeOperator.class.getSimpleName()),
- operatorList);
+ LastQueryMergeOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+ return new LastQueryMergeOperator(operatorContext, operatorList);
}
private Map<String, List<InputLocation>> makeLayout(PlanNode node) {
@@ -1397,6 +1449,8 @@ public class LocalExecutionPlanner {
// whether we need to update last cache
private boolean needUpdateLastCache;
+ private final RuleBasedTimeSliceAllocator timeSliceAllocator;
+
public LocalExecutionPlanContext(
TypeProvider typeProvider, FragmentInstanceContext instanceContext) {
this.typeProvider = typeProvider;
@@ -1404,6 +1458,7 @@ public class LocalExecutionPlanner {
this.paths = new ArrayList<>();
this.allSensorsMap = new HashMap<>();
this.sourceOperators = new ArrayList<>();
+ this.timeSliceAllocator = new RuleBasedTimeSliceAllocator();
}
public LocalExecutionPlanContext(FragmentInstanceContext instanceContext) {
@@ -1411,6 +1466,9 @@ public class LocalExecutionPlanner {
this.paths = new ArrayList<>();
this.allSensorsMap = new HashMap<>();
this.sourceOperators = new ArrayList<>();
+
+ // only used in `order by heat`
+ this.timeSliceAllocator = new RuleBasedTimeSliceAllocator();
}
private int getNextOperatorId() {
@@ -1476,5 +1534,9 @@ public class LocalExecutionPlanner {
public TypeProvider getTypeProvider() {
return typeProvider;
}
+
+ public RuleBasedTimeSliceAllocator getTimeSliceAllocator() {
+ return timeSliceAllocator;
+ }
}
}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregationOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregationOperatorTest.java
index 3b940a78a8..a8380c27f1 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregationOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregationOperatorTest.java
@@ -46,6 +46,7 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import io.airlift.units.Duration;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -55,12 +56,15 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
import static org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
import static org.junit.Assert.assertEquals;
public class AggregationOperatorTest {
+ public static Duration TEST_TIME_SLICE = new Duration(500, TimeUnit.MILLISECONDS);
+
private static final String AGGREGATION_OPERATOR_TEST_SG = "root.AggregationOperatorTest";
private final List<String> deviceIds = new ArrayList<>();
private final List<MeasurementSchema> measurementSchemas = new ArrayList<>();
@@ -177,14 +181,17 @@ public class AggregationOperatorTest {
int count = 0;
while (aggregationOperator.hasNext()) {
TsBlock resultTsBlock = aggregationOperator.next();
- assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[0][count], resultTsBlock.getColumn(0).getLong(0));
- assertEquals(result[1][count], resultTsBlock.getColumn(1).getDouble(0), 0.0001);
- assertEquals(result[2][count], resultTsBlock.getColumn(2).getLong(0));
- assertEquals(result[3][count], resultTsBlock.getColumn(3).getLong(0));
- assertEquals(result[4][count], resultTsBlock.getColumn(4).getInt(0));
- assertEquals(result[5][count], resultTsBlock.getColumn(5).getInt(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[0][count], resultTsBlock.getColumn(0).getLong(pos));
+ assertEquals(result[1][count], resultTsBlock.getColumn(1).getDouble(pos), 0.0001);
+ assertEquals(result[2][count], resultTsBlock.getColumn(2).getLong(pos));
+ assertEquals(result[3][count], resultTsBlock.getColumn(3).getLong(pos));
+ assertEquals(result[4][count], resultTsBlock.getColumn(4).getInt(pos));
+ assertEquals(result[5][count], resultTsBlock.getColumn(5).getInt(pos));
+ count++;
+ }
}
assertEquals(4, count);
}
@@ -216,11 +223,14 @@ public class AggregationOperatorTest {
int count = 0;
while (aggregationOperator.hasNext()) {
TsBlock resultTsBlock = aggregationOperator.next();
- assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[0][count], resultTsBlock.getColumn(0).getDouble(0), 0.001);
- assertEquals((int) result[1][count], resultTsBlock.getColumn(1).getInt(0));
- assertEquals((int) result[2][count], resultTsBlock.getColumn(2).getInt(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[0][count], resultTsBlock.getColumn(0).getDouble(pos), 0.001);
+ assertEquals((int) result[1][count], resultTsBlock.getColumn(1).getInt(pos));
+ assertEquals((int) result[2][count], resultTsBlock.getColumn(2).getInt(pos));
+ count++;
+ }
}
assertEquals(4, count);
}
@@ -252,6 +262,12 @@ public class AggregationOperatorTest {
PlanNodeId planNodeId3 = new PlanNodeId("3");
fragmentInstanceContext.addOperatorContext(
3, planNodeId3, AggregationOperator.class.getSimpleName());
+ fragmentInstanceContext
+ .getOperatorContexts()
+ .forEach(
+ operatorContext -> {
+ operatorContext.setMaxRunTime(TEST_TIME_SLICE);
+ });
MeasurementPath measurementPath1 =
new MeasurementPath(AGGREGATION_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AlignedSeriesAggregationScanOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AlignedSeriesAggregationScanOperatorTest.java
index fbd76ab819..2277299102 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AlignedSeriesAggregationScanOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AlignedSeriesAggregationScanOperatorTest.java
@@ -62,6 +62,7 @@ import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;
import static org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationOperatorTest.TEST_TIME_SLICE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -380,11 +381,14 @@ public class AlignedSeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(0));
- for (int i = 0; i < measurementSchemas.size(); i++) {
- assertEquals(result[count], resultTsBlock.getColumn(i).getLong(0));
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
+ for (int i = 0; i < measurementSchemas.size(); i++) {
+ assertEquals(result[count], resultTsBlock.getColumn(i).getLong(pos));
+ }
+ count++;
}
- count++;
}
assertEquals(4, count);
}
@@ -411,11 +415,14 @@ public class AlignedSeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(0));
- for (int i = 0; i < measurementSchemas.size(); i++) {
- assertEquals(result[count], resultTsBlock.getColumn(i).getLong(0));
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
+ for (int i = 0; i < measurementSchemas.size(); i++) {
+ assertEquals(result[count], resultTsBlock.getColumn(i).getLong(pos));
+ }
+ count++;
}
- count++;
}
assertEquals(4, count);
}
@@ -445,12 +452,15 @@ public class AlignedSeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[0][count], resultTsBlock.getColumn(0).getInt(0));
- assertEquals(result[1][count], resultTsBlock.getColumn(1).getInt(0));
- assertEquals(result[2][count], resultTsBlock.getColumn(2).getInt(0));
- assertEquals(result[3][count], resultTsBlock.getColumn(3).getInt(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[0][count], resultTsBlock.getColumn(0).getInt(pos));
+ assertEquals(result[1][count], resultTsBlock.getColumn(1).getInt(pos));
+ assertEquals(result[2][count], resultTsBlock.getColumn(2).getInt(pos));
+ assertEquals(result[3][count], resultTsBlock.getColumn(3).getInt(pos));
+ count++;
+ }
}
assertEquals(4, count);
}
@@ -480,12 +490,15 @@ public class AlignedSeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(100 * (3 - count), resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[0][3 - count], resultTsBlock.getColumn(0).getInt(0));
- assertEquals(result[1][3 - count], resultTsBlock.getColumn(1).getInt(0));
- assertEquals(result[2][3 - count], resultTsBlock.getColumn(2).getInt(0));
- assertEquals(result[3][3 - count], resultTsBlock.getColumn(3).getInt(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(100 * (3 - count), resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[0][3 - count], resultTsBlock.getColumn(0).getInt(pos));
+ assertEquals(result[1][3 - count], resultTsBlock.getColumn(1).getInt(pos));
+ assertEquals(result[2][3 - count], resultTsBlock.getColumn(2).getInt(pos));
+ assertEquals(result[3][3 - count], resultTsBlock.getColumn(3).getInt(pos));
+ count++;
+ }
}
assertEquals(4, count);
}
@@ -505,9 +518,12 @@ public class AlignedSeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(50 * count, resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[count], resultTsBlock.getColumn(0).getLong(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(50 * count, resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[count], resultTsBlock.getColumn(0).getLong(pos));
+ count++;
+ }
}
assertEquals(result.length, count);
}
@@ -528,9 +544,12 @@ public class AlignedSeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[count], resultTsBlock.getColumn(0).getLong(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[count], resultTsBlock.getColumn(0).getLong(pos));
+ count++;
+ }
}
assertEquals(timeColumn.length, count);
}
@@ -561,12 +580,15 @@ public class AlignedSeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[0][count], resultTsBlock.getColumn(0).getInt(0));
- assertEquals(result[1][count], resultTsBlock.getColumn(1).getInt(0));
- assertEquals(result[2][count], resultTsBlock.getColumn(2).getInt(0));
- assertEquals(result[3][count], resultTsBlock.getColumn(3).getInt(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[0][count], resultTsBlock.getColumn(0).getInt(pos));
+ assertEquals(result[1][count], resultTsBlock.getColumn(1).getInt(pos));
+ assertEquals(result[2][count], resultTsBlock.getColumn(2).getInt(pos));
+ assertEquals(result[3][count], resultTsBlock.getColumn(3).getInt(pos));
+ count++;
+ }
}
assertEquals(timeColumn.length, count);
}
@@ -597,6 +619,12 @@ public class AlignedSeriesAggregationScanOperatorTest {
PlanNodeId planNodeId = new PlanNodeId("1");
fragmentInstanceContext.addOperatorContext(
1, planNodeId, SeriesScanOperator.class.getSimpleName());
+ fragmentInstanceContext
+ .getOperatorContexts()
+ .forEach(
+ operatorContext -> {
+ operatorContext.setMaxRunTime(TEST_TIME_SLICE);
+ });
AlignedSeriesAggregationScanOperator seriesAggregationScanOperator =
new AlignedSeriesAggregationScanOperator(
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/LastQueryMergeOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/LastQueryMergeOperatorTest.java
index 03fa1a1584..c0ced72d91 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/LastQueryMergeOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/LastQueryMergeOperatorTest.java
@@ -55,6 +55,7 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import static org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationOperatorTest.TEST_TIME_SLICE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -118,6 +119,13 @@ public class LastQueryMergeOperatorTest {
fragmentInstanceContext.addOperatorContext(
5, planNodeId5, LastQueryMergeOperator.class.getSimpleName());
+ fragmentInstanceContext
+ .getOperatorContexts()
+ .forEach(
+ operatorContext -> {
+ operatorContext.setMaxRunTime(TEST_TIME_SLICE);
+ });
+
SeriesAggregationScanOperator seriesAggregationScanOperator1 =
new SeriesAggregationScanOperator(
planNodeId1,
@@ -233,6 +241,13 @@ public class LastQueryMergeOperatorTest {
fragmentInstanceContext.addOperatorContext(
6, planNodeId6, LastQueryMergeOperator.class.getSimpleName());
+ fragmentInstanceContext
+ .getOperatorContexts()
+ .forEach(
+ operatorContext -> {
+ operatorContext.setMaxRunTime(TEST_TIME_SLICE);
+ });
+
SeriesAggregationScanOperator seriesAggregationScanOperator1 =
new SeriesAggregationScanOperator(
planNodeId1,
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregationOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregationOperatorTest.java
index 46b85b85f7..9cf7c6331f 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregationOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregationOperatorTest.java
@@ -63,6 +63,7 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import static org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationOperatorTest.TEST_TIME_SLICE;
import static org.junit.Assert.assertEquals;
public class RawDataAggregationOperatorTest {
@@ -326,6 +327,13 @@ public class RawDataAggregationOperatorTest {
3, new PlanNodeId("3"), TimeJoinOperator.class.getSimpleName());
fragmentInstanceContext.addOperatorContext(
4, new PlanNodeId("4"), RawDataAggregationOperatorTest.class.getSimpleName());
+ fragmentInstanceContext
+ .getOperatorContexts()
+ .forEach(
+ operatorContext -> {
+ operatorContext.setMaxRunTime(TEST_TIME_SLICE);
+ });
+
SeriesScanOperator seriesScanOperator1 =
new SeriesScanOperator(
planNodeId1,
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SeriesAggregationScanOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SeriesAggregationScanOperatorTest.java
index 7e95c4ac2c..71fd167404 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SeriesAggregationScanOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SeriesAggregationScanOperatorTest.java
@@ -58,6 +58,7 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import static org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationOperatorTest.TEST_TIME_SLICE;
import static org.junit.Assert.assertEquals;
public class SeriesAggregationScanOperatorTest {
@@ -291,9 +292,12 @@ public class SeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[count], resultTsBlock.getColumn(0).getLong(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[count], resultTsBlock.getColumn(0).getLong(pos));
+ count++;
+ }
}
assertEquals(4, count);
}
@@ -312,9 +316,12 @@ public class SeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[count], resultTsBlock.getColumn(0).getLong(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[count], resultTsBlock.getColumn(0).getLong(pos));
+ count++;
+ }
}
assertEquals(4, count);
}
@@ -342,12 +349,15 @@ public class SeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[0][count], resultTsBlock.getColumn(0).getInt(0));
- assertEquals(result[1][count], resultTsBlock.getColumn(1).getInt(0));
- assertEquals(result[2][count], resultTsBlock.getColumn(2).getInt(0));
- assertEquals(result[3][count], resultTsBlock.getColumn(3).getInt(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[0][count], resultTsBlock.getColumn(0).getInt(pos));
+ assertEquals(result[1][count], resultTsBlock.getColumn(1).getInt(pos));
+ assertEquals(result[2][count], resultTsBlock.getColumn(2).getInt(pos));
+ assertEquals(result[3][count], resultTsBlock.getColumn(3).getInt(pos));
+ count++;
+ }
}
assertEquals(4, count);
}
@@ -375,12 +385,15 @@ public class SeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(100 * (3 - count), resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[0][3 - count], resultTsBlock.getColumn(0).getInt(0));
- assertEquals(result[1][3 - count], resultTsBlock.getColumn(1).getInt(0));
- assertEquals(result[2][3 - count], resultTsBlock.getColumn(2).getInt(0));
- assertEquals(result[3][3 - count], resultTsBlock.getColumn(3).getInt(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(100 * (3 - count), resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[0][3 - count], resultTsBlock.getColumn(0).getInt(pos));
+ assertEquals(result[1][3 - count], resultTsBlock.getColumn(1).getInt(pos));
+ assertEquals(result[2][3 - count], resultTsBlock.getColumn(2).getInt(pos));
+ assertEquals(result[3][3 - count], resultTsBlock.getColumn(3).getInt(pos));
+ count++;
+ }
}
assertEquals(4, count);
}
@@ -398,9 +411,12 @@ public class SeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(50 * count, resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[count], resultTsBlock.getColumn(0).getLong(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(50 * count, resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[count], resultTsBlock.getColumn(0).getLong(pos));
+ count++;
+ }
}
assertEquals(result.length, count);
}
@@ -419,9 +435,12 @@ public class SeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[count], resultTsBlock.getColumn(0).getLong(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[count], resultTsBlock.getColumn(0).getLong(pos));
+ count++;
+ }
}
assertEquals(timeColumn.length, count);
}
@@ -450,12 +469,15 @@ public class SeriesAggregationScanOperatorTest {
int count = 0;
while (seriesAggregationScanOperator.hasNext()) {
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
- assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(0));
- assertEquals(result[0][count], resultTsBlock.getColumn(0).getInt(0));
- assertEquals(result[1][count], resultTsBlock.getColumn(1).getInt(0));
- assertEquals(result[2][count], resultTsBlock.getColumn(2).getInt(0));
- assertEquals(result[3][count], resultTsBlock.getColumn(3).getInt(0));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(pos));
+ assertEquals(result[0][count], resultTsBlock.getColumn(0).getInt(pos));
+ assertEquals(result[1][count], resultTsBlock.getColumn(1).getInt(pos));
+ assertEquals(result[2][count], resultTsBlock.getColumn(2).getInt(pos));
+ assertEquals(result[3][count], resultTsBlock.getColumn(3).getInt(pos));
+ count++;
+ }
}
assertEquals(timeColumn.length, count);
}
@@ -479,6 +501,12 @@ public class SeriesAggregationScanOperatorTest {
PlanNodeId planNodeId = new PlanNodeId("1");
fragmentInstanceContext.addOperatorContext(
1, planNodeId, SeriesAggregationScanOperator.class.getSimpleName());
+ fragmentInstanceContext
+ .getOperatorContexts()
+ .forEach(
+ operatorContext -> {
+ operatorContext.setMaxRunTime(TEST_TIME_SLICE);
+ });
SeriesAggregationScanOperator seriesAggregationScanOperator =
new SeriesAggregationScanOperator(
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SlidingWindowAggregationOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SlidingWindowAggregationOperatorTest.java
index b4f849c7b6..81ab99e1ff 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SlidingWindowAggregationOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SlidingWindowAggregationOperatorTest.java
@@ -27,7 +27,6 @@ import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.metadata.path.MeasurementPath;
import org.apache.iotdb.db.mpp.aggregation.AccumulatorFactory;
import org.apache.iotdb.db.mpp.aggregation.Aggregator;
-import org.apache.iotdb.db.mpp.aggregation.slidingwindow.SlidingWindowAggregator;
import org.apache.iotdb.db.mpp.aggregation.slidingwindow.SlidingWindowAggregatorFactory;
import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
import org.apache.iotdb.db.mpp.common.PlanFragmentId;
@@ -61,6 +60,7 @@ import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;
import static org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationOperatorTest.TEST_TIME_SLICE;
public class SlidingWindowAggregationOperatorTest {
@@ -148,8 +148,11 @@ public class SlidingWindowAggregationOperatorTest {
continue;
}
Assert.assertEquals(rootAggregationTypes.size(), resultTsBlock.getValueColumnCount());
- Assert.assertEquals(retArray[count], getResultString(resultTsBlock));
- count++;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ Assert.assertEquals(retArray[count], getResultString(resultTsBlock, pos));
+ count++;
+ }
}
Assert.assertEquals(retArray.length, count);
@@ -161,32 +164,35 @@ public class SlidingWindowAggregationOperatorTest {
continue;
}
Assert.assertEquals(rootAggregationTypes.size(), resultTsBlock.getValueColumnCount());
- Assert.assertEquals(retArray[count - 1], getResultString(resultTsBlock));
- count--;
+ int positionCount = resultTsBlock.getPositionCount();
+ for (int pos = 0; pos < positionCount; pos++) {
+ Assert.assertEquals(retArray[count - 1], getResultString(resultTsBlock, pos));
+ count--;
+ }
}
Assert.assertEquals(0, count);
}
- private String getResultString(TsBlock resultTsBlock) {
- return resultTsBlock.getTimeColumn().getLong(0)
+ private String getResultString(TsBlock resultTsBlock, int pos) {
+ return resultTsBlock.getTimeColumn().getLong(pos)
+ ","
- + resultTsBlock.getColumn(0).getLong(0)
+ + resultTsBlock.getColumn(0).getLong(pos)
+ ","
- + resultTsBlock.getColumn(1).getDouble(0)
+ + resultTsBlock.getColumn(1).getDouble(pos)
+ ","
- + resultTsBlock.getColumn(2).getDouble(0)
+ + resultTsBlock.getColumn(2).getDouble(pos)
+ ","
- + resultTsBlock.getColumn(3).getInt(0)
+ + resultTsBlock.getColumn(3).getInt(pos)
+ ","
- + resultTsBlock.getColumn(4).getLong(0)
+ + resultTsBlock.getColumn(4).getLong(pos)
+ ","
- + resultTsBlock.getColumn(5).getLong(0)
+ + resultTsBlock.getColumn(5).getLong(pos)
+ ","
- + resultTsBlock.getColumn(6).getInt(0)
+ + resultTsBlock.getColumn(6).getInt(pos)
+ ","
- + resultTsBlock.getColumn(7).getInt(0)
+ + resultTsBlock.getColumn(7).getInt(pos)
+ ","
- + resultTsBlock.getColumn(8).getInt(0);
+ + resultTsBlock.getColumn(8).getInt(pos);
}
private SlidingWindowAggregationOperator initSlidingWindowAggregationOperator(boolean ascending)
@@ -204,6 +210,12 @@ public class SlidingWindowAggregationOperatorTest {
0, sourceId, SeriesAggregationScanOperator.class.getSimpleName());
fragmentInstanceContext.addOperatorContext(
1, queryId.genPlanNodeId(), SlidingWindowAggregationOperator.class.getSimpleName());
+ fragmentInstanceContext
+ .getOperatorContexts()
+ .forEach(
+ operatorContext -> {
+ operatorContext.setMaxRunTime(TEST_TIME_SLICE);
+ });
MeasurementPath d0s0 =
new MeasurementPath(AGGREGATION_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
@@ -226,7 +238,7 @@ public class SlidingWindowAggregationOperatorTest {
seriesAggregationScanOperator.initQueryDataSource(
new QueryDataSource(seqResources, unSeqResources));
- List<SlidingWindowAggregator> finalAggregators = new ArrayList<>();
+ List<Aggregator> finalAggregators = new ArrayList<>();
for (int i = 0; i < rootAggregationTypes.size(); i++) {
finalAggregators.add(
SlidingWindowAggregatorFactory.createSlidingWindowAggregator(
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/UpdateLastCacheOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/UpdateLastCacheOperatorTest.java
index b78cb7b1ac..581f1e703e 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/UpdateLastCacheOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/UpdateLastCacheOperatorTest.java
@@ -54,6 +54,7 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import static org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationOperatorTest.TEST_TIME_SLICE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -195,6 +196,13 @@ public class UpdateLastCacheOperatorTest {
fragmentInstanceContext.addOperatorContext(
2, planNodeId2, UpdateLastCacheOperator.class.getSimpleName());
+ fragmentInstanceContext
+ .getOperatorContexts()
+ .forEach(
+ operatorContext -> {
+ operatorContext.setMaxRunTime(TEST_TIME_SLICE);
+ });
+
SeriesAggregationScanOperator seriesAggregationScanOperator =
new SeriesAggregationScanOperator(
planNodeId1,
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java
index 5d36d6286b..a157d57058 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java
@@ -150,6 +150,12 @@ public class TSFileConfig implements Serializable {
/** The amount of data iterate each time */
private int batchSize = 1000;
+ /** Maximum capacity of a TsBlock */
+ private int maxTsBlockSizeInBytes = 1024 * 1024;
+
+ /** Maximum number of lines in a single TsBlock */
+ private int maxTsBlockLineNumber = 1000;
+
public TSFileConfig() {}
public int getGroupSizeInByte() {
@@ -431,4 +437,20 @@ public class TSFileConfig implements Serializable {
public void setFreqEncodingBlockSize(int freqEncodingBlockSize) {
this.freqEncodingBlockSize = freqEncodingBlockSize;
}
+
+ public int getMaxTsBlockSizeInBytes() {
+ return maxTsBlockSizeInBytes;
+ }
+
+ public void setMaxTsBlockSizeInBytes(int maxTsBlockSizeInBytes) {
+ this.maxTsBlockSizeInBytes = maxTsBlockSizeInBytes;
+ }
+
+ public int getMaxTsBlockLineNumber() {
+ return maxTsBlockLineNumber;
+ }
+
+ public void setMaxTsBlockLineNumber(int maxTsBlockLineNumber) {
+ this.maxTsBlockLineNumber = maxTsBlockLineNumber;
+ }
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/Statistics.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/Statistics.java
index 012fbe729d..197d9848f8 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/Statistics.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/Statistics.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.tsfile.file.metadata.statistics;
import org.apache.iotdb.tsfile.exception.filter.StatisticsClassException;
import org.apache.iotdb.tsfile.exception.write.UnknownColumnTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.ReadWriteForEncodingUtils;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
@@ -373,6 +374,10 @@ public abstract class Statistics<T extends Serializable> {
public abstract long calculateRamSize();
+ public boolean containedByTimeFilter(Filter timeFilter) {
+ return timeFilter == null || timeFilter.containStartEndTime(getStartTime(), getEndTime());
+ }
+
@Override
public String toString() {
return "startTime: " + startTime + " endTime: " + endTime + " count: " + count;
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlock.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlock.java
index 7e4d836bcd..c7206efb57 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlock.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlock.java
@@ -182,6 +182,10 @@ public class TsBlock {
return new TsBlock(subTimeColumn, subValueColumns);
}
+ public TsBlock skipFirst() {
+ return this.subTsBlock(1);
+ }
+
public long getTimeByIndex(int index) {
return timeColumn.getLong(index);
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilder.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilder.java
index b02e897235..b2b47bf43b 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilder.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilder.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.tsfile.read.common.block;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.block.column.BinaryColumnBuilder;
import org.apache.iotdb.tsfile.read.common.block.column.BooleanColumnBuilder;
@@ -44,6 +45,9 @@ public class TsBlockBuilder {
// This could be any other small number.
private static final int DEFAULT_INITIAL_EXPECTED_ENTRIES = 8;
+ private static final int MAX_LINE_NUMBER =
+ TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber();
+
private TimeColumnBuilder timeColumnBuilder;
private ColumnBuilder[] valueColumnBuilders;
private List<TSDataType> types;
@@ -254,7 +258,7 @@ public class TsBlockBuilder {
}
public boolean isFull() {
- return declaredPositions == Integer.MAX_VALUE || tsBlockBuilderStatus.isFull();
+ return declaredPositions == MAX_LINE_NUMBER || tsBlockBuilderStatus.isFull();
}
public boolean isEmpty() {
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilderStatus.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilderStatus.java
index a48e9cd0ff..15d585800c 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilderStatus.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilderStatus.java
@@ -18,11 +18,13 @@
*/
package org.apache.iotdb.tsfile.read.common.block;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilderStatus;
public class TsBlockBuilderStatus {
- public static final int DEFAULT_MAX_TSBLOCK_SIZE_IN_BYTES = 1024 * 1024;
+ public static final int DEFAULT_MAX_TSBLOCK_SIZE_IN_BYTES =
+ TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes();
private final int maxTsBlockSizeInBytes;
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockUtil.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockUtil.java
new file mode 100644
index 0000000000..1aa79eb4a5
--- /dev/null
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockUtil.java
@@ -0,0 +1,59 @@
+/*
+ * 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.tsfile.read.common.block;
+
+import org.apache.iotdb.tsfile.read.common.TimeRange;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumn;
+
+public class TsBlockUtil {
+
+ private TsBlockUtil() {
+ // forbidding instantiation
+ }
+
+ /** Skip lines at the beginning of the tsBlock that are not in the time range. */
+ public static TsBlock skipPointsOutOfTimeRange(
+ TsBlock tsBlock, TimeRange targetTimeRange, boolean ascending) {
+ TimeColumn timeColumn = tsBlock.getTimeColumn();
+ long targetTime = ascending ? targetTimeRange.getMin() : targetTimeRange.getMax();
+ int left = 0, right = timeColumn.getPositionCount() - 1, mid;
+ // if ascending, find the first greater than or equal to targetTime
+ // else, find the first less than or equal to targetTime
+ while (left < right) {
+ mid = (left + right) >> 1;
+ if (timeColumn.getLongWithoutCheck(mid) < targetTime) {
+ if (ascending) {
+ left = mid + 1;
+ } else {
+ right = mid;
+ }
+ } else if (timeColumn.getLongWithoutCheck(mid) > targetTime) {
+ if (ascending) {
+ right = mid;
+ } else {
+ left = mid + 1;
+ }
+ } else if (timeColumn.getLongWithoutCheck(mid) == targetTime) {
+ return tsBlock.subTsBlock(mid);
+ }
+ }
+ return tsBlock.subTsBlock(left);
+ }
+}