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 565b14dbb1 [IOTDB-3383] Refactor calcInputLocationList() in
LocalExecutionPlanner (#6152)
565b14dbb1 is described below
commit 565b14dbb104670b336bf928fdc9ebe3bf8608f0
Author: liuminghui233 <[email protected]>
AuthorDate: Mon Jun 6 09:22:32 2022 +0800
[IOTDB-3383] Refactor calcInputLocationList() in LocalExecutionPlanner
(#6152)
---
.../TimeRangeIteratorFactory.java | 4 +-
.../operator/process/AggregationOperator.java | 24 ++++++++---
.../source/SeriesAggregationScanOperator.java | 6 ++-
.../apache/iotdb/db/mpp/plan/analyze/Analyzer.java | 3 +-
.../mpp/plan/analyze/GroupByLevelController.java | 6 ++-
.../db/mpp/plan/planner/LocalExecutionPlanner.java | 46 +++++++++++-----------
.../plan/parameter/AggregationDescriptor.java | 13 +++---
.../operator/AggregationOperatorTest.java | 3 +-
.../plan/analyze/AggregationDescriptorTest.java | 41 ++++++++++---------
9 files changed, 86 insertions(+), 60 deletions(-)
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
index 7a270eb6cd..1dbe7a04f8 100644
---
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
@@ -40,10 +40,10 @@ public class TimeRangeIteratorFactory {
boolean isIntervalByMonth,
boolean isSlidingStepByMonth,
boolean leftCRightO,
- boolean isPreAggr) {
+ boolean outputPartialTimeWindow) {
long tmpInterval = isIntervalByMonth ? interval * MS_TO_MONTH : interval;
long tmpSlidingStep = isSlidingStepByMonth ? slidingStep * MS_TO_MONTH :
slidingStep;
- if (isPreAggr && tmpInterval > tmpSlidingStep) {
+ if (outputPartialTimeWindow && tmpInterval > tmpSlidingStep) {
if (!isIntervalByMonth && !isSlidingStepByMonth) {
return new PreAggrWindowIterator(
startTime, endTime, interval, slidingStep, isAscending,
leftCRightO);
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java
index 3b5beb32eb..4dd1d9c411 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java
@@ -53,7 +53,7 @@ public class AggregationOperator implements ProcessOperator {
private final TsBlock[] inputTsBlocks;
private final TsBlockBuilder tsBlockBuilder;
- private ITimeRangeIterator timeRangeIterator;
+ private final ITimeRangeIterator timeRangeIterator;
// current interval of aggregation window [curStartTime, curEndTime)
private TimeRange curTimeRange;
@@ -62,7 +62,8 @@ public class AggregationOperator implements ProcessOperator {
List<Aggregator> aggregators,
List<Operator> children,
boolean ascending,
- GroupByTimeParameter groupByTimeParameter) {
+ GroupByTimeParameter groupByTimeParameter,
+ boolean outputPartialTimeWindow) {
this.operatorContext = operatorContext;
this.aggregators = aggregators;
this.children = children;
@@ -74,7 +75,8 @@ public class AggregationOperator implements ProcessOperator {
dataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
}
tsBlockBuilder = new TsBlockBuilder(dataTypes);
- this.timeRangeIterator = initTimeRangeIterator(groupByTimeParameter,
ascending, true);
+ this.timeRangeIterator =
+ initTimeRangeIterator(groupByTimeParameter, ascending,
outputPartialTimeWindow);
}
@Override
@@ -96,9 +98,17 @@ public class AggregationOperator implements ProcessOperator {
@Override
public TsBlock next() {
// update input tsBlock
- curTimeRange = timeRangeIterator.nextTimeRange();
+ if (curTimeRange == null && timeRangeIterator.hasNextTimeRange()) {
+ curTimeRange = timeRangeIterator.nextTimeRange();
+ }
for (int i = 0; i < inputOperatorsCount; i++) {
+ if (inputTsBlocks[i] != null) {
+ continue;
+ }
inputTsBlocks[i] = children.get(i).next();
+ if (inputTsBlocks[i] == null) {
+ return null;
+ }
}
// consume current input tsBlocks
for (Aggregator aggregator : aggregators) {
@@ -106,12 +116,16 @@ public class AggregationOperator implements
ProcessOperator {
aggregator.processTsBlocks(inputTsBlocks);
}
// output result from aggregator
+ curTimeRange = null;
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ inputTsBlocks[i] = null;
+ }
return updateResultTsBlockFromAggregators(tsBlockBuilder, aggregators,
timeRangeIterator);
}
@Override
public boolean hasNext() {
- return timeRangeIterator.hasNextTimeRange();
+ return curTimeRange != null || timeRangeIterator.hasNextTimeRange();
}
@Override
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregationScanOperator.java
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregationScanOperator.java
index 807b1a2ebd..3b107b01e7 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregationScanOperator.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregationScanOperator.java
@@ -110,7 +110,9 @@ public class SeriesAggregationScanOperator implements
DataSourceOperator {
* timestamp, so it doesn't matter what the time range returns.
*/
public static ITimeRangeIterator initTimeRangeIterator(
- GroupByTimeParameter groupByTimeParameter, boolean ascending, boolean
isPreAggr) {
+ GroupByTimeParameter groupByTimeParameter,
+ boolean ascending,
+ boolean outputPartialTimeWindow) {
if (groupByTimeParameter == null) {
return new SingleTimeWindowIterator(0, Long.MAX_VALUE);
} else {
@@ -123,7 +125,7 @@ public class SeriesAggregationScanOperator implements
DataSourceOperator {
groupByTimeParameter.isIntervalByMonth(),
groupByTimeParameter.isSlidingStepByMonth(),
groupByTimeParameter.isLeftCRightO(),
- isPreAggr);
+ outputPartialTimeWindow);
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java
index 8418241072..2e8ac3a942 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java
@@ -655,7 +655,8 @@ public class Analyzer {
Set<Expression> transformExpressions,
Map<Expression, Expression> rawPathToGroupedPathMap) {
GroupByLevelController groupByLevelController =
- new
GroupByLevelController(queryStatement.getGroupByLevelComponent().getLevels());
+ new GroupByLevelController(
+ queryStatement.getGroupByLevelComponent().getLevels(),
typeProvider);
for (Pair<Expression, String> measurementWithAlias : outputExpressions) {
groupByLevelController.control(measurementWithAlias.left,
measurementWithAlias.right);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/GroupByLevelController.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/GroupByLevelController.java
index 6055bc930c..a13353518e 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/GroupByLevelController.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/GroupByLevelController.java
@@ -61,12 +61,15 @@ public class GroupByLevelController {
*/
private final Map<String, String> aliasToColumnMap;
- public GroupByLevelController(int[] levels) {
+ private final TypeProvider typeProvider;
+
+ public GroupByLevelController(int[] levels, TypeProvider typeProvider) {
this.levels = levels;
this.groupedPathMap = new LinkedHashMap<>();
this.rawPathToGroupedPathMap = new HashMap<>();
this.columnToAliasMap = new HashMap<>();
this.aliasToColumnMap = new HashMap<>();
+ this.typeProvider = typeProvider;
}
public void control(Expression expression, String alias) {
@@ -77,6 +80,7 @@ public class GroupByLevelController {
PartialPath rawPath = ((TimeSeriesOperand)
expression.getExpressions().get(0)).getPath();
PartialPath groupedPath = generatePartialPathByLevel(rawPath.getNodes(),
levels);
+ typeProvider.setType(groupedPath.getFullPath(), rawPath.getSeriesType());
Expression rawPathExpression = new TimeSeriesOperand(rawPath);
Expression groupedPathExpression = new TimeSeriesOperand(groupedPath);
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 f85a8eb809..70108b5932 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
@@ -827,12 +827,7 @@ public class LocalExecutionPlanner {
List<Aggregator> aggregators = new ArrayList<>();
Map<String, List<InputLocation>> layout = makeLayout(node);
for (GroupByLevelDescriptor descriptor :
node.getGroupByLevelDescriptors()) {
- List<String> inputColumnNames = descriptor.getInputColumnNames();
- List<InputLocation[]> inputLocationList = new
ArrayList<>(inputColumnNames.size());
- inputColumnNames.forEach(
- inputColumnName ->
- inputLocationList.add(layout.get(inputColumnName).toArray(new
InputLocation[0])));
-
+ List<InputLocation[]> inputLocationList =
calcInputLocationList(descriptor, layout);
aggregators.add(
new Aggregator(
AccumulatorFactory.createAccumulator(
@@ -840,7 +835,7 @@ public class LocalExecutionPlanner {
context
.getTypeProvider()
// get the type of first inputExpression
- .getType(inputColumnNames.get(0)),
+
.getType(descriptor.getInputExpressions().get(0).getExpressionString()),
ascending),
descriptor.getStep(),
inputLocationList));
@@ -851,7 +846,7 @@ public class LocalExecutionPlanner {
node.getPlanNodeId(),
AggregationOperator.class.getSimpleName());
return new AggregationOperator(
- operatorContext, aggregators, children, ascending,
node.getGroupByTimeParameter());
+ operatorContext, aggregators, children, ascending,
node.getGroupByTimeParameter(), false);
}
@Override
@@ -958,26 +953,33 @@ public class LocalExecutionPlanner {
node.getPlanNodeId(),
AggregationOperator.class.getSimpleName());
return new AggregationOperator(
- operatorContext, aggregators, children, ascending,
node.getGroupByTimeParameter());
+ operatorContext,
+ aggregators,
+ children,
+ ascending,
+ node.getGroupByTimeParameter(),
+ true);
}
}
private List<InputLocation[]> calcInputLocationList(
AggregationDescriptor descriptor, Map<String, List<InputLocation>>
layout) {
- List<String> inputColumnNames = descriptor.getInputColumnNames();
- // it may include double parts
- List<List<InputLocation>> inputLocationParts = new
ArrayList<>(inputColumnNames.size());
- inputColumnNames.forEach(o -> inputLocationParts.add(layout.get(o)));
-
+ List<List<String>> inputColumnNames =
descriptor.getInputColumnNamesList();
List<InputLocation[]> inputLocationList = new ArrayList<>();
- for (int i = 0; i < inputLocationParts.get(0).size(); i++) {
- if (inputColumnNames.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)
- });
+
+ for (List<String> inputColumnNamesOfOneInput : inputColumnNames) {
+ // it may include double parts
+ List<List<InputLocation>> inputLocationParts = new ArrayList<>();
+ inputColumnNamesOfOneInput.forEach(o ->
inputLocationParts.add(layout.get(o)));
+ for (int i = 0; i < inputLocationParts.get(0).size(); i++) {
+ if (inputColumnNamesOfOneInput.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)
+ });
+ }
}
}
return inputLocationList;
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 942562bee7..f97add0369 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
@@ -25,6 +25,7 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -72,20 +73,16 @@ public class AggregationDescriptor {
return outputColumnNames;
}
- public List<String> getInputColumnNames() {
+ public List<List<String>> getInputColumnNamesList() {
if (step.isInputRaw()) {
return inputExpressions.stream()
- .map(Expression::getExpressionString)
+ .map(expression ->
Collections.singletonList(expression.getExpressionString()))
.collect(Collectors.toList());
}
- List<AggregationType> inputAggregationTypes =
getActualAggregationTypes(step.isInputPartial());
- List<String> inputColumnNames = new ArrayList<>();
+ List<List<String>> inputColumnNames = new ArrayList<>();
for (Expression expression : inputExpressions) {
- for (AggregationType funcName : inputAggregationTypes) {
- inputColumnNames.add(
- funcName.toString().toLowerCase() + "(" +
expression.getExpressionString() + ")");
- }
+ inputColumnNames.add(getInputColumnNames(expression));
}
return inputColumnNames;
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregationOperatorTest.java
b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregationOperatorTest.java
index 35cd0e25bb..3b940a78a8 100644
---
a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregationOperatorTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AggregationOperatorTest.java
@@ -316,6 +316,7 @@ public class AggregationOperatorTest {
finalAggregators,
children,
true,
- groupByTimeParameter);
+ groupByTimeParameter,
+ true);
}
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/mpp/plan/analyze/AggregationDescriptorTest.java
b/server/src/test/java/org/apache/iotdb/db/mpp/plan/analyze/AggregationDescriptorTest.java
index fb7b72515b..4b8bbf6b00 100644
---
a/server/src/test/java/org/apache/iotdb/db/mpp/plan/analyze/AggregationDescriptorTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/mpp/plan/analyze/AggregationDescriptorTest.java
@@ -145,20 +145,19 @@ public class AggregationDescriptorTest {
@Test
public void testInputColumnNames() {
- List<String> expectedInputColumnNames =
+ List<List<List<String>>> expectedInputColumnNames =
Arrays.asList(
- "root.sg.d1.s1",
- "count(root.sg.d1.s1)",
- "sum(root.sg.d1.s1)",
- "last_value(root.sg.d1.s1)",
- "max_time(root.sg.d1.s1)",
- "max_value(root.sg.d1.s1)");
+
Collections.singletonList(Collections.singletonList("root.sg.d1.s1")),
+
Collections.singletonList(Collections.singletonList("root.sg.d1.s1")),
+ Collections.singletonList(Arrays.asList("count(root.sg.d1.s1)",
"sum(root.sg.d1.s1)")),
+ Collections.singletonList(
+ Arrays.asList("last_value(root.sg.d1.s1)",
"max_time(root.sg.d1.s1)")),
+
Collections.singletonList(Collections.singletonList("max_value(root.sg.d1.s1)")),
+
Collections.singletonList(Collections.singletonList("count(root.sg.d1.s1)")));
Assert.assertEquals(
expectedInputColumnNames,
aggregationDescriptorList.stream()
- .map(AggregationDescriptor::getInputColumnNames)
- .flatMap(List::stream)
- .distinct()
+ .map(AggregationDescriptor::getInputColumnNamesList)
.collect(Collectors.toList()));
}
@@ -177,18 +176,24 @@ public class AggregationDescriptorTest {
@Test
public void testInputColumnNamesInGroupByLevel() {
- List<String> expectedInputColumnNames =
+ List<List<List<String>>> expectedInputColumnNames =
Arrays.asList(
- "count(root.sg.d2.s1)",
- "count(root.sg.d1.s1)",
- "sum(root.sg.d1.s1)",
- "sum(root.sg.d2.s1)");
+ Arrays.asList(
+ Collections.singletonList("count(root.sg.d2.s1)"),
+ Collections.singletonList("count(root.sg.d1.s1)")),
+ Arrays.asList(
+ Arrays.asList("count(root.sg.d1.s1)", "sum(root.sg.d1.s1)"),
+ Arrays.asList("count(root.sg.d2.s1)", "sum(root.sg.d2.s1)")),
+ Arrays.asList(
+ Collections.singletonList("count(root.sg.d2.s1)"),
+ Collections.singletonList("count(root.sg.d1.s1)")),
+ Arrays.asList(
+ Arrays.asList("count(root.sg.d1.s1)", "sum(root.sg.d1.s1)"),
+ Arrays.asList("count(root.sg.d2.s1)", "sum(root.sg.d2.s1)")));
Assert.assertEquals(
expectedInputColumnNames,
groupByLevelDescriptorList.stream()
- .map(GroupByLevelDescriptor::getInputColumnNames)
- .flatMap(List::stream)
- .distinct()
+ .map(GroupByLevelDescriptor::getInputColumnNamesList)
.collect(Collectors.toList()));
}
