This is an automated email from the ASF dual-hosted git repository.
xiangweiwei pushed a commit to branch aggregationVector
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/aggregationVector by this push:
new b54ad6f implement part 1 of aggregation executor
b54ad6f is described below
commit b54ad6f83b5a4ee7a83276bf81b513512c7e6657
Author: Alima777 <[email protected]>
AuthorDate: Tue Sep 14 20:56:27 2021 +0800
implement part 1 of aggregation executor
---
.../query/aggregate/ClusterAggregateExecutor.java | 4 +-
.../iotdb/db/metadata/VectorPartialPath.java | 2 +-
.../db/query/executor/AggregationExecutor.java | 80 ++++++++++++++++++++--
.../query/executor/ISeriesAggregationExecutor.java | 5 ++
4 files changed, 84 insertions(+), 7 deletions(-)
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/aggregate/ClusterAggregateExecutor.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/aggregate/ClusterAggregateExecutor.java
index a007ee9..65b6db3 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/query/aggregate/ClusterAggregateExecutor.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/aggregate/ClusterAggregateExecutor.java
@@ -62,7 +62,7 @@ public class ClusterAggregateExecutor extends
AggregationExecutor {
protected void aggregateOneSeries(
Map.Entry<PartialPath, List<Integer>> pathToAggrIndexes,
AggregateResult[] aggregateResultList,
- Set<String> measurements,
+ Set<String> allMeasurementsInDevice,
Filter timeFilter,
QueryContext context)
throws StorageEngineException {
@@ -75,7 +75,7 @@ public class ClusterAggregateExecutor extends
AggregationExecutor {
}
List<AggregateResult> aggregateResult =
aggregator.getAggregateResult(
- seriesPath, measurements, aggregationNames, tsDataType,
timeFilter, context, ascending);
+ seriesPath, allMeasurementsInDevice, aggregationNames, tsDataType,
timeFilter, context, ascending);
int rstIndex = 0;
for (int i : pathToAggrIndexes.getValue()) {
aggregateResultList[i] = aggregateResult.get(rstIndex++);
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/VectorPartialPath.java
b/server/src/main/java/org/apache/iotdb/db/metadata/VectorPartialPath.java
index b6bb353..7f93ded 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/VectorPartialPath.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/VectorPartialPath.java
@@ -75,7 +75,7 @@ public class VectorPartialPath extends PartialPath {
}
@Override
- public PartialPath copy() {
+ public VectorPartialPath copy() {
VectorPartialPath result = new VectorPartialPath();
result.nodes = nodes;
result.fullPath = fullPath;
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
b/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
index c01f3c3..c2e1a49 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.query.executor;
+import java.util.LinkedHashMap;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
@@ -27,6 +28,7 @@ import
org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.metadata.VectorPartialPath;
import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
@@ -100,8 +102,7 @@ public class AggregationExecutor {
}
// TODO use multi-thread
- Map<PartialPath, List<Integer>> pathToAggrIndexesMap =
- groupAggregationsBySeries(selectedSeries);
+ List<ISeriesAggregationExecutor> seriesAggregationExecutor =
groupSeriesToExecutor(selectedSeries);
AggregateResult[] aggregateResultList = new
AggregateResult[selectedSeries.size()];
// TODO-Cluster: group the paths by storage group to reduce communications
List<StorageGroupProcessor> list =
@@ -132,7 +133,7 @@ public class AggregationExecutor {
protected void aggregateOneSeries(
Map.Entry<PartialPath, List<Integer>> pathToAggrIndexes,
AggregateResult[] aggregateResultList,
- Set<String> measurements,
+ Set<String> allMeasurementsInDevice,
Filter timeFilter,
QueryContext context)
throws IOException, QueryProcessException, StorageEngineException {
@@ -156,7 +157,7 @@ public class AggregationExecutor {
}
aggregateOneSeries(
seriesPath,
- measurements,
+ allMeasurementsInDevice,
context,
timeFilter,
tsDataType,
@@ -511,4 +512,75 @@ public class AggregationExecutor {
}
return pathToAggrIndexesMap;
}
+
+ /**
+ * Merge same series, Group all the subSensors of one vector into one
VectorPartialPath and convert to ISeriesAggregationExecutor.
+ * For example: Given: paths: s1, vector.s1, vector.s2, s1 and aggregations:
count, count, count, sum.
+ * Then: SeriesAggregationExecutor s1 -> [0, 3],
VectorSeriesAggregationExecutor vector[s1, s2], Map{s1 -> 1, s2 -> 2}
+ *
+ * @param selectedSeries selected series
+ * @return ISeriesAggregationExecutor list
+ */
+ private List<ISeriesAggregationExecutor> groupSeriesToExecutor(
+ List<PartialPath> selectedSeries) {
+ Map<String, ISeriesAggregationExecutor> pathToSeriesExecutor = new
LinkedHashMap<>();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ PartialPath path = selectedSeries.get(i);
+ String pathName = path.getFullPath();
+ if (!pathToSeriesExecutor.containsKey(pathName)) {
+ ISeriesAggregationExecutor seriesExecutor =
getSeriesAggregationExecutor(path);
+ pathToSeriesExecutor.put(pathName, seriesExecutor);
+ } else {
+ // if pathToSeriesExecutor contains node
+ ISeriesAggregationExecutor seriesExecutor =
pathToSeriesExecutor.get(pathName);
+ if (seriesExecutor instanceof SeriesAggregationExecutor) {
+ ((SeriesAggregationExecutor) seriesExecutor).addSeriesIndex(i);
+ } else if (seriesExecutor instanceof VectorSeriesAggregationExecutor){
+ ((VectorSeriesAggregationExecutor)
seriesExecutor).addSubSensorIndex((((VectorPartialPath) path).getSubSensor(0)),
i);
+ }
+ }
+ }
+ return new ArrayList<>(pathToSeriesExecutor.values());
+ }
+
+ private ISeriesAggregationExecutor getSeriesAggregationExecutor(PartialPath
path) {
+ if (path instanceof VectorPartialPath) {
+ return new VectorSeriesAggregationExecutor((VectorPartialPath) path);
+ } else {
+ return new SeriesAggregationExecutor(path);
+ }
+ }
+
+ public class SeriesAggregationExecutor implements ISeriesAggregationExecutor
{
+ private PartialPath path;
+ private TSDataType dataType;
+ private List<Integer> seriesIndex = new ArrayList<>();
+ private IReaderByTimestamp readerByTimestamp;
+
+ public SeriesAggregationExecutor(PartialPath path) {
+ this.path = path.copy();
+ }
+
+ public void addSeriesIndex(int index) {
+ seriesIndex.add(index);
+ }
+ }
+
+ public class VectorSeriesAggregationExecutor implements
ISeriesAggregationExecutor {
+ private VectorPartialPath vectorPath;
+ private List<TSDataType> dataTypes;
+ private Map<String, List<Integer>> subSensorIndexMap = new HashMap<>();
+ private IReaderByTimestamp readerByTimestamp;
+
+ public VectorSeriesAggregationExecutor(VectorPartialPath vectorPath) {
+ this.vectorPath = vectorPath.copy();
+ }
+
+ public void addSubSensorIndex(String subSensor, int index) {
+ subSensorIndexMap.computeIfAbsent(subSensor, k -> {
+ vectorPath.addSubSensor(k);
+ return new ArrayList<>();
+ }).add(index);
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/executor/ISeriesAggregationExecutor.java
b/server/src/main/java/org/apache/iotdb/db/query/executor/ISeriesAggregationExecutor.java
new file mode 100644
index 0000000..120f4e7
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/query/executor/ISeriesAggregationExecutor.java
@@ -0,0 +1,5 @@
+package org.apache.iotdb.db.query.executor;
+
+public interface ISeriesAggregationExecutor {
+
+}
