This is an automated email from the ASF dual-hosted git repository.
liudw pushed a commit to branch reverse_reader
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/reverse_reader by this push:
new b479e5c add 'groupByTime' order by time logic
b479e5c is described below
commit b479e5cefd565e70191c93051690d600a8c21cfb
Author: liudw <[email protected]>
AuthorDate: Sat Aug 22 01:30:00 2020 +0800
add 'groupByTime' order by time logic
---
.../iotdb/db/qp/logical/crud/QueryOperator.java | 2 +-
.../iotdb/db/qp/physical/crud/QueryPlan.java | 2 +-
.../db/query/aggregation/AggregateResult.java | 7 +++--
.../db/query/aggregation/impl/AvgAggrResult.java | 7 +++--
.../db/query/aggregation/impl/CountAggrResult.java | 5 ++--
.../aggregation/impl/FirstValueAggrResult.java | 9 ++++--
.../aggregation/impl/LastValueAggrResult.java | 15 ++++++----
.../query/aggregation/impl/MaxTimeAggrResult.java | 7 +++--
.../query/aggregation/impl/MaxValueAggrResult.java | 13 +++++----
.../query/aggregation/impl/MinTimeAggrResult.java | 9 ++++--
.../query/aggregation/impl/MinValueAggrResult.java | 11 ++++---
.../db/query/aggregation/impl/SumAggrResult.java | 7 +++--
.../dataset/groupby/GroupByEngineDataSet.java | 34 +++++++++++++++++-----
.../groupby/GroupByWithoutValueFilterDataSet.java | 2 +-
.../dataset/groupby/LocalGroupByExecutor.java | 25 +++++++++++-----
.../db/query/executor/RawDataQueryExecutor.java | 19 ++++++++----
.../query/reader/series/SeriesAggregateReader.java | 2 +-
.../reader/series/SeriesRawDataBatchReader.java | 4 +--
.../query/timegenerator/ServerTimeGenerator.java | 10 +++++--
.../iotdb/db/engine/storagegroup/TTLTest.java | 5 ++--
.../tsfile/read/query/dataset/QueryDataSet.java | 8 +++++
21 files changed, 137 insertions(+), 66 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java
index c6bde39..7707961 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java
@@ -55,7 +55,7 @@ public class QueryOperator extends SFWOperator {
private String column;
- private boolean ascending;
+ private boolean ascending = true;
public QueryOperator(int tokenIntType) {
super(tokenIntType);
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/QueryPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/QueryPlan.java
index 8aa9773..89ab0f2 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/QueryPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/QueryPlan.java
@@ -36,7 +36,7 @@ public abstract class QueryPlan extends PhysicalPlan {
private int rowLimit = 0;
private int rowOffset = 0;
- private boolean ascending;
+ private boolean ascending = true;
private Map<String, Integer> pathToIndex = new HashMap<>();
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java
index 832e69a..ed798e5 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java
@@ -78,10 +78,11 @@ public abstract class AggregateResult {
* Aggregate results cannot be calculated using Statistics directly, using
the data in each page
*
* @param dataInThisPage the data in Page
- * @param bound calculate points whose time < bound
+ * @param minBound calculate points whose time > bound
+ * @param maxBound calculate points whose time < bound
*/
- public abstract void updateResultFromPageData(BatchData dataInThisPage, long
bound)
- throws IOException;
+ public abstract void updateResultFromPageData(BatchData dataInThisPage, long
minBound,
+ long maxBound) throws IOException;
/**
* <p> This method calculates the aggregation using common timestamps of the
cross series
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/AvgAggrResult.java
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/AvgAggrResult.java
index 2eaf975..1d16cca 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/AvgAggrResult.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/AvgAggrResult.java
@@ -74,13 +74,14 @@ public class AvgAggrResult extends AggregateResult {
@Override
public void updateResultFromPageData(BatchData dataInThisPage) throws
IOException {
- updateResultFromPageData(dataInThisPage, Long.MAX_VALUE);
+ updateResultFromPageData(dataInThisPage, Long.MIN_VALUE, Long.MAX_VALUE);
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage, long bound)
throws IOException {
+ public void updateResultFromPageData(BatchData dataInThisPage, long
minBound, long maxBound)
+ throws IOException {
while (dataInThisPage.hasCurrent()) {
- if (dataInThisPage.currentTime() >= bound) {
+ if (dataInThisPage.currentTime() >= maxBound ||
dataInThisPage.currentTime() < minBound) {
break;
}
updateAvg(seriesDataType, dataInThisPage.currentValue());
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/CountAggrResult.java
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/CountAggrResult.java
index 7b96304..b15294f 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/CountAggrResult.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/CountAggrResult.java
@@ -58,9 +58,10 @@ public class CountAggrResult extends AggregateResult {
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage, long bound) {
+ public void updateResultFromPageData(BatchData dataInThisPage, long
minBound, long maxBound)
+ throws IOException {
while (dataInThisPage.hasCurrent()) {
- if (dataInThisPage.currentTime() >= bound) {
+ if (dataInThisPage.currentTime() >= maxBound ||
dataInThisPage.currentTime() < minBound) {
break;
}
long preValue = getLongValue();
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/FirstValueAggrResult.java
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/FirstValueAggrResult.java
index 2dc3e2b..0e2dc7a 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/FirstValueAggrResult.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/FirstValueAggrResult.java
@@ -75,11 +75,14 @@ public class FirstValueAggrResult extends AggregateResult {
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage, long bound) {
+ public void updateResultFromPageData(BatchData dataInThisPage, long
minBound, long maxBound)
+ throws IOException {
if (hasResult()) {
return;
}
- if (dataInThisPage.hasCurrent() && dataInThisPage.currentTime() < bound) {
+ if (dataInThisPage.hasCurrent()
+ && dataInThisPage.currentTime() < maxBound
+ && dataInThisPage.currentTime() >= minBound) {
setValue(dataInThisPage.currentValue());
timestamp = dataInThisPage.currentTime();
dataInThisPage.next();
@@ -111,7 +114,7 @@ public class FirstValueAggrResult extends AggregateResult {
@Override
public void merge(AggregateResult another) {
FirstValueAggrResult anotherFirst = (FirstValueAggrResult) another;
- if(this.getValue() == null || this.timestamp > anotherFirst.timestamp){
+ if (this.getValue() == null || this.timestamp > anotherFirst.timestamp) {
setValue(anotherFirst.getValue());
timestamp = anotherFirst.timestamp;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/LastValueAggrResult.java
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/LastValueAggrResult.java
index 13a6a67..9375da0 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/LastValueAggrResult.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/LastValueAggrResult.java
@@ -59,15 +59,18 @@ public class LastValueAggrResult extends AggregateResult {
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage) {
- updateResultFromPageData(dataInThisPage, Long.MAX_VALUE);
+ public void updateResultFromPageData(BatchData dataInThisPage) throws
IOException {
+ updateResultFromPageData(dataInThisPage, Long.MIN_VALUE, Long.MAX_VALUE);
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage, long bound) {
+ public void updateResultFromPageData(BatchData dataInThisPage, long
minBound, long maxBound)
+ throws IOException {
long time = Long.MIN_VALUE;
Object lastVal = null;
- while (dataInThisPage.hasCurrent() && dataInThisPage.currentTime() <
bound) {
+ while (dataInThisPage.hasCurrent()
+ && dataInThisPage.currentTime() < maxBound
+ && dataInThisPage.currentTime() >= minBound) {
time = dataInThisPage.currentTime();
lastVal = dataInThisPage.currentValue();
dataInThisPage.next();
@@ -106,8 +109,8 @@ public class LastValueAggrResult extends AggregateResult {
@Override
public void merge(AggregateResult another) {
LastValueAggrResult anotherLast = (LastValueAggrResult) another;
- if(this.getValue() == null || this.timestamp < anotherLast.timestamp){
- this.setValue( anotherLast.getValue() );
+ if (this.getValue() == null || this.timestamp < anotherLast.timestamp) {
+ this.setValue(anotherLast.getValue());
this.timestamp = anotherLast.timestamp;
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MaxTimeAggrResult.java
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MaxTimeAggrResult.java
index 69dc286..78d834a 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MaxTimeAggrResult.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MaxTimeAggrResult.java
@@ -58,8 +58,11 @@ public class MaxTimeAggrResult extends AggregateResult {
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage, long bound) {
- while (dataInThisPage.hasCurrent() && dataInThisPage.currentTime() <
bound) {
+ public void updateResultFromPageData(BatchData dataInThisPage, long
minBound, long maxBound)
+ throws IOException {
+ while (dataInThisPage.hasCurrent()
+ && dataInThisPage.currentTime() < maxBound
+ && dataInThisPage.currentTime() >= minBound) {
updateMaxTimeResult(dataInThisPage.currentTime());
dataInThisPage.next();
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MaxValueAggrResult.java
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MaxValueAggrResult.java
index 567a1e9..3c996aa 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MaxValueAggrResult.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MaxValueAggrResult.java
@@ -48,15 +48,18 @@ public class MaxValueAggrResult extends AggregateResult {
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage) {
- updateResultFromPageData(dataInThisPage, Long.MAX_VALUE);
+ public void updateResultFromPageData(BatchData dataInThisPage) throws
IOException {
+ updateResultFromPageData(dataInThisPage, Long.MIN_VALUE, Long.MAX_VALUE);
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage, long bound) {
+ public void updateResultFromPageData(BatchData dataInThisPage, long
minBound, long maxBound)
+ throws IOException {
Comparable<Object> maxVal = null;
- while (dataInThisPage.hasCurrent() && dataInThisPage.currentTime() <
bound) {
+ while (dataInThisPage.hasCurrent()
+ && dataInThisPage.currentTime() < maxBound
+ && dataInThisPage.currentTime() >= minBound) {
if (maxVal == null || maxVal.compareTo(dataInThisPage.currentValue()) <
0) {
maxVal = (Comparable<Object>) dataInThisPage.currentValue();
}
@@ -88,7 +91,7 @@ public class MaxValueAggrResult extends AggregateResult {
@Override
public void merge(AggregateResult another) {
- this.updateResult((Comparable<Object>)another.getResult());
+ this.updateResult((Comparable<Object>) another.getResult());
}
@Override
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MinTimeAggrResult.java
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MinTimeAggrResult.java
index 27cbb29..9d15523 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MinTimeAggrResult.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MinTimeAggrResult.java
@@ -52,15 +52,18 @@ public class MinTimeAggrResult extends AggregateResult {
@Override
public void updateResultFromPageData(BatchData dataInThisPage) throws
IOException {
- updateResultFromPageData(dataInThisPage, Long.MAX_VALUE);
+ updateResultFromPageData(dataInThisPage, Long.MIN_VALUE, Long.MAX_VALUE);
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage, long bound)
throws IOException {
+ public void updateResultFromPageData(BatchData dataInThisPage, long
minBound, long maxBound)
+ throws IOException {
if (hasResult()) {
return;
}
- if (dataInThisPage.hasCurrent() && dataInThisPage.currentTime() < bound) {
+ if (dataInThisPage.hasCurrent()
+ && dataInThisPage.currentTime() < maxBound
+ && dataInThisPage.currentTime() >= minBound) {
setLongValue(dataInThisPage.currentTime());
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MinValueAggrResult.java
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MinValueAggrResult.java
index 7a20bbe..354ebc6 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MinValueAggrResult.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MinValueAggrResult.java
@@ -48,13 +48,16 @@ public class MinValueAggrResult extends AggregateResult {
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage) {
- updateResultFromPageData(dataInThisPage, Long.MAX_VALUE);
+ public void updateResultFromPageData(BatchData dataInThisPage) throws
IOException {
+ updateResultFromPageData(dataInThisPage, Long.MIN_VALUE, Long.MAX_VALUE);
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage, long bound) {
- while (dataInThisPage.hasCurrent() && dataInThisPage.currentTime() <
bound) {
+ public void updateResultFromPageData(BatchData dataInThisPage, long
minBound, long maxBound)
+ throws IOException {
+ while (dataInThisPage.hasCurrent()
+ && dataInThisPage.currentTime() < maxBound
+ && dataInThisPage.currentTime() >= minBound) {
updateResult((Comparable<Object>) dataInThisPage.currentValue());
dataInThisPage.next();
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/SumAggrResult.java
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/SumAggrResult.java
index 57d921e..ceb57c0 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/SumAggrResult.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/SumAggrResult.java
@@ -55,13 +55,14 @@ public class SumAggrResult extends AggregateResult {
@Override
public void updateResultFromPageData(BatchData dataInThisPage) throws
IOException {
- updateResultFromPageData(dataInThisPage, Long.MAX_VALUE);
+ updateResultFromPageData(dataInThisPage, Long.MIN_VALUE, Long.MAX_VALUE);
}
@Override
- public void updateResultFromPageData(BatchData dataInThisPage, long bound)
throws IOException {
+ public void updateResultFromPageData(BatchData dataInThisPage, long
minBound, long maxBound)
+ throws IOException {
while (dataInThisPage.hasCurrent()) {
- if (dataInThisPage.currentTime() >= bound) {
+ if (dataInThisPage.currentTime() >= maxBound ||
dataInThisPage.currentTime() < minBound) {
break;
}
updateSum(dataInThisPage.currentValue());
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByEngineDataSet.java
b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByEngineDataSet.java
index df1a6cf..725671e 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByEngineDataSet.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByEngineDataSet.java
@@ -43,6 +43,9 @@ public abstract class GroupByEngineDataSet extends
QueryDataSet {
protected boolean leftCRightO;
+ protected boolean ascending;
+ protected long curSteps;
+
public GroupByEngineDataSet() {
}
@@ -50,7 +53,8 @@ public abstract class GroupByEngineDataSet extends
QueryDataSet {
* groupBy query.
*/
public GroupByEngineDataSet(QueryContext context, GroupByTimePlan
groupByTimePlan) {
- super(groupByTimePlan.getDeduplicatedPaths(),
groupByTimePlan.getDeduplicatedDataTypes());
+ super(groupByTimePlan.getDeduplicatedPaths(),
groupByTimePlan.getDeduplicatedDataTypes(),
+ groupByTimePlan.isAscending());
this.queryId = context.getQueryId();
this.interval = groupByTimePlan.getInterval();
this.slidingStep = groupByTimePlan.getSlidingStep();
@@ -59,8 +63,15 @@ public abstract class GroupByEngineDataSet extends
QueryDataSet {
this.leftCRightO = groupByTimePlan.isLeftCRightO();
// init group by time partition
this.hasCachedTimeInterval = false;
+ this.ascending = groupByTimePlan.isAscending();
this.curStartTime = this.startTime - slidingStep;
this.curEndTime = -1;
+ if (!groupByTimePlan.isAscending()) {
+ this.curSteps = this.endTime / slidingStep;
+ this.curStartTime = curSteps * slidingStep - 1;
+ this.curEndTime = Math.min(curStartTime + interval, this.endTime);
+ hasCachedTimeInterval = true;
+ }
}
@Override
@@ -71,14 +82,23 @@ public abstract class GroupByEngineDataSet extends
QueryDataSet {
}
curStartTime += slidingStep;
- //This is an open interval , [0-100)
- if (curStartTime < endTime) {
- hasCachedTimeInterval = true;
- curEndTime = Math.min(curStartTime + interval, endTime);
- return true;
- } else {
+ if (ascending) {
+ //This is an open interval , [0-100)
+ if (curStartTime < endTime) {
+ hasCachedTimeInterval = true;
+ curEndTime = Math.min(curStartTime + interval, endTime);
+ return true;
+ }
return false;
}
+
+ curSteps--;
+ if (curSteps > 0) {
+ hasCachedTimeInterval = true;
+ curStartTime = slidingStep * curSteps - 1;
+ curEndTime = curStartTime + interval;
+ }
+ return false;
}
@Override
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
index a2635f1..5ecf772 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
@@ -86,7 +86,7 @@ public class GroupByWithoutValueFilterDataSet extends
GroupByEngineDataSet {
//init GroupByExecutor
pathExecutors.put(path,
getGroupByExecutor(path,
groupByTimePlan.getAllMeasurementsInDevice(path.getDevice()),
- dataTypes.get(i), context, timeFilter, null,
groupByTimePlan.isAlignByTime()));
+ dataTypes.get(i), context, timeFilter, null,
groupByTimePlan.isAscending()));
resultIndexes.put(path, new ArrayList<>());
}
resultIndexes.get(path).add(i);
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java
b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java
index 9571608..d845998 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java
@@ -52,6 +52,7 @@ public class LocalGroupByExecutor implements GroupByExecutor {
// used for resetting the batch data to the last index
private int lastReadCurArrayIndex;
private int lastReadCurListIndex;
+ private boolean ascending;
private QueryDataSource queryDataSource;
@@ -68,6 +69,7 @@ public class LocalGroupByExecutor implements GroupByExecutor {
timeRange = new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE);
lastReadCurArrayIndex = 0;
lastReadCurListIndex = 0;
+ this.ascending = ascending;
}
public boolean isEmpty() {
@@ -110,18 +112,25 @@ public class LocalGroupByExecutor implements
GroupByExecutor {
}
// lazy reset batch data for calculation
batchData.resetBatchData(lastReadCurArrayIndex, lastReadCurListIndex);
- // skip points that cannot be calculated
- while (batchData.hasCurrent() && batchData.currentTime() < curStartTime)
{
- batchData.next();
+ if (ascending) {
+ // skip points that cannot be calculated
+ while (batchData.hasCurrent() && batchData.currentTime() <
curStartTime) {
+ batchData.next();
+ }
+ } else {
+ while (batchData.hasCurrent() && batchData.currentTime() > curEndTime)
{
+ batchData.next();
+ }
}
+
if (batchData.hasCurrent()) {
- result.updateResultFromPageData(batchData, curEndTime);
+ result.updateResultFromPageData(batchData, curStartTime, curEndTime);
}
}
lastReadCurArrayIndex = batchData.getReadCurArrayIndex();
lastReadCurListIndex = batchData.getReadCurListIndex();
// can calc for next interval
- if (batchData.getMaxTimestamp() >= curEndTime) {
+ if (batchData.hasCurrent()) {
preCachedData = batchData;
}
}
@@ -236,9 +245,9 @@ public class LocalGroupByExecutor implements
GroupByExecutor {
return true;
}
- // reset the last position to zero
- lastReadCurArrayIndex = 0;
- lastReadCurListIndex = 0;
+ // reset the last position to current Index
+ lastReadCurArrayIndex = batchData.getReadCurArrayIndex();
+ lastReadCurListIndex = batchData.getReadCurListIndex();
calcFromBatch(batchData, curStartTime, curEndTime);
// judge whether the calculation finished
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
b/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
index 5899643..52b3624 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
@@ -54,11 +54,13 @@ public class RawDataQueryExecutor {
protected List<Path> deduplicatedPaths;
protected List<TSDataType> deduplicatedDataTypes;
protected IExpression optimizedExpression;
+ protected boolean ascending;
public RawDataQueryExecutor(RawDataQueryPlan queryPlan) {
this.deduplicatedPaths = queryPlan.getDeduplicatedPaths();
this.deduplicatedDataTypes = queryPlan.getDeduplicatedDataTypes();
this.optimizedExpression = queryPlan.getExpression();
+ this.ascending = queryPlan.isAscending();
}
/**
@@ -86,7 +88,8 @@ public class RawDataQueryExecutor {
readersOfSelectedSeries);
}
- protected List<ManagedSeriesReader> initManagedSeriesReader(QueryContext
context, RawDataQueryPlan queryPlan)
+ protected List<ManagedSeriesReader> initManagedSeriesReader(QueryContext
context,
+ RawDataQueryPlan queryPlan)
throws StorageEngineException, QueryProcessException {
Filter timeFilter = null;
if (optimizedExpression != null) {
@@ -102,8 +105,9 @@ public class RawDataQueryExecutor {
.getQueryDataSource(path, context, timeFilter);
timeFilter = queryDataSource.updateFilterUsingTTL(timeFilter);
- ManagedSeriesReader reader = new SeriesRawDataBatchReader(path,
queryPlan.getAllMeasurementsInDevice(path.getDevice()), dataType, context,
- queryDataSource, timeFilter, null, null);
+ ManagedSeriesReader reader = new SeriesRawDataBatchReader(path,
+ queryPlan.getAllMeasurementsInDevice(path.getDevice()), dataType,
context,
+ queryDataSource, timeFilter, null, null, ascending);
readersOfSelectedSeries.add(reader);
}
return readersOfSelectedSeries;
@@ -120,7 +124,8 @@ public class RawDataQueryExecutor {
TimeGenerator timestampGenerator = getTimeGenerator(
optimizedExpression, context, queryPlan);
- List<Boolean> cached = markFilterdPaths(optimizedExpression,
deduplicatedPaths, timestampGenerator.hasOrNode());
+ List<Boolean> cached = markFilterdPaths(optimizedExpression,
deduplicatedPaths,
+ timestampGenerator.hasOrNode());
List<IReaderByTimestamp> readersOfSelectedSeries = new ArrayList<>();
for (int i = 0; i < deduplicatedPaths.size(); i++) {
@@ -129,7 +134,8 @@ public class RawDataQueryExecutor {
continue;
}
Path path = deduplicatedPaths.get(i);
- IReaderByTimestamp seriesReaderByTimestamp = getReaderByTimestamp(path,
queryPlan.getAllMeasurementsInDevice(path.getDevice()),
+ IReaderByTimestamp seriesReaderByTimestamp = getReaderByTimestamp(path,
+ queryPlan.getAllMeasurementsInDevice(path.getDevice()),
deduplicatedDataTypes.get(i), context);
readersOfSelectedSeries.add(seriesReaderByTimestamp);
}
@@ -137,7 +143,8 @@ public class RawDataQueryExecutor {
timestampGenerator, readersOfSelectedSeries, cached);
}
- protected IReaderByTimestamp getReaderByTimestamp(Path path, Set<String>
allSensors, TSDataType dataType,
+ protected IReaderByTimestamp getReaderByTimestamp(Path path, Set<String>
allSensors,
+ TSDataType dataType,
QueryContext context) throws StorageEngineException,
QueryProcessException {
return new SeriesReaderByTimestamp(path, allSensors,
dataType, context,
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesAggregateReader.java
b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesAggregateReader.java
index d25539d..f8a5e43 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesAggregateReader.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesAggregateReader.java
@@ -114,7 +114,7 @@ public class SeriesAggregateReader implements
IAggregateReader {
@Override
public BatchData nextPage() throws IOException {
- return seriesReader.nextPage();
+ return seriesReader.nextPage().flip();
}
private boolean containedByTimeFilter(Statistics statistics) {
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesRawDataBatchReader.java
b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesRawDataBatchReader.java
index 6f96c47..82276d3 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesRawDataBatchReader.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesRawDataBatchReader.java
@@ -49,9 +49,9 @@ public class SeriesRawDataBatchReader implements
ManagedSeriesReader {
public SeriesRawDataBatchReader(Path seriesPath, Set<String> allSensors,
TSDataType dataType,
QueryContext context, QueryDataSource dataSource, Filter timeFilter,
Filter valueFilter,
- TsFileFilter fileFilter) {
+ TsFileFilter fileFilter, boolean ascending) {
this.seriesReader = new SeriesReader(seriesPath, allSensors, dataType,
context, dataSource,
- timeFilter, valueFilter, fileFilter, true);
+ timeFilter, valueFilter, fileFilter, ascending);
}
@TestOnly
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
b/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
index ee1cf91..210373d 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
@@ -51,7 +51,8 @@ public class ServerTimeGenerator extends TimeGenerator {
/**
* Constructor of EngineTimeGenerator.
*/
- public ServerTimeGenerator(IExpression expression, QueryContext context,
RawDataQueryPlan queryPlan)
+ public ServerTimeGenerator(IExpression expression, QueryContext context,
+ RawDataQueryPlan queryPlan)
throws StorageEngineException {
this.context = context;
this.queryPlan = queryPlan;
@@ -71,13 +72,16 @@ public class ServerTimeGenerator extends TimeGenerator {
QueryDataSource queryDataSource;
try {
dataType = IoTDB.metaManager.getSeriesType(path.getFullPath());
- queryDataSource =
QueryResourceManager.getInstance().getQueryDataSource(path, context, filter);
+ queryDataSource = QueryResourceManager.getInstance()
+ .getQueryDataSource(path, context, filter);
// update filter by TTL
filter = queryDataSource.updateFilterUsingTTL(filter);
} catch (Exception e) {
throw new IOException(e);
}
- return new SeriesRawDataBatchReader(path,
queryPlan.getAllMeasurementsInDevice(path.getDevice()), dataType, context,
queryDataSource, null, filter, null);
+ return new SeriesRawDataBatchReader(path,
+ queryPlan.getAllMeasurementsInDevice(path.getDevice()), dataType,
context, queryDataSource,
+ null, filter, null, true);
}
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
index fa0abb2..8b6f9e0 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
@@ -98,7 +98,8 @@ public class TTLTest {
IoTDBDescriptor.getInstance().getConfig().setPartitionInterval(prevPartitionInterval);
}
- private void insertToStorageGroupProcessor(InsertRowPlan insertPlan) throws
WriteProcessException {
+ private void insertToStorageGroupProcessor(InsertRowPlan insertPlan)
+ throws WriteProcessException {
insertPlan.setDeviceMNode(deviceMNode);
storageGroupProcessor.insert(insertPlan);
}
@@ -218,7 +219,7 @@ public class TTLTest {
Set<String> allSensors = new HashSet<>();
allSensors.add(s1);
IBatchReader reader = new SeriesRawDataBatchReader(path, allSensors,
TSDataType.INT64,
- EnvironmentUtils.TEST_QUERY_CONTEXT, dataSource, null, null, null);
+ EnvironmentUtils.TEST_QUERY_CONTEXT, dataSource, null, null, null,
true);
int cnt = 0;
while (reader.hasNextBatch()) {
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/query/dataset/QueryDataSet.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/query/dataset/QueryDataSet.java
index bebae01..84e2a29 100644
---
a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/query/dataset/QueryDataSet.java
+++
b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/query/dataset/QueryDataSet.java
@@ -32,6 +32,7 @@ public abstract class QueryDataSet {
protected int rowLimit = 0; // rowLimit > 0 means the LIMIT constraint exists
protected int rowOffset = 0;
protected int alreadyReturnedRowNum = 0;
+ protected boolean ascending;
public QueryDataSet() {
}
@@ -39,6 +40,13 @@ public abstract class QueryDataSet {
public QueryDataSet(List<Path> paths, List<TSDataType> dataTypes) {
this.paths = paths;
this.dataTypes = dataTypes;
+ this.ascending = true;
+ }
+
+ public QueryDataSet(List<Path> paths, List<TSDataType> dataTypes, boolean
ascending) {
+ this.paths = paths;
+ this.dataTypes = dataTypes;
+ this.ascending = ascending;
}
public boolean hasNext() throws IOException {
