You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2022/05/17 03:19:21 UTC
[iotdb] branch master updated: [IOTDB-2844] Implementaion of AggregationOperator and RawDataAggregationOperator (#5846)
This is an automated email from the ASF dual-hosted git repository.
xiangweiwei 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 7a7e390680 [IOTDB-2844] Implementaion of AggregationOperator and RawDataAggregationOperator (#5846)
7a7e390680 is described below
commit 7a7e390680c512ed107d92ceb93ce26d2d550227
Author: Xiangwei Wei <34...@users.noreply.github.com>
AuthorDate: Tue May 17 11:19:15 2022 +0800
[IOTDB-2844] Implementaion of AggregationOperator and RawDataAggregationOperator (#5846)
Co-authored-by: Jinrui.Zhang <xi...@gmail.com>
---
.../iotdb/db/mpp/aggregation/Accumulator.java | 3 +-
.../iotdb/db/mpp/aggregation/Aggregator.java | 39 ++-
.../iotdb/db/mpp/aggregation/AvgAccumulator.java | 38 ++-
.../iotdb/db/mpp/aggregation/CountAccumulator.java | 8 +-
.../db/mpp/aggregation/ExtremeAccumulator.java | 23 +-
.../db/mpp/aggregation/FirstValueAccumulator.java | 25 +-
.../mpp/aggregation/FirstValueDescAccumulator.java | 12 +-
.../db/mpp/aggregation/LastValueAccumulator.java | 32 +-
.../mpp/aggregation/LastValueDescAccumulator.java | 48 +--
.../db/mpp/aggregation/MaxTimeAccumulator.java | 24 +-
.../db/mpp/aggregation/MaxTimeDescAccumulator.java | 17 +-
.../db/mpp/aggregation/MaxValueAccumulator.java | 22 +-
.../db/mpp/aggregation/MinTimeAccumulator.java | 20 +-
.../db/mpp/aggregation/MinTimeDescAccumulator.java | 2 +-
.../db/mpp/aggregation/MinValueAccumulator.java | 22 +-
.../iotdb/db/mpp/aggregation/SumAccumulator.java | 34 +-
.../timerangeiterator/AggrWindowIterator.java | 172 ++++++++++
.../timerangeiterator/ITimeRangeIterator.java | 58 ++++
.../timerangeiterator/PreAggrWindowIterator.java | 169 ++++++++++
.../PreAggrWindowWithNaturalMonthIterator.java | 143 ++++++++
.../SingleTimeWindowIterator.java | 7 +-
.../TimeRangeIteratorFactory.java | 73 ++++
.../operator/process/AggregateOperator.java | 90 ++++-
.../operator/process/RawDataAggregateOperator.java | 194 +++++++++++
.../source/SeriesAggregateScanOperator.java | 107 +++---
.../db/mpp/execution/schedule/DriverScheduler.java | 2 +-
.../db/mpp/plan/planner/DistributionPlanner.java | 51 ++-
.../db/mpp/plan/planner/LocalExecutionPlanner.java | 75 ++++-
.../db/mpp/plan/planner/plan/node/PlanVisitor.java | 2 +-
.../planner/plan/node/process/AggregationNode.java | 64 +++-
.../plan/node/process/GroupByLevelNode.java | 6 +-
.../node/source/SeriesAggregationScanNode.java | 12 +-
.../plan/parameter/AggregationDescriptor.java | 61 +++-
.../planner/plan/parameter/AggregationStep.java | 44 ++-
.../iotdb/db/query/expression/Expression.java | 2 +-
.../iotdb/db/mpp/aggregation/AccumulatorTest.java | 116 +++++--
.../db/mpp/aggregation/TimeRangeIteratorTest.java | 298 +++++++++++++++++
.../execution/operator/AggregateOperatorTest.java | 321 ++++++++++++++++++
.../operator/RawDataAggregateOperatorTest.java | 368 +++++++++++++++++++++
.../operator/SeriesAggregateScanOperatorTest.java | 25 +-
.../node/process/AggregationNodeSerdeTest.java | 228 +++++++++++++
.../apache/iotdb/tsfile/read/common/TimeRange.java | 20 +-
42 files changed, 2799 insertions(+), 278 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/Accumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/Accumulator.java
index 40c8a08d50..cfbdd7774e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/Accumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/Accumulator.java
@@ -49,7 +49,8 @@ public interface Accumulator {
/**
* For aggregation function like COUNT, SUM, partialResult should be single, so its output column
- * is single too; But for AVG, last_value, it should be double column with dictionary order.
+ * is single too; But for AVG(COUNT and SUM), LAST_VALUE(LAST_VALUE and MAX_TIME), the output
+ * columns should be double in dictionary order.
*/
void outputIntermediate(ColumnBuilder[] tsBlockBuilder);
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 a0a7b87f1b..6077648e59 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
@@ -55,24 +55,38 @@ public class Aggregator {
this.inputLocationList = inputLocationList;
}
- // Used for SeriesAggregateScanOperator
+ // Used for SeriesAggregateScanOperator and RawDataAggregateOperator
public void processTsBlock(TsBlock tsBlock) {
checkArgument(
- step.isInputRaw(), "Step in SeriesAggregateScanOperator can only process raw input");
+ step.isInputRaw(),
+ "Step in SeriesAggregateScanOperator and RawDataAggregateOperator can only process raw input");
// TODO Aligned TimeSeries
- accumulator.addInput(tsBlock.getTimeAndValueColumn(0), timeRange);
+ if (inputLocationList == null) {
+ accumulator.addInput(tsBlock.getTimeAndValueColumn(0), timeRange);
+ } else {
+ for (InputLocation[] inputLocations : inputLocationList) {
+ checkArgument(
+ inputLocations[0].getTsBlockIndex() == 0,
+ "RawDataAggregateOperator can only process one tsBlock input.");
+ Column[] timeValueColumn = new Column[2];
+ timeValueColumn[0] = tsBlock.getTimeColumn();
+ timeValueColumn[1] = tsBlock.getColumn(inputLocations[0].getValueColumnIndex());
+ accumulator.addInput(timeValueColumn, timeRange);
+ }
+ }
}
- // Used for aggregateOperator
+ // Used for AggregateOperator
public void processTsBlocks(TsBlock[] tsBlock) {
- for (InputLocation[] inputLocations : inputLocationList) {
- if (step.isInputRaw()) {
- TsBlock rawTsBlock = tsBlock[inputLocations[0].getTsBlockIndex()];
- Column[] timeValueColumn = new Column[2];
- timeValueColumn[0] = rawTsBlock.getTimeColumn();
- timeValueColumn[1] = rawTsBlock.getColumn(inputLocations[0].getValueColumnIndex());
- accumulator.addInput(timeValueColumn, timeRange);
- } else {
+ checkArgument(!step.isInputRaw(), "Step in AggregateOperator cannot process raw input");
+ if (step.isInputFinal()) {
+ checkArgument(inputLocationList.size() == 1, "Final output can only be single column");
+ Column finalResult =
+ tsBlock[inputLocationList.get(0)[0].getTsBlockIndex()].getColumn(
+ inputLocationList.get(0)[0].getValueColumnIndex());
+ accumulator.setFinal(finalResult);
+ } else {
+ for (InputLocation[] inputLocations : inputLocationList) {
Column[] columns = new Column[inputLocations.length];
for (int i = 0; i < inputLocations.length; i++) {
columns[i] =
@@ -105,6 +119,7 @@ public class Aggregator {
}
public void reset() {
+ timeRange = new TimeRange(0, Long.MAX_VALUE);
accumulator.reset();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/AvgAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/AvgAccumulator.java
index 3c726f9603..09dcf14628 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/AvgAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/AvgAccumulator.java
@@ -35,6 +35,7 @@ public class AvgAccumulator implements Accumulator {
private TSDataType seriesDataType;
private long countValue;
private double sumValue;
+ private boolean initResult = false;
public AvgAccumulator(TSDataType seriesDataType) {
this.seriesDataType = seriesDataType;
@@ -67,12 +68,17 @@ public class AvgAccumulator implements Accumulator {
@Override
public void addIntermediate(Column[] partialResult) {
checkArgument(partialResult.length == 2, "partialResult of Avg should be 2");
+ if (partialResult[0].isNull(0)) {
+ return;
+ }
+ initResult = true;
countValue += partialResult[0].getLong(0);
sumValue += partialResult[1].getDouble(0);
}
@Override
public void addStatistics(Statistics statistics) {
+ initResult = true;
countValue += statistics.getCount();
if (statistics instanceof IntegerStatistics) {
sumValue += statistics.getSumLongValue();
@@ -85,6 +91,10 @@ public class AvgAccumulator implements Accumulator {
@Override
public void setFinal(Column finalResult) {
reset();
+ if (finalResult.isNull(0)) {
+ return;
+ }
+ initResult = true;
countValue = 1;
sumValue = finalResult.getDouble(0);
}
@@ -92,17 +102,27 @@ public class AvgAccumulator implements Accumulator {
@Override
public void outputIntermediate(ColumnBuilder[] columnBuilders) {
checkArgument(columnBuilders.length == 2, "partialResult of Avg should be 2");
- columnBuilders[0].writeLong(countValue);
- columnBuilders[1].writeDouble(sumValue);
+ if (!initResult) {
+ columnBuilders[0].appendNull();
+ columnBuilders[1].appendNull();
+ } else {
+ columnBuilders[0].writeLong(countValue);
+ columnBuilders[1].writeDouble(sumValue);
+ }
}
@Override
public void outputFinal(ColumnBuilder columnBuilder) {
- columnBuilder.writeDouble(sumValue / countValue);
+ if (!initResult) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.writeDouble(sumValue / countValue);
+ }
}
@Override
public void reset() {
+ initResult = false;
this.countValue = 0;
this.sumValue = 0.0;
}
@@ -126,10 +146,11 @@ public class AvgAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
+ initResult = true;
countValue++;
sumValue += column[1].getInt(i);
}
@@ -140,10 +161,11 @@ public class AvgAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
+ initResult = true;
countValue++;
sumValue += column[1].getLong(i);
}
@@ -154,10 +176,11 @@ public class AvgAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
+ initResult = true;
countValue++;
sumValue += column[1].getFloat(i);
}
@@ -168,10 +191,11 @@ public class AvgAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
+ initResult = true;
countValue++;
sumValue += column[1].getDouble(i);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/CountAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/CountAccumulator.java
index c2825b00bb..eeae6a0377 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/CountAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/CountAccumulator.java
@@ -40,7 +40,7 @@ public class CountAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
@@ -53,6 +53,9 @@ public class CountAccumulator implements Accumulator {
@Override
public void addIntermediate(Column[] partialResult) {
checkArgument(partialResult.length == 1, "partialResult of Count should be 1");
+ if (partialResult[0].isNull(0)) {
+ return;
+ }
countValue += partialResult[0].getLong(0);
}
@@ -64,6 +67,9 @@ public class CountAccumulator implements Accumulator {
// finalResult should be single column, like: | finalCountValue |
@Override
public void setFinal(Column finalResult) {
+ if (finalResult.isNull(0)) {
+ return;
+ }
countValue = finalResult.getLong(0);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/ExtremeAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/ExtremeAccumulator.java
index e70a7f484a..7dc5788243 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/ExtremeAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/ExtremeAccumulator.java
@@ -68,6 +68,9 @@ public class ExtremeAccumulator implements Accumulator {
@Override
public void addIntermediate(Column[] partialResult) {
checkArgument(partialResult.length == 1, "partialResult of ExtremeValue should be 1");
+ if (partialResult[0].isNull(0)) {
+ return;
+ }
switch (seriesDataType) {
case INT32:
updateIntResult(partialResult[0].getInt(0));
@@ -118,6 +121,10 @@ public class ExtremeAccumulator implements Accumulator {
@Override
public void setFinal(Column finalResult) {
+ if (finalResult.isNull(0)) {
+ return;
+ }
+ initResult = true;
extremeResult.setObject(finalResult.getObject(0));
}
@@ -125,6 +132,10 @@ public class ExtremeAccumulator implements Accumulator {
@Override
public void outputIntermediate(ColumnBuilder[] columnBuilders) {
checkArgument(columnBuilders.length == 1, "partialResult of ExtremeValue should be 1");
+ if (!initResult) {
+ columnBuilders[0].appendNull();
+ return;
+ }
switch (seriesDataType) {
case INT32:
columnBuilders[0].writeInt(extremeResult.getInt());
@@ -148,6 +159,10 @@ public class ExtremeAccumulator implements Accumulator {
@Override
public void outputFinal(ColumnBuilder columnBuilder) {
+ if (!initResult) {
+ columnBuilder.appendNull();
+ return;
+ }
switch (seriesDataType) {
case INT32:
columnBuilder.writeInt(extremeResult.getInt());
@@ -194,7 +209,7 @@ public class ExtremeAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
@@ -220,7 +235,7 @@ public class ExtremeAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
@@ -246,7 +261,7 @@ public class ExtremeAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
@@ -272,7 +287,7 @@ public class ExtremeAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/FirstValueAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/FirstValueAccumulator.java
index 45bce52036..ac0f4a423f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/FirstValueAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/FirstValueAccumulator.java
@@ -73,6 +73,9 @@ public class FirstValueAccumulator implements Accumulator {
@Override
public void addIntermediate(Column[] partialResult) {
checkArgument(partialResult.length == 2, "partialResult of FirstValue should be 2");
+ if (partialResult[0].isNull(0)) {
+ return;
+ }
switch (seriesDataType) {
case INT32:
updateIntFirstValue(partialResult[0].getInt(0), partialResult[1].getLong(0));
@@ -127,6 +130,7 @@ public class FirstValueAccumulator implements Accumulator {
@Override
public void setFinal(Column finalResult) {
reset();
+ hasCandidateResult = true;
firstValue.setObject(finalResult.getObject(0));
}
@@ -134,6 +138,11 @@ public class FirstValueAccumulator implements Accumulator {
@Override
public void outputIntermediate(ColumnBuilder[] columnBuilders) {
checkArgument(columnBuilders.length == 2, "partialResult of FirstValue should be 2");
+ if (!hasCandidateResult) {
+ columnBuilders[0].appendNull();
+ columnBuilders[1].appendNull();
+ return;
+ }
switch (seriesDataType) {
case INT32:
columnBuilders[0].writeInt(firstValue.getInt());
@@ -162,6 +171,10 @@ public class FirstValueAccumulator implements Accumulator {
@Override
public void outputFinal(ColumnBuilder columnBuilder) {
+ if (!hasCandidateResult) {
+ columnBuilder.appendNull();
+ return;
+ }
switch (seriesDataType) {
case INT32:
columnBuilder.writeInt(firstValue.getInt());
@@ -212,7 +225,7 @@ public class FirstValueAccumulator implements Accumulator {
protected void addIntInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateIntFirstValue(column[1].getInt(i), curTime);
break;
}
@@ -230,7 +243,7 @@ public class FirstValueAccumulator implements Accumulator {
protected void addLongInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateLongFirstValue(column[1].getLong(i), curTime);
break;
}
@@ -248,7 +261,7 @@ public class FirstValueAccumulator implements Accumulator {
protected void addFloatInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateFloatFirstValue(column[1].getFloat(i), curTime);
break;
}
@@ -266,7 +279,7 @@ public class FirstValueAccumulator implements Accumulator {
protected void addDoubleInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateDoubleFirstValue(column[1].getDouble(i), curTime);
break;
}
@@ -284,7 +297,7 @@ public class FirstValueAccumulator implements Accumulator {
protected void addBooleanInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateBooleanFirstValue(column[1].getBoolean(i), curTime);
break;
}
@@ -302,7 +315,7 @@ public class FirstValueAccumulator implements Accumulator {
protected void addBinaryInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateBinaryFirstValue(column[1].getBinary(i), curTime);
break;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/FirstValueDescAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/FirstValueDescAccumulator.java
index 08c04b8185..595e3277c6 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/FirstValueDescAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/FirstValueDescAccumulator.java
@@ -38,7 +38,7 @@ public class FirstValueDescAccumulator extends FirstValueAccumulator {
protected void addIntInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateIntFirstValue(column[1].getInt(i), curTime);
}
}
@@ -47,7 +47,7 @@ public class FirstValueDescAccumulator extends FirstValueAccumulator {
protected void addLongInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateLongFirstValue(column[1].getLong(i), curTime);
}
}
@@ -56,7 +56,7 @@ public class FirstValueDescAccumulator extends FirstValueAccumulator {
protected void addFloatInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateFloatFirstValue(column[1].getFloat(i), curTime);
}
}
@@ -65,7 +65,7 @@ public class FirstValueDescAccumulator extends FirstValueAccumulator {
protected void addDoubleInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateDoubleFirstValue(column[1].getDouble(i), curTime);
}
}
@@ -74,7 +74,7 @@ public class FirstValueDescAccumulator extends FirstValueAccumulator {
protected void addBooleanInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateBooleanFirstValue(column[1].getBoolean(i), curTime);
}
}
@@ -83,7 +83,7 @@ public class FirstValueDescAccumulator extends FirstValueAccumulator {
protected void addBinaryInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateBinaryFirstValue(column[1].getBinary(i), curTime);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/LastValueAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/LastValueAccumulator.java
index db418a490e..b884af9d56 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/LastValueAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/LastValueAccumulator.java
@@ -35,6 +35,7 @@ public class LastValueAccumulator implements Accumulator {
protected final TSDataType seriesDataType;
protected TsPrimitiveType lastValue;
protected long maxTime = Long.MIN_VALUE;
+ protected boolean initResult = false;
public LastValueAccumulator(TSDataType seriesDataType) {
this.seriesDataType = seriesDataType;
@@ -72,6 +73,9 @@ public class LastValueAccumulator implements Accumulator {
@Override
public void addIntermediate(Column[] partialResult) {
checkArgument(partialResult.length == 2, "partialResult of LastValue should be 2");
+ if (partialResult[0].isNull(0)) {
+ return;
+ }
switch (seriesDataType) {
case INT32:
updateIntLastValue(partialResult[0].getInt(0), partialResult[1].getLong(0));
@@ -133,6 +137,11 @@ public class LastValueAccumulator implements Accumulator {
@Override
public void outputIntermediate(ColumnBuilder[] columnBuilders) {
checkArgument(columnBuilders.length == 2, "partialResult of LastValue should be 2");
+ if (!initResult) {
+ columnBuilders[0].appendNull();
+ columnBuilders[1].appendNull();
+ return;
+ }
switch (seriesDataType) {
case INT32:
columnBuilders[0].writeInt(lastValue.getInt());
@@ -161,6 +170,10 @@ public class LastValueAccumulator implements Accumulator {
@Override
public void outputFinal(ColumnBuilder columnBuilder) {
+ if (!initResult) {
+ columnBuilder.appendNull();
+ return;
+ }
switch (seriesDataType) {
case INT32:
columnBuilder.writeInt(lastValue.getInt());
@@ -188,6 +201,7 @@ public class LastValueAccumulator implements Accumulator {
@Override
public void reset() {
+ initResult = false;
this.maxTime = Long.MIN_VALUE;
this.lastValue.reset();
}
@@ -210,13 +224,14 @@ public class LastValueAccumulator implements Accumulator {
protected void addIntInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateIntLastValue(column[1].getInt(i), curTime);
}
}
}
protected void updateIntLastValue(int value, long curTime) {
+ initResult = true;
if (curTime > maxTime) {
maxTime = curTime;
lastValue.setInt(value);
@@ -226,13 +241,14 @@ public class LastValueAccumulator implements Accumulator {
protected void addLongInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateLongLastValue(column[1].getLong(i), curTime);
}
}
}
protected void updateLongLastValue(long value, long curTime) {
+ initResult = true;
if (curTime > maxTime) {
maxTime = curTime;
lastValue.setLong(value);
@@ -242,13 +258,14 @@ public class LastValueAccumulator implements Accumulator {
protected void addFloatInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateFloatLastValue(column[1].getFloat(i), curTime);
}
}
}
protected void updateFloatLastValue(float value, long curTime) {
+ initResult = true;
if (curTime > maxTime) {
maxTime = curTime;
lastValue.setFloat(value);
@@ -258,13 +275,14 @@ public class LastValueAccumulator implements Accumulator {
protected void addDoubleInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateDoubleLastValue(column[1].getDouble(i), curTime);
}
}
}
protected void updateDoubleLastValue(double value, long curTime) {
+ initResult = true;
if (curTime > maxTime) {
maxTime = curTime;
lastValue.setDouble(value);
@@ -274,13 +292,14 @@ public class LastValueAccumulator implements Accumulator {
protected void addBooleanInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateBooleanLastValue(column[1].getBoolean(i), curTime);
}
}
}
protected void updateBooleanLastValue(boolean value, long curTime) {
+ initResult = true;
if (curTime > maxTime) {
maxTime = curTime;
lastValue.setBoolean(value);
@@ -290,13 +309,14 @@ public class LastValueAccumulator implements Accumulator {
protected void addBinaryInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateBinaryLastValue(column[1].getBinary(i), curTime);
}
}
}
protected void updateBinaryLastValue(Binary value, long curTime) {
+ initResult = true;
if (curTime > maxTime) {
maxTime = curTime;
lastValue.setBinary(value);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/LastValueDescAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/LastValueDescAccumulator.java
index 3360bad7a3..0339a2b102 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/LastValueDescAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/LastValueDescAccumulator.java
@@ -22,31 +22,27 @@ package org.apache.iotdb.db.mpp.aggregation;
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.column.Column;
-import org.apache.iotdb.tsfile.utils.Binary;
public class LastValueDescAccumulator extends LastValueAccumulator {
- private boolean hasCandidateResult = false;
-
public LastValueDescAccumulator(TSDataType seriesDataType) {
super(seriesDataType);
}
@Override
public boolean hasFinalResult() {
- return hasCandidateResult;
+ return initResult;
}
@Override
public void reset() {
- hasCandidateResult = false;
super.reset();
}
protected void addIntInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateIntLastValue(column[1].getInt(i), curTime);
break;
}
@@ -56,7 +52,7 @@ public class LastValueDescAccumulator extends LastValueAccumulator {
protected void addLongInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateLongLastValue(column[1].getLong(i), curTime);
break;
}
@@ -66,7 +62,7 @@ public class LastValueDescAccumulator extends LastValueAccumulator {
protected void addFloatInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateFloatLastValue(column[1].getFloat(i), curTime);
break;
}
@@ -76,7 +72,7 @@ public class LastValueDescAccumulator extends LastValueAccumulator {
protected void addDoubleInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateDoubleLastValue(column[1].getDouble(i), curTime);
break;
}
@@ -86,7 +82,7 @@ public class LastValueDescAccumulator extends LastValueAccumulator {
protected void addBooleanInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateBooleanLastValue(column[1].getBoolean(i), curTime);
break;
}
@@ -96,40 +92,10 @@ public class LastValueDescAccumulator extends LastValueAccumulator {
protected void addBinaryInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateBinaryLastValue(column[1].getBinary(i), curTime);
break;
}
}
}
-
- protected void updateIntLastValue(int value, long curTime) {
- hasCandidateResult = true;
- super.updateIntLastValue(value, curTime);
- }
-
- protected void updateLongLastValue(long value, long curTime) {
- hasCandidateResult = true;
- super.updateLongLastValue(value, curTime);
- }
-
- protected void updateFloatLastValue(float value, long curTime) {
- hasCandidateResult = true;
- super.updateFloatLastValue(value, curTime);
- }
-
- protected void updateDoubleLastValue(double value, long curTime) {
- hasCandidateResult = true;
- super.updateDoubleLastValue(value, curTime);
- }
-
- protected void updateBooleanLastValue(boolean value, long curTime) {
- hasCandidateResult = true;
- super.updateBooleanLastValue(value, curTime);
- }
-
- protected void updateBinaryLastValue(Binary value, long curTime) {
- hasCandidateResult = true;
- super.updateBinaryLastValue(value, curTime);
- }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxTimeAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxTimeAccumulator.java
index ce47210337..2b8eb22de7 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxTimeAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxTimeAccumulator.java
@@ -30,6 +30,7 @@ import static com.google.common.base.Preconditions.checkArgument;
public class MaxTimeAccumulator implements Accumulator {
protected long maxTime = Long.MIN_VALUE;
+ protected boolean initResult = false;
public MaxTimeAccumulator() {}
@@ -39,7 +40,7 @@ public class MaxTimeAccumulator implements Accumulator {
public void addInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateMaxTime(curTime);
}
}
@@ -49,6 +50,9 @@ public class MaxTimeAccumulator implements Accumulator {
@Override
public void addIntermediate(Column[] partialResult) {
checkArgument(partialResult.length == 1, "partialResult of MaxTime should be 1");
+ if (partialResult[0].isNull(0)) {
+ return;
+ }
updateMaxTime(partialResult[0].getLong(0));
}
@@ -60,6 +64,10 @@ public class MaxTimeAccumulator implements Accumulator {
// finalResult should be single column, like: | finalMaxTime |
@Override
public void setFinal(Column finalResult) {
+ if (finalResult.isNull(0)) {
+ return;
+ }
+ initResult = true;
maxTime = finalResult.getLong(0);
}
@@ -67,16 +75,25 @@ public class MaxTimeAccumulator implements Accumulator {
@Override
public void outputIntermediate(ColumnBuilder[] columnBuilders) {
checkArgument(columnBuilders.length == 1, "partialResult of MaxTime should be 1");
- columnBuilders[0].writeLong(maxTime);
+ if (!initResult) {
+ columnBuilders[0].appendNull();
+ } else {
+ columnBuilders[0].writeLong(maxTime);
+ }
}
@Override
public void outputFinal(ColumnBuilder columnBuilder) {
- columnBuilder.writeLong(maxTime);
+ if (!initResult) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.writeLong(maxTime);
+ }
}
@Override
public void reset() {
+ initResult = false;
this.maxTime = Long.MIN_VALUE;
}
@@ -96,6 +113,7 @@ public class MaxTimeAccumulator implements Accumulator {
}
protected void updateMaxTime(long curTime) {
+ initResult = true;
maxTime = Math.max(maxTime, curTime);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxTimeDescAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxTimeDescAccumulator.java
index b753e9ca3b..eae6e9f859 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxTimeDescAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxTimeDescAccumulator.java
@@ -24,15 +24,13 @@ import org.apache.iotdb.tsfile.read.common.block.column.Column;
public class MaxTimeDescAccumulator extends MaxTimeAccumulator {
- private boolean hasCandidateResult = false;
-
// Column should be like: | Time | Value |
// Value is used to judge isNull()
@Override
public void addInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateMaxTime(curTime);
break;
}
@@ -41,17 +39,6 @@ public class MaxTimeDescAccumulator extends MaxTimeAccumulator {
@Override
public boolean hasFinalResult() {
- return hasCandidateResult;
- }
-
- @Override
- public void reset() {
- hasCandidateResult = false;
- super.reset();
- }
-
- protected void updateMaxTime(long curTime) {
- hasCandidateResult = true;
- super.updateMaxTime(curTime);
+ return initResult;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxValueAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxValueAccumulator.java
index c07480a13d..343a03f093 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxValueAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MaxValueAccumulator.java
@@ -69,6 +69,9 @@ public class MaxValueAccumulator implements Accumulator {
@Override
public void addIntermediate(Column[] partialResult) {
checkArgument(partialResult.length == 1, "partialResult of MaxValue should be 1");
+ if (partialResult[0].isNull(0)) {
+ return;
+ }
switch (seriesDataType) {
case INT32:
updateIntResult(partialResult[0].getInt(0));
@@ -116,6 +119,9 @@ public class MaxValueAccumulator implements Accumulator {
// finalResult should be single column, like: | finalCountValue |
@Override
public void setFinal(Column finalResult) {
+ if (finalResult.isNull(0)) {
+ return;
+ }
maxResult.setObject(finalResult.getObject(0));
}
@@ -123,6 +129,10 @@ public class MaxValueAccumulator implements Accumulator {
@Override
public void outputIntermediate(ColumnBuilder[] columnBuilders) {
checkArgument(columnBuilders.length == 1, "partialResult of MaxValue should be 1");
+ if (!initResult) {
+ columnBuilders[0].appendNull();
+ return;
+ }
switch (seriesDataType) {
case INT32:
columnBuilders[0].writeInt(maxResult.getInt());
@@ -146,6 +156,10 @@ public class MaxValueAccumulator implements Accumulator {
@Override
public void outputFinal(ColumnBuilder columnBuilder) {
+ if (!initResult) {
+ columnBuilder.appendNull();
+ return;
+ }
switch (seriesDataType) {
case INT32:
columnBuilder.writeInt(maxResult.getInt());
@@ -192,7 +206,7 @@ public class MaxValueAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
@@ -212,7 +226,7 @@ public class MaxValueAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
@@ -232,7 +246,7 @@ public class MaxValueAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
@@ -252,7 +266,7 @@ public class MaxValueAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinTimeAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinTimeAccumulator.java
index a6af85093d..d2a3a8e837 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinTimeAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinTimeAccumulator.java
@@ -40,7 +40,7 @@ public class MinTimeAccumulator implements Accumulator {
public void addInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax() && !column[1].isNull(i)) {
+ if (timeRange.contains(curTime) && !column[1].isNull(i)) {
updateMinTime(curTime);
break;
}
@@ -51,6 +51,9 @@ public class MinTimeAccumulator implements Accumulator {
@Override
public void addIntermediate(Column[] partialResult) {
checkArgument(partialResult.length == 1, "partialResult of MinTime should be 1");
+ if (partialResult[0].isNull(0)) {
+ return;
+ }
updateMinTime(partialResult[0].getLong(0));
}
@@ -62,6 +65,9 @@ public class MinTimeAccumulator implements Accumulator {
// finalResult should be single column, like: | finalMinTime |
@Override
public void setFinal(Column finalResult) {
+ if (finalResult.isNull(0)) {
+ return;
+ }
minTime = finalResult.getLong(0);
}
@@ -69,12 +75,20 @@ public class MinTimeAccumulator implements Accumulator {
@Override
public void outputIntermediate(ColumnBuilder[] columnBuilders) {
checkArgument(columnBuilders.length == 1, "partialResult of MinTime should be 1");
- columnBuilders[0].writeLong(minTime);
+ if (!hasCandidateResult) {
+ columnBuilders[0].appendNull();
+ } else {
+ columnBuilders[0].writeLong(minTime);
+ }
}
@Override
public void outputFinal(ColumnBuilder columnBuilder) {
- columnBuilder.writeLong(minTime);
+ if (!hasCandidateResult) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.writeLong(minTime);
+ }
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinTimeDescAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinTimeDescAccumulator.java
index e97fa3ca70..615a05686c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinTimeDescAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinTimeDescAccumulator.java
@@ -28,7 +28,7 @@ public class MinTimeDescAccumulator extends MinTimeAccumulator {
public void addInput(Column[] column, TimeRange timeRange) {
for (int i = 0; i < column[0].getPositionCount(); i++) {
long curTime = column[0].getLong(i);
- if (curTime >= timeRange.getMin() && curTime < timeRange.getMax()) {
+ if (timeRange.contains(curTime)) {
updateMinTime(curTime);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinValueAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinValueAccumulator.java
index 8b24de07ea..b70d153104 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinValueAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/MinValueAccumulator.java
@@ -69,6 +69,9 @@ public class MinValueAccumulator implements Accumulator {
@Override
public void addIntermediate(Column[] partialResult) {
checkArgument(partialResult.length == 1, "partialResult of MinValue should be 1");
+ if (partialResult[0].isNull(0)) {
+ return;
+ }
switch (seriesDataType) {
case INT32:
updateIntResult(partialResult[0].getInt(0));
@@ -116,6 +119,9 @@ public class MinValueAccumulator implements Accumulator {
// finalResult should be single column, like: | finalCountValue |
@Override
public void setFinal(Column finalResult) {
+ if (finalResult.isNull(0)) {
+ return;
+ }
minResult.setObject(finalResult.getObject(0));
}
@@ -123,6 +129,10 @@ public class MinValueAccumulator implements Accumulator {
@Override
public void outputIntermediate(ColumnBuilder[] columnBuilders) {
checkArgument(columnBuilders.length == 1, "partialResult of MinValue should be 1");
+ if (!initResult) {
+ columnBuilders[0].appendNull();
+ return;
+ }
switch (seriesDataType) {
case INT32:
columnBuilders[0].writeInt(minResult.getInt());
@@ -146,6 +156,10 @@ public class MinValueAccumulator implements Accumulator {
@Override
public void outputFinal(ColumnBuilder columnBuilder) {
+ if (!initResult) {
+ columnBuilder.appendNull();
+ return;
+ }
switch (seriesDataType) {
case INT32:
columnBuilder.writeInt(minResult.getInt());
@@ -192,7 +206,7 @@ public class MinValueAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
@@ -212,7 +226,7 @@ public class MinValueAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
@@ -232,7 +246,7 @@ public class MinValueAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
@@ -252,7 +266,7 @@ public class MinValueAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/SumAccumulator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/SumAccumulator.java
index 132fb017a9..057292b624 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/SumAccumulator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/SumAccumulator.java
@@ -34,6 +34,7 @@ public class SumAccumulator implements Accumulator {
private TSDataType seriesDataType;
private double sumValue = 0;
+ private boolean initResult = false;
public SumAccumulator(TSDataType seriesDataType) {
this.seriesDataType = seriesDataType;
@@ -67,11 +68,16 @@ public class SumAccumulator implements Accumulator {
@Override
public void addIntermediate(Column[] partialResult) {
checkArgument(partialResult.length == 1, "partialResult of Sum should be 1");
+ if (partialResult[0].isNull(0)) {
+ return;
+ }
+ initResult = true;
sumValue += partialResult[0].getDouble(0);
}
@Override
public void addStatistics(Statistics statistics) {
+ initResult = true;
if (statistics instanceof IntegerStatistics) {
sumValue += statistics.getSumLongValue();
} else {
@@ -83,6 +89,9 @@ public class SumAccumulator implements Accumulator {
@Override
public void setFinal(Column finalResult) {
reset();
+ if (finalResult.isNull(0)) {
+ return;
+ }
sumValue = finalResult.getDouble(0);
}
@@ -90,16 +99,25 @@ public class SumAccumulator implements Accumulator {
@Override
public void outputIntermediate(ColumnBuilder[] columnBuilders) {
checkArgument(columnBuilders.length == 1, "partialResult of Sum should be 1");
- columnBuilders[0].writeDouble(sumValue);
+ if (!initResult) {
+ columnBuilders[0].appendNull();
+ } else {
+ columnBuilders[0].writeDouble(sumValue);
+ }
}
@Override
public void outputFinal(ColumnBuilder columnBuilder) {
- columnBuilder.writeDouble(sumValue);
+ if (!initResult) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.writeDouble(sumValue);
+ }
}
@Override
public void reset() {
+ initResult = false;
this.sumValue = 0;
}
@@ -122,10 +140,11 @@ public class SumAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
+ initResult = true;
sumValue += column[1].getInt(i);
}
}
@@ -135,10 +154,11 @@ public class SumAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
+ initResult = true;
sumValue += column[1].getLong(i);
}
}
@@ -148,10 +168,11 @@ public class SumAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
+ initResult = true;
sumValue += column[1].getFloat(i);
}
}
@@ -161,10 +182,11 @@ public class SumAccumulator implements Accumulator {
TimeColumn timeColumn = (TimeColumn) column[0];
for (int i = 0; i < timeColumn.getPositionCount(); i++) {
long curTime = timeColumn.getLong(i);
- if (curTime >= timeRange.getMax() || curTime < timeRange.getMin()) {
+ if (curTime > timeRange.getMax() || curTime < timeRange.getMin()) {
break;
}
if (!column[1].isNull(i)) {
+ initResult = true;
sumValue += column[1].getDouble(i);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/AggrWindowIterator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/AggrWindowIterator.java
new file mode 100644
index 0000000000..d53678bad2
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/AggrWindowIterator.java
@@ -0,0 +1,172 @@
+/*
+ * 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.aggregation.timerangeiterator;
+
+import org.apache.iotdb.db.qp.utils.DatetimeUtils;
+import org.apache.iotdb.tsfile.read.common.TimeRange;
+
+import static org.apache.iotdb.db.qp.utils.DatetimeUtils.MS_TO_MONTH;
+
+/**
+ * This class iteratively generates aggregated time windows.
+ *
+ * <p>For example, startTime = 0, endTime = 10, interval = 5, slidingStep = 3, leftCloseRightOpen,
+ * return [0,5],[3,7],[6,9],[9,9]
+ */
+public class AggrWindowIterator implements ITimeRangeIterator {
+
+ private final long startTime;
+ private final long endTime;
+ private final long interval;
+ private final long slidingStep;
+
+ private final boolean isAscending;
+ private final boolean isSlidingStepByMonth;
+ private final boolean isIntervalByMonth;
+ private final boolean leftCRightO;
+
+ private TimeRange curTimeRange;
+
+ public AggrWindowIterator(
+ long startTime,
+ long endTime,
+ long interval,
+ long slidingStep,
+ boolean isAscending,
+ boolean isSlidingStepByMonth,
+ boolean isIntervalByMonth,
+ boolean leftCRightO) {
+ this.startTime = startTime;
+ this.endTime = endTime;
+ this.interval = interval;
+ this.slidingStep = slidingStep;
+ this.isAscending = isAscending;
+ this.isSlidingStepByMonth = isSlidingStepByMonth;
+ this.isIntervalByMonth = isIntervalByMonth;
+ this.leftCRightO = leftCRightO;
+ }
+
+ @Override
+ public TimeRange getFirstTimeRange() {
+ if (isAscending) {
+ return getLeftmostTimeRange();
+ } else {
+ return getRightmostTimeRange();
+ }
+ }
+
+ private TimeRange getLeftmostTimeRange() {
+ long retEndTime;
+ if (isIntervalByMonth) {
+ // calculate interval length by natural month based on startTime
+ // ie. startTIme = 1/31, interval = 1mo, curEndTime will be set to 2/29
+ retEndTime = Math.min(DatetimeUtils.calcIntervalByMonth(startTime, interval), endTime);
+ } else {
+ retEndTime = Math.min(startTime + interval, endTime);
+ }
+ return new TimeRange(startTime, retEndTime);
+ }
+
+ private TimeRange getRightmostTimeRange() {
+ long retStartTime;
+ long retEndTime;
+ long queryRange = endTime - startTime;
+ long intervalNum;
+
+ if (isSlidingStepByMonth) {
+ intervalNum = (long) Math.ceil(queryRange / (double) (slidingStep * MS_TO_MONTH));
+ retStartTime = DatetimeUtils.calcIntervalByMonth(startTime, intervalNum * slidingStep);
+ while (retStartTime >= endTime) {
+ intervalNum -= 1;
+ retStartTime = DatetimeUtils.calcIntervalByMonth(startTime, intervalNum * slidingStep);
+ }
+ } else {
+ intervalNum = (long) Math.ceil(queryRange / (double) slidingStep);
+ retStartTime = slidingStep * (intervalNum - 1) + startTime;
+ }
+
+ if (isIntervalByMonth) {
+ // calculate interval length by natural month based on curStartTime
+ // ie. startTIme = 1/31, interval = 1mo, curEndTime will be set to 2/29
+ retEndTime = Math.min(DatetimeUtils.calcIntervalByMonth(retStartTime, interval), endTime);
+ } else {
+ retEndTime = Math.min(retStartTime + interval, endTime);
+ }
+ return new TimeRange(retStartTime, retEndTime);
+ }
+
+ @Override
+ public boolean hasNextTimeRange() {
+ if (curTimeRange == null) {
+ curTimeRange = getFirstTimeRange();
+ return true;
+ }
+
+ long retStartTime, retEndTime;
+ long curStartTime = curTimeRange.getMin();
+ if (isAscending) {
+ if (isSlidingStepByMonth) {
+ retStartTime = DatetimeUtils.calcIntervalByMonth(curStartTime, (int) (slidingStep));
+ } else {
+ retStartTime = curStartTime + slidingStep;
+ }
+ // This is an open interval , [0-100)
+ if (retStartTime >= endTime) {
+ return false;
+ }
+ } else {
+ if (isSlidingStepByMonth) {
+ retStartTime = DatetimeUtils.calcIntervalByMonth(curStartTime, (int) (-slidingStep));
+ } else {
+ retStartTime = curStartTime - slidingStep;
+ }
+ if (retStartTime < startTime) {
+ return false;
+ }
+ }
+
+ if (isIntervalByMonth) {
+ retEndTime = DatetimeUtils.calcIntervalByMonth(retStartTime, (int) (interval));
+ } else {
+ retEndTime = retStartTime + interval;
+ }
+ retEndTime = Math.min(retEndTime, endTime);
+ curTimeRange = new TimeRange(retStartTime, retEndTime);
+ return true;
+ }
+
+ @Override
+ public TimeRange nextTimeRange() {
+ if (curTimeRange != null || hasNextTimeRange()) {
+ return getFinalTimeRange(curTimeRange, leftCRightO);
+ }
+ return null;
+ }
+
+ @Override
+ public boolean isAscending() {
+ return isAscending;
+ }
+
+ @Override
+ public long currentOutputTime() {
+ return leftCRightO ? curTimeRange.getMin() : curTimeRange.getMax();
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/ITimeRangeIterator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/ITimeRangeIterator.java
new file mode 100644
index 0000000000..ccc999e810
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/ITimeRangeIterator.java
@@ -0,0 +1,58 @@
+/*
+ * 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.aggregation.timerangeiterator;
+
+import org.apache.iotdb.tsfile.read.common.TimeRange;
+
+/**
+ * This interface used for iteratively generating aggregated time windows in GROUP BY query.
+ *
+ * <p>It will return a leftCloseRightClose time window, by decreasing maxTime if leftCloseRightOpen
+ * and increasing minTime if leftOpenRightClose.
+ */
+public interface ITimeRangeIterator {
+
+ /** return the first time range by sorting order */
+ TimeRange getFirstTimeRange();
+
+ /** @return whether current iterator has next time range */
+ boolean hasNextTimeRange();
+ /**
+ * return the next time range according to curStartTime (the start time of the last returned time
+ * range)
+ */
+ TimeRange nextTimeRange();
+
+ boolean isAscending();
+
+ default TimeRange getFinalTimeRange(TimeRange timeRange, boolean leftCRightO) {
+ return leftCRightO
+ ? new TimeRange(timeRange.getMin(), timeRange.getMax() - 1)
+ : new TimeRange(timeRange.getMin() + 1, timeRange.getMax());
+ }
+
+ /**
+ * As there is only one timestamp can be output for a time range, this method will return the
+ * output time based on leftCloseRightOpen or not.
+ *
+ * @return minTime if leftCloseRightOpen, else maxTime.
+ */
+ long currentOutputTime();
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/PreAggrWindowIterator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/PreAggrWindowIterator.java
new file mode 100644
index 0000000000..ab018e3cb7
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/PreAggrWindowIterator.java
@@ -0,0 +1,169 @@
+/*
+ * 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.aggregation.timerangeiterator;
+
+import org.apache.iotdb.tsfile.read.common.TimeRange;
+
+/**
+ * This class iteratively generates pre-aggregated time windows.
+ *
+ * <p>For example, startTime = 0, endTime = 11, interval = 5, slidingStep = 3, return
+ * [0,2),[2,3),[3,5),[5,6),[6,8),[8,9),[9,10)
+ */
+public class PreAggrWindowIterator implements ITimeRangeIterator {
+
+ private final long startTime;
+ private final long endTime;
+ private final long interval;
+ private final long slidingStep;
+
+ private final boolean isAscending;
+ private final boolean leftCRightO;
+
+ private long curInterval;
+ private long curSlidingStep;
+ private boolean isIntervalCyclicChange = false;
+ private int intervalCnt = 0;
+
+ private TimeRange curTimeRange;
+
+ public PreAggrWindowIterator(
+ long startTime,
+ long endTime,
+ long interval,
+ long slidingStep,
+ boolean isAscending,
+ boolean leftCRightO) {
+ this.startTime = startTime;
+ this.endTime = endTime;
+ this.interval = interval;
+ this.slidingStep = slidingStep;
+ this.isAscending = isAscending;
+ this.leftCRightO = leftCRightO;
+ initIntervalAndStep();
+ }
+
+ @Override
+ public TimeRange getFirstTimeRange() {
+ if (isAscending) {
+ return getLeftmostTimeRange();
+ } else {
+ return getRightmostTimeRange();
+ }
+ }
+
+ private TimeRange getLeftmostTimeRange() {
+ long retEndTime = Math.min(startTime + curInterval, endTime);
+ updateIntervalAndStep();
+ return new TimeRange(startTime, retEndTime);
+ }
+
+ private TimeRange getRightmostTimeRange() {
+ long retStartTime;
+ long retEndTime;
+ long intervalNum = (long) Math.ceil((endTime - startTime) / (double) slidingStep);
+ retStartTime = slidingStep * (intervalNum - 1) + startTime;
+ if (isIntervalCyclicChange && endTime - retStartTime > interval % slidingStep) {
+ retStartTime += interval % slidingStep;
+ updateIntervalAndStep();
+ }
+ retEndTime = Math.min(retStartTime + curInterval, endTime);
+ updateIntervalAndStep();
+ return new TimeRange(retStartTime, retEndTime);
+ }
+
+ @Override
+ public boolean hasNextTimeRange() {
+ if (curTimeRange == null) {
+ curTimeRange = getFirstTimeRange();
+ return true;
+ }
+
+ long retStartTime, retEndTime;
+ long curStartTime = curTimeRange.getMin();
+ if (isAscending) {
+ retStartTime = curStartTime + curSlidingStep;
+ // This is an open interval , [0-100)
+ if (retStartTime >= endTime) {
+ return false;
+ }
+ } else {
+ retStartTime = curStartTime - curSlidingStep;
+ if (retStartTime < startTime) {
+ return false;
+ }
+ }
+ retEndTime = Math.min(retStartTime + curInterval, endTime);
+ updateIntervalAndStep();
+ curTimeRange = new TimeRange(retStartTime, retEndTime);
+ return true;
+ }
+
+ @Override
+ public TimeRange nextTimeRange() {
+ if (curTimeRange != null || hasNextTimeRange()) {
+ return getFinalTimeRange(curTimeRange, leftCRightO);
+ }
+ return null;
+ }
+
+ private void initIntervalAndStep() {
+ if (slidingStep >= interval) {
+ curInterval = interval;
+ curSlidingStep = slidingStep;
+ } else if (interval % slidingStep == 0) {
+ curInterval = slidingStep;
+ curSlidingStep = slidingStep;
+ } else {
+ isIntervalCyclicChange = true;
+ curInterval = interval % slidingStep;
+ curSlidingStep = curInterval;
+ }
+ }
+
+ private void updateIntervalAndStep() {
+ if (!isIntervalCyclicChange) {
+ return;
+ }
+
+ if (isAscending) {
+ curSlidingStep = curInterval;
+ }
+ intervalCnt++;
+ if ((intervalCnt & 1) == 1) {
+ curInterval = slidingStep - interval % slidingStep;
+ } else {
+ curInterval = interval % slidingStep;
+ }
+ if (!isAscending) {
+ curSlidingStep = curInterval;
+ }
+ }
+
+ @Override
+ public boolean isAscending() {
+ return isAscending;
+ }
+
+ @Override
+ public long currentOutputTime() {
+ return leftCRightO ? curTimeRange.getMin() : curTimeRange.getMax();
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/PreAggrWindowWithNaturalMonthIterator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/PreAggrWindowWithNaturalMonthIterator.java
new file mode 100644
index 0000000000..0a0d630811
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/PreAggrWindowWithNaturalMonthIterator.java
@@ -0,0 +1,143 @@
+/*
+ * 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.aggregation.timerangeiterator;
+
+import org.apache.iotdb.db.utils.datastructure.TimeSelector;
+import org.apache.iotdb.tsfile.read.common.TimeRange;
+
+public class PreAggrWindowWithNaturalMonthIterator implements ITimeRangeIterator {
+
+ private static final int HEAP_MAX_SIZE = 100;
+
+ private final boolean isAscending;
+ private final boolean leftCRightO;
+ private final TimeSelector timeBoundaryHeap;
+
+ private final AggrWindowIterator aggrWindowIterator;
+ private long curStartTimeForIterator;
+
+ private long lastEndTime;
+ private TimeRange curTimeRange;
+
+ public PreAggrWindowWithNaturalMonthIterator(
+ long startTime,
+ long endTime,
+ long interval,
+ long slidingStep,
+ boolean isAscending,
+ boolean isSlidingStepByMonth,
+ boolean isIntervalByMonth,
+ boolean leftCRightO) {
+ this.isAscending = isAscending;
+ this.timeBoundaryHeap = new TimeSelector(HEAP_MAX_SIZE, isAscending);
+ this.aggrWindowIterator =
+ new AggrWindowIterator(
+ startTime,
+ endTime,
+ interval,
+ slidingStep,
+ isAscending,
+ isSlidingStepByMonth,
+ isIntervalByMonth,
+ leftCRightO);
+ this.leftCRightO = leftCRightO;
+ initHeap();
+ }
+
+ @Override
+ public TimeRange getFirstTimeRange() {
+ long retStartTime = timeBoundaryHeap.pollFirst();
+ lastEndTime = timeBoundaryHeap.first();
+ return new TimeRange(retStartTime, lastEndTime);
+ }
+
+ @Override
+ public boolean hasNextTimeRange() {
+ if (curTimeRange == null) {
+ curTimeRange = getFirstTimeRange();
+ return true;
+ }
+
+ if (lastEndTime >= curStartTimeForIterator) {
+ tryToExpandHeap();
+ }
+ if (timeBoundaryHeap.isEmpty()) {
+ return false;
+ }
+ long retStartTime = timeBoundaryHeap.pollFirst();
+ if (retStartTime >= curStartTimeForIterator) {
+ tryToExpandHeap();
+ }
+ if (timeBoundaryHeap.isEmpty()) {
+ return false;
+ }
+ lastEndTime = timeBoundaryHeap.first();
+ curTimeRange = new TimeRange(retStartTime, lastEndTime);
+ return true;
+ }
+
+ @Override
+ public TimeRange nextTimeRange() {
+ if (curTimeRange != null || hasNextTimeRange()) {
+ return getFinalTimeRange(curTimeRange, leftCRightO);
+ }
+ return null;
+ }
+
+ private void initHeap() {
+ TimeRange firstTimeRange = aggrWindowIterator.nextTimeRange();
+ if (leftCRightO) {
+ timeBoundaryHeap.add(firstTimeRange.getMin());
+ timeBoundaryHeap.add(firstTimeRange.getMax() + 1);
+ curStartTimeForIterator = firstTimeRange.getMin();
+ } else {
+ timeBoundaryHeap.add(firstTimeRange.getMin() - 1);
+ timeBoundaryHeap.add(firstTimeRange.getMax());
+ curStartTimeForIterator = firstTimeRange.getMin() - 1;
+ }
+ tryToExpandHeap();
+ }
+
+ private void tryToExpandHeap() {
+ TimeRange timeRangeToExpand = null;
+ while (aggrWindowIterator.hasNextTimeRange() && timeBoundaryHeap.size() < HEAP_MAX_SIZE) {
+ timeRangeToExpand = aggrWindowIterator.nextTimeRange();
+ if (leftCRightO) {
+ timeBoundaryHeap.add(timeRangeToExpand.getMin());
+ timeBoundaryHeap.add(timeRangeToExpand.getMax() + 1);
+ curStartTimeForIterator = timeRangeToExpand.getMin();
+ } else {
+ timeBoundaryHeap.add(timeRangeToExpand.getMin() - 1);
+ timeBoundaryHeap.add(timeRangeToExpand.getMax());
+ curStartTimeForIterator = timeRangeToExpand.getMin() - 1;
+ }
+ }
+ }
+
+ @Override
+ public boolean isAscending() {
+ return isAscending;
+ }
+
+ @Override
+ public long currentOutputTime() {
+ return leftCRightO ? curTimeRange.getMin() : curTimeRange.getMax();
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/timerangeiterator/SingleTimeWindowIterator.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/SingleTimeWindowIterator.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/utils/timerangeiterator/SingleTimeWindowIterator.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/SingleTimeWindowIterator.java
index 3470f98834..2145b51f45 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/timerangeiterator/SingleTimeWindowIterator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/SingleTimeWindowIterator.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.utils.timerangeiterator;
+package org.apache.iotdb.db.mpp.aggregation.timerangeiterator;
import org.apache.iotdb.tsfile.read.common.TimeRange;
@@ -62,4 +62,9 @@ public class SingleTimeWindowIterator implements ITimeRangeIterator {
public boolean isAscending() {
return false;
}
+
+ @Override
+ public long currentOutputTime() {
+ return curTimeRange.getMin();
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/TimeRangeIteratorFactory.java b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/TimeRangeIteratorFactory.java
new file mode 100644
index 0000000000..7a270eb6cd
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/aggregation/timerangeiterator/TimeRangeIteratorFactory.java
@@ -0,0 +1,73 @@
+/*
+ * 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.aggregation.timerangeiterator;
+
+import static org.apache.iotdb.db.qp.utils.DatetimeUtils.MS_TO_MONTH;
+
+public class TimeRangeIteratorFactory {
+
+ private TimeRangeIteratorFactory() {}
+
+ /**
+ * The method returns different implements of ITimeRangeIterator depending on the parameters.
+ *
+ * <p>Note: interval and slidingStep stand for the milliseconds if not grouped by month, or the
+ * month count if grouped by month.
+ */
+ public static ITimeRangeIterator getTimeRangeIterator(
+ long startTime,
+ long endTime,
+ long interval,
+ long slidingStep,
+ boolean isAscending,
+ boolean isIntervalByMonth,
+ boolean isSlidingStepByMonth,
+ boolean leftCRightO,
+ boolean isPreAggr) {
+ long tmpInterval = isIntervalByMonth ? interval * MS_TO_MONTH : interval;
+ long tmpSlidingStep = isSlidingStepByMonth ? slidingStep * MS_TO_MONTH : slidingStep;
+ if (isPreAggr && tmpInterval > tmpSlidingStep) {
+ if (!isIntervalByMonth && !isSlidingStepByMonth) {
+ return new PreAggrWindowIterator(
+ startTime, endTime, interval, slidingStep, isAscending, leftCRightO);
+ } else {
+ return new PreAggrWindowWithNaturalMonthIterator(
+ startTime,
+ endTime,
+ interval,
+ slidingStep,
+ isAscending,
+ isSlidingStepByMonth,
+ isIntervalByMonth,
+ leftCRightO);
+ }
+ } else {
+ return new AggrWindowIterator(
+ startTime,
+ endTime,
+ interval,
+ slidingStep,
+ isAscending,
+ isSlidingStepByMonth,
+ isIntervalByMonth,
+ leftCRightO);
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregateOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregateOperator.java
index 71e7817b94..c860cd7450 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregateOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregateOperator.java
@@ -19,25 +19,62 @@
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 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 static org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregateScanOperator.initTimeRangeIterator;
+
+/**
+ * AggregateOperator 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.
+ */
public class AggregateOperator implements ProcessOperator {
private final OperatorContext operatorContext;
private final List<Aggregator> aggregators;
private final List<Operator> children;
+ private final int inputOperatorsCount;
+ private final TsBlock[] inputTsBlocks;
+ private final TsBlockBuilder tsBlockBuilder;
+
+ private ITimeRangeIterator timeRangeIterator;
+ // current interval of aggregation window [curStartTime, curEndTime)
+ private TimeRange curTimeRange;
+
public AggregateOperator(
- OperatorContext operatorContext, List<Aggregator> aggregators, List<Operator> children) {
+ OperatorContext operatorContext,
+ List<Aggregator> aggregators,
+ List<Operator> children,
+ boolean ascending,
+ GroupByTimeParameter groupByTimeParameter) {
this.operatorContext = operatorContext;
this.aggregators = aggregators;
this.children = children;
+
+ this.inputOperatorsCount = children.size();
+ this.inputTsBlocks = new TsBlock[inputOperatorsCount];
+ List<TSDataType> dataTypes = new ArrayList<>();
+ for (Aggregator aggregator : aggregators) {
+ dataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
+ }
+ tsBlockBuilder = new TsBlockBuilder(dataTypes);
+ this.timeRangeIterator = initTimeRangeIterator(groupByTimeParameter, ascending);
}
@Override
@@ -47,26 +84,67 @@ public class AggregateOperator implements ProcessOperator {
@Override
public ListenableFuture<Void> isBlocked() {
- return ProcessOperator.super.isBlocked();
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ ListenableFuture<Void> blocked = children.get(i).isBlocked();
+ if (!blocked.isDone()) {
+ return blocked;
+ }
+ }
+ return NOT_BLOCKED;
}
@Override
public TsBlock next() {
- return null;
+ // update input tsBlock
+ curTimeRange = timeRangeIterator.nextTimeRange();
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ inputTsBlocks[i] = children.get(i).next();
+ }
+ // consume current input tsBlocks
+ for (Aggregator aggregator : aggregators) {
+ aggregator.reset();
+ aggregator.processTsBlocks(inputTsBlocks);
+ }
+ // output result from aggregator
+ return updateResultTsBlockFromAggregators(tsBlockBuilder, aggregators, timeRangeIterator);
}
@Override
public boolean hasNext() {
- return false;
+ return timeRangeIterator.hasNextTimeRange();
}
@Override
public void close() throws Exception {
- ProcessOperator.super.close();
+ for (Operator child : children) {
+ child.close();
+ }
}
@Override
public boolean isFinished() {
- return false;
+ return !this.hasNext();
+ }
+
+ public static TsBlock updateResultTsBlockFromAggregators(
+ TsBlockBuilder tsBlockBuilder,
+ List<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;
+ 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 tsBlockBuilder.build();
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/RawDataAggregateOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/RawDataAggregateOperator.java
new file mode 100644
index 0000000000..45f376eaa2
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/RawDataAggregateOperator.java
@@ -0,0 +1,194 @@
+/*
+ * 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.TsBlock.TsBlockSingleColumnIterator;
+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 org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregateScanOperator.initTimeRangeIterator;
+
+/**
+ * RawDataAggregateOperator is used to process raw data tsBlock input calculating using value
+ * filter. It's possible that there is more than one tsBlock input in one time interval. And it's
+ * also possible that one tsBlock can cover multiple time intervals too.
+ *
+ * <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.
+ */
+public class RawDataAggregateOperator 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 RawDataAggregateOperator(
+ OperatorContext operatorContext,
+ List<Aggregator> aggregators,
+ 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);
+ }
+
+ @Override
+ public OperatorContext getOperatorContext() {
+ return operatorContext;
+ }
+
+ @Override
+ public ListenableFuture<Void> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ public TsBlock next() {
+ // 1. Clear previous aggregation result
+ curTimeRange = timeRangeIterator.nextTimeRange();
+ for (Aggregator aggregator : aggregators) {
+ aggregator.reset();
+ aggregator.setTimeRange(curTimeRange);
+ }
+
+ // 2. Calculate aggregation result based on current time window
+ while (!calcFromCacheData(curTimeRange)) {
+ if (child.hasNext()) {
+ preCachedData = child.next();
+ } else {
+ break;
+ }
+ }
+
+ // 3. Update result using aggregators
+ return AggregateOperator.updateResultTsBlockFromAggregators(
+ tsBlockBuilder, aggregators, timeRangeIterator);
+ }
+
+ @Override
+ public boolean hasNext() {
+ return 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) {
+ // check if the batchData does not contain points in current interval
+ if (preCachedData != null && satisfied(preCachedData, curTimeRange, ascending)) {
+ // skip points that cannot be calculated
+ preCachedData = skipOutOfTimeRangePoints(preCachedData, curTimeRange, ascending);
+
+ for (Aggregator aggregator : aggregators) {
+ // current agg method has been calculated
+ if (aggregator.hasFinalResult()) {
+ continue;
+ }
+
+ aggregator.processTsBlock(preCachedData);
+ }
+ }
+ // 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) {
+ TsBlockSingleColumnIterator tsBlockIterator = tsBlock.getTsBlockSingleColumnIterator();
+ if (ascending) {
+ while (tsBlockIterator.hasNext() && tsBlockIterator.currentTime() < curTimeRange.getMin()) {
+ tsBlockIterator.next();
+ }
+ } else {
+ while (tsBlockIterator.hasNext() && tsBlockIterator.currentTime() > curTimeRange.getMax()) {
+ tsBlockIterator.next();
+ }
+ }
+ return tsBlock.subTsBlock(tsBlockIterator.getRowIndex());
+ }
+
+ private boolean satisfied(TsBlock tsBlock, TimeRange timeRange, boolean ascending) {
+ TsBlockSingleColumnIterator tsBlockIterator = tsBlock.getTsBlockSingleColumnIterator();
+ if (tsBlockIterator == null || !tsBlockIterator.hasNext()) {
+ return false;
+ }
+
+ return ascending
+ ? (tsBlockIterator.getEndTime() >= timeRange.getMin()
+ && tsBlockIterator.currentTime() <= timeRange.getMax())
+ : (tsBlockIterator.getEndTime() <= timeRange.getMax()
+ && tsBlockIterator.currentTime() >= timeRange.getMin());
+ }
+
+ public static boolean isEndCalc(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/source/SeriesAggregateScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregateScanOperator.java
index fdc50a808b..92010c87ec 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregateScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregateScanOperator.java
@@ -21,20 +21,19 @@ 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.AggregateOperator;
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.db.utils.timerangeiterator.ITimeRangeIterator;
-import org.apache.iotdb.db.utils.timerangeiterator.SingleTimeWindowIterator;
-import org.apache.iotdb.db.utils.timerangeiterator.TimeRangeIteratorFactory;
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.common.block.column.ColumnBuilder;
-import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import com.google.common.util.concurrent.ListenableFuture;
@@ -45,6 +44,9 @@ import java.util.Arrays;
import java.util.List;
import java.util.Set;
+import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregateOperator.isEndCalc;
+import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregateOperator.skipOutOfTimeRangePoints;
+
/**
* This operator is responsible to do the aggregation calculation for one series based on global
* time range and time split parameter.
@@ -101,7 +103,7 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
dataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
}
tsBlockBuilder = new TsBlockBuilder(dataTypes);
- this.timeRangeIterator = initTimeRangeIterator(groupByTimeParameter);
+ this.timeRangeIterator = initTimeRangeIterator(groupByTimeParameter, ascending);
}
/**
@@ -109,7 +111,8 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
* 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 ITimeRangeIterator initTimeRangeIterator(GroupByTimeParameter groupByTimeParameter) {
+ public static ITimeRangeIterator initTimeRangeIterator(
+ GroupByTimeParameter groupByTimeParameter, boolean ascending) {
if (groupByTimeParameter == null) {
return new SingleTimeWindowIterator(0, Long.MAX_VALUE);
} else {
@@ -121,6 +124,7 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
ascending,
groupByTimeParameter.isIntervalByMonth(),
groupByTimeParameter.isSlidingStepByMonth(),
+ groupByTimeParameter.isLeftCRightO(),
groupByTimeParameter.getInterval() > groupByTimeParameter.getSlidingStep());
}
}
@@ -164,28 +168,28 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
// 2. Calculate aggregation result based on current time window
if (calcFromCacheData(curTimeRange)) {
- updateResultTsBlockUsingAggregateResult();
+ updateResultTsBlockFromAggregators();
return true;
}
// read page data firstly
if (readAndCalcFromPage(curTimeRange)) {
- updateResultTsBlockUsingAggregateResult();
+ updateResultTsBlockFromAggregators();
return true;
}
// read chunk data secondly
if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockUsingAggregateResult();
+ updateResultTsBlockFromAggregators();
return true;
}
// read from file first
while (seriesScanUtil.hasNextFile()) {
Statistics fileStatistics = seriesScanUtil.currentFileStatistics();
- if (fileStatistics.getStartTime() >= curTimeRange.getMax()) {
+ if (fileStatistics.getStartTime() > curTimeRange.getMax()) {
if (ascending) {
- updateResultTsBlockUsingAggregateResult();
+ updateResultTsBlockFromAggregators();
return true;
} else {
seriesScanUtil.skipCurrentFile();
@@ -202,35 +206,22 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
// read chunk
if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockUsingAggregateResult();
+ updateResultTsBlockFromAggregators();
return true;
}
}
- updateResultTsBlockUsingAggregateResult();
+ updateResultTsBlockFromAggregators();
return true;
} catch (IOException e) {
throw new RuntimeException("Error while scanning the file", e);
}
}
- private void updateResultTsBlockUsingAggregateResult() {
- tsBlockBuilder.reset();
- TimeColumnBuilder timeColumnBuilder = tsBlockBuilder.getTimeColumnBuilder();
- // Use start time of current time range as time column
- timeColumnBuilder.writeLong(curTimeRange.getMin());
- 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();
- resultTsBlock = tsBlockBuilder.build();
+ private void updateResultTsBlockFromAggregators() {
+ resultTsBlock =
+ AggregateOperator.updateResultTsBlockFromAggregators(
+ tsBlockBuilder, aggregators, timeRangeIterator);
hasCachedTsBlock = true;
}
@@ -242,7 +233,7 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
@Override
public boolean isFinished() {
- return finished || (finished = hasNext());
+ return finished || (finished = !hasNext());
}
@Override
@@ -261,20 +252,20 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
// The result is calculated from the cache
return (preCachedData != null
&& (ascending
- ? preCachedData.getEndTime() >= curTimeRange.getMax()
- : preCachedData.getStartTime() < curTimeRange.getMin()))
- || isEndCalc();
+ ? 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)) {
+ if (tsBlock == null || !satisfied(tsBlock, curTimeRange, ascending)) {
return;
}
// skip points that cannot be calculated
- tsBlock = skipOutOfTimeRangePoints(tsBlock, curTimeRange);
+ tsBlock = skipOutOfTimeRangePoints(tsBlock, curTimeRange, ascending);
for (Aggregator aggregator : aggregators) {
// current agg method has been calculated
@@ -291,22 +282,7 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
}
}
- // skip points that cannot be calculated
- private TsBlock skipOutOfTimeRangePoints(TsBlock tsBlock, TimeRange curTimeRange) {
- TsBlockSingleColumnIterator tsBlockIterator = tsBlock.getTsBlockSingleColumnIterator();
- if (ascending) {
- while (tsBlockIterator.hasNext() && tsBlockIterator.currentTime() < curTimeRange.getMin()) {
- tsBlockIterator.next();
- }
- } else {
- while (tsBlockIterator.hasNext() && tsBlockIterator.currentTime() >= curTimeRange.getMax()) {
- tsBlockIterator.next();
- }
- }
- return tsBlock.subTsBlock(tsBlockIterator.getRowIndex());
- }
-
- private boolean satisfied(TsBlock tsBlock, TimeRange timeRange) {
+ private boolean satisfied(TsBlock tsBlock, TimeRange timeRange, boolean ascending) {
TsBlockSingleColumnIterator tsBlockIterator = tsBlock.getTsBlockSingleColumnIterator();
if (tsBlockIterator == null || !tsBlockIterator.hasNext()) {
return false;
@@ -314,11 +290,11 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
if (ascending
&& (tsBlockIterator.getEndTime() < timeRange.getMin()
- || tsBlockIterator.currentTime() >= timeRange.getMax())) {
+ || tsBlockIterator.currentTime() > timeRange.getMax())) {
return false;
}
if (!ascending
- && (tsBlockIterator.getStartTime() >= timeRange.getMax()
+ && (tsBlockIterator.getEndTime() > timeRange.getMax()
|| tsBlockIterator.currentTime() < timeRange.getMin())) {
preCachedData = tsBlock;
return false;
@@ -326,15 +302,6 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
return true;
}
- private boolean isEndCalc() {
- for (Aggregator aggregator : aggregators) {
- if (!aggregator.hasFinalResult()) {
- return false;
- }
- }
- return true;
- }
-
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
private boolean readAndCalcFromPage(TimeRange curTimeRange) throws IOException {
while (seriesScanUtil.hasNextPage()) {
@@ -342,7 +309,7 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
// must be non overlapped page
if (pageStatistics != null) {
// There is no more eligible points in current time range
- if (pageStatistics.getStartTime() >= curTimeRange.getMax()) {
+ if (pageStatistics.getStartTime() > curTimeRange.getMax()) {
if (ascending) {
return true;
} else {
@@ -355,7 +322,7 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
&& curTimeRange.contains(pageStatistics.getStartTime(), pageStatistics.getEndTime())) {
calcFromStatistics(pageStatistics);
seriesScanUtil.skipCurrentPage();
- if (isEndCalc()) {
+ if (isEndCalc(aggregators)) {
return true;
}
continue;
@@ -373,7 +340,7 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
// lastReadIndex = tsBlockIterator.getRowIndex();
// stop calc and cached current batchData
- if (ascending && tsBlockIterator.currentTime() >= curTimeRange.getMax()) {
+ if (ascending && tsBlockIterator.currentTime() > curTimeRange.getMax()) {
preCachedData = tsBlock;
return true;
}
@@ -382,10 +349,10 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
calcFromBatch(tsBlock, curTimeRange);
// judge whether the calculation finished
- if (isEndCalc()
+ if (isEndCalc(aggregators)
|| (tsBlockIterator.hasNext()
&& (ascending
- ? tsBlockIterator.currentTime() >= curTimeRange.getMax()
+ ? tsBlockIterator.currentTime() > curTimeRange.getMax()
: tsBlockIterator.currentTime() < curTimeRange.getMin()))) {
return true;
}
@@ -396,7 +363,7 @@ public class SeriesAggregateScanOperator implements DataSourceOperator {
private boolean readAndCalcFromChunk(TimeRange curTimeRange) throws IOException {
while (seriesScanUtil.hasNextChunk()) {
Statistics chunkStatistics = seriesScanUtil.currentChunkStatistics();
- if (chunkStatistics.getStartTime() >= curTimeRange.getMax()) {
+ if (chunkStatistics.getStartTime() > curTimeRange.getMax()) {
if (ascending) {
return true;
} else {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java
index 9d82308138..b1a5cc66fb 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java
@@ -66,7 +66,7 @@ public class DriverScheduler implements IDriverScheduler, IService {
private static final int MAX_CAPACITY = 1000; // TODO: load from config files
private static final int WORKER_THREAD_NUM = 4; // TODO: load from config files
- private static final int QUERY_TIMEOUT_MS = 10_000; // TODO: load from config files or requests
+ private static final int QUERY_TIMEOUT_MS = 60_000; // TODO: load from config files or requests
private final ThreadGroup workerGroups;
private final List<AbstractDriverThread> threads;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/DistributionPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/DistributionPlanner.java
index 775601c0b6..f0eb1fb387 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/DistributionPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/DistributionPlanner.java
@@ -39,11 +39,14 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaFetchM
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaFetchScanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryMergeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.AggregationNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.TimeJoinNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.sink.FragmentSinkNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationDescriptor;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationStep;
import org.apache.iotdb.db.mpp.plan.statement.crud.QueryStatement;
import java.util.ArrayList;
@@ -231,6 +234,51 @@ public class DistributionPlanner {
return timeJoinNode;
}
+ @Override
+ public PlanNode visitSeriesAggregationScan(
+ SeriesAggregationScanNode node, DistributionPlanContext context) {
+ List<TRegionReplicaSet> dataDistribution =
+ analysis.getPartitionInfo(node.getSeriesPath(), node.getTimeFilter());
+ if (dataDistribution.size() == 1) {
+ node.setRegionReplicaSet(dataDistribution.get(0));
+ return node;
+ }
+
+ List<AggregationDescriptor> leafAggDescriptorList = new ArrayList<>();
+ node.getAggregationDescriptorList()
+ .forEach(
+ descriptor -> {
+ leafAggDescriptorList.add(
+ new AggregationDescriptor(
+ descriptor.getAggregationType(),
+ AggregationStep.PARTIAL,
+ descriptor.getInputExpressions()));
+ });
+
+ List<AggregationDescriptor> rootAggDescriptorList = new ArrayList<>();
+ node.getAggregationDescriptorList()
+ .forEach(
+ descriptor -> {
+ rootAggDescriptorList.add(
+ new AggregationDescriptor(
+ descriptor.getAggregationType(),
+ AggregationStep.FINAL,
+ descriptor.getInputExpressions()));
+ });
+
+ AggregationNode aggregationNode =
+ new AggregationNode(
+ context.queryContext.getQueryId().genPlanNodeId(), rootAggDescriptorList);
+ for (TRegionReplicaSet dataRegion : dataDistribution) {
+ SeriesAggregationScanNode split = (SeriesAggregationScanNode) node.clone();
+ split.setAggregationDescriptorList(leafAggDescriptorList);
+ split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ split.setRegionReplicaSet(dataRegion);
+ aggregationNode.addChild(split);
+ }
+ return aggregationNode;
+ }
+
@Override
public PlanNode visitSchemaFetchMerge(
SchemaFetchMergeNode node, DistributionPlanContext context) {
@@ -433,7 +481,8 @@ public class DistributionPlanner {
}
@Override
- public PlanNode visitSeriesAggregate(SeriesAggregationScanNode node, NodeGroupContext context) {
+ public PlanNode visitSeriesAggregationScan(
+ SeriesAggregationScanNode node, NodeGroupContext context) {
context.putNodeDistribution(
node.getPlanNodeId(),
new NodeDistribution(NodeDistributionType.NO_CHILD, node.getRegionReplicaSet()));
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 6a8c5956e2..da95d2b075 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
@@ -37,11 +37,13 @@ import org.apache.iotdb.db.mpp.execution.driver.SchemaDriverContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.execution.operator.Operator;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
+import org.apache.iotdb.db.mpp.execution.operator.process.AggregateOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.DeviceMergeOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.DeviceViewOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.FilterOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.LimitOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.OffsetOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregateOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.TimeJoinOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.TransformOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.merge.AscTimeComparator;
@@ -95,6 +97,7 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.sink.FragmentSinkNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesScanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationDescriptor;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.InputLocation;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.OutputColumn;
import org.apache.iotdb.db.mpp.plan.statement.component.OrderBy;
@@ -356,7 +359,7 @@ public class LocalExecutionPlanner {
}
@Override
- public Operator visitSeriesAggregate(
+ public Operator visitSeriesAggregationScan(
SeriesAggregationScanNode node, LocalExecutionPlanContext context) {
PartialPath seriesPath = node.getSeriesPath();
boolean ascending = node.getScanOrder() == OrderBy.TIMESTAMP_ASC;
@@ -370,10 +373,13 @@ public class LocalExecutionPlanner {
node.getAggregationDescriptorList()
.forEach(
o ->
- new Aggregator(
- AccumulatorFactory.createAccumulator(
- o.getAggregationType(), node.getSeriesPath().getSeriesType(), ascending),
- o.getStep()));
+ aggregators.add(
+ new Aggregator(
+ AccumulatorFactory.createAccumulator(
+ o.getAggregationType(),
+ node.getSeriesPath().getSeriesType(),
+ ascending),
+ o.getStep())));
SeriesAggregateScanOperator aggregateScanOperator =
new SeriesAggregateScanOperator(
node.getPlanNodeId(),
@@ -534,7 +540,64 @@ public class LocalExecutionPlanner {
@Override
public Operator visitRowBasedSeriesAggregate(
AggregationNode node, LocalExecutionPlanContext context) {
- return super.visitRowBasedSeriesAggregate(node, context);
+ checkArgument(
+ node.getAggregationDescriptorList().size() >= 1,
+ "Aggregation descriptorList cannot be empty");
+ OperatorContext operatorContext =
+ context.instanceContext.addOperatorContext(
+ context.getNextOperatorId(),
+ node.getPlanNodeId(),
+ DeviceViewNode.class.getSimpleName());
+ List<Operator> children =
+ node.getChildren().stream()
+ .map(child -> child.accept(this, context))
+ .collect(Collectors.toList());
+ boolean ascending = node.getScanOrder() == OrderBy.TIMESTAMP_ASC;
+ List<Aggregator> aggregators = new ArrayList<>();
+ Map<String, List<InputLocation>> layout = makeLayout(node);
+ for (AggregationDescriptor descriptor : node.getAggregationDescriptorList()) {
+ List<String> outputColumnNames = descriptor.getOutputColumnNames();
+ // it may include double parts
+ List<List<InputLocation>> inputLocationParts = new ArrayList<>(outputColumnNames.size());
+ outputColumnNames.forEach(o -> inputLocationParts.add(layout.get(o)));
+
+ List<InputLocation[]> inputLocationList = new ArrayList<>();
+ for (int i = 0; i < inputLocationParts.get(0).size(); i++) {
+ if (outputColumnNames.size() == 1) {
+ inputLocationList.add(new InputLocation[] {inputLocationParts.get(0).get(i)});
+ } else {
+ inputLocationList.add(
+ new InputLocation[] {
+ inputLocationParts.get(0).get(i), inputLocationParts.get(1).get(i)
+ });
+ }
+ }
+
+ aggregators.add(
+ new Aggregator(
+ AccumulatorFactory.createAccumulator(
+ descriptor.getAggregationType(),
+ context
+ .getTypeProvider()
+ // get the type of first inputExpression
+ .getType(descriptor.getInputExpressions().get(0).toString()),
+ ascending),
+ descriptor.getStep(),
+ inputLocationList));
+ }
+ boolean inputRaw = node.getAggregationDescriptorList().get(0).getStep().isInputRaw();
+ if (inputRaw) {
+ checkArgument(children.size() == 1, "rawDataAggregateOperator can only accept one input");
+ return new RawDataAggregateOperator(
+ operatorContext,
+ aggregators,
+ children.get(0),
+ ascending,
+ node.getGroupByTimeParameter());
+ } else {
+ return new AggregateOperator(
+ operatorContext, aggregators, children, ascending, node.getGroupByTimeParameter());
+ }
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
index b11dac5a26..61470a7f20 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
@@ -65,7 +65,7 @@ public abstract class PlanVisitor<R, C> {
return visitPlan(node, context);
}
- public R visitSeriesAggregate(SeriesAggregationScanNode node, C context) {
+ public R visitSeriesAggregationScan(SeriesAggregationScanNode node, C context) {
return visitPlan(node, context);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/AggregationNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/AggregationNode.java
index 7693abf79d..e02958ea62 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/AggregationNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/AggregationNode.java
@@ -24,13 +24,16 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationDescriptor;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
+import org.apache.iotdb.db.mpp.plan.statement.component.OrderBy;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import javax.annotation.Nullable;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
@@ -43,11 +46,12 @@ public class AggregationNode extends ProcessNode {
// The list of aggregate functions, each AggregateDescriptor will be output as one column of
// result TsBlock
- protected final List<AggregationDescriptor> aggregationDescriptorList;
+ protected List<AggregationDescriptor> aggregationDescriptorList;
// The parameter of `group by time`.
// Its value will be null if there is no `group by time` clause.
@Nullable protected GroupByTimeParameter groupByTimeParameter;
+ protected OrderBy scanOrder = OrderBy.TIMESTAMP_ASC;
protected List<PlanNode> children;
@@ -55,9 +59,7 @@ public class AggregationNode extends ProcessNode {
PlanNodeId id,
List<PlanNode> children,
List<AggregationDescriptor> aggregationDescriptorList) {
- super(id);
- this.children = children;
- this.aggregationDescriptorList = aggregationDescriptorList;
+ this(id, children, aggregationDescriptorList, null);
}
public AggregationNode(
@@ -67,14 +69,12 @@ public class AggregationNode extends ProcessNode {
@Nullable GroupByTimeParameter groupByTimeParameter) {
super(id);
this.children = children;
- this.aggregationDescriptorList = aggregationDescriptorList;
+ this.aggregationDescriptorList = getDeduplicatedDescriptors(aggregationDescriptorList);
this.groupByTimeParameter = groupByTimeParameter;
}
public AggregationNode(PlanNodeId id, List<AggregationDescriptor> aggregationDescriptorList) {
- super(id);
- this.aggregationDescriptorList = aggregationDescriptorList;
- this.children = new ArrayList<>();
+ this(id, aggregationDescriptorList, null);
}
public AggregationNode(
@@ -82,7 +82,7 @@ public class AggregationNode extends ProcessNode {
List<AggregationDescriptor> aggregationDescriptorList,
@Nullable GroupByTimeParameter groupByTimeParameter) {
super(id);
- this.aggregationDescriptorList = aggregationDescriptorList;
+ this.aggregationDescriptorList = getDeduplicatedDescriptors(aggregationDescriptorList);
this.groupByTimeParameter = groupByTimeParameter;
this.children = new ArrayList<>();
}
@@ -96,6 +96,10 @@ public class AggregationNode extends ProcessNode {
return groupByTimeParameter;
}
+ public OrderBy getScanOrder() {
+ return scanOrder;
+ }
+
@Override
public List<PlanNode> getChildren() {
return children;
@@ -183,4 +187,46 @@ public class AggregationNode extends ProcessNode {
return Objects.hash(
super.hashCode(), aggregationDescriptorList, groupByTimeParameter, children);
}
+
+ /**
+ * If aggregation function COUNT and AVG for one time series appears at the same time, and outputs
+ * intermediate result, the output columns will be like | COUNT | COUNT | SUM |. In this
+ * situation, one COUNT column is not needed. Therefore, when COUNT(or SUM) appears with AVG and
+ * outputs intermediate result(if they output final result, they will be all necessary), we need
+ * to REMOVE the COUNT aggregation, and only keep AVG function no matter their appearing order.
+ *
+ * <p>The related functions include AVG(COUNT AND SUM), FIRST_VALUE(FIRST_VALUE AND MIN_TIME),
+ * LAST_VALUE(LAST_VALUE AND MAX_TIME).
+ */
+ public static List<AggregationDescriptor> getDeduplicatedDescriptors(
+ List<AggregationDescriptor> aggregationDescriptors) {
+ Map<String, Integer> columnToIndexMap = new HashMap<>();
+ boolean[] removedIndexes = new boolean[aggregationDescriptors.size()];
+ for (int i = 0; i < aggregationDescriptors.size(); i++) {
+ AggregationDescriptor descriptor = aggregationDescriptors.get(i);
+ if (descriptor.getStep().isOutputPartial()) {
+ List<String> outputColumnNames = descriptor.getOutputColumnNames();
+ for (String outputColumn : outputColumnNames) {
+ // if encountering repeated column
+ if (columnToIndexMap.containsKey(outputColumn)) {
+ // if self is double outputs, then remove the former, else remove self
+ if (outputColumnNames.size() == 2) {
+ removedIndexes[columnToIndexMap.get(outputColumn)] = true;
+ } else {
+ removedIndexes[i] = true;
+ }
+ } else {
+ columnToIndexMap.put(outputColumn, i);
+ }
+ }
+ }
+ }
+ List<AggregationDescriptor> deduplicatedDescriptors = new ArrayList<>();
+ for (int i = 0; i < aggregationDescriptors.size(); i++) {
+ if (!removedIndexes[i]) {
+ deduplicatedDescriptors.add(aggregationDescriptors.get(i));
+ }
+ }
+ return deduplicatedDescriptors;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java
index e072d6d3f7..034456c692 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java
@@ -30,7 +30,6 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
-import java.util.stream.Collectors;
/**
* This node is responsible for the final aggregation merge operation. It will process the data from
@@ -97,10 +96,7 @@ public class GroupByLevelNode extends AggregationNode {
@Override
public List<String> getOutputColumnNames() {
- return aggregationDescriptorList.stream()
- .map(AggregationDescriptor::getOutputColumnNames)
- .flatMap(List::stream)
- .collect(Collectors.toList());
+ return outputColumnNames;
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesAggregationScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesAggregationScanNode.java
index abf3e5b8d9..ced35c4fb4 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesAggregationScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesAggregationScanNode.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.AggregationNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationDescriptor;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
import org.apache.iotdb.db.mpp.plan.statement.component.OrderBy;
@@ -65,7 +66,7 @@ public class SeriesAggregationScanNode extends SourceNode {
// The list of aggregate functions, each AggregateDescriptor will be output as one column in
// result TsBlock
- private final List<AggregationDescriptor> aggregationDescriptorList;
+ private List<AggregationDescriptor> aggregationDescriptorList;
// The order to traverse the data.
// Currently, we only support TIMESTAMP_ASC and TIMESTAMP_DESC here.
@@ -88,7 +89,8 @@ public class SeriesAggregationScanNode extends SourceNode {
List<AggregationDescriptor> aggregationDescriptorList) {
super(id);
this.seriesPath = seriesPath;
- this.aggregationDescriptorList = aggregationDescriptorList;
+ this.aggregationDescriptorList =
+ AggregationNode.getDeduplicatedDescriptors(aggregationDescriptorList);
}
public SeriesAggregationScanNode(
@@ -176,6 +178,10 @@ public class SeriesAggregationScanNode extends SourceNode {
.collect(Collectors.toList());
}
+ public void setAggregationDescriptorList(List<AggregationDescriptor> aggregationDescriptorList) {
+ this.aggregationDescriptorList = aggregationDescriptorList;
+ }
+
@Override
public void open() throws Exception {}
@@ -194,7 +200,7 @@ public class SeriesAggregationScanNode extends SourceNode {
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
- return visitor.visitSeriesAggregate(this, context);
+ return visitor.visitSeriesAggregationScan(this, context);
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java
index 1f1923a6b2..74fd995005 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java
@@ -44,6 +44,8 @@ public class AggregationDescriptor {
*/
private final List<Expression> inputExpressions;
+ private String parametersString;
+
public AggregationDescriptor(
AggregationType aggregationType, AggregationStep step, List<Expression> inputExpressions) {
this.aggregationType = aggregationType;
@@ -52,7 +54,60 @@ public class AggregationDescriptor {
}
public List<String> getOutputColumnNames() {
- return new ArrayList<>();
+ List<AggregationType> outputAggregationTypes = new ArrayList<>();
+ if (step.isOutputPartial()) {
+ switch (aggregationType) {
+ case AVG:
+ outputAggregationTypes.add(AggregationType.COUNT);
+ outputAggregationTypes.add(AggregationType.SUM);
+ break;
+ case FIRST_VALUE:
+ outputAggregationTypes.add(AggregationType.FIRST_VALUE);
+ outputAggregationTypes.add(AggregationType.MIN_TIME);
+ break;
+ case LAST_VALUE:
+ outputAggregationTypes.add(AggregationType.LAST_VALUE);
+ outputAggregationTypes.add(AggregationType.MAX_TIME);
+ break;
+ default:
+ outputAggregationTypes.add(aggregationType);
+ }
+ } else {
+ outputAggregationTypes.add(aggregationType);
+ }
+ List<String> outputColumnNames = new ArrayList<>();
+ for (AggregationType outputType : outputAggregationTypes) {
+ outputColumnNames.add(
+ outputType.toString().toLowerCase() + "(" + getParametersString() + ")");
+ }
+ return outputColumnNames;
+ }
+
+ /**
+ * Generates the parameter part of the function column name.
+ *
+ * <p>Example:
+ *
+ * <p>Full column name -> udf(root.sg.d.s1, sin(root.sg.d.s1))
+ *
+ * <p>The parameter part -> root.sg.d.s1, sin(root.sg.d.s1)
+ */
+ public String getParametersString() {
+ if (parametersString == null) {
+ StringBuilder builder = new StringBuilder();
+ if (!inputExpressions.isEmpty()) {
+ builder.append(inputExpressions.get(0).toString());
+ for (int i = 1; i < inputExpressions.size(); ++i) {
+ builder.append(", ").append(inputExpressions.get(i).toString());
+ }
+ }
+ parametersString = builder.toString();
+ }
+ return parametersString;
+ }
+
+ public List<Expression> getInputExpressions() {
+ return inputExpressions;
}
public AggregationType getAggregationType() {
@@ -67,10 +122,6 @@ public class AggregationDescriptor {
this.step = step;
}
- public List<Expression> getInputExpressions() {
- return inputExpressions;
- }
-
public void serialize(ByteBuffer byteBuffer) {
ReadWriteIOUtils.write(aggregationType.ordinal(), byteBuffer);
step.serialize(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationStep.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationStep.java
index 8eab8d0f62..3c3af3cd83 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationStep.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationStep.java
@@ -31,48 +31,46 @@ import java.nio.ByteBuffer;
* <ul>
* <li>Raw: raw data, as input only
* <li>Partial: intermediate aggregation result
- * <li>Final: final aggregation result, as output only
+ * <li>Final: final aggregation result
* </ul>
*/
public enum AggregationStep {
// input Raw, output Partial
- PARTIAL(true, true),
+ PARTIAL(InputType.RAW, true),
// input Partial, output Final
- FINAL(false, false),
+ FINAL(InputType.PARTIAL, false),
// input Partial, output Partial
- INTERMEDIATE(false, true),
+ INTERMEDIATE(InputType.PARTIAL, true),
// input Raw, output Final
- SINGLE(true, false);
+ SINGLE(InputType.RAW, false),
+ // input final, output final
+ STATIC(InputType.FINAL, false);
- private final boolean inputRaw;
+ private enum InputType {
+ RAW,
+ PARTIAL,
+ FINAL
+ }
+
+ private final InputType inputType;
private final boolean outputPartial;
- AggregationStep(boolean inputRaw, boolean outputPartial) {
- this.inputRaw = inputRaw;
+ AggregationStep(InputType inputType, boolean outputPartial) {
+ this.inputType = inputType;
this.outputPartial = outputPartial;
}
public boolean isInputRaw() {
- return inputRaw;
+ return inputType == InputType.RAW;
}
- public boolean isOutputPartial() {
- return outputPartial;
+ public boolean isInputFinal() {
+ return inputType == InputType.FINAL;
}
- public static AggregationStep partialOutput(AggregationStep step) {
- if (step.isInputRaw()) {
- return AggregationStep.PARTIAL;
- }
- return AggregationStep.INTERMEDIATE;
- }
-
- public static AggregationStep partialInput(AggregationStep step) {
- if (step.isOutputPartial()) {
- return AggregationStep.INTERMEDIATE;
- }
- return AggregationStep.FINAL;
+ public boolean isOutputPartial() {
+ return outputPartial;
}
public void serialize(ByteBuffer byteBuffer) {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/expression/Expression.java b/server/src/main/java/org/apache/iotdb/db/query/expression/Expression.java
index 28698ab133..ee460b61b4 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/expression/Expression.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/expression/Expression.java
@@ -71,7 +71,7 @@ import java.util.Set;
public abstract class Expression {
/////////////////////////////////////////////////////////////////////////////////////////////////
- // Expression type inferring for execution plan generation
+ // Expression type inferring for execution plan generation //////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////
public boolean isBuiltInAggregationFunctionExpression() {
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/aggregation/AccumulatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/aggregation/AccumulatorTest.java
index 6fe08cd345..f329609a29 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/aggregation/AccumulatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/aggregation/AccumulatorTest.java
@@ -73,10 +73,19 @@ public class AccumulatorTest {
Assert.assertEquals(TSDataType.INT64, avgAccumulator.getIntermediateType()[0]);
Assert.assertEquals(TSDataType.DOUBLE, avgAccumulator.getIntermediateType()[1]);
Assert.assertEquals(TSDataType.DOUBLE, avgAccumulator.getFinalType());
+ // check returning null while no data
+ ColumnBuilder[] intermediateResult = new ColumnBuilder[2];
+ intermediateResult[0] = new LongColumnBuilder(null, 1);
+ intermediateResult[1] = new DoubleColumnBuilder(null, 1);
+ avgAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertTrue(intermediateResult[0].build().isNull(0));
+ Assert.assertTrue(intermediateResult[1].build().isNull(0));
+ ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ avgAccumulator.outputFinal(finalResult);
+ Assert.assertTrue(finalResult.build().isNull(0));
avgAccumulator.addInput(rawData.getTimeAndValueColumn(0), defaultTimeRange);
Assert.assertFalse(avgAccumulator.hasFinalResult());
- ColumnBuilder[] intermediateResult = new ColumnBuilder[2];
intermediateResult[0] = new LongColumnBuilder(null, 1);
intermediateResult[1] = new DoubleColumnBuilder(null, 1);
avgAccumulator.outputIntermediate(intermediateResult);
@@ -86,7 +95,7 @@ public class AccumulatorTest {
// add intermediate result as input
avgAccumulator.addIntermediate(
new Column[] {intermediateResult[0].build(), intermediateResult[1].build()});
- ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ finalResult = new DoubleColumnBuilder(null, 1);
avgAccumulator.outputFinal(finalResult);
Assert.assertEquals(49.5d, finalResult.build().getDouble(0), 0.001);
@@ -103,17 +112,24 @@ public class AccumulatorTest {
AccumulatorFactory.createAccumulator(AggregationType.COUNT, TSDataType.DOUBLE, true);
Assert.assertEquals(TSDataType.INT64, countAccumulator.getIntermediateType()[0]);
Assert.assertEquals(TSDataType.INT64, countAccumulator.getFinalType());
+ // check returning null while no data
+ ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
+ intermediateResult[0] = new LongColumnBuilder(null, 1);
+ countAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertEquals(0, intermediateResult[0].build().getLong(0));
+ ColumnBuilder finalResult = new LongColumnBuilder(null, 1);
+ countAccumulator.outputFinal(finalResult);
+ Assert.assertEquals(0, finalResult.build().getLong(0));
countAccumulator.addInput(rawData.getTimeAndValueColumn(0), defaultTimeRange);
Assert.assertFalse(countAccumulator.hasFinalResult());
- ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
intermediateResult[0] = new LongColumnBuilder(null, 1);
countAccumulator.outputIntermediate(intermediateResult);
Assert.assertEquals(100, intermediateResult[0].build().getLong(0));
// add intermediate result as input
countAccumulator.addIntermediate(new Column[] {intermediateResult[0].build()});
- ColumnBuilder finalResult = new LongColumnBuilder(null, 1);
+ finalResult = new LongColumnBuilder(null, 1);
countAccumulator.outputFinal(finalResult);
Assert.assertEquals(200, finalResult.build().getLong(0));
@@ -130,17 +146,24 @@ public class AccumulatorTest {
AccumulatorFactory.createAccumulator(AggregationType.EXTREME, TSDataType.DOUBLE, true);
Assert.assertEquals(TSDataType.DOUBLE, extremeAccumulator.getIntermediateType()[0]);
Assert.assertEquals(TSDataType.DOUBLE, extremeAccumulator.getFinalType());
+ // check returning null while no data
+ ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
+ intermediateResult[0] = new DoubleColumnBuilder(null, 1);
+ extremeAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertTrue(intermediateResult[0].build().isNull(0));
+ ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ extremeAccumulator.outputFinal(finalResult);
+ Assert.assertTrue(finalResult.build().isNull(0));
extremeAccumulator.addInput(rawData.getTimeAndValueColumn(0), defaultTimeRange);
Assert.assertFalse(extremeAccumulator.hasFinalResult());
- ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
intermediateResult[0] = new DoubleColumnBuilder(null, 1);
extremeAccumulator.outputIntermediate(intermediateResult);
Assert.assertEquals(99d, intermediateResult[0].build().getDouble(0), 0.001);
// add intermediate result as input
extremeAccumulator.addIntermediate(new Column[] {intermediateResult[0].build()});
- ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ finalResult = new DoubleColumnBuilder(null, 1);
extremeAccumulator.outputFinal(finalResult);
Assert.assertEquals(99d, finalResult.build().getDouble(0), 0.001);
@@ -158,10 +181,19 @@ public class AccumulatorTest {
Assert.assertEquals(TSDataType.DOUBLE, firstValueAccumulator.getIntermediateType()[0]);
Assert.assertEquals(TSDataType.INT64, firstValueAccumulator.getIntermediateType()[1]);
Assert.assertEquals(TSDataType.DOUBLE, firstValueAccumulator.getFinalType());
+ // check returning null while no data
+ ColumnBuilder[] intermediateResult = new ColumnBuilder[2];
+ intermediateResult[0] = new DoubleColumnBuilder(null, 1);
+ intermediateResult[1] = new LongColumnBuilder(null, 1);
+ firstValueAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertTrue(intermediateResult[0].build().isNull(0));
+ Assert.assertTrue(intermediateResult[1].build().isNull(0));
+ ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ firstValueAccumulator.outputFinal(finalResult);
+ Assert.assertTrue(finalResult.build().isNull(0));
firstValueAccumulator.addInput(rawData.getTimeAndValueColumn(0), defaultTimeRange);
Assert.assertTrue(firstValueAccumulator.hasFinalResult());
- ColumnBuilder[] intermediateResult = new ColumnBuilder[2];
intermediateResult[0] = new DoubleColumnBuilder(null, 1);
intermediateResult[1] = new LongColumnBuilder(null, 1);
firstValueAccumulator.outputIntermediate(intermediateResult);
@@ -171,7 +203,7 @@ public class AccumulatorTest {
// add intermediate result as input
firstValueAccumulator.addIntermediate(
new Column[] {intermediateResult[0].build(), intermediateResult[1].build()});
- ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ finalResult = new DoubleColumnBuilder(null, 1);
firstValueAccumulator.outputFinal(finalResult);
Assert.assertEquals(0L, finalResult.build().getDouble(0), 0.001);
@@ -189,9 +221,18 @@ public class AccumulatorTest {
Assert.assertEquals(TSDataType.DOUBLE, lastValueAccumulator.getIntermediateType()[0]);
Assert.assertEquals(TSDataType.INT64, lastValueAccumulator.getIntermediateType()[1]);
Assert.assertEquals(TSDataType.DOUBLE, lastValueAccumulator.getFinalType());
+ // check returning null while no data
+ ColumnBuilder[] intermediateResult = new ColumnBuilder[2];
+ intermediateResult[0] = new DoubleColumnBuilder(null, 1);
+ intermediateResult[1] = new LongColumnBuilder(null, 1);
+ lastValueAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertTrue(intermediateResult[0].build().isNull(0));
+ Assert.assertTrue(intermediateResult[1].build().isNull(0));
+ ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ lastValueAccumulator.outputFinal(finalResult);
+ Assert.assertTrue(finalResult.build().isNull(0));
lastValueAccumulator.addInput(rawData.getTimeAndValueColumn(0), defaultTimeRange);
- ColumnBuilder[] intermediateResult = new ColumnBuilder[2];
intermediateResult[0] = new DoubleColumnBuilder(null, 1);
intermediateResult[1] = new LongColumnBuilder(null, 1);
lastValueAccumulator.outputIntermediate(intermediateResult);
@@ -201,7 +242,7 @@ public class AccumulatorTest {
// add intermediate result as input
lastValueAccumulator.addIntermediate(
new Column[] {intermediateResult[0].build(), intermediateResult[1].build()});
- ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ finalResult = new DoubleColumnBuilder(null, 1);
lastValueAccumulator.outputFinal(finalResult);
Assert.assertEquals(99L, finalResult.build().getDouble(0), 0.001);
@@ -218,17 +259,24 @@ public class AccumulatorTest {
AccumulatorFactory.createAccumulator(AggregationType.MAX_TIME, TSDataType.DOUBLE, true);
Assert.assertEquals(TSDataType.INT64, maxTimeAccumulator.getIntermediateType()[0]);
Assert.assertEquals(TSDataType.INT64, maxTimeAccumulator.getFinalType());
+ // check returning null while no data
+ ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
+ intermediateResult[0] = new LongColumnBuilder(null, 1);
+ maxTimeAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertTrue(intermediateResult[0].build().isNull(0));
+ ColumnBuilder finalResult = new LongColumnBuilder(null, 1);
+ maxTimeAccumulator.outputFinal(finalResult);
+ Assert.assertTrue(finalResult.build().isNull(0));
maxTimeAccumulator.addInput(rawData.getTimeAndValueColumn(0), defaultTimeRange);
Assert.assertFalse(maxTimeAccumulator.hasFinalResult());
- ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
intermediateResult[0] = new LongColumnBuilder(null, 1);
maxTimeAccumulator.outputIntermediate(intermediateResult);
Assert.assertEquals(99, intermediateResult[0].build().getLong(0));
// add intermediate result as input
maxTimeAccumulator.addIntermediate(new Column[] {intermediateResult[0].build()});
- ColumnBuilder finalResult = new LongColumnBuilder(null, 1);
+ finalResult = new LongColumnBuilder(null, 1);
maxTimeAccumulator.outputFinal(finalResult);
Assert.assertEquals(99, finalResult.build().getLong(0));
@@ -245,17 +293,24 @@ public class AccumulatorTest {
AccumulatorFactory.createAccumulator(AggregationType.MIN_TIME, TSDataType.DOUBLE, true);
Assert.assertEquals(TSDataType.INT64, minTimeAccumulator.getIntermediateType()[0]);
Assert.assertEquals(TSDataType.INT64, minTimeAccumulator.getFinalType());
+ // check returning null while no data
+ ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
+ intermediateResult[0] = new LongColumnBuilder(null, 1);
+ minTimeAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertTrue(intermediateResult[0].build().isNull(0));
+ ColumnBuilder finalResult = new LongColumnBuilder(null, 1);
+ minTimeAccumulator.outputFinal(finalResult);
+ Assert.assertTrue(finalResult.build().isNull(0));
minTimeAccumulator.addInput(rawData.getTimeAndValueColumn(0), defaultTimeRange);
Assert.assertTrue(minTimeAccumulator.hasFinalResult());
- ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
intermediateResult[0] = new LongColumnBuilder(null, 1);
minTimeAccumulator.outputIntermediate(intermediateResult);
Assert.assertEquals(0, intermediateResult[0].build().getLong(0));
// add intermediate result as input
minTimeAccumulator.addIntermediate(new Column[] {intermediateResult[0].build()});
- ColumnBuilder finalResult = new LongColumnBuilder(null, 1);
+ finalResult = new LongColumnBuilder(null, 1);
minTimeAccumulator.outputFinal(finalResult);
Assert.assertEquals(0, finalResult.build().getLong(0));
@@ -272,17 +327,24 @@ public class AccumulatorTest {
AccumulatorFactory.createAccumulator(AggregationType.MAX_VALUE, TSDataType.DOUBLE, true);
Assert.assertEquals(TSDataType.DOUBLE, extremeAccumulator.getIntermediateType()[0]);
Assert.assertEquals(TSDataType.DOUBLE, extremeAccumulator.getFinalType());
+ // check returning null while no data
+ ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
+ intermediateResult[0] = new DoubleColumnBuilder(null, 1);
+ extremeAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertTrue(intermediateResult[0].build().isNull(0));
+ ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ extremeAccumulator.outputFinal(finalResult);
+ Assert.assertTrue(finalResult.build().isNull(0));
extremeAccumulator.addInput(rawData.getTimeAndValueColumn(0), defaultTimeRange);
Assert.assertFalse(extremeAccumulator.hasFinalResult());
- ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
intermediateResult[0] = new DoubleColumnBuilder(null, 1);
extremeAccumulator.outputIntermediate(intermediateResult);
Assert.assertEquals(99d, intermediateResult[0].build().getDouble(0), 0.001);
// add intermediate result as input
extremeAccumulator.addIntermediate(new Column[] {intermediateResult[0].build()});
- ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ finalResult = new DoubleColumnBuilder(null, 1);
extremeAccumulator.outputFinal(finalResult);
Assert.assertEquals(99d, finalResult.build().getDouble(0), 0.001);
@@ -299,17 +361,24 @@ public class AccumulatorTest {
AccumulatorFactory.createAccumulator(AggregationType.MIN_VALUE, TSDataType.DOUBLE, true);
Assert.assertEquals(TSDataType.DOUBLE, extremeAccumulator.getIntermediateType()[0]);
Assert.assertEquals(TSDataType.DOUBLE, extremeAccumulator.getFinalType());
+ // check returning null while no data
+ ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
+ intermediateResult[0] = new DoubleColumnBuilder(null, 1);
+ extremeAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertTrue(intermediateResult[0].build().isNull(0));
+ ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ extremeAccumulator.outputFinal(finalResult);
+ Assert.assertTrue(finalResult.build().isNull(0));
extremeAccumulator.addInput(rawData.getTimeAndValueColumn(0), defaultTimeRange);
Assert.assertFalse(extremeAccumulator.hasFinalResult());
- ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
intermediateResult[0] = new DoubleColumnBuilder(null, 1);
extremeAccumulator.outputIntermediate(intermediateResult);
Assert.assertEquals(0d, intermediateResult[0].build().getDouble(0), 0.001);
// add intermediate result as input
extremeAccumulator.addIntermediate(new Column[] {intermediateResult[0].build()});
- ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ finalResult = new DoubleColumnBuilder(null, 1);
extremeAccumulator.outputFinal(finalResult);
Assert.assertEquals(0d, finalResult.build().getDouble(0), 0.001);
@@ -326,17 +395,24 @@ public class AccumulatorTest {
AccumulatorFactory.createAccumulator(AggregationType.SUM, TSDataType.DOUBLE, true);
Assert.assertEquals(TSDataType.DOUBLE, sumAccumulator.getIntermediateType()[0]);
Assert.assertEquals(TSDataType.DOUBLE, sumAccumulator.getFinalType());
+ // check returning null while no data
+ ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
+ intermediateResult[0] = new DoubleColumnBuilder(null, 1);
+ sumAccumulator.outputIntermediate(intermediateResult);
+ Assert.assertTrue(intermediateResult[0].build().isNull(0));
+ ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ sumAccumulator.outputFinal(finalResult);
+ Assert.assertTrue(finalResult.build().isNull(0));
sumAccumulator.addInput(rawData.getTimeAndValueColumn(0), defaultTimeRange);
Assert.assertFalse(sumAccumulator.hasFinalResult());
- ColumnBuilder[] intermediateResult = new ColumnBuilder[1];
intermediateResult[0] = new DoubleColumnBuilder(null, 1);
sumAccumulator.outputIntermediate(intermediateResult);
Assert.assertEquals(4950d, intermediateResult[0].build().getDouble(0), 0.001);
// add intermediate result as input
sumAccumulator.addIntermediate(new Column[] {intermediateResult[0].build()});
- ColumnBuilder finalResult = new DoubleColumnBuilder(null, 1);
+ finalResult = new DoubleColumnBuilder(null, 1);
sumAccumulator.outputFinal(finalResult);
Assert.assertEquals(9900d, finalResult.build().getDouble(0), 0.001);
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/aggregation/TimeRangeIteratorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/aggregation/TimeRangeIteratorTest.java
new file mode 100644
index 0000000000..48a3f53553
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/aggregation/TimeRangeIteratorTest.java
@@ -0,0 +1,298 @@
+/*
+ * 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.aggregation;
+
+import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.ITimeRangeIterator;
+import org.apache.iotdb.db.mpp.aggregation.timerangeiterator.TimeRangeIteratorFactory;
+import org.apache.iotdb.tsfile.read.common.TimeRange;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TimeRangeIteratorTest {
+
+ @Test
+ public void testNotSplitTimeRange() {
+ String[] res = {
+ "[ 0 : 3 ]",
+ "[ 3 : 6 ]",
+ "[ 6 : 9 ]",
+ "[ 9 : 12 ]",
+ "[ 12 : 15 ]",
+ "[ 15 : 18 ]",
+ "[ 18 : 21 ]",
+ "[ 21 : 24 ]",
+ "[ 24 : 27 ]",
+ "[ 27 : 30 ]",
+ "[ 30 : 31 ]"
+ };
+
+ long startTime = 0, endTime = 32, interval = 4, slidingStep = 3;
+
+ ITimeRangeIterator timeRangeIterator =
+ TimeRangeIteratorFactory.getTimeRangeIterator(
+ startTime, endTime, interval, slidingStep, true, false, false, true, false);
+
+ checkRes(timeRangeIterator, res);
+
+ ITimeRangeIterator descTimeRangeIterator =
+ TimeRangeIteratorFactory.getTimeRangeIterator(
+ startTime, endTime, interval, slidingStep, false, false, false, true, false);
+
+ checkRes(descTimeRangeIterator, res);
+ }
+
+ @Test
+ public void testSplitTimeRange() {
+ String[] res4_1 = {
+ "[ 0 : 0 ]",
+ "[ 1 : 1 ]",
+ "[ 2 : 2 ]",
+ "[ 3 : 3 ]",
+ "[ 4 : 4 ]",
+ "[ 5 : 5 ]",
+ "[ 6 : 6 ]",
+ "[ 7 : 7 ]",
+ "[ 8 : 8 ]",
+ "[ 9 : 9 ]",
+ "[ 10 : 10 ]",
+ "[ 11 : 11 ]",
+ "[ 12 : 12 ]",
+ "[ 13 : 13 ]",
+ "[ 14 : 14 ]",
+ "[ 15 : 15 ]",
+ "[ 16 : 16 ]",
+ "[ 17 : 17 ]",
+ "[ 18 : 18 ]",
+ "[ 19 : 19 ]",
+ "[ 20 : 20 ]",
+ "[ 21 : 21 ]",
+ "[ 22 : 22 ]",
+ "[ 23 : 23 ]",
+ "[ 24 : 24 ]",
+ "[ 25 : 25 ]",
+ "[ 26 : 26 ]",
+ "[ 27 : 27 ]",
+ "[ 28 : 28 ]",
+ "[ 29 : 29 ]",
+ "[ 30 : 30 ]",
+ "[ 31 : 31 ]"
+ };
+ String[] res4_2 = {
+ "[ 0 : 1 ]",
+ "[ 2 : 3 ]",
+ "[ 4 : 5 ]",
+ "[ 6 : 7 ]",
+ "[ 8 : 9 ]",
+ "[ 10 : 11 ]",
+ "[ 12 : 13 ]",
+ "[ 14 : 15 ]",
+ "[ 16 : 17 ]",
+ "[ 18 : 19 ]",
+ "[ 20 : 21 ]",
+ "[ 22 : 23 ]",
+ "[ 24 : 25 ]",
+ "[ 26 : 27 ]",
+ "[ 28 : 29 ]",
+ "[ 30 : 31 ]"
+ };
+ String[] res4_3 = {
+ "[ 0 : 0 ]",
+ "[ 1 : 2 ]",
+ "[ 3 : 3 ]",
+ "[ 4 : 5 ]",
+ "[ 6 : 6 ]",
+ "[ 7 : 8 ]",
+ "[ 9 : 9 ]",
+ "[ 10 : 11 ]",
+ "[ 12 : 12 ]",
+ "[ 13 : 14 ]",
+ "[ 15 : 15 ]",
+ "[ 16 : 17 ]",
+ "[ 18 : 18 ]",
+ "[ 19 : 20 ]",
+ "[ 21 : 21 ]",
+ "[ 22 : 23 ]",
+ "[ 24 : 24 ]",
+ "[ 25 : 26 ]",
+ "[ 27 : 27 ]",
+ "[ 28 : 29 ]",
+ "[ 30 : 30 ]",
+ "[ 31 : 31 ]"
+ };
+ String[] res4_4 = {
+ "[ 0 : 3 ]",
+ "[ 4 : 7 ]",
+ "[ 8 : 11 ]",
+ "[ 12 : 15 ]",
+ "[ 16 : 19 ]",
+ "[ 20 : 23 ]",
+ "[ 24 : 27 ]",
+ "[ 28 : 31 ]"
+ };
+ String[] res4_5 = {
+ "[ 0 : 3 ]",
+ "[ 5 : 8 ]",
+ "[ 10 : 13 ]",
+ "[ 15 : 18 ]",
+ "[ 20 : 23 ]",
+ "[ 25 : 28 ]",
+ "[ 30 : 31 ]"
+ };
+ String[] res4_6 = {
+ "[ 0 : 3 ]", "[ 6 : 9 ]", "[ 12 : 15 ]", "[ 18 : 21 ]", "[ 24 : 27 ]", "[ 30 : 31 ]"
+ };
+
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 1, true, false, false, true, true),
+ res4_1);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 2, true, false, false, true, true),
+ res4_2);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 3, true, false, false, true, true),
+ res4_3);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 4, true, false, false, true, true),
+ res4_4);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 5, true, false, false, true, true),
+ res4_5);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 6, true, false, false, true, true),
+ res4_6);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 1, false, false, false, true, true),
+ res4_1);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 2, false, false, false, true, true),
+ res4_2);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 3, false, false, false, true, true),
+ res4_3);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 4, false, false, false, true, true),
+ res4_4);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 5, false, false, false, true, true),
+ res4_5);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(0, 32, 4, 6, false, false, false, true, true),
+ res4_6);
+ }
+
+ @Test
+ public void testNaturalMonthTimeRange() {
+ String[] res1 = {
+ "[ 1604102400000 : 1606694399999 ]",
+ "[ 1606694400000 : 1609372799999 ]",
+ "[ 1609372800000 : 1612051199999 ]",
+ "[ 1612051200000 : 1614470399999 ]",
+ "[ 1614470400000 : 1617148799999 ]"
+ };
+ String[] res2 = {
+ "[ 1604102400000 : 1604966399999 ]",
+ "[ 1606694400000 : 1607558399999 ]",
+ "[ 1609372800000 : 1610236799999 ]",
+ "[ 1612051200000 : 1612915199999 ]",
+ "[ 1614470400000 : 1615334399999 ]"
+ };
+ String[] res3 = {
+ "[ 1604102400000 : 1606694399999 ]",
+ "[ 1604966400000 : 1607558399999 ]",
+ "[ 1605830400000 : 1608422399999 ]",
+ "[ 1606694400000 : 1609372799999 ]",
+ "[ 1607558400000 : 1610236799999 ]",
+ "[ 1608422400000 : 1611100799999 ]",
+ "[ 1609286400000 : 1611964799999 ]",
+ "[ 1610150400000 : 1612828799999 ]",
+ "[ 1611014400000 : 1613692799999 ]",
+ "[ 1611878400000 : 1614470399999 ]",
+ "[ 1612742400000 : 1615161599999 ]",
+ "[ 1613606400000 : 1616025599999 ]",
+ "[ 1614470400000 : 1617148799999 ]",
+ "[ 1615334400000 : 1617148799999 ]",
+ "[ 1616198400000 : 1617148799999 ]",
+ "[ 1617062400000 : 1617148799999 ]"
+ };
+ String[] res4 = {
+ "[ 1604102400000 : 1604966399999 ]",
+ "[ 1604966400000 : 1605830399999 ]",
+ "[ 1605830400000 : 1606694399999 ]",
+ "[ 1606694400000 : 1607558399999 ]",
+ "[ 1607558400000 : 1608422399999 ]",
+ "[ 1608422400000 : 1609286399999 ]",
+ "[ 1609286400000 : 1609372799999 ]",
+ "[ 1609372800000 : 1610150399999 ]",
+ "[ 1610150400000 : 1610236799999 ]",
+ "[ 1610236800000 : 1611014399999 ]",
+ "[ 1611014400000 : 1611100799999 ]",
+ "[ 1611100800000 : 1611878399999 ]",
+ "[ 1611878400000 : 1611964799999 ]",
+ "[ 1611964800000 : 1612742399999 ]",
+ "[ 1612742400000 : 1612828799999 ]",
+ "[ 1612828800000 : 1613606399999 ]",
+ "[ 1613606400000 : 1613692799999 ]",
+ "[ 1613692800000 : 1614470399999 ]",
+ "[ 1614470400000 : 1615161599999 ]",
+ "[ 1615161600000 : 1615334399999 ]",
+ "[ 1615334400000 : 1616025599999 ]",
+ "[ 1616025600000 : 1616198399999 ]",
+ "[ 1616198400000 : 1617062399999 ]",
+ "[ 1617062400000 : 1617148799999 ]"
+ };
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(
+ 1604102400000L, 1617148800000L, 1, 1, true, true, true, true, false),
+ res1);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(
+ 1604102400000L, 1617148800000L, 1, 1, true, true, true, true, true),
+ res1);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(
+ 1604102400000L, 1617148800000L, 864000000, 1, true, false, true, true, false),
+ res2);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(
+ 1604102400000L, 1617148800000L, 864000000, 1, true, false, true, true, true),
+ res2);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(
+ 1604102400000L, 1617148800000L, 1, 864000000, true, true, false, true, false),
+ res3);
+ checkRes(
+ TimeRangeIteratorFactory.getTimeRangeIterator(
+ 1604102400000L, 1617148800000L, 1, 864000000, true, true, false, true, true),
+ res4);
+ }
+
+ private void checkRes(ITimeRangeIterator timeRangeIterator, String[] res) {
+ boolean isAscending = timeRangeIterator.isAscending();
+ int cnt = isAscending ? 0 : res.length - 1;
+
+ // test next time ranges
+ while (timeRangeIterator.hasNextTimeRange()) {
+ TimeRange curTimeRange = timeRangeIterator.nextTimeRange();
+ Assert.assertEquals(res[cnt], curTimeRange.toString());
+ cnt += isAscending ? 1 : -1;
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregateOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregateOperatorTest.java
new file mode 100644
index 0000000000..d4a3392e97
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregateOperatorTest.java
@@ -0,0 +1,321 @@
+/*
+ * 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.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.metadata.path.MeasurementPath;
+import org.apache.iotdb.db.mpp.aggregation.Accumulator;
+import org.apache.iotdb.db.mpp.aggregation.AccumulatorFactory;
+import org.apache.iotdb.db.mpp.aggregation.Aggregator;
+import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
+import org.apache.iotdb.db.mpp.common.PlanFragmentId;
+import org.apache.iotdb.db.mpp.common.QueryId;
+import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
+import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceStateMachine;
+import org.apache.iotdb.db.mpp.execution.operator.process.AggregateOperator;
+import org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregateScanOperator;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationStep;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.InputLocation;
+import org.apache.iotdb.db.query.aggregation.AggregationType;
+import org.apache.iotdb.db.query.reader.series.SeriesReaderTestUtil;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+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 org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+import static org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
+import static org.junit.Assert.assertEquals;
+
+public class AggregateOperatorTest {
+
+ private static final String AGGREGATE_OPERATOR_TEST_SG = "root.AggregateOperatorTest";
+ private final List<String> deviceIds = new ArrayList<>();
+ private final List<MeasurementSchema> measurementSchemas = new ArrayList<>();
+
+ private final List<TsFileResource> seqResources = new ArrayList<>();
+ private final List<TsFileResource> unSeqResources = new ArrayList<>();
+ private ExecutorService instanceNotificationExecutor =
+ IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");;
+
+ @Before
+ public void setUp() throws MetadataException, IOException, WriteProcessException {
+ SeriesReaderTestUtil.setUp(
+ measurementSchemas, deviceIds, seqResources, unSeqResources, AGGREGATE_OPERATOR_TEST_SG);
+ this.instanceNotificationExecutor =
+ IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ SeriesReaderTestUtil.tearDown(seqResources, unSeqResources);
+ instanceNotificationExecutor.shutdown();
+ }
+
+ /** Try to aggregate unary intermediate result of one time series without group by interval. */
+ @Test
+ public void testAggregateIntermediateResult1() throws IllegalPathException {
+ List<AggregationType> aggregationTypes = new ArrayList<>();
+ aggregationTypes.add(AggregationType.COUNT);
+ aggregationTypes.add(AggregationType.SUM);
+ aggregationTypes.add(AggregationType.MIN_TIME);
+ aggregationTypes.add(AggregationType.MAX_TIME);
+ aggregationTypes.add(AggregationType.MAX_VALUE);
+ aggregationTypes.add(AggregationType.MIN_VALUE);
+ List<List<InputLocation[]>> inputLocations = new ArrayList<>();
+ for (int i = 0; i < aggregationTypes.size(); i++) {
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(0, i)});
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(1, i)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+ AggregateOperator aggregateOperator =
+ initAggregateOperator(aggregationTypes, null, inputLocations);
+ int count = 0;
+ while (aggregateOperator.hasNext()) {
+ TsBlock resultTsBlock = aggregateOperator.next();
+ assertEquals(500, resultTsBlock.getColumn(0).getLong(0));
+ assertEquals(6524750.0, resultTsBlock.getColumn(1).getDouble(0), 0.0001);
+ assertEquals(0, resultTsBlock.getColumn(2).getLong(0));
+ assertEquals(499, resultTsBlock.getColumn(3).getLong(0));
+ assertEquals(20199, resultTsBlock.getColumn(4).getInt(0));
+ assertEquals(260, resultTsBlock.getColumn(5).getInt(0));
+ count++;
+ }
+ assertEquals(1, count);
+ }
+
+ /** Try to aggregate binary intermediate result of one time series without group by interval. */
+ @Test
+ public void testAggregateIntermediateResult2() throws IllegalPathException {
+ List<AggregationType> aggregationTypes = new ArrayList<>();
+ aggregationTypes.add(AggregationType.AVG);
+ aggregationTypes.add(AggregationType.FIRST_VALUE);
+ aggregationTypes.add(AggregationType.LAST_VALUE);
+ List<List<InputLocation[]>> inputLocations = new ArrayList<>();
+ for (int i = 0; i < aggregationTypes.size(); i++) {
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(
+ new InputLocation[] {new InputLocation(0, 2 * i), new InputLocation(0, 2 * i + 1)});
+ inputLocationForOneAggregator.add(
+ new InputLocation[] {new InputLocation(1, 2 * i), new InputLocation(1, 2 * i + 1)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+ AggregateOperator aggregateOperator =
+ initAggregateOperator(aggregationTypes, null, inputLocations);
+ int count = 0;
+ while (aggregateOperator.hasNext()) {
+ TsBlock resultTsBlock = aggregateOperator.next();
+ assertEquals(13049.5, resultTsBlock.getColumn(0).getDouble(0), 0.001);
+ assertEquals(20000, resultTsBlock.getColumn(1).getInt(0));
+ assertEquals(10499, resultTsBlock.getColumn(2).getInt(0));
+ count++;
+ }
+ assertEquals(1, count);
+ }
+
+ @Test
+ public void testGroupByIntermediateResult1() throws IllegalPathException {
+ int[][] result =
+ new int[][] {
+ {100, 100, 100, 99},
+ {2004950, 2014950, 624950, 834551},
+ {0, 100, 200, 300},
+ {99, 199, 299, 398},
+ {20099, 20199, 10259, 10379},
+ {20000, 20100, 260, 380}
+ };
+ GroupByTimeParameter groupByTimeParameter = new GroupByTimeParameter(0, 399, 100, 100, true);
+ List<AggregationType> aggregationTypes = new ArrayList<>();
+ aggregationTypes.add(AggregationType.COUNT);
+ aggregationTypes.add(AggregationType.SUM);
+ aggregationTypes.add(AggregationType.MIN_TIME);
+ aggregationTypes.add(AggregationType.MAX_TIME);
+ aggregationTypes.add(AggregationType.MAX_VALUE);
+ aggregationTypes.add(AggregationType.MIN_VALUE);
+ List<List<InputLocation[]>> inputLocations = new ArrayList<>();
+ for (int i = 0; i < aggregationTypes.size(); i++) {
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(0, i)});
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(1, i)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+ AggregateOperator aggregateOperator =
+ initAggregateOperator(aggregationTypes, groupByTimeParameter, inputLocations);
+ int count = 0;
+ while (aggregateOperator.hasNext()) {
+ TsBlock resultTsBlock = aggregateOperator.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++;
+ }
+ assertEquals(4, count);
+ }
+
+ @Test
+ public void testGroupByIntermediateResult2() throws IllegalPathException {
+ double[][] result =
+ new double[][] {
+ {20049.5, 20149.5, 6249.5, 8429.808},
+ {20000, 20100, 10200, 10300},
+ {20099, 20199, 299, 398},
+ };
+ List<AggregationType> aggregationTypes = new ArrayList<>();
+ aggregationTypes.add(AggregationType.AVG);
+ aggregationTypes.add(AggregationType.FIRST_VALUE);
+ aggregationTypes.add(AggregationType.LAST_VALUE);
+ GroupByTimeParameter groupByTimeParameter = new GroupByTimeParameter(0, 399, 100, 100, true);
+ List<List<InputLocation[]>> inputLocations = new ArrayList<>();
+ for (int i = 0; i < aggregationTypes.size(); i++) {
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(
+ new InputLocation[] {new InputLocation(0, 2 * i), new InputLocation(0, 2 * i + 1)});
+ inputLocationForOneAggregator.add(
+ new InputLocation[] {new InputLocation(1, 2 * i), new InputLocation(1, 2 * i + 1)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+ AggregateOperator aggregateOperator =
+ initAggregateOperator(aggregationTypes, groupByTimeParameter, inputLocations);
+ int count = 0;
+ while (aggregateOperator.hasNext()) {
+ TsBlock resultTsBlock = aggregateOperator.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++;
+ }
+ assertEquals(4, count);
+ }
+
+ /**
+ * @param aggregationTypes Aggregation function used in test
+ * @param groupByTimeParameter group by time parameter
+ * @param inputLocations each inputLocation is used in one aggregator
+ */
+ private AggregateOperator initAggregateOperator(
+ List<AggregationType> aggregationTypes,
+ GroupByTimeParameter groupByTimeParameter,
+ List<List<InputLocation[]>> inputLocations)
+ throws IllegalPathException {
+ // Construct operator tree
+ QueryId queryId = new QueryId("stub_query");
+ FragmentInstanceId instanceId =
+ new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance");
+ FragmentInstanceStateMachine stateMachine =
+ new FragmentInstanceStateMachine(instanceId, instanceNotificationExecutor);
+ FragmentInstanceContext fragmentInstanceContext =
+ createFragmentInstanceContext(instanceId, stateMachine);
+ PlanNodeId planNodeId1 = new PlanNodeId("1");
+ fragmentInstanceContext.addOperatorContext(
+ 1, planNodeId1, SeriesAggregateScanOperator.class.getSimpleName());
+ PlanNodeId planNodeId2 = new PlanNodeId("2");
+ fragmentInstanceContext.addOperatorContext(
+ 2, planNodeId2, SeriesAggregateScanOperator.class.getSimpleName());
+ PlanNodeId planNodeId3 = new PlanNodeId("3");
+ fragmentInstanceContext.addOperatorContext(
+ 3, planNodeId3, AggregateOperator.class.getSimpleName());
+
+ MeasurementPath measurementPath1 =
+ new MeasurementPath(AGGREGATE_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
+ List<Aggregator> aggregators = new ArrayList<>();
+ AccumulatorFactory.createAccumulators(aggregationTypes, TSDataType.INT32, true)
+ .forEach(o -> aggregators.add(new Aggregator(o, AggregationStep.PARTIAL)));
+ SeriesAggregateScanOperator seriesAggregateScanOperator1 =
+ new SeriesAggregateScanOperator(
+ planNodeId1,
+ measurementPath1,
+ Collections.singleton("sensor0"),
+ fragmentInstanceContext.getOperatorContexts().get(0),
+ aggregators,
+ null,
+ true,
+ groupByTimeParameter);
+ List<TsFileResource> seqResources1 = new ArrayList<>();
+ List<TsFileResource> unSeqResources1 = new ArrayList<>();
+ seqResources1.add(seqResources.get(0));
+ seqResources1.add(seqResources.get(1));
+ seqResources1.add(seqResources.get(3));
+ unSeqResources1.add(unSeqResources.get(0));
+ unSeqResources1.add(unSeqResources.get(1));
+ unSeqResources1.add(unSeqResources.get(3));
+ unSeqResources1.add(unSeqResources.get(5));
+ seriesAggregateScanOperator1.initQueryDataSource(
+ new QueryDataSource(seqResources1, unSeqResources1));
+
+ SeriesAggregateScanOperator seriesAggregateScanOperator2 =
+ new SeriesAggregateScanOperator(
+ planNodeId2,
+ measurementPath1,
+ Collections.singleton("sensor0"),
+ fragmentInstanceContext.getOperatorContexts().get(1),
+ aggregators,
+ null,
+ true,
+ groupByTimeParameter);
+ List<TsFileResource> seqResources2 = new ArrayList<>();
+ List<TsFileResource> unSeqResources2 = new ArrayList<>();
+ seqResources2.add(seqResources.get(2));
+ seqResources2.add(seqResources.get(4));
+ unSeqResources2.add(unSeqResources.get(2));
+ unSeqResources2.add(unSeqResources.get(4));
+ seriesAggregateScanOperator2.initQueryDataSource(
+ new QueryDataSource(seqResources2, unSeqResources2));
+
+ List<Operator> children = new ArrayList<>();
+ children.add(seriesAggregateScanOperator1);
+ children.add(seriesAggregateScanOperator2);
+
+ List<Aggregator> finalAggregators = new ArrayList<>();
+ List<Accumulator> accumulators =
+ AccumulatorFactory.createAccumulators(aggregationTypes, TSDataType.INT32, true);
+ for (int i = 0; i < accumulators.size(); i++) {
+ finalAggregators.add(
+ new Aggregator(accumulators.get(i), AggregationStep.FINAL, inputLocations.get(i)));
+ }
+
+ return new AggregateOperator(
+ fragmentInstanceContext.getOperatorContexts().get(2),
+ finalAggregators,
+ children,
+ true,
+ groupByTimeParameter);
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregateOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregateOperatorTest.java
new file mode 100644
index 0000000000..0de2bd6b35
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregateOperatorTest.java
@@ -0,0 +1,368 @@
+/*
+ * 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.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.metadata.path.MeasurementPath;
+import org.apache.iotdb.db.mpp.aggregation.Accumulator;
+import org.apache.iotdb.db.mpp.aggregation.AccumulatorFactory;
+import org.apache.iotdb.db.mpp.aggregation.Aggregator;
+import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
+import org.apache.iotdb.db.mpp.common.PlanFragmentId;
+import org.apache.iotdb.db.mpp.common.QueryId;
+import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
+import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceStateMachine;
+import org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregateOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.merge.AscTimeComparator;
+import org.apache.iotdb.db.mpp.execution.operator.process.merge.SingleColumnMerger;
+import org.apache.iotdb.db.mpp.execution.operator.source.SeriesScanOperator;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationStep;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.InputLocation;
+import org.apache.iotdb.db.mpp.plan.statement.component.OrderBy;
+import org.apache.iotdb.db.query.aggregation.AggregationType;
+import org.apache.iotdb.db.query.reader.series.SeriesReaderTestUtil;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+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 org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+
+import static org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
+import static org.junit.Assert.assertEquals;
+
+public class RawDataAggregateOperatorTest {
+
+ private static final String AGGREGATE_OPERATOR_TEST_SG = "root.RawDataAggregateOperatorTest";
+ private final List<String> deviceIds = new ArrayList<>();
+ private final List<MeasurementSchema> measurementSchemas = new ArrayList<>();
+
+ private final List<TsFileResource> seqResources = new ArrayList<>();
+ private final List<TsFileResource> unSeqResources = new ArrayList<>();
+ private ExecutorService instanceNotificationExecutor;
+
+ @Before
+ public void setUp() throws MetadataException, IOException, WriteProcessException {
+ SeriesReaderTestUtil.setUp(
+ measurementSchemas, deviceIds, seqResources, unSeqResources, AGGREGATE_OPERATOR_TEST_SG);
+ this.instanceNotificationExecutor =
+ IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ SeriesReaderTestUtil.tearDown(seqResources, unSeqResources);
+ instanceNotificationExecutor.shutdown();
+ }
+
+ /**
+ * Test aggregating raw data without group by interval.
+ *
+ * <p>Example SQL: select count(s0), sum(s0), min_time(s0), max_time(s0), min_value(s0),
+ * max_value(s0), count(s1), sum(s1), min_time(s1), max_time(s1), min_value(s1), max_value(s1)
+ * from root.sg.d0 where s1 > 10 and s2 < 15
+ *
+ * <p>For convenience, we don't use FilterOperator in test though rawDataAggregateOperator is
+ * always used with value filter.
+ */
+ @Test
+ public void aggregateRawDataTest1() throws IllegalPathException {
+ List<AggregationType> aggregationTypes = new ArrayList<>();
+ List<List<InputLocation[]>> inputLocations = new ArrayList<>();
+ for (int i = 0; i < 2; i++) {
+ aggregationTypes.add(AggregationType.COUNT);
+ aggregationTypes.add(AggregationType.SUM);
+ aggregationTypes.add(AggregationType.MIN_TIME);
+ aggregationTypes.add(AggregationType.MAX_TIME);
+ aggregationTypes.add(AggregationType.MAX_VALUE);
+ aggregationTypes.add(AggregationType.MIN_VALUE);
+ for (int j = 0; j < 6; j++) {
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(0, i)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+ }
+
+ RawDataAggregateOperator rawDataAggregateOperator =
+ initRawDataAggregateOperator(aggregationTypes, null, inputLocations);
+ int count = 0;
+ while (rawDataAggregateOperator.hasNext()) {
+ TsBlock resultTsBlock = rawDataAggregateOperator.next();
+ for (int i = 0; i < 2; i++) {
+ assertEquals(500, resultTsBlock.getColumn(6 * i).getLong(0));
+ assertEquals(6524750.0, resultTsBlock.getColumn(6 * i + 1).getDouble(0), 0.0001);
+ assertEquals(0, resultTsBlock.getColumn(6 * i + 2).getLong(0));
+ assertEquals(499, resultTsBlock.getColumn(6 * i + 3).getLong(0));
+ assertEquals(20199, resultTsBlock.getColumn(6 * i + 4).getInt(0));
+ assertEquals(260, resultTsBlock.getColumn(6 * i + 5).getInt(0));
+ }
+ count++;
+ }
+ assertEquals(1, count);
+ }
+
+ /**
+ * Test aggregating raw data without group by interval.
+ *
+ * <p>Example SQL: select avg(s0), avg(s1), first_value(s0), first_value(s1), last_value(s0),
+ * last_value(s1) from root.sg.d0 where s1 > 10 and s2 < 15
+ *
+ * <p>For convenience, we don't use FilterOperator in test though rawDataAggregateOperator is
+ * always used with value filter.
+ */
+ @Test
+ public void aggregateRawDataTest2() throws IllegalPathException {
+ List<AggregationType> aggregationTypes = new ArrayList<>();
+ List<List<InputLocation[]>> inputLocations = new ArrayList<>();
+ for (int i = 0; i < 2; i++) {
+ aggregationTypes.add(AggregationType.AVG);
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(0, i)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+ for (int i = 0; i < 2; i++) {
+ aggregationTypes.add(AggregationType.FIRST_VALUE);
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(0, i)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+ for (int i = 0; i < 2; i++) {
+ aggregationTypes.add(AggregationType.LAST_VALUE);
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(0, i)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+
+ RawDataAggregateOperator rawDataAggregateOperator =
+ initRawDataAggregateOperator(aggregationTypes, null, inputLocations);
+ int count = 0;
+ while (rawDataAggregateOperator.hasNext()) {
+ TsBlock resultTsBlock = rawDataAggregateOperator.next();
+ for (int i = 0; i < 2; i++) {
+ assertEquals(13049.5, resultTsBlock.getColumn(i).getDouble(0), 0.001);
+ }
+ for (int i = 2; i < 4; i++) {
+ assertEquals(20000, resultTsBlock.getColumn(i).getInt(0));
+ }
+ for (int i = 4; i < 6; i++) {
+ assertEquals(10499, resultTsBlock.getColumn(i).getInt(0));
+ }
+ count++;
+ }
+ assertEquals(1, count);
+ }
+
+ /** Test aggregating raw data by time interval. */
+ @Test
+ public void groupByRawDataTest1() throws IllegalPathException {
+ int[][] result =
+ new int[][] {
+ {100, 100, 100, 99},
+ {2004950, 2014950, 624950, 834551},
+ {0, 100, 200, 300},
+ {99, 199, 299, 398},
+ {20099, 20199, 10259, 10379},
+ {20000, 20100, 260, 380}
+ };
+ List<AggregationType> aggregationTypes = new ArrayList<>();
+ List<List<InputLocation[]>> inputLocations = new ArrayList<>();
+ for (int i = 0; i < 2; i++) {
+ aggregationTypes.add(AggregationType.COUNT);
+ aggregationTypes.add(AggregationType.SUM);
+ aggregationTypes.add(AggregationType.MIN_TIME);
+ aggregationTypes.add(AggregationType.MAX_TIME);
+ aggregationTypes.add(AggregationType.MAX_VALUE);
+ aggregationTypes.add(AggregationType.MIN_VALUE);
+ for (int j = 0; j < 6; j++) {
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(0, i)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+ }
+ GroupByTimeParameter groupByTimeParameter = new GroupByTimeParameter(0, 399, 100, 100, true);
+
+ RawDataAggregateOperator rawDataAggregateOperator =
+ initRawDataAggregateOperator(aggregationTypes, groupByTimeParameter, inputLocations);
+ int count = 0;
+ while (rawDataAggregateOperator.hasNext()) {
+ TsBlock resultTsBlock = rawDataAggregateOperator.next();
+ assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(0));
+ for (int i = 0; i < 2; i++) {
+ assertEquals(result[0][count], resultTsBlock.getColumn(6 * i).getLong(0));
+ assertEquals(result[1][count], resultTsBlock.getColumn(6 * i + 1).getDouble(0), 0.0001);
+ assertEquals(result[2][count], resultTsBlock.getColumn(6 * i + 2).getLong(0));
+ assertEquals(result[3][count], resultTsBlock.getColumn(6 * i + 3).getLong(0));
+ assertEquals(result[4][count], resultTsBlock.getColumn(6 * i + 4).getInt(0));
+ assertEquals(result[5][count], resultTsBlock.getColumn(6 * i + 5).getInt(0));
+ }
+ count++;
+ }
+ assertEquals(4, count);
+ }
+
+ /** Test aggregating raw data by time interval. */
+ @Test
+ public void groupByRawDataTest2() throws IllegalPathException {
+ double[][] result =
+ new double[][] {
+ {20049.5, 20149.5, 6249.5, 8429.808},
+ {20000, 20100, 10200, 10300},
+ {20099, 20199, 299, 398},
+ };
+ List<AggregationType> aggregationTypes = new ArrayList<>();
+ List<List<InputLocation[]>> inputLocations = new ArrayList<>();
+ for (int i = 0; i < 2; i++) {
+ aggregationTypes.add(AggregationType.AVG);
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(0, i)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+ for (int i = 0; i < 2; i++) {
+ aggregationTypes.add(AggregationType.FIRST_VALUE);
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(0, i)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+ for (int i = 0; i < 2; i++) {
+ aggregationTypes.add(AggregationType.LAST_VALUE);
+ List<InputLocation[]> inputLocationForOneAggregator = new ArrayList<>();
+ inputLocationForOneAggregator.add(new InputLocation[] {new InputLocation(0, i)});
+ inputLocations.add(inputLocationForOneAggregator);
+ }
+ GroupByTimeParameter groupByTimeParameter = new GroupByTimeParameter(0, 399, 100, 100, true);
+ RawDataAggregateOperator rawDataAggregateOperator =
+ initRawDataAggregateOperator(aggregationTypes, groupByTimeParameter, inputLocations);
+ int count = 0;
+ while (rawDataAggregateOperator.hasNext()) {
+ TsBlock resultTsBlock = rawDataAggregateOperator.next();
+ assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(0));
+ for (int i = 0; i < 2; i++) {
+ assertEquals(result[0][count], resultTsBlock.getColumn(i).getDouble(0), 0.001);
+ }
+ for (int i = 2; i < 4; i++) {
+ assertEquals((int) result[1][count], resultTsBlock.getColumn(i).getInt(0));
+ }
+ for (int i = 4; i < 6; i++) {
+ assertEquals((int) result[2][count], resultTsBlock.getColumn(i).getInt(0));
+ }
+ count++;
+ }
+ assertEquals(4, count);
+ }
+
+ private RawDataAggregateOperator initRawDataAggregateOperator(
+ List<AggregationType> aggregationTypes,
+ GroupByTimeParameter groupByTimeParameter,
+ List<List<InputLocation[]>> inputLocations)
+ throws IllegalPathException {
+ ExecutorService instanceNotificationExecutor =
+ IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
+
+ MeasurementPath measurementPath1 =
+ new MeasurementPath(AGGREGATE_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
+ Set<String> allSensors = new HashSet<>();
+ allSensors.add("sensor0");
+ allSensors.add("sensor1");
+ QueryId queryId = new QueryId("stub_query");
+ FragmentInstanceId instanceId =
+ new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance");
+ FragmentInstanceStateMachine stateMachine =
+ new FragmentInstanceStateMachine(instanceId, instanceNotificationExecutor);
+ FragmentInstanceContext fragmentInstanceContext =
+ createFragmentInstanceContext(instanceId, stateMachine);
+ PlanNodeId planNodeId1 = new PlanNodeId("1");
+ fragmentInstanceContext.addOperatorContext(
+ 1, planNodeId1, SeriesScanOperator.class.getSimpleName());
+ PlanNodeId planNodeId2 = new PlanNodeId("2");
+ fragmentInstanceContext.addOperatorContext(
+ 2, planNodeId2, SeriesScanOperator.class.getSimpleName());
+ fragmentInstanceContext.addOperatorContext(
+ 3, new PlanNodeId("3"), TimeJoinOperator.class.getSimpleName());
+ fragmentInstanceContext.addOperatorContext(
+ 4, new PlanNodeId("4"), RawDataAggregateOperatorTest.class.getSimpleName());
+ SeriesScanOperator seriesScanOperator1 =
+ new SeriesScanOperator(
+ planNodeId1,
+ measurementPath1,
+ allSensors,
+ TSDataType.INT32,
+ fragmentInstanceContext.getOperatorContexts().get(0),
+ null,
+ null,
+ true);
+ seriesScanOperator1.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
+
+ MeasurementPath measurementPath2 =
+ new MeasurementPath(AGGREGATE_OPERATOR_TEST_SG + ".device0.sensor1", TSDataType.INT32);
+ SeriesScanOperator seriesScanOperator2 =
+ new SeriesScanOperator(
+ planNodeId2,
+ measurementPath2,
+ allSensors,
+ TSDataType.INT32,
+ fragmentInstanceContext.getOperatorContexts().get(1),
+ null,
+ null,
+ true);
+ seriesScanOperator2.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
+
+ TimeJoinOperator timeJoinOperator =
+ new TimeJoinOperator(
+ fragmentInstanceContext.getOperatorContexts().get(2),
+ Arrays.asList(seriesScanOperator1, seriesScanOperator2),
+ OrderBy.TIMESTAMP_ASC,
+ Arrays.asList(TSDataType.INT32, TSDataType.INT32),
+ Arrays.asList(
+ new SingleColumnMerger(new InputLocation(0, 0), new AscTimeComparator()),
+ new SingleColumnMerger(new InputLocation(1, 0), new AscTimeComparator())),
+ new AscTimeComparator());
+
+ List<Aggregator> aggregators = new ArrayList<>();
+ List<Accumulator> accumulators =
+ AccumulatorFactory.createAccumulators(aggregationTypes, TSDataType.INT32, true);
+ for (int i = 0; i < accumulators.size(); i++) {
+ aggregators.add(
+ new Aggregator(accumulators.get(i), AggregationStep.SINGLE, inputLocations.get(i)));
+ }
+ return new RawDataAggregateOperator(
+ fragmentInstanceContext.getOperatorContexts().get(3),
+ aggregators,
+ timeJoinOperator,
+ true,
+ groupByTimeParameter);
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SeriesAggregateScanOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SeriesAggregateScanOperatorTest.java
index a109b52609..1c7a77ded6 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SeriesAggregateScanOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SeriesAggregateScanOperatorTest.java
@@ -104,6 +104,23 @@ public class SeriesAggregateScanOperatorTest {
assertEquals(1, count);
}
+ @Test
+ public void testAggregationWithoutTimeFilterOrderByTimeDesc() throws IllegalPathException {
+ List<AggregationType> aggregationTypes = Collections.singletonList(AggregationType.COUNT);
+ List<Aggregator> aggregators = new ArrayList<>();
+ AccumulatorFactory.createAccumulators(aggregationTypes, TSDataType.INT32, true)
+ .forEach(o -> aggregators.add(new Aggregator(o, AggregationStep.SINGLE)));
+ SeriesAggregateScanOperator seriesAggregateScanOperator =
+ initSeriesAggregateScanOperator(aggregators, null, false, null);
+ int count = 0;
+ while (seriesAggregateScanOperator.hasNext()) {
+ TsBlock resultTsBlock = seriesAggregateScanOperator.next();
+ assertEquals(500, resultTsBlock.getColumn(0).getLong(0));
+ count++;
+ }
+ assertEquals(1, count);
+ }
+
@Test
public void testMultiAggregationFuncWithoutTimeFilter1() throws IllegalPathException {
List<AggregationType> aggregationTypes = new ArrayList<>();
@@ -266,7 +283,7 @@ public class SeriesAggregateScanOperatorTest {
@Test
public void testGroupByWithoutGlobalTimeFilter() throws IllegalPathException {
- int[] result = new int[] {100, 100, 100, 100};
+ int[] result = new int[] {100, 100, 100, 99};
GroupByTimeParameter groupByTimeParameter = new GroupByTimeParameter(0, 399, 100, 100, true);
List<AggregationType> aggregationTypes = Collections.singletonList(AggregationType.COUNT);
List<Aggregator> aggregators = new ArrayList<>();
@@ -310,7 +327,7 @@ public class SeriesAggregateScanOperatorTest {
int[][] result =
new int[][] {
{20000, 20100, 10200, 10300},
- {20099, 20199, 299, 399},
+ {20099, 20199, 299, 398},
{20099, 20199, 10259, 10379},
{20000, 20100, 260, 380}
};
@@ -343,7 +360,7 @@ public class SeriesAggregateScanOperatorTest {
int[][] result =
new int[][] {
{20000, 20100, 10200, 10300},
- {20099, 20199, 299, 399},
+ {20099, 20199, 299, 398},
{20099, 20199, 10259, 10379},
{20000, 20100, 260, 380}
};
@@ -373,7 +390,7 @@ public class SeriesAggregateScanOperatorTest {
@Test
public void testGroupBySlidingTimeWindow() throws IllegalPathException {
- int[] result = new int[] {50, 50, 50, 50, 50, 50, 50, 50};
+ int[] result = new int[] {50, 50, 50, 50, 50, 50, 50, 49};
GroupByTimeParameter groupByTimeParameter = new GroupByTimeParameter(0, 399, 100, 50, true);
List<AggregationType> aggregationTypes = Collections.singletonList(AggregationType.COUNT);
List<Aggregator> aggregators = new ArrayList<>();
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/process/AggregationNodeSerdeTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/process/AggregationNodeSerdeTest.java
index e1ead56759..1ebddfd736 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/process/AggregationNodeSerdeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/process/AggregationNodeSerdeTest.java
@@ -43,6 +43,7 @@ import org.junit.Test;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import static org.apache.iotdb.tsfile.read.filter.factory.FilterType.VALUE_FILTER;
import static org.junit.Assert.assertEquals;
@@ -85,4 +86,231 @@ public class AggregationNodeSerdeTest {
byteBuffer.flip();
assertEquals(PlanNodeDeserializeHelper.deserialize(byteBuffer), aggregationNode);
}
+
+ /** Test AVG with COUNT and SUM (former or later). */
+ @Test
+ public void getDeduplicatedDescriptorsTest1() {
+ PartialPath seriesPath1 = new PartialPath(new String[] {"root", "sg", "d1", "s1"});
+ List<AggregationType> aggregationTypeList = new ArrayList<>();
+ aggregationTypeList.add(AggregationType.COUNT);
+ aggregationTypeList.add(AggregationType.AVG);
+ aggregationTypeList.add(AggregationType.SUM);
+ List<AggregationDescriptor> descriptorList = new ArrayList<>();
+ for (AggregationType aggregationType : aggregationTypeList) {
+ descriptorList.add(
+ new AggregationDescriptor(
+ aggregationType,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ }
+ List<AggregationDescriptor> deduplicatedDescriptors =
+ AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(1, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.AVG, deduplicatedDescriptors.get(0).getAggregationType());
+
+ aggregationTypeList = new ArrayList<>();
+ aggregationTypeList.add(AggregationType.COUNT);
+ aggregationTypeList.add(AggregationType.SUM);
+ aggregationTypeList.add(AggregationType.AVG);
+ descriptorList = new ArrayList<>();
+ for (AggregationType aggregationType : aggregationTypeList) {
+ descriptorList.add(
+ new AggregationDescriptor(
+ aggregationType,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ }
+ deduplicatedDescriptors = AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(1, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.AVG, deduplicatedDescriptors.get(0).getAggregationType());
+
+ aggregationTypeList = new ArrayList<>();
+ aggregationTypeList.add(AggregationType.AVG);
+ aggregationTypeList.add(AggregationType.COUNT);
+ aggregationTypeList.add(AggregationType.SUM);
+ descriptorList = new ArrayList<>();
+ for (AggregationType aggregationType : aggregationTypeList) {
+ descriptorList.add(
+ new AggregationDescriptor(
+ aggregationType,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ }
+ deduplicatedDescriptors = AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(1, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.AVG, deduplicatedDescriptors.get(0).getAggregationType());
+
+ // try output final
+ descriptorList = new ArrayList<>();
+ for (AggregationType aggregationType : aggregationTypeList) {
+ descriptorList.add(
+ new AggregationDescriptor(
+ aggregationType,
+ AggregationStep.SINGLE,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ }
+ deduplicatedDescriptors = AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(3, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.AVG, deduplicatedDescriptors.get(0).getAggregationType());
+ assertEquals(AggregationType.COUNT, deduplicatedDescriptors.get(1).getAggregationType());
+ assertEquals(AggregationType.SUM, deduplicatedDescriptors.get(2).getAggregationType());
+ }
+
+ /** Test FIRST_VALUE with MIN_TIME (former or later). */
+ @Test
+ public void getDeduplicatedDescriptorsTest2() {
+ PartialPath seriesPath1 = new PartialPath(new String[] {"root", "sg", "d1", "s1"});
+ List<AggregationType> aggregationTypeList = new ArrayList<>();
+ aggregationTypeList.add(AggregationType.FIRST_VALUE);
+ aggregationTypeList.add(AggregationType.MIN_TIME);
+ List<AggregationDescriptor> descriptorList = new ArrayList<>();
+ for (AggregationType aggregationType : aggregationTypeList) {
+ descriptorList.add(
+ new AggregationDescriptor(
+ aggregationType,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ }
+ List<AggregationDescriptor> deduplicatedDescriptors =
+ AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(1, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.FIRST_VALUE, deduplicatedDescriptors.get(0).getAggregationType());
+
+ aggregationTypeList = new ArrayList<>();
+ aggregationTypeList.add(AggregationType.MIN_TIME);
+ aggregationTypeList.add(AggregationType.FIRST_VALUE);
+ descriptorList = new ArrayList<>();
+ for (AggregationType aggregationType : aggregationTypeList) {
+ descriptorList.add(
+ new AggregationDescriptor(
+ aggregationType,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ }
+ deduplicatedDescriptors = AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(1, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.FIRST_VALUE, deduplicatedDescriptors.get(0).getAggregationType());
+
+ // try output final
+ descriptorList = new ArrayList<>();
+ for (AggregationType aggregationType : aggregationTypeList) {
+ descriptorList.add(
+ new AggregationDescriptor(
+ aggregationType,
+ AggregationStep.SINGLE,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ }
+ deduplicatedDescriptors = AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(2, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.MIN_TIME, deduplicatedDescriptors.get(0).getAggregationType());
+ assertEquals(AggregationType.FIRST_VALUE, deduplicatedDescriptors.get(1).getAggregationType());
+ }
+
+ /** Test LAST_VALUE with MAX_TIME (former or later). */
+ @Test
+ public void getDeduplicatedDescriptorsTest3() {
+ PartialPath seriesPath1 = new PartialPath(new String[] {"root", "sg", "d1", "s1"});
+ List<AggregationType> aggregationTypeList = new ArrayList<>();
+ aggregationTypeList.add(AggregationType.LAST_VALUE);
+ aggregationTypeList.add(AggregationType.MAX_TIME);
+ List<AggregationDescriptor> descriptorList = new ArrayList<>();
+ for (AggregationType aggregationType : aggregationTypeList) {
+ descriptorList.add(
+ new AggregationDescriptor(
+ aggregationType,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ }
+ List<AggregationDescriptor> deduplicatedDescriptors =
+ AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(1, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.LAST_VALUE, deduplicatedDescriptors.get(0).getAggregationType());
+
+ aggregationTypeList = new ArrayList<>();
+ aggregationTypeList.add(AggregationType.MAX_TIME);
+ aggregationTypeList.add(AggregationType.LAST_VALUE);
+ descriptorList = new ArrayList<>();
+ for (AggregationType aggregationType : aggregationTypeList) {
+ descriptorList.add(
+ new AggregationDescriptor(
+ aggregationType,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ }
+ deduplicatedDescriptors = AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(1, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.LAST_VALUE, deduplicatedDescriptors.get(0).getAggregationType());
+
+ // try output final
+ descriptorList = new ArrayList<>();
+ for (AggregationType aggregationType : aggregationTypeList) {
+ descriptorList.add(
+ new AggregationDescriptor(
+ aggregationType,
+ AggregationStep.SINGLE,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ }
+ deduplicatedDescriptors = AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(2, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.MAX_TIME, deduplicatedDescriptors.get(0).getAggregationType());
+ assertEquals(AggregationType.LAST_VALUE, deduplicatedDescriptors.get(1).getAggregationType());
+ }
+
+ /** Test AVG with COUNT but work on different time series. */
+ @Test
+ public void getDeduplicatedDescriptorsTest4() {
+ PartialPath seriesPath1 = new PartialPath(new String[] {"root", "sg", "d1", "s1"});
+ PartialPath seriesPath2 = new PartialPath(new String[] {"root", "sg", "d1", "s2"});
+ List<AggregationDescriptor> descriptorList = new ArrayList<>();
+ descriptorList.add(
+ new AggregationDescriptor(
+ AggregationType.AVG,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ descriptorList.add(
+ new AggregationDescriptor(
+ AggregationType.COUNT,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath2))));
+
+ List<AggregationDescriptor> deduplicatedDescriptors =
+ AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(2, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.AVG, deduplicatedDescriptors.get(0).getAggregationType());
+ assertEquals(AggregationType.COUNT, deduplicatedDescriptors.get(1).getAggregationType());
+
+ descriptorList = new ArrayList<>();
+ descriptorList.add(
+ new AggregationDescriptor(
+ AggregationType.FIRST_VALUE,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ descriptorList.add(
+ new AggregationDescriptor(
+ AggregationType.MIN_TIME,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath2))));
+
+ deduplicatedDescriptors = AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(2, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.FIRST_VALUE, deduplicatedDescriptors.get(0).getAggregationType());
+ assertEquals(AggregationType.MIN_TIME, deduplicatedDescriptors.get(1).getAggregationType());
+
+ descriptorList = new ArrayList<>();
+ descriptorList.add(
+ new AggregationDescriptor(
+ AggregationType.LAST_VALUE,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath1))));
+ descriptorList.add(
+ new AggregationDescriptor(
+ AggregationType.MAX_TIME,
+ AggregationStep.PARTIAL,
+ Collections.singletonList(new TimeSeriesOperand(seriesPath2))));
+
+ deduplicatedDescriptors = AggregationNode.getDeduplicatedDescriptors(descriptorList);
+ assertEquals(2, deduplicatedDescriptors.size());
+ assertEquals(AggregationType.LAST_VALUE, deduplicatedDescriptors.get(0).getAggregationType());
+ assertEquals(AggregationType.MAX_TIME, deduplicatedDescriptors.get(1).getAggregationType());
+ }
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TimeRange.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TimeRange.java
index d4cc33b7fc..969a13d637 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TimeRange.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TimeRange.java
@@ -98,11 +98,27 @@ public class TimeRange implements Comparable<TimeRange> {
}
public boolean contains(long min, long max) {
- return this.min <= min && this.max >= max;
+ if (leftClose && rightClose) {
+ return this.min <= min && this.max >= max;
+ } else if (leftClose) {
+ return this.min <= min && this.max > max;
+ } else if (rightClose) {
+ return this.min < min && this.max >= max;
+ } else {
+ return this.min < min && this.max > max;
+ }
}
public boolean contains(long time) {
- return this.min <= time && time <= this.max;
+ if (leftClose && rightClose) {
+ return time >= this.min && time <= this.max;
+ } else if (leftClose) {
+ return time >= this.min && time < this.max;
+ } else if (rightClose) {
+ return time > this.min && time <= this.max;
+ } else {
+ return time > this.min && time < this.max;
+ }
}
/**