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 f706aacccaf Fix multiple bugs in window function
f706aacccaf is described below
commit f706aacccafa8c47877d9397a0db37b814defd79
Author: Zhihao Shen <[email protected]>
AuthorDate: Fri Jun 20 22:03:54 2025 +0800
Fix multiple bugs in window function
---
.../relational/it/db/it/IoTDBWindowFunctionIT.java | 170 ++++++++++++++++++++-
.../process/window/TableWindowOperator.java | 1 -
.../window/partition/PartitionExecutor.java | 6 +-
.../process/window/partition/frame/RangeFrame.java | 6 +-
.../process/window/utils/RowComparator.java | 20 +++
.../relational/analyzer/ExpressionAnalyzer.java | 5 +-
.../plan/relational/planner/QueryPlanner.java | 42 +++--
.../distribute/TableDistributedPlanGenerator.java | 11 ++
.../planner/optimizations/SortElimination.java | 60 ++++++--
9 files changed, 285 insertions(+), 36 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowFunctionIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowFunctionIT.java
index 2f5b04a11f5..1be98fa7bf4 100644
---
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowFunctionIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowFunctionIT.java
@@ -41,7 +41,7 @@ import static org.junit.Assert.fail;
@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
public class IoTDBWindowFunctionIT {
private static final String DATABASE_NAME = "test";
- private static final String[] sqls =
+ private static final String[] sqlsWithoutNulls =
new String[] {
"CREATE DATABASE " + DATABASE_NAME,
"USE " + DATABASE_NAME,
@@ -55,11 +55,31 @@ public class IoTDBWindowFunctionIT {
"FLUSH",
"CLEAR ATTRIBUTE CACHE",
};
+ private static final String[] sqlsWithNulls =
+ new String[] {
+ "create table demo2 (device string tag, value double field)",
+ "insert into demo2 values (2021-01-01T09:04:00, 'd1', null)",
+ "insert into demo2 values (2021-01-01T09:05:00, 'd1', 3)",
+ "insert into demo2 values (2021-01-01T09:07:00, 'd1', 5)",
+ "insert into demo2 values (2021-01-01T09:09:00, 'd1', 3)",
+ "insert into demo2 values (2021-01-01T09:10:00, 'd1', 1)",
+ "insert into demo2 values (2021-01-01T09:06:00, 'd2', null)",
+ "insert into demo2 values (2021-01-01T09:08:00, 'd2', 2)",
+ "insert into demo2 values (2021-01-01T09:15:00, 'd2', 4)",
+ "insert into demo2 values (2021-01-01T09:20:00, null, null)",
+ "insert into demo2 values (2021-01-01T09:21:00, null, 1)",
+ "insert into demo2 values (2021-01-01T09:22:00, null, 2)",
+ "FLUSH",
+ "CLEAR ATTRIBUTE CACHE",
+ };
private static void insertData() {
try (Connection connection = EnvFactory.getEnv().getTableConnection();
Statement statement = connection.createStatement()) {
- for (String sql : sqls) {
+ for (String sql : sqlsWithoutNulls) {
+ statement.execute(sql);
+ }
+ for (String sql : sqlsWithNulls) {
statement.execute(sql);
}
} catch (Exception e) {
@@ -78,6 +98,25 @@ public class IoTDBWindowFunctionIT {
EnvFactory.getEnv().cleanClusterEnvironment();
}
+ @Test
+ public void testEmptyOver() {
+ String[] expectedHeader = new String[] {"time", "device", "value", "cnt"};
+ String[] retArray =
+ new String[] {
+ "2021-01-01T09:05:00.000Z,d1,3.0,6,",
+ "2021-01-01T09:07:00.000Z,d1,5.0,6,",
+ "2021-01-01T09:09:00.000Z,d1,3.0,6,",
+ "2021-01-01T09:10:00.000Z,d1,1.0,6,",
+ "2021-01-01T09:08:00.000Z,d2,2.0,6,",
+ "2021-01-01T09:15:00.000Z,d2,4.0,6,",
+ };
+ tableResultSetEqualTest(
+ "SELECT *, count(value) OVER () AS cnt FROM demo ORDER BY device",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
@Test
public void testPartitionBy() {
String[] expectedHeader = new String[] {"time", "device", "value", "cnt"};
@@ -97,8 +136,75 @@ public class IoTDBWindowFunctionIT {
DATABASE_NAME);
}
+ @Test
+ public void testPartitionByWithNulls() {
+ String[] expectedHeader = new String[] {"time", "device", "value", "cnt"};
+ String[] retArray =
+ new String[] {
+ "2021-01-01T09:04:00.000Z,d1,null,4,",
+ "2021-01-01T09:05:00.000Z,d1,3.0,4,",
+ "2021-01-01T09:07:00.000Z,d1,5.0,4,",
+ "2021-01-01T09:09:00.000Z,d1,3.0,4,",
+ "2021-01-01T09:10:00.000Z,d1,1.0,4,",
+ "2021-01-01T09:06:00.000Z,d2,null,2,",
+ "2021-01-01T09:08:00.000Z,d2,2.0,2,",
+ "2021-01-01T09:15:00.000Z,d2,4.0,2,",
+ "2021-01-01T09:20:00.000Z,null,null,2,",
+ "2021-01-01T09:21:00.000Z,null,1.0,2,",
+ "2021-01-01T09:22:00.000Z,null,2.0,2,",
+ };
+ tableResultSetEqualTest(
+ "SELECT *, count(value) OVER (PARTITION BY device) AS cnt FROM demo2
ORDER BY device",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
@Test
public void testOrderBy() {
+ String[] expectedHeader = new String[] {"time", "device", "value", "cnt"};
+ String[] retArray =
+ new String[] {
+ "2021-01-01T09:10:00.000Z,d1,1.0,1,",
+ "2021-01-01T09:08:00.000Z,d2,2.0,2,",
+ "2021-01-01T09:05:00.000Z,d1,3.0,4,",
+ "2021-01-01T09:09:00.000Z,d1,3.0,4,",
+ "2021-01-01T09:15:00.000Z,d2,4.0,5,",
+ "2021-01-01T09:07:00.000Z,d1,5.0,6,",
+ };
+ tableResultSetEqualTest(
+ "SELECT *, count(value) OVER (ORDER BY value) AS cnt FROM demo",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testOrderByWithNulls() {
+ String[] expectedHeader = new String[] {"time", "device", "value", "cnt"};
+ String[] retArray =
+ new String[] {
+ "2021-01-01T09:10:00.000Z,d1,1.0,2,",
+ "2021-01-01T09:21:00.000Z,null,1.0,2,",
+ "2021-01-01T09:08:00.000Z,d2,2.0,4,",
+ "2021-01-01T09:22:00.000Z,null,2.0,4,",
+ "2021-01-01T09:05:00.000Z,d1,3.0,6,",
+ "2021-01-01T09:09:00.000Z,d1,3.0,6,",
+ "2021-01-01T09:15:00.000Z,d2,4.0,7,",
+ "2021-01-01T09:07:00.000Z,d1,5.0,8,",
+ "2021-01-01T09:04:00.000Z,d1,null,8,",
+ "2021-01-01T09:06:00.000Z,d2,null,8,",
+ "2021-01-01T09:20:00.000Z,null,null,8,",
+ };
+ tableResultSetEqualTest(
+ "SELECT *, count(value) OVER (ORDER BY value) AS cnt FROM demo2 ORDER
BY value, device",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testPartitionByAndOrderBy() {
String[] expectedHeader = new String[] {"time", "device", "value", "rnk"};
String[] retArray =
new String[] {
@@ -116,6 +222,30 @@ public class IoTDBWindowFunctionIT {
DATABASE_NAME);
}
+ @Test
+ public void testPartitionByAndOrderByWithNulls() {
+ String[] expectedHeader = new String[] {"time", "device", "value", "cnt"};
+ String[] retArray =
+ new String[] {
+ "2021-01-01T09:10:00.000Z,d1,1.0,1,",
+ "2021-01-01T09:05:00.000Z,d1,3.0,3,",
+ "2021-01-01T09:09:00.000Z,d1,3.0,3,",
+ "2021-01-01T09:07:00.000Z,d1,5.0,4,",
+ "2021-01-01T09:04:00.000Z,d1,null,4,",
+ "2021-01-01T09:08:00.000Z,d2,2.0,1,",
+ "2021-01-01T09:15:00.000Z,d2,4.0,2,",
+ "2021-01-01T09:06:00.000Z,d2,null,2,",
+ "2021-01-01T09:21:00.000Z,null,1.0,1,",
+ "2021-01-01T09:22:00.000Z,null,2.0,2,",
+ "2021-01-01T09:20:00.000Z,null,null,2,",
+ };
+ tableResultSetEqualTest(
+ "SELECT *, count(value) OVER (PARTITION BY device ORDER BY value) AS
cnt FROM demo2 ORDER BY device",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
@Test
public void testRowsFraming() {
String[] expectedHeader = new String[] {"time", "device", "value", "cnt"};
@@ -410,4 +540,40 @@ public class IoTDBWindowFunctionIT {
retArray,
DATABASE_NAME);
}
+
+ @Test
+ public void testNegativeRowsFrameOffset() {
+ tableAssertTestFail(
+ "SELECT *, count(value) OVER (PARTITION BY device ORDER BY time ROWS
-1 PRECEDING) AS cnt FROM demo",
+ "Window frame offset value must not be negative or null",
+ DATABASE_NAME);
+
+ tableAssertTestFail(
+ "SELECT *, count(value) OVER (PARTITION BY device ORDER BY time ROWS
BETWEEN -2 PRECEDING AND -1 FOLLOWING) AS cnt FROM demo",
+ "Window frame offset value must not be negative or null",
+ DATABASE_NAME);
+
+ tableAssertTestFail(
+ "SELECT *, count(value) OVER (PARTITION BY device ORDER BY time ROWS
BETWEEN 1 PRECEDING AND -1 FOLLOWING) AS cnt FROM demo",
+ "Window frame offset value must not be negative or null",
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testNegativeGroupsFrameOffset() {
+ tableAssertTestFail(
+ "SELECT *, count(value) OVER (PARTITION BY device ORDER BY time GROUPS
-1 PRECEDING) AS cnt FROM demo",
+ "Window frame offset value must not be negative or null",
+ DATABASE_NAME);
+
+ tableAssertTestFail(
+ "SELECT *, count(value) OVER (PARTITION BY device ORDER BY time GROUPS
BETWEEN -2 PRECEDING AND -1 FOLLOWING) AS cnt FROM demo",
+ "Window frame offset value must not be negative or null",
+ DATABASE_NAME);
+
+ tableAssertTestFail(
+ "SELECT *, count(value) OVER (PARTITION BY device ORDER BY time GROUPS
BETWEEN 1 PRECEDING AND -1 FOLLOWING) AS cnt FROM demo",
+ "Window frame offset value must not be negative or null",
+ DATABASE_NAME);
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/TableWindowOperator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/TableWindowOperator.java
index cb7fb15d470..e1737ffdbcd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/TableWindowOperator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/TableWindowOperator.java
@@ -294,7 +294,6 @@ public class TableWindowOperator implements ProcessOperator
{
partitionExecutors.addLast(partitionExecutor);
partitionStartInCurrentBlock = partitionEndInCurrentBlock;
- partitionEndInCurrentBlock = partitionStartInCurrentBlock + 1;
} else {
// Last partition of TsBlock
// The beginning of next TsBlock may have rows in this partition
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/partition/PartitionExecutor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/partition/PartitionExecutor.java
index bf136c12391..9075592a7e1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/partition/PartitionExecutor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/partition/PartitionExecutor.java
@@ -147,7 +147,11 @@ public final class PartitionExecutor {
for (int i = 0; i < outputChannels.size(); i++) {
Column column = tsBlock.getColumn(outputChannels.get(i));
ColumnBuilder columnBuilder = builder.getColumnBuilder(i);
- columnBuilder.write(column, offsetInTsBlock);
+ if (column.isNull(offsetInTsBlock)) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.write(column, offsetInTsBlock);
+ }
channel++;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/partition/frame/RangeFrame.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/partition/frame/RangeFrame.java
index b4b4615be15..64981fb6232 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/partition/frame/RangeFrame.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/partition/frame/RangeFrame.java
@@ -40,6 +40,7 @@ public class RangeFrame implements Frame {
private final FrameInfo frameInfo;
private boolean noOrderBy = false;
+ private List<ColumnList> allSortedColumns;
private ColumnList column;
private TSDataType dataType;
@@ -64,7 +65,7 @@ public class RangeFrame implements Frame {
}
// Only one sort key is allowed in range frame
- checkArgument(sortedColumns.size() == 1);
+ this.allSortedColumns = sortedColumns;
this.column = sortedColumns.get(0);
this.dataType = column.getDataType();
this.peerGroupComparator = comparator;
@@ -87,7 +88,8 @@ public class RangeFrame implements Frame {
|| frameInfo.getStartType() == UNBOUNDED_PRECEDING
&& frameInfo.getEndType() == CURRENT_ROW) {
if (currentPosition == 0
- || !peerGroupComparator.equal(column, currentPosition - 1,
currentPosition)) {
+ || !peerGroupComparator.equalColumnLists(
+ allSortedColumns, currentPosition - 1, currentPosition)) {
// New peer group
int frameStart = frameInfo.getStartType() == CURRENT_ROW ?
peerGroupStart : 0;
int frameEnd = frameInfo.getEndType() == CURRENT_ROW ? peerGroupEnd -
1 : partitionSize - 1;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/utils/RowComparator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/utils/RowComparator.java
index cd0c2ffdcff..ac6f2258809 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/utils/RowComparator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/utils/RowComparator.java
@@ -50,6 +50,14 @@ public class RowComparator {
}
private boolean equal(Column column, TSDataType dataType, int offset1, int
offset2) {
+ if (offset1 == offset2) {
+ return true;
+ }
+
+ if (column.isNull(offset1) || column.isNull(offset2)) {
+ return column.isNull(offset1) && column.isNull(offset2);
+ }
+
switch (dataType) {
case BOOLEAN:
boolean bool1 = column.getBoolean(offset1);
@@ -121,6 +129,14 @@ public class RowComparator {
}
private boolean equal(ColumnList column, TSDataType dataType, int offset1,
int offset2) {
+ if (offset1 == offset2) {
+ return true;
+ }
+
+ if (column.isNull(offset1) || column.isNull(offset2)) {
+ return column.isNull(offset1) && column.isNull(offset2);
+ }
+
switch (dataType) {
case BOOLEAN:
boolean bool1 = column.getBoolean(offset1);
@@ -181,6 +197,10 @@ public class RowComparator {
Column column1 = columns1.get(i);
Column column2 = columns2.get(i);
+ if (column1.isNull(offset1) || column2.isNull(offset2)) {
+ return column1.isNull(offset1) && column2.isNull(offset2);
+ }
+
switch (dataType) {
case BOOLEAN:
boolean bool1 = column1.getBoolean(offset1);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/ExpressionAnalyzer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/ExpressionAnalyzer.java
index 9032929d2c7..fad9bc32d2f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/ExpressionAnalyzer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/ExpressionAnalyzer.java
@@ -1778,8 +1778,9 @@ public class ExpressionAnalyzer {
"Window frame of type RANGE PRECEDING or FOLLOWING
requires ORDER BY"));
if (orderBy.getSortItems().size() != 1) {
throw new SemanticException(
- "Window frame of type RANGE PRECEDING or FOLLOWING requires single
sort item in ORDER BY (actual: %s)",
- orderBy.getSortItems().size());
+ String.format(
+ "Window frame of type RANGE PRECEDING or FOLLOWING requires
single sort item in ORDER BY (actual: %s)",
+ orderBy.getSortItems().size()));
}
Expression sortKey =
Iterables.getOnlyElement(orderBy.getSortItems()).getSortKey();
Type sortKeyType;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/QueryPlanner.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/QueryPlanner.java
index 4df3d4f6562..cb5085d7aa2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/QueryPlanner.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/QueryPlanner.java
@@ -394,12 +394,12 @@ public class QueryPlanner {
window.getFrame().get().getEnd().flatMap(FrameBound::getValue);
// process frame start
- FrameOffsetPlanAndSymbol plan = planFrameOffset(subPlan,
startValue.map(coercions::get));
+ FrameOffsetPlanAndSymbol plan = planFrameOffset(subPlan, startValue,
coercions);
subPlan = plan.getSubPlan();
frameStart = plan.getFrameOffsetSymbol();
// process frame end
- plan = planFrameOffset(subPlan, endValue.map(coercions::get));
+ plan = planFrameOffset(subPlan, endValue, coercions);
subPlan = plan.getSubPlan();
frameEnd = plan.getFrameOffsetSymbol();
} else if (window.getFrame().isPresent()) {
@@ -509,22 +509,26 @@ public class QueryPlanner {
}
}
});
- GroupNode groupNode =
- new GroupNode(
- queryIdAllocator.genPlanNodeId(),
- subPlan.getRoot(),
- new OrderingScheme(sortSymbols, sortOrderings),
- sortKeyOffset);
- PlanBuilder planBuilder =
- new PlanBuilder(
-
subPlan.getTranslations().withAdditionalMappings(mappings.buildOrThrow()),
groupNode);
+
+ PlanBuilder planBuilder = null;
+ if (!sortSymbols.isEmpty()) {
+ GroupNode groupNode =
+ new GroupNode(
+ queryIdAllocator.genPlanNodeId(),
+ subPlan.getRoot(),
+ new OrderingScheme(sortSymbols, sortOrderings),
+ sortKeyOffset);
+ planBuilder =
+ new PlanBuilder(
+
subPlan.getTranslations().withAdditionalMappings(mappings.buildOrThrow()),
groupNode);
+ }
// create window node
return new PlanBuilder(
subPlan.getTranslations().withAdditionalMappings(mappings.buildOrThrow()),
new WindowNode(
queryIdAllocator.genPlanNodeId(),
- planBuilder.getRoot(),
+ planBuilder != null ? planBuilder.getRoot() : subPlan.getRoot(),
specification,
functions.buildOrThrow(),
Optional.empty(),
@@ -570,7 +574,7 @@ public class QueryPlanner {
return new FrameBoundPlanAndSymbols(subPlan, Optional.empty(),
Optional.empty());
}
- // First, append filter to validate offset values. They mustn't be
negative or null.
+ // Append filter to validate offset values. They mustn't be negative or
null.
Symbol offsetSymbol = coercions.get(frameOffset.get());
Expression zeroOffset =
zeroOfType(symbolAllocator.getTypes().getTableModelType(offsetSymbol));
Expression predicate =
@@ -655,12 +659,20 @@ public class QueryPlanner {
}
private FrameOffsetPlanAndSymbol planFrameOffset(
- PlanBuilder subPlan, Optional<Symbol> frameOffset) {
+ PlanBuilder subPlan, Optional<Expression> frameOffset, PlanAndMappings
coercions) {
if (!frameOffset.isPresent()) {
return new FrameOffsetPlanAndSymbol(subPlan, Optional.empty());
}
- Symbol offsetSymbol = frameOffset.get();
+ // Report error if frame offsets are literals and they are negative or null
+ if (frameOffset.get() instanceof LongLiteral) {
+ long frameOffsetValue = ((LongLiteral)
frameOffset.get()).getParsedValue();
+ if (frameOffsetValue < 0) {
+ throw new SemanticException("Window frame offset value must not be
negative or null");
+ }
+ }
+
+ Symbol offsetSymbol = frameOffset.map(coercions::get).get();
Type offsetType =
symbolAllocator.getTypes().getTableModelType(offsetSymbol);
// Append filter to validate offset values. They mustn't be negative or
null.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
index 31526d7c0e6..7f43fc7446b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
@@ -562,9 +562,15 @@ public class TableDistributedPlanGenerator
@Override
public List<PlanNode> visitPatternRecognition(PatternRecognitionNode node,
PlanContext context) {
context.clearExpectedOrderingScheme();
+ if (node.getPartitionBy().isEmpty()) {
+ Optional<OrderingScheme> orderingScheme = node.getOrderingScheme();
+ orderingScheme.ifPresent(scheme ->
nodeOrderingMap.put(node.getPlanNodeId(), scheme));
+ }
+
if (node.getChildren().isEmpty()) {
return Collections.singletonList(node);
}
+
boolean canSplitPushDown = (node.getChild() instanceof GroupNode);
List<PlanNode> childrenNodes = node.getChild().accept(this, context);
if (childrenNodes.size() == 1) {
@@ -1648,6 +1654,11 @@ public class TableDistributedPlanGenerator
@Override
public List<PlanNode> visitWindowFunction(WindowNode node, PlanContext
context) {
context.clearExpectedOrderingScheme();
+ if (node.getSpecification().getPartitionBy().isEmpty()) {
+ Optional<OrderingScheme> orderingScheme =
node.getSpecification().getOrderingScheme();
+ orderingScheme.ifPresent(scheme ->
nodeOrderingMap.put(node.getPlanNodeId(), scheme));
+ }
+
if (node.getChildren().isEmpty()) {
return Collections.singletonList(node);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java
index b785637bb1e..e581678a58f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java
@@ -26,9 +26,11 @@ import
org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FillNode;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode;
+import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.PatternRecognitionNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.StreamSortNode;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.WindowNode;
import java.util.Collections;
@@ -66,14 +68,14 @@ public class SortElimination implements PlanOptimizer {
public PlanNode visitSort(SortNode node, Context context) {
Context newContext = new Context();
PlanNode child = node.getChild().accept(this, newContext);
- context.setHasSeenFill(newContext.hasSeenFill);
+ context.setCannotEliminateSort(newContext.cannotEliminateSort);
OrderingScheme orderingScheme = node.getOrderingScheme();
- if (!context.hasSeenFill()
+ if (context.canEliminateSort()
&& newContext.getTotalDeviceEntrySize() == 1
&&
orderingScheme.getOrderBy().get(0).getName().equals(context.getTimeColumnName()))
{
return child;
}
- return !context.hasSeenFill() && node.isOrderByAllIdsAndTime()
+ return context.canEliminateSort() && node.isOrderByAllIdsAndTime()
? child
: node.replaceChildren(Collections.singletonList(child));
}
@@ -82,8 +84,8 @@ public class SortElimination implements PlanOptimizer {
public PlanNode visitStreamSort(StreamSortNode node, Context context) {
Context newContext = new Context();
PlanNode child = node.getChild().accept(this, newContext);
- context.setHasSeenFill(newContext.hasSeenFill);
- return !context.hasSeenFill()
+ context.setCannotEliminateSort(newContext.cannotEliminateSort);
+ return context.canEliminateSort()
&& (node.isOrderByAllIdsAndTime()
|| node.getStreamCompareKeyEndIndex()
== node.getOrderingScheme().getOrderBy().size() - 1)
@@ -104,7 +106,7 @@ public class SortElimination implements PlanOptimizer {
for (PlanNode child : node.getChildren()) {
newNode.addChild(child.accept(this, context));
}
- context.setHasSeenFill(!(node instanceof ValueFillNode));
+ context.setCannotEliminateSort(!(node instanceof ValueFillNode));
return newNode;
}
@@ -114,7 +116,36 @@ public class SortElimination implements PlanOptimizer {
for (PlanNode child : node.getChildren()) {
newNode.addChild(child.accept(this, context));
}
- context.setHasSeenFill(true);
+ context.setCannotEliminateSort(true);
+ return newNode;
+ }
+
+ @Override
+ public PlanNode visitWindowFunction(WindowNode node, Context context) {
+ PlanNode newNode = node.clone();
+ for (PlanNode child : node.getChildren()) {
+ newNode.addChild(child.accept(this, context));
+ }
+
+ // We can continue to eliminate sort when there is only PARTITION BY
+ if (node.getSpecification().getPartitionBy().isEmpty()
+ || node.getSpecification().getOrderingScheme().isPresent()) {
+ context.setCannotEliminateSort(true);
+ }
+ return newNode;
+ }
+
+ @Override
+ public PlanNode visitPatternRecognition(PatternRecognitionNode node,
Context context) {
+ PlanNode newNode = node.clone();
+ for (PlanNode child : node.getChildren()) {
+ newNode.addChild(child.accept(this, context));
+ }
+
+ // Same as window function
+ if (node.getPartitionBy().isEmpty() ||
node.getOrderingScheme().isPresent()) {
+ context.setCannotEliminateSort(true);
+ }
return newNode;
}
}
@@ -122,8 +153,11 @@ public class SortElimination implements PlanOptimizer {
private static class Context {
private int totalDeviceEntrySize = 0;
- // has seen linear fill, previous fill or gapfill
- private boolean hasSeenFill = false;
+ // There are 3 situations where sort cannot be eliminated
+ // 1. Query plan has linear fill, previous fill or gapfill
+ // 2. Query plan has window function and it has ordering scheme
+ // 3. Query plan has pattern recognition and it has ordering scheme
+ private boolean cannotEliminateSort = false;
private String timeColumnName = null;
@@ -137,12 +171,12 @@ public class SortElimination implements PlanOptimizer {
return totalDeviceEntrySize;
}
- public boolean hasSeenFill() {
- return hasSeenFill;
+ public boolean canEliminateSort() {
+ return !cannotEliminateSort;
}
- public void setHasSeenFill(boolean hasSeenFill) {
- this.hasSeenFill = hasSeenFill;
+ public void setCannotEliminateSort(boolean cannotEliminateSort) {
+ this.cannotEliminateSort = cannotEliminateSort;
}
public String getTimeColumnName() {
