You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/07/12 13:08:54 UTC
[GitHub] [iotdb] JackieTien97 commented on a diff in pull request #6622: [IOTDB-3714] Refactor operators related to aggregate query to support batch processing
JackieTien97 commented on code in PR #6622:
URL: https://github.com/apache/iotdb/pull/6622#discussion_r918847673
##########
server/src/main/java/org/apache/iotdb/db/mpp/aggregation/Aggregator.java:
##########
@@ -108,16 +111,12 @@ public void outputResult(ColumnBuilder[] columnBuilder) {
}
}
- public void processStatistics(Statistics statistics) {
- accumulator.addStatistics(statistics);
- }
-
- /** Used for AlignedSeriesAggregateScanOperator. */
+ /** Used for SeriesAggregateScanOperator. */
public void processStatistics(Statistics[] statistics) {
for (InputLocation[] inputLocations : inputLocationList) {
checkArgument(
inputLocations[0].getTsBlockIndex() == 0,
- "AlignedSeriesAggregateScanOperator can only process one tsBlock input.");
+ "SeriesAggregateScanOperator can only process one tsBlock input.");
Review Comment:
not only SeriesAggregateScanOperator but also AlignedSeriesAggregateScanOperator
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/AggregationUtil.java:
##########
@@ -0,0 +1,158 @@
+/*
+ * 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.TimeColumnBuilder;
+
+import java.util.List;
+
+import static org.apache.iotdb.tsfile.read.common.block.TsBlockUtil.skipPointsOutOfTimeRange;
+
+public class AggregationUtil {
+
+ private AggregationUtil() {
+ // forbidding instantiation
+ }
+
+ /**
+ * 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);
+ }
+ }
+
+ /**
+ * Calculate aggregation value on the time range from the tsBlock containing raw data.
+ *
+ * @return whether the aggregation value of the current time range has been calculated
+ */
+ public static boolean calculateAggregationFromRawData(
+ TsBlock inputTsBlock,
+ List<Aggregator> aggregators,
+ TimeRange curTimeRange,
+ boolean ascending) {
+ if (inputTsBlock == null || inputTsBlock.isEmpty()) {
+ return false;
+ }
+
+ // check if the tsBlock does not contain points in current interval
+ if (satisfiedTimeRange(inputTsBlock, curTimeRange, ascending)) {
+ // skip points that cannot be calculated
+ if ((ascending && inputTsBlock.getStartTime() < curTimeRange.getMin())
+ || (!ascending && inputTsBlock.getStartTime() > curTimeRange.getMax())) {
+ inputTsBlock = skipPointsOutOfTimeRange(inputTsBlock, curTimeRange, ascending);
+ }
+
+ int lastReadRowIndex = 0;
+ for (Aggregator aggregator : aggregators) {
+ // current agg method has been calculated
+ if (aggregator.hasFinalResult()) {
+ continue;
+ }
+
+ lastReadRowIndex = Math.max(lastReadRowIndex, aggregator.processTsBlock(inputTsBlock));
+ }
+ if (lastReadRowIndex >= inputTsBlock.getPositionCount()) {
+ inputTsBlock = null;
+ } else {
+ inputTsBlock = inputTsBlock.subTsBlock(lastReadRowIndex);
+ }
Review Comment:
Updating inputTsBlock here won't change anything about the caller's real inputTsBlock. So it may need iterate from start again.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanUtil.java:
##########
@@ -57,6 +59,60 @@ public AlignedSeriesScanUtil(
.getSchemaList().stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
}
+ @Override
+ protected Statistics currentFileStatistics(int index) throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics by index from alignedTimeSeriesMetaData");
+ }
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentFileTimeStatistics() throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
+ }
Review Comment:
same as above.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanUtil.java:
##########
@@ -57,6 +59,60 @@ public AlignedSeriesScanUtil(
.getSchemaList().stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
}
+ @Override
+ protected Statistics currentFileStatistics(int index) throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics by index from alignedTimeSeriesMetaData");
+ }
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentFileTimeStatistics() throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
+ }
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getTimeStatistics();
+ }
+
+ @Override
+ protected Statistics currentChunkStatistics(int index) throws IOException {
+ if (!(firstChunkMetadata instanceof AlignedChunkMetadata)) {
+ throw new IOException("Can only get statistics by index from alignedChunkMetaData");
+ }
Review Comment:
same as above.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanUtil.java:
##########
@@ -57,6 +59,60 @@ public AlignedSeriesScanUtil(
.getSchemaList().stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
}
+ @Override
+ protected Statistics currentFileStatistics(int index) throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics by index from alignedTimeSeriesMetaData");
+ }
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentFileTimeStatistics() throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
+ }
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getTimeStatistics();
+ }
+
+ @Override
+ protected Statistics currentChunkStatistics(int index) throws IOException {
+ if (!(firstChunkMetadata instanceof AlignedChunkMetadata)) {
+ throw new IOException("Can only get statistics by index from alignedChunkMetaData");
+ }
+ return ((AlignedChunkMetadata) firstChunkMetadata).getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentChunkTimeStatistics() throws IOException {
+ if (!(firstChunkMetadata instanceof AlignedChunkMetadata)) {
+ throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
+ }
+ return ((AlignedChunkMetadata) firstChunkMetadata).getTimeStatistics();
+ }
+
+ @Override
+ protected Statistics currentPageStatistics(int index) throws IOException {
+ if (firstPageReader == null) {
+ return null;
+ }
+ if (!(firstPageReader.isAlignedPageReader())) {
+ throw new IOException("Can only get statistics by index from AlignedPageReader");
+ }
Review Comment:
same as above.
##########
server/src/main/java/org/apache/iotdb/db/mpp/aggregation/Aggregator.java:
##########
@@ -108,16 +111,12 @@ public void outputResult(ColumnBuilder[] columnBuilder) {
}
}
- public void processStatistics(Statistics statistics) {
- accumulator.addStatistics(statistics);
- }
-
- /** Used for AlignedSeriesAggregateScanOperator. */
+ /** Used for SeriesAggregateScanOperator. */
public void processStatistics(Statistics[] statistics) {
for (InputLocation[] inputLocations : inputLocationList) {
checkArgument(
inputLocations[0].getTsBlockIndex() == 0,
- "AlignedSeriesAggregateScanOperator can only process one tsBlock input.");
+ "SeriesAggregateScanOperator can only process one tsBlock input.");
Review Comment:
and put this check into constrcutor, don't do this check each time processStatistics() is called.
##########
server/src/main/java/org/apache/iotdb/db/mpp/aggregation/Aggregator.java:
##########
@@ -141,6 +140,7 @@ public boolean hasFinalResult() {
}
public void updateTimeRange(TimeRange curTimeRange) {
+ reset();
Review Comment:
Why previously we don't need this reset()? Is it a previous bug?
##########
tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java:
##########
@@ -150,6 +150,12 @@ public class TSFileConfig implements Serializable {
/** The amount of data iterate each time */
private int batchSize = 1000;
+ /** Maximum capacity of a TsBlock */
+ private int maxTsBlockSizeInBytes = 1024 * 1024;
+
+ /** Maximum number of lines in a single TsBlock */
+ private int maxTsBlockLineNumber = 1000;
Review Comment:
put them into IoTDBConfig, it has nothing to do with tsfile config.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanUtil.java:
##########
@@ -57,6 +59,60 @@ public AlignedSeriesScanUtil(
.getSchemaList().stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
}
+ @Override
+ protected Statistics currentFileStatistics(int index) throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics by index from alignedTimeSeriesMetaData");
+ }
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentFileTimeStatistics() throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
+ }
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getTimeStatistics();
+ }
+
+ @Override
+ protected Statistics currentChunkStatistics(int index) throws IOException {
+ if (!(firstChunkMetadata instanceof AlignedChunkMetadata)) {
+ throw new IOException("Can only get statistics by index from alignedChunkMetaData");
+ }
+ return ((AlignedChunkMetadata) firstChunkMetadata).getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentChunkTimeStatistics() throws IOException {
+ if (!(firstChunkMetadata instanceof AlignedChunkMetadata)) {
+ throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
+ }
+ return ((AlignedChunkMetadata) firstChunkMetadata).getTimeStatistics();
+ }
+
+ @Override
+ protected Statistics currentPageStatistics(int index) throws IOException {
+ if (firstPageReader == null) {
+ return null;
+ }
+ if (!(firstPageReader.isAlignedPageReader())) {
+ throw new IOException("Can only get statistics by index from AlignedPageReader");
+ }
+ return firstPageReader.getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentPageTimeStatistics() throws IOException {
+ if (firstPageReader == null) {
+ return null;
+ }
+ if (!(firstPageReader.isAlignedPageReader())) {
+ throw new IOException("Can only get statistics of time column from AlignedPageReader");
+ }
Review Comment:
same as above.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanUtil.java:
##########
@@ -57,6 +59,60 @@ public AlignedSeriesScanUtil(
.getSchemaList().stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
}
+ @Override
+ protected Statistics currentFileStatistics(int index) throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics by index from alignedTimeSeriesMetaData");
+ }
Review Comment:
no need to do this check, firstTimeSeriesMetadata in AlignedSeriesScanUtil must be AlignedTimeSeriesMetadata.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AlignedSeriesScanUtil.java:
##########
@@ -57,6 +59,60 @@ public AlignedSeriesScanUtil(
.getSchemaList().stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
}
+ @Override
+ protected Statistics currentFileStatistics(int index) throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics by index from alignedTimeSeriesMetaData");
+ }
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentFileTimeStatistics() throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
+ }
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getTimeStatistics();
+ }
+
+ @Override
+ protected Statistics currentChunkStatistics(int index) throws IOException {
+ if (!(firstChunkMetadata instanceof AlignedChunkMetadata)) {
+ throw new IOException("Can only get statistics by index from alignedChunkMetaData");
+ }
+ return ((AlignedChunkMetadata) firstChunkMetadata).getStatistics(index);
+ }
+
+ @Override
+ protected Statistics currentChunkTimeStatistics() throws IOException {
+ if (!(firstChunkMetadata instanceof AlignedChunkMetadata)) {
+ throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
+ }
Review Comment:
same as above.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/SeriesAggregationScanOperator.java:
##########
@@ -96,337 +57,6 @@ public SeriesAggregationScanOperator(
timeFilter,
null,
ascending);
- this.aggregators = aggregators;
- List<TSDataType> dataTypes = new ArrayList<>();
- for (Aggregator aggregator : aggregators) {
- dataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
- }
- tsBlockBuilder = new TsBlockBuilder(dataTypes);
- this.timeRangeIterator = initTimeRangeIterator(groupByTimeParameter, ascending, true);
- 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;
- }
-
- @Override
- public TsBlock next() {
- if (hasCachedTsBlock || hasNext()) {
- hasCachedTsBlock = false;
- return resultTsBlock;
- }
- return null;
- }
-
- @Override
- public boolean hasNext() {
- if (hasCachedTsBlock) {
- return true;
- }
- try {
- if (!timeRangeIterator.hasNextTimeRange()) {
- return false;
- }
- curTimeRange = timeRangeIterator.nextTimeRange();
-
- // 1. Clear previous aggregation result
- for (Aggregator aggregator : aggregators) {
- aggregator.reset();
- aggregator.updateTimeRange(curTimeRange);
- }
-
- // 2. Calculate aggregation result based on current time window
- if (calcFromCacheData(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
-
- // read page data firstly
- if (readAndCalcFromPage(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
-
- // read chunk data secondly
- if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
-
- // read from file first
- while (seriesScanUtil.hasNextFile()) {
- if (canUseCurrentFileStatistics()) {
- Statistics fileStatistics = seriesScanUtil.currentFileStatistics();
- if (fileStatistics.getStartTime() > curTimeRange.getMax()) {
- if (ascending) {
- updateResultTsBlockFromAggregators();
- return true;
- } else {
- seriesScanUtil.skipCurrentFile();
- continue;
- }
- }
- // calc from fileMetaData
- if (curTimeRange.contains(fileStatistics.getStartTime(), fileStatistics.getEndTime())) {
- calcFromStatistics(fileStatistics);
- seriesScanUtil.skipCurrentFile();
- if (isEndCalc(aggregators) && !isGroupByQuery) {
- break;
- } else {
- continue;
- }
- }
- }
-
- // read chunk
- if (readAndCalcFromChunk(curTimeRange)) {
- updateResultTsBlockFromAggregators();
- return true;
- }
- }
-
- updateResultTsBlockFromAggregators();
- return true;
- } catch (IOException e) {
- throw new RuntimeException("Error while scanning the file", e);
- }
- }
-
- private void updateResultTsBlockFromAggregators() {
- resultTsBlock =
- AggregationOperator.updateResultTsBlockFromAggregators(
- tsBlockBuilder, aggregators, timeRangeIterator);
- hasCachedTsBlock = true;
- }
-
- @Override
- public boolean isFinished() {
- return finished || (finished = !hasNext());
- }
-
- @Override
- public PlanNodeId getSourceId() {
- return sourceId;
- }
-
- @Override
- public void initQueryDataSource(QueryDataSource dataSource) {
- seriesScanUtil.initQueryDataSource(dataSource);
- }
-
- /** @return if already get the result */
- private boolean calcFromCacheData(TimeRange curTimeRange) throws IOException {
- calcFromBatch(preCachedData, curTimeRange);
- // The result is calculated from the cache
- return (preCachedData != null
- && (ascending
- ? preCachedData.getEndTime() > curTimeRange.getMax()
- : preCachedData.getEndTime() < curTimeRange.getMin()))
- || isEndCalc(aggregators);
- }
-
- @SuppressWarnings("squid:S3776")
- private void calcFromBatch(TsBlock tsBlock, TimeRange curTimeRange) {
- // check if the batchData does not contain points in current interval
- if (tsBlock != null && satisfied(tsBlock, curTimeRange, ascending)) {
- // skip points that cannot be calculated
- if ((ascending && tsBlock.getStartTime() < curTimeRange.getMin())
- || (!ascending && tsBlock.getStartTime() > curTimeRange.getMax())) {
- tsBlock = skipOutOfTimeRangePoints(tsBlock, curTimeRange, ascending);
- }
-
- int lastReadRowIndex = 0;
- for (Aggregator aggregator : aggregators) {
- // current agg method has been calculated
- if (aggregator.hasFinalResult()) {
- continue;
- }
-
- lastReadRowIndex = Math.max(lastReadRowIndex, aggregator.processTsBlock(tsBlock));
- }
- if (lastReadRowIndex >= tsBlock.getPositionCount()) {
- tsBlock = null;
- } else {
- tsBlock = tsBlock.subTsBlock(lastReadRowIndex);
- }
-
- // can calc for next interval
- if (tsBlock != null && tsBlock.getTsBlockSingleColumnIterator().hasNext()) {
- preCachedData = tsBlock;
- }
- }
- }
-
- private boolean satisfied(TsBlock tsBlock, TimeRange timeRange, boolean ascending) {
- TsBlockSingleColumnIterator tsBlockIterator = tsBlock.getTsBlockSingleColumnIterator();
- if (tsBlockIterator == null || !tsBlockIterator.hasNext()) {
- return false;
- }
-
- if (ascending
- && (tsBlockIterator.getEndTime() < timeRange.getMin()
- || tsBlockIterator.currentTime() > timeRange.getMax())) {
- return false;
- }
- if (!ascending
- && (tsBlockIterator.getEndTime() > timeRange.getMax()
- || tsBlockIterator.currentTime() < timeRange.getMin())) {
- preCachedData = tsBlock;
- return false;
- }
- return true;
- }
-
- @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
- private boolean readAndCalcFromPage(TimeRange curTimeRange) throws IOException {
- while (seriesScanUtil.hasNextPage()) {
- if (canUseCurrentPageStatistics()) {
- Statistics pageStatistics = seriesScanUtil.currentPageStatistics();
- // There is no more eligible points in current time range
- if (pageStatistics.getStartTime() > curTimeRange.getMax()) {
- if (ascending) {
- return true;
- } else {
- seriesScanUtil.skipCurrentPage();
- continue;
- }
- }
- // can use pageHeader
- if (curTimeRange.contains(pageStatistics.getStartTime(), pageStatistics.getEndTime())) {
- calcFromStatistics(pageStatistics);
- seriesScanUtil.skipCurrentPage();
- if (isEndCalc(aggregators) && !isGroupByQuery) {
- return true;
- } else {
- continue;
- }
- }
- }
-
- // calc from page data
- TsBlock tsBlock = seriesScanUtil.nextPage();
- TsBlockSingleColumnIterator tsBlockIterator = tsBlock.getTsBlockSingleColumnIterator();
- if (tsBlockIterator == null || !tsBlockIterator.hasNext()) {
- continue;
- }
-
- // stop calc and cached current batchData
- if (ascending && tsBlockIterator.currentTime() > curTimeRange.getMax()) {
- preCachedData = tsBlock;
- return true;
- }
-
- // calc from batch data
- calcFromBatch(tsBlock, curTimeRange);
-
- // judge whether the calculation finished
- boolean isTsBlockOutOfBound =
- ascending
- ? tsBlock.getEndTime() > curTimeRange.getMax()
- : tsBlock.getEndTime() < curTimeRange.getMin();
- if (isEndCalc(aggregators) || isTsBlockOutOfBound) {
- return true;
- }
- }
- return false;
- }
-
- private boolean readAndCalcFromChunk(TimeRange curTimeRange) throws IOException {
- while (seriesScanUtil.hasNextChunk()) {
- if (canUseCurrentChunkStatistics()) {
- Statistics chunkStatistics = seriesScanUtil.currentChunkStatistics();
- if (chunkStatistics.getStartTime() > curTimeRange.getMax()) {
- if (ascending) {
- return true;
- } else {
- seriesScanUtil.skipCurrentChunk();
- continue;
- }
- }
- // calc from chunkMetaData
- if (curTimeRange.contains(chunkStatistics.getStartTime(), chunkStatistics.getEndTime())) {
- calcFromStatistics(chunkStatistics);
- seriesScanUtil.skipCurrentChunk();
- if (isEndCalc(aggregators) && !isGroupByQuery) {
- return true;
- } else {
- continue;
- }
- }
- }
-
- // read page
- if (readAndCalcFromPage(curTimeRange)) {
- return true;
- }
- }
- return false;
- }
-
- private void calcFromStatistics(Statistics statistics) {
- for (int i = 0; i < aggregators.size(); i++) {
- Aggregator aggregator = aggregators.get(i);
- if (aggregator.hasFinalResult()) {
- continue;
- }
- aggregator.processStatistics(statistics);
- }
- }
-
- public boolean canUseCurrentFileStatistics() throws IOException {
- Statistics fileStatistics = seriesScanUtil.currentFileStatistics();
- return !seriesScanUtil.isFileOverlapped()
- && containedByTimeFilter(fileStatistics)
- && !seriesScanUtil.currentFileModified();
- }
-
- public boolean canUseCurrentChunkStatistics() throws IOException {
- Statistics chunkStatistics = seriesScanUtil.currentChunkStatistics();
- return !seriesScanUtil.isChunkOverlapped()
- && containedByTimeFilter(chunkStatistics)
- && !seriesScanUtil.currentChunkModified();
- }
-
- public boolean canUseCurrentPageStatistics() throws IOException {
- Statistics currentPageStatistics = seriesScanUtil.currentPageStatistics();
- if (currentPageStatistics == null) {
- return false;
- }
- return !seriesScanUtil.isPageOverlapped()
- && containedByTimeFilter(currentPageStatistics)
- && !seriesScanUtil.currentPageModified();
- }
-
- private boolean containedByTimeFilter(Statistics statistics) {
- Filter timeFilter = seriesScanUtil.getTimeFilter();
- return timeFilter == null
- || timeFilter.containStartEndTime(statistics.getStartTime(), statistics.getEndTime());
+ this.subSensorSize = 1;
Review Comment:
put this field into super's constructor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org