You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by hu...@apache.org on 2022/07/07 15:48:14 UTC
[iotdb] 01/14: move some static method to AggregationUtil
This is an automated email from the ASF dual-hosted git repository.
hui pushed a commit to branch lmh/aggrOpRefactor
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit df14f23bef4c8b8c6cf7b3bf6515d0477f7480ac
Author: Minghui Liu <li...@foxmail.com>
AuthorDate: Fri Jul 1 15:56:47 2022 +0800
move some static method to AggregationUtil
---
.../db/mpp/execution/operator/AggregationUtil.java | 135 +++++++++++++++++++++
.../operator/process/AggregationOperator.java | 27 +----
.../process/RawDataAggregationOperator.java | 62 +---------
.../process/SlidingWindowAggregationOperator.java | 8 +-
.../AlignedSeriesAggregationScanOperator.java | 25 ++--
.../source/SeriesAggregationScanOperator.java | 51 ++------
6 files changed, 171 insertions(+), 137 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/AggregationUtil.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/AggregationUtil.java
new file mode 100644
index 0000000000..a7ec92ec8c
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/AggregationUtil.java
@@ -0,0 +1,135 @@
+/*
+ * 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.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.plan.planner.plan.parameter.GroupByTimeParameter;
+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.TimeColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
+
+import java.util.List;
+
+public class AggregationUtil {
+
+ public static TsBlock updateResultTsBlockFromAggregators(
+ TsBlockBuilder tsBlockBuilder,
+ List<? extends 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();
+ }
+
+ /**
+ * If groupByTimeParameter is null, which means it's an aggregation query without down sampling.
+ * 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 static ITimeRangeIterator initTimeRangeIterator(
+ GroupByTimeParameter groupByTimeParameter,
+ boolean ascending,
+ boolean outputPartialTimeWindow) {
+ if (groupByTimeParameter == null) {
+ return new SingleTimeWindowIterator(0, Long.MAX_VALUE);
+ } else {
+ return TimeRangeIteratorFactory.getTimeRangeIterator(
+ groupByTimeParameter.getStartTime(),
+ groupByTimeParameter.getEndTime(),
+ groupByTimeParameter.getInterval(),
+ groupByTimeParameter.getSlidingStep(),
+ ascending,
+ groupByTimeParameter.isIntervalByMonth(),
+ groupByTimeParameter.isSlidingStepByMonth(),
+ groupByTimeParameter.isLeftCRightO(),
+ outputPartialTimeWindow);
+ }
+ }
+
+ // skip points that cannot be calculated
+ public static TsBlock skipOutOfTimeRangePoints(
+ TsBlock tsBlock, TimeRange curTimeRange, boolean ascending) {
+ TimeColumn timeColumn = tsBlock.getTimeColumn();
+ long targetTime = ascending ? curTimeRange.getMin() : curTimeRange.getMax();
+ int left = 0, right = timeColumn.getPositionCount() - 1, mid;
+ // if ascending, find the first greater than or equal to targetTime
+ // else, find the first less than or equal to targetTime
+ while (left < right) {
+ mid = (left + right) >> 1;
+ if (timeColumn.getLongWithoutCheck(mid) < targetTime) {
+ if (ascending) {
+ left = mid + 1;
+ } else {
+ right = mid;
+ }
+ } else if (timeColumn.getLongWithoutCheck(mid) > targetTime) {
+ if (ascending) {
+ right = mid;
+ } else {
+ left = mid + 1;
+ }
+ } else if (timeColumn.getLongWithoutCheck(mid) == targetTime) {
+ return tsBlock.subTsBlock(mid);
+ }
+ }
+ return tsBlock.subTsBlock(left);
+ }
+
+ public static boolean satisfied(TsBlock tsBlock, TimeRange timeRange, boolean ascending) {
+ TsBlock.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/process/AggregationOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java
index e4de130719..0027328e1e 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
@@ -27,8 +27,6 @@ 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;
@@ -36,7 +34,8 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
-import static org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregationScanOperator.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.updateResultTsBlockFromAggregators;
/**
* AggregationOperator can process the situation: aggregation of intermediate aggregate result, it
@@ -139,26 +138,4 @@ public class AggregationOperator implements ProcessOperator {
public boolean isFinished() {
return !this.hasNext();
}
-
- public static TsBlock updateResultTsBlockFromAggregators(
- TsBlockBuilder tsBlockBuilder,
- List<? extends 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/RawDataAggregationOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/RawDataAggregationOperator.java
index 01b0625c60..a64a924426 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/RawDataAggregationOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/RawDataAggregationOperator.java
@@ -27,9 +27,7 @@ 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 org.apache.iotdb.tsfile.read.common.block.column.TimeColumn;
import com.google.common.util.concurrent.ListenableFuture;
@@ -37,7 +35,11 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
-import static org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregationScanOperator.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.isEndCalc;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.satisfied;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.skipOutOfTimeRangePoints;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.updateResultTsBlockFromAggregators;
/**
* RawDataAggregationOperator is used to process raw data tsBlock input calculating using value
@@ -122,8 +124,7 @@ public class RawDataAggregationOperator implements ProcessOperator {
// 2. Update result using aggregators
curTimeRange = null;
- return AggregationOperator.updateResultTsBlockFromAggregators(
- tsBlockBuilder, aggregators, timeRangeIterator);
+ return updateResultTsBlockFromAggregators(tsBlockBuilder, aggregators, timeRangeIterator);
}
@Override
@@ -176,55 +177,4 @@ public class RawDataAggregationOperator implements ProcessOperator {
: preCachedData.getEndTime() < curTimeRange.getMin()))
|| isEndCalc(aggregators);
}
-
- // skip points that cannot be calculated
- public static TsBlock skipOutOfTimeRangePoints(
- TsBlock tsBlock, TimeRange curTimeRange, boolean ascending) {
- TimeColumn timeColumn = tsBlock.getTimeColumn();
- long targetTime = ascending ? curTimeRange.getMin() : curTimeRange.getMax();
- int left = 0, right = timeColumn.getPositionCount() - 1, mid;
- // if ascending, find the first greater than or equal to targetTime
- // else, find the first less than or equal to targetTime
- while (left < right) {
- mid = (left + right) >> 1;
- if (timeColumn.getLongWithoutCheck(mid) < targetTime) {
- if (ascending) {
- left = mid + 1;
- } else {
- right = mid;
- }
- } else if (timeColumn.getLongWithoutCheck(mid) > targetTime) {
- if (ascending) {
- right = mid;
- } else {
- left = mid + 1;
- }
- } else if (timeColumn.getLongWithoutCheck(mid) == targetTime) {
- return tsBlock.subTsBlock(mid);
- }
- }
- return tsBlock.subTsBlock(left);
- }
-
- public static 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/process/SlidingWindowAggregationOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SlidingWindowAggregationOperator.java
index 2685e3e465..9e1a3fcf14 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SlidingWindowAggregationOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SlidingWindowAggregationOperator.java
@@ -37,10 +37,10 @@ import java.util.Arrays;
import java.util.List;
import static com.google.common.base.Preconditions.checkArgument;
-import static org.apache.iotdb.db.mpp.execution.operator.process.AggregationOperator.updateResultTsBlockFromAggregators;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.satisfied;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.skipOutOfTimeRangePoints;
-import static org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregationScanOperator.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.satisfied;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.skipOutOfTimeRangePoints;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.updateResultTsBlockFromAggregators;
public class SlidingWindowAggregationOperator implements ProcessOperator {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesAggregationScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesAggregationScanOperator.java
index 375b203976..f90d8a26f3 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesAggregationScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesAggregationScanOperator.java
@@ -24,7 +24,6 @@ import org.apache.iotdb.db.metadata.path.AlignedPath;
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.OperatorContext;
-import org.apache.iotdb.db.mpp.execution.operator.process.AggregationOperator;
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.tsfile.file.metadata.enums.TSDataType;
@@ -41,9 +40,10 @@ import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.isEndCalc;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.skipOutOfTimeRangePoints;
-import static org.apache.iotdb.db.mpp.execution.operator.source.SeriesAggregationScanOperator.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.isEndCalc;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.skipOutOfTimeRangePoints;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.updateResultTsBlockFromAggregators;
/** This operator is responsible to do the aggregation calculation especially for aligned series. */
public class AlignedSeriesAggregationScanOperator implements DataSourceOperator {
@@ -132,19 +132,19 @@ public class AlignedSeriesAggregationScanOperator implements DataSourceOperator
// 2. Calculate aggregation result based on current time window
if (calcFromCacheData(curTimeRange)) {
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
}
// read page data firstly
if (readAndCalcFromPage(curTimeRange)) {
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
}
// read chunk data secondly
if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
}
@@ -154,7 +154,7 @@ public class AlignedSeriesAggregationScanOperator implements DataSourceOperator
Statistics fileTimeStatistics = alignedSeriesScanUtil.currentFileTimeStatistics();
if (fileTimeStatistics.getStartTime() > curTimeRange.getMax()) {
if (ascending) {
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
} else {
alignedSeriesScanUtil.skipCurrentFile();
@@ -180,12 +180,12 @@ public class AlignedSeriesAggregationScanOperator implements DataSourceOperator
// read chunk
if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
}
}
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
} catch (IOException e) {
throw new RuntimeException("Error while scanning the file", e);
@@ -207,10 +207,9 @@ public class AlignedSeriesAggregationScanOperator implements DataSourceOperator
return sourceId;
}
- private void updateResultTsBlockFromAggregators() {
+ private void updateResultTsBlock() {
resultTsBlock =
- AggregationOperator.updateResultTsBlockFromAggregators(
- tsBlockBuilder, aggregators, timeRangeIterator);
+ updateResultTsBlockFromAggregators(tsBlockBuilder, aggregators, timeRangeIterator);
hasCachedTsBlock = true;
}
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 56b1a33b4a..42a9776015 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
@@ -22,10 +22,7 @@ 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.AggregationOperator;
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.tsfile.file.metadata.enums.TSDataType;
@@ -42,8 +39,10 @@ import java.util.Arrays;
import java.util.List;
import java.util.Set;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.isEndCalc;
-import static org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator.skipOutOfTimeRangePoints;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.initTimeRangeIterator;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.isEndCalc;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.skipOutOfTimeRangePoints;
+import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.updateResultTsBlockFromAggregators;
/**
* This operator is responsible to do the aggregation calculation for one series based on global
@@ -106,31 +105,6 @@ public class SeriesAggregationScanOperator implements DataSourceOperator {
this.isGroupByQuery = groupByTimeParameter != null;
}
- /**
- * If groupByTimeParameter is null, which means it's an aggregation query without down sampling.
- * 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 static ITimeRangeIterator initTimeRangeIterator(
- GroupByTimeParameter groupByTimeParameter,
- boolean ascending,
- boolean outputPartialTimeWindow) {
- if (groupByTimeParameter == null) {
- return new SingleTimeWindowIterator(0, Long.MAX_VALUE);
- } else {
- return TimeRangeIteratorFactory.getTimeRangeIterator(
- groupByTimeParameter.getStartTime(),
- groupByTimeParameter.getEndTime(),
- groupByTimeParameter.getInterval(),
- groupByTimeParameter.getSlidingStep(),
- ascending,
- groupByTimeParameter.isIntervalByMonth(),
- groupByTimeParameter.isSlidingStepByMonth(),
- groupByTimeParameter.isLeftCRightO(),
- outputPartialTimeWindow);
- }
- }
-
@Override
public OperatorContext getOperatorContext() {
return operatorContext;
@@ -164,19 +138,19 @@ public class SeriesAggregationScanOperator implements DataSourceOperator {
// 2. Calculate aggregation result based on current time window
if (calcFromCacheData(curTimeRange)) {
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
}
// read page data firstly
if (readAndCalcFromPage(curTimeRange)) {
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
}
// read chunk data secondly
if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
}
@@ -186,7 +160,7 @@ public class SeriesAggregationScanOperator implements DataSourceOperator {
Statistics fileStatistics = seriesScanUtil.currentFileStatistics();
if (fileStatistics.getStartTime() > curTimeRange.getMax()) {
if (ascending) {
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
} else {
seriesScanUtil.skipCurrentFile();
@@ -207,22 +181,21 @@ public class SeriesAggregationScanOperator implements DataSourceOperator {
// read chunk
if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
}
}
- updateResultTsBlockFromAggregators();
+ updateResultTsBlock();
return true;
} catch (IOException e) {
throw new RuntimeException("Error while scanning the file", e);
}
}
- private void updateResultTsBlockFromAggregators() {
+ private void updateResultTsBlock() {
resultTsBlock =
- AggregationOperator.updateResultTsBlockFromAggregators(
- tsBlockBuilder, aggregators, timeRangeIterator);
+ updateResultTsBlockFromAggregators(tsBlockBuilder, aggregators, timeRangeIterator);
hasCachedTsBlock = true;
}