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:32 UTC
(iotdb) branch beyyes/fix_dop created (now 55daa6f6a95)
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a change to branch beyyes/fix_dop
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 55daa6f6a95 fix npe in TopKOperator
This branch includes the following new commits:
new 55daa6f6a95 fix npe in TopKOperator
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: fix npe in TopKOperator
Posted by ca...@apache.org.
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();
+ }
+ }
}