You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2023/11/28 02:49:33 UTC
(iotdb) 01/01: fix npe in TopKOperator
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch beyyes/fix_dop
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 55daa6f6a95f088831b1db5a00cb5e3d754341d6
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Tue Nov 28 10:48:22 2023 +0800
fix npe in TopKOperator
---
.../IoTDBAlignByDeviceWithTemplateIT.java | 16 ++
.../execution/operator/process/TopKOperator.java | 25 +-
.../queryengine/plan/analyze/TemplatedAnalyze.java | 1 +
.../execution/operator/TopKOperatorTest.java | 297 +++++++++++++++++----
4 files changed, 280 insertions(+), 59 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBAlignByDeviceWithTemplateIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBAlignByDeviceWithTemplateIT.java
index 21ff962ad6a..78cda6b40e6 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBAlignByDeviceWithTemplateIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBAlignByDeviceWithTemplateIT.java
@@ -577,6 +577,22 @@ public class IoTDBAlignByDeviceWithTemplateIT {
"select count(s1+1) from root.sg1.** align by device;", expectedHeader, retArray);
}
+ @Test
+ public void emptyResultTest() {
+ String[] expectedHeader = new String[] {"Time,Device,s3,s1,s2"};
+ String[] retArray = new String[] {};
+ resultSetEqualTest(
+ "SELECT * FROM root.sg1.** where time>=now()-1d and time<=now() "
+ + "ORDER BY TIME DESC ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+ resultSetEqualTest(
+ "SELECT * FROM root.sg2.** where time>=now()-1d and time<=now() "
+ + "ORDER BY TIME DESC ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+ }
+
private static void insertData() {
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java
index d14741a31d6..d59067463f9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java
@@ -80,6 +80,8 @@ public class TopKOperator implements ProcessOperator {
// the data of every childOperator is in order
private final boolean childrenDataInOrder;
+ public static int operatorBatchUpperBound = 100000;
+
public TopKOperator(
OperatorContext operatorContext,
List<Operator> deviceOperators,
@@ -98,7 +100,10 @@ public class TopKOperator implements ProcessOperator {
initResultTsBlock();
- deviceBatchStep = 10000 % topValue == 0 ? 10000 / topValue : 10000 / topValue + 1;
+ deviceBatchStep =
+ operatorBatchUpperBound % topValue == 0
+ ? operatorBatchUpperBound / topValue
+ : operatorBatchUpperBound / topValue + 1;
canCallNext = new boolean[deviceOperators.size()];
}
@@ -137,7 +142,14 @@ public class TopKOperator implements ProcessOperator {
@Override
public boolean hasNext() throws Exception {
- return !(deviceIndex >= deviceOperators.size() && resultReturnSize >= topKResult.length);
+ if (deviceIndex >= deviceOperators.size()) {
+ if (topKResult == null) {
+ return false;
+ }
+
+ return resultReturnSize < topKResult.length;
+ }
+ return true;
}
@Override
@@ -289,10 +301,13 @@ public class TopKOperator implements ProcessOperator {
}
tsBlockBuilder.reset();
+
+ if (topKResult == null || topKResult.length == 0) {
+ return tsBlockBuilder.build();
+ }
+
ColumnBuilder[] valueColumnBuilders = tsBlockBuilder.getValueColumnBuilders();
- for (int i = resultReturnSize, size = (topKResult == null ? 0 : topKResult.length);
- i < size;
- i++) {
+ for (int i = resultReturnSize; i < topKResult.length; i++) {
MergeSortKey mergeSortKey = topKResult[i];
TsBlock targetBlock = mergeSortKey.tsBlock;
tsBlockBuilder
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyze.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyze.java
index a66b1518c92..5ec7997fdfc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyze.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyze.java
@@ -171,6 +171,7 @@ public class TemplatedAnalyze {
// generate result set header according to output expressions
analyzeOutput(analysis, queryStatement, outputExpressions);
+ context.generateGlobalTimeFilter(analysis);
// fetch partition information
analyzeDataPartition(analysis, schemaTree, partitionFetcher, context.getGlobalTimeFilter());
return true;
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java
index 2a48c94159f..62ba3e68971 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java
@@ -197,9 +197,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath1));
seriesScanOperator1.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator1
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
MeasurementPath measurementPath2 =
new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
@@ -211,9 +209,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath2));
seriesScanOperator2.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator2
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
MeasurementPath measurementPath3 =
new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
@@ -225,9 +221,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath3));
seriesScanOperator3.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator3
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
MeasurementPath measurementPath4 =
new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
@@ -239,9 +233,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath4));
seriesScanOperator4.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator4
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
MeasurementPath measurementPath5 =
new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
@@ -253,9 +245,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath5));
seriesScanOperator5.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator5
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
List<TSDataType> tsDataTypes =
new LinkedList<>(
@@ -292,9 +282,7 @@ public class TopKOperatorTest {
? new AscTimeComparator()
: new DescTimeComparator())),
timeOrdering == Ordering.ASC ? new AscTimeComparator() : new DescTimeComparator());
- timeJoinOperator1
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
SingleDeviceViewOperator singleDeviceViewOperator2 =
new SingleDeviceViewOperator(
@@ -322,9 +310,7 @@ public class TopKOperatorTest {
? new AscTimeComparator()
: new DescTimeComparator())),
timeOrdering == Ordering.ASC ? new AscTimeComparator() : new DescTimeComparator());
- timeJoinOperator2
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
SingleDeviceViewOperator singleDeviceViewOperator3 =
new SingleDeviceViewOperator(
@@ -348,7 +334,7 @@ public class TopKOperatorTest {
Arrays.asList(TSDataType.INT64, TSDataType.TEXT)),
limitValue,
true);
- topKOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
return topKOperator;
} catch (IllegalPathException e) {
e.printStackTrace();
@@ -913,9 +899,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath1));
seriesScanOperator1.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator1
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
MeasurementPath measurementPath2 =
new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
@@ -927,9 +911,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath2));
seriesScanOperator2.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator2
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
MeasurementPath measurementPath3 =
new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
@@ -941,9 +923,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath3));
seriesScanOperator3.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator3
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
MeasurementPath measurementPath4 =
new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
@@ -955,9 +935,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath4));
seriesScanOperator4.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator4
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
MeasurementPath measurementPath5 =
new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
@@ -969,9 +947,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath5));
seriesScanOperator5.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator5
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
MeasurementPath measurementPath6 =
new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device3.sensor0", TSDataType.INT32);
@@ -983,9 +959,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath6));
seriesScanOperator6.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator6
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
MeasurementPath measurementPath7 =
new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device3.sensor1", TSDataType.INT32);
@@ -997,9 +971,7 @@ public class TopKOperatorTest {
timeOrdering,
SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath7));
seriesScanOperator7.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- seriesScanOperator7
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
List<TSDataType> tsDataTypes =
new LinkedList<>(Arrays.asList(TSDataType.TEXT, TSDataType.INT32, TSDataType.INT32));
@@ -1021,9 +993,7 @@ public class TopKOperatorTest {
? new AscTimeComparator()
: new DescTimeComparator())),
timeOrdering == Ordering.ASC ? new AscTimeComparator() : new DescTimeComparator());
- timeJoinOperator1
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
RowBasedTimeJoinOperator timeJoinOperator2 =
new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(8),
@@ -1042,9 +1012,7 @@ public class TopKOperatorTest {
? new AscTimeComparator()
: new DescTimeComparator())),
timeOrdering == Ordering.ASC ? new AscTimeComparator() : new DescTimeComparator());
- timeJoinOperator2
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
RowBasedTimeJoinOperator timeJoinOperator3 =
new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(9),
@@ -1063,9 +1031,7 @@ public class TopKOperatorTest {
? new AscTimeComparator()
: new DescTimeComparator())),
timeOrdering == Ordering.ASC ? new AscTimeComparator() : new DescTimeComparator());
- timeJoinOperator3
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
List<String> devices = new ArrayList<>(Arrays.asList(DEVICE0, DEVICE1, DEVICE2, DEVICE3));
if (deviceOrdering == Ordering.DESC) {
Collections.reverse(devices);
@@ -1111,7 +1077,7 @@ public class TopKOperatorTest {
Arrays.asList(TSDataType.TEXT, TSDataType.INT64)),
limitValue,
false);
- topKOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
return topKOperator;
} catch (IllegalPathException e) {
e.printStackTrace();
@@ -1343,7 +1309,7 @@ public class TopKOperatorTest {
Collections.singletonList(TSDataType.INT64)),
20,
true);
- topKOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
int index = 0;
while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
@@ -1358,4 +1324,227 @@ public class TopKOperatorTest {
}
assertEquals(index, ans.length);
}
+
+ // ----------------------------------------------------------------------------------------------
+ // order by time - all result of SeriesScan is empty
+ // ----------------------------------------------------------------------------------------------
+ // TopKOperator
+ // ____________|_______________
+ // / | \
+ // SingleDeviceViewOperator SingleDeviceViewOperator SingleDeviceViewOperator
+ // / | \
+ // SeriesScanOperator TimeJoinOperator TimeJoinOperator
+ // / \ / \
+ // SeriesScanOperator SeriesScanOperator SeriesScanOperator SeriesScanOperator
+ // ----------------------------------------------------------------------------------------------
+ public TopKOperator emptyTopKOperatorTest(
+ Ordering timeOrdering, Ordering deviceOrdering, int limitValue) {
+ ExecutorService instanceNotificationExecutor =
+ IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
+ try {
+ // Construct operator tree
+ QueryId queryId = new QueryId("stub_query");
+
+ FragmentInstanceId instanceId =
+ new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance");
+ FragmentInstanceStateMachine stateMachine =
+ new FragmentInstanceStateMachine(instanceId, instanceNotificationExecutor);
+ FragmentInstanceContext fragmentInstanceContext =
+ createFragmentInstanceContext(instanceId, stateMachine);
+ DriverContext driverContext = new DriverContext(fragmentInstanceContext, 0);
+ PlanNodeId planNodeId1 = new PlanNodeId("1");
+ driverContext.addOperatorContext(1, planNodeId1, SeriesScanOperator.class.getSimpleName());
+ PlanNodeId planNodeId2 = new PlanNodeId("2");
+ driverContext.addOperatorContext(2, planNodeId2, SeriesScanOperator.class.getSimpleName());
+ PlanNodeId planNodeId3 = new PlanNodeId("3");
+ driverContext.addOperatorContext(3, planNodeId3, SeriesScanOperator.class.getSimpleName());
+ PlanNodeId planNodeId4 = new PlanNodeId("4");
+ driverContext.addOperatorContext(4, planNodeId4, SeriesScanOperator.class.getSimpleName());
+ PlanNodeId planNodeId5 = new PlanNodeId("5");
+ driverContext.addOperatorContext(5, planNodeId5, SeriesScanOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 6, new PlanNodeId("6"), SingleDeviceViewOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 7, new PlanNodeId("7"), RowBasedTimeJoinOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 8, new PlanNodeId("8"), SingleDeviceViewOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 9, new PlanNodeId("9"), RowBasedTimeJoinOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 10, new PlanNodeId("10"), SingleDeviceViewOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 11, new PlanNodeId("11"), TopKOperator.class.getSimpleName());
+
+ MeasurementPath measurementPath1 =
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
+ SeriesScanOperator seriesScanOperator1 =
+ new SeriesScanOperator(
+ driverContext.getOperatorContexts().get(0),
+ planNodeId1,
+ measurementPath1,
+ timeOrdering,
+ SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath1));
+ seriesScanOperator1.initQueryDataSource(
+ new QueryDataSource(Collections.emptyList(), Collections.emptyList()));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+
+ MeasurementPath measurementPath2 =
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
+ SeriesScanOperator seriesScanOperator2 =
+ new SeriesScanOperator(
+ driverContext.getOperatorContexts().get(1),
+ planNodeId2,
+ measurementPath2,
+ timeOrdering,
+ SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath2));
+ seriesScanOperator2.initQueryDataSource(
+ new QueryDataSource(Collections.emptyList(), Collections.emptyList()));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+
+ MeasurementPath measurementPath3 =
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
+ SeriesScanOperator seriesScanOperator3 =
+ new SeriesScanOperator(
+ driverContext.getOperatorContexts().get(2),
+ planNodeId3,
+ measurementPath3,
+ timeOrdering,
+ SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath3));
+ seriesScanOperator3.initQueryDataSource(
+ new QueryDataSource(Collections.emptyList(), Collections.emptyList()));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+
+ MeasurementPath measurementPath4 =
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
+ SeriesScanOperator seriesScanOperator4 =
+ new SeriesScanOperator(
+ driverContext.getOperatorContexts().get(3),
+ planNodeId4,
+ measurementPath4,
+ timeOrdering,
+ SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath4));
+ seriesScanOperator4.initQueryDataSource(
+ new QueryDataSource(Collections.emptyList(), Collections.emptyList()));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+
+ MeasurementPath measurementPath5 =
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
+ SeriesScanOperator seriesScanOperator5 =
+ new SeriesScanOperator(
+ driverContext.getOperatorContexts().get(4),
+ planNodeId5,
+ measurementPath5,
+ timeOrdering,
+ SeriesScanOptions.getDefaultSeriesScanOptions(measurementPath5));
+ seriesScanOperator5.initQueryDataSource(
+ new QueryDataSource(Collections.emptyList(), Collections.emptyList()));
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+
+ List<TSDataType> tsDataTypes =
+ new LinkedList<>(
+ Arrays.asList(
+ TSDataType.TEXT,
+ TSDataType.INT32,
+ TSDataType.INT32,
+ TSDataType.INT32,
+ TSDataType.INT32,
+ TSDataType.INT32));
+ SingleDeviceViewOperator singleDeviceViewOperator1 =
+ new SingleDeviceViewOperator(
+ driverContext.getOperatorContexts().get(5),
+ DEVICE0,
+ seriesScanOperator1,
+ Collections.singletonList(1),
+ tsDataTypes);
+
+ RowBasedTimeJoinOperator timeJoinOperator1 =
+ new RowBasedTimeJoinOperator(
+ driverContext.getOperatorContexts().get(6),
+ Arrays.asList(seriesScanOperator2, seriesScanOperator3),
+ timeOrdering,
+ Arrays.asList(TSDataType.INT32, TSDataType.INT32),
+ Arrays.asList(
+ new SingleColumnMerger(
+ new InputLocation(0, 0),
+ timeOrdering == Ordering.ASC
+ ? new AscTimeComparator()
+ : new DescTimeComparator()),
+ new SingleColumnMerger(
+ new InputLocation(1, 0),
+ timeOrdering == Ordering.ASC
+ ? new AscTimeComparator()
+ : new DescTimeComparator())),
+ timeOrdering == Ordering.ASC ? new AscTimeComparator() : new DescTimeComparator());
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+
+ SingleDeviceViewOperator singleDeviceViewOperator2 =
+ new SingleDeviceViewOperator(
+ driverContext.getOperatorContexts().get(7),
+ DEVICE1,
+ timeJoinOperator1,
+ Arrays.asList(2, 3),
+ tsDataTypes);
+
+ RowBasedTimeJoinOperator timeJoinOperator2 =
+ new RowBasedTimeJoinOperator(
+ driverContext.getOperatorContexts().get(8),
+ Arrays.asList(seriesScanOperator4, seriesScanOperator5),
+ timeOrdering,
+ Arrays.asList(TSDataType.INT32, TSDataType.INT32),
+ Arrays.asList(
+ new SingleColumnMerger(
+ new InputLocation(0, 0),
+ timeOrdering == Ordering.ASC
+ ? new AscTimeComparator()
+ : new DescTimeComparator()),
+ new SingleColumnMerger(
+ new InputLocation(1, 0),
+ timeOrdering == Ordering.ASC
+ ? new AscTimeComparator()
+ : new DescTimeComparator())),
+ timeOrdering == Ordering.ASC ? new AscTimeComparator() : new DescTimeComparator());
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+
+ SingleDeviceViewOperator singleDeviceViewOperator3 =
+ new SingleDeviceViewOperator(
+ driverContext.getOperatorContexts().get(9),
+ DEVICE2,
+ timeJoinOperator2,
+ Arrays.asList(4, 5),
+ tsDataTypes);
+
+ TopKOperator topKOperator =
+ new TopKOperator(
+ driverContext.getOperatorContexts().get(10),
+ Arrays.asList(
+ singleDeviceViewOperator1, singleDeviceViewOperator2, singleDeviceViewOperator3),
+ tsDataTypes,
+ MergeSortComparator.getComparator(
+ Arrays.asList(
+ new SortItem(OrderByKey.TIME, timeOrdering),
+ new SortItem(OrderByKey.DEVICE, deviceOrdering)),
+ Arrays.asList(-1, 0),
+ Arrays.asList(TSDataType.INT64, TSDataType.TEXT)),
+ limitValue,
+ true);
+ OperatorContext.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ return topKOperator;
+ } catch (IllegalPathException e) {
+ e.printStackTrace();
+ fail();
+ return null;
+ }
+ }
+
+ @Test
+ public void testEmptyTopKOperator() throws Exception {
+ TopKOperator topKOperator = emptyTopKOperatorTest(Ordering.ASC, Ordering.ASC, limitValue);
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
+ if (tsBlock == null || tsBlock.isEmpty()) {
+ continue;
+ }
+ fail();
+ }
+ }
}