You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2022/05/05 07:25:18 UTC
[iotdb] branch master updated: implement visitSeriesAggregate in localExecutionPlanner (#5801)
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new a67a260d23 implement visitSeriesAggregate in localExecutionPlanner (#5801)
a67a260d23 is described below
commit a67a260d23abec347801b0086131437c823cdf09
Author: Xiangwei Wei <34...@users.noreply.github.com>
AuthorDate: Thu May 5 15:25:14 2022 +0800
implement visitSeriesAggregate in localExecutionPlanner (#5801)
---
.../iotdb/db/mpp/plan/planner/LocalExecutionPlanner.java | 12 +++++++++++-
1 file changed, 11 insertions(+), 1 deletion(-)
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 fbcdf7deaf..18c137d818 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
@@ -22,6 +22,8 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.metadata.path.AlignedPath;
import org.apache.iotdb.db.metadata.schemaregion.ISchemaRegion;
+import org.apache.iotdb.db.mpp.aggregation.AccumulatorFactory;
+import org.apache.iotdb.db.mpp.aggregation.Aggregator;
import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
import org.apache.iotdb.db.mpp.execution.datatransfer.DataBlockManager;
import org.apache.iotdb.db.mpp.execution.datatransfer.DataBlockService;
@@ -352,13 +354,21 @@ public class LocalExecutionPlanner {
node.getPlanNodeId(),
SeriesAggregationScanNode.class.getSimpleName());
+ List<Aggregator> aggregators = new ArrayList<>();
+ node.getAggregationDescriptorList()
+ .forEach(
+ o ->
+ new Aggregator(
+ AccumulatorFactory.createAccumulator(
+ o.getAggregationType(), node.getSeriesPath().getSeriesType(), ascending),
+ o.getStep()));
SeriesAggregateScanOperator aggregateScanOperator =
new SeriesAggregateScanOperator(
node.getPlanNodeId(),
seriesPath,
node.getAllSensors(),
operatorContext,
- null,
+ aggregators,
node.getTimeFilter(),
ascending,
node.getGroupByTimeParameter());