This is an automated email from the ASF dual-hosted git repository. xiangweiwei pushed a commit to branch aggregationOp in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit a34d51ee2839cd8c579ce6353b4133ae8410a838 Author: Alima777 <[email protected]> AuthorDate: Mon May 9 20:42:45 2022 +0800 add some tests --- .../iotdb/db/mpp/aggregation/AvgAccumulator.java | 7 + .../iotdb/db/mpp/aggregation/CountAccumulator.java | 4 + .../db/mpp/aggregation/ExtremeAccumulator.java | 3 + .../db/mpp/aggregation/FirstValueAccumulator.java | 3 + .../db/mpp/aggregation/LastValueAccumulator.java | 3 + .../db/mpp/aggregation/MaxTimeAccumulator.java | 4 + .../db/mpp/aggregation/MaxValueAccumulator.java | 3 + .../db/mpp/aggregation/MinTimeAccumulator.java | 3 + .../db/mpp/aggregation/MinValueAccumulator.java | 14 + .../iotdb/db/mpp/aggregation/SumAccumulator.java | 17 ++ .../execution/operator/AggregateOperatorTest.java | 319 +++++++++++++++++++++ .../operator/RawDataAggregateOperator.java | 21 ++ 12 files changed, 401 insertions(+) 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..28ef1e79ec 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,6 +68,9 @@ 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; + } countValue += partialResult[0].getLong(0); sumValue += partialResult[1].getDouble(0); } @@ -130,6 +134,7 @@ public class AvgAccumulator implements Accumulator { break; } if (!column[1].isNull(i)) { + initResult = true; countValue++; sumValue += column[1].getInt(i); } @@ -144,6 +149,7 @@ public class AvgAccumulator implements Accumulator { break; } if (!column[1].isNull(i)) { + initResult = true; countValue++; sumValue += column[1].getLong(i); } @@ -158,6 +164,7 @@ public class AvgAccumulator implements Accumulator { break; } if (!column[1].isNull(i)) { + initResult = true; countValue++; sumValue += column[1].getFloat(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..4750cdf01e 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 @@ -31,6 +31,7 @@ import static com.google.common.base.Preconditions.checkArgument; public class CountAccumulator implements Accumulator { private long countValue = 0; + private boolean initResult = false; public CountAccumulator() {} @@ -53,6 +54,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); } 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..78e7c986c7 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)); 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..908bbf4605 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)); 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..f85dd8f5be 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 @@ -72,6 +72,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)); 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..cdd2dfd563 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; + private boolean initResult = false; public MaxTimeAccumulator() {} @@ -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)); } 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..606fce0a8a 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)); 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..6969e82aa5 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 @@ -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)); } 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..6450e32ebc 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()); 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..5672e970fb 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,6 +68,9 @@ 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; + } sumValue += partialResult[0].getDouble(0); } @@ -83,6 +87,9 @@ public class SumAccumulator implements Accumulator { @Override public void setFinal(Column finalResult) { reset(); + if (finalResult.isNull(0)) { + return; + } sumValue = finalResult.getDouble(0); } @@ -90,11 +97,17 @@ public class SumAccumulator implements Accumulator { @Override public void outputIntermediate(ColumnBuilder[] columnBuilders) { checkArgument(columnBuilders.length == 1, "partialResult of Sum should be 1"); + if (!initResult) { + columnBuilders[0].appendNull(); + } columnBuilders[0].writeDouble(sumValue); } @Override public void outputFinal(ColumnBuilder columnBuilder) { + if (!initResult) { + columnBuilder.appendNull(); + } columnBuilder.writeDouble(sumValue); } @@ -126,6 +139,7 @@ public class SumAccumulator implements Accumulator { break; } if (!column[1].isNull(i)) { + initResult = true; sumValue += column[1].getInt(i); } } @@ -139,6 +153,7 @@ public class SumAccumulator implements Accumulator { break; } if (!column[1].isNull(i)) { + initResult = true; sumValue += column[1].getLong(i); } } @@ -152,6 +167,7 @@ public class SumAccumulator implements Accumulator { break; } if (!column[1].isNull(i)) { + initResult = true; sumValue += column[1].getFloat(i); } } @@ -165,6 +181,7 @@ public class SumAccumulator implements Accumulator { break; } if (!column[1].isNull(i)) { + initResult = true; sumValue += column[1].getDouble(i); } } 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..39e62f7c02 --- /dev/null +++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregateOperatorTest.java @@ -0,0 +1,319 @@ +/* + * 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; + + @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, 100}, + {2004950, 2014950, 624950, 834950}, + {0, 100, 200, 300}, + {99, 199, 299, 399}, + {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 { + int[][] result = + new int[][] { + {20000, 20100, 10200, 10300}, + {20099, 20199, 299, 399} + }; + List<AggregationType> aggregationTypes = new ArrayList<>(); + 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).getInt(0)); + assertEquals(result[1][count], resultTsBlock.getColumn(1).getInt(0)); + count++; + } + assertEquals(4, count); + } + + /** + * @param aggregationTypes Aggregation function used in test + * @param groupByTimeParameter + * @param inputLocations each inputLocation is used in one aggregator + */ + private AggregateOperator initAggregateOperator( + List<AggregationType> aggregationTypes, + GroupByTimeParameter groupByTimeParameter, + List<List<InputLocation[]>> inputLocations) + throws IllegalPathException { + ExecutorService instanceNotificationExecutor = + IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification"); + // 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/RawDataAggregateOperator.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregateOperator.java new file mode 100644 index 0000000000..121959dc7f --- /dev/null +++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregateOperator.java @@ -0,0 +1,21 @@ +/* + * 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;public class RawDataAggregateOperator { +}
