You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2023/10/17 01:55:48 UTC
[iotdb] branch master updated: [IOTDB-6175] Add TopKNode for `order by time|expression limit N align by device` query and push down LIMIT value.
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 37a5212f23a [IOTDB-6175] Add TopKNode for `order by time|expression limit N align by device` query and push down LIMIT value.
37a5212f23a is described below
commit 37a5212f23a8cfa96094f1452c6c5ee531007850
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Tue Oct 17 09:55:43 2023 +0800
[IOTDB-6175] Add TopKNode for `order by time|expression limit N align by device` query and push down LIMIT value.
---
.../IoTDBOrderByLimitOffsetAlignByDeviceIT.java | 102 ++
.../IoTDBOrderByWithAlignByDeviceIT.java | 290 +++++-
.../db/it/withoutNull/IoTDBWithoutAllNullIT.java | 4 +-
.../execution/operator/process/TopKOperator.java | 323 +++++++
.../db/queryengine/plan/analyze/Analysis.java | 11 +
.../plan/planner/LocalExecutionPlanner.java | 4 +-
.../plan/planner/LogicalPlanBuilder.java | 209 +++-
.../plan/planner/LogicalPlanVisitor.java | 31 +-
.../plan/planner/OperatorTreeGenerator.java | 40 +
.../planner/distribution/DistributionPlanner.java | 16 +-
.../planner/distribution/ExchangeNodeAdder.java | 112 ++-
.../plan/planner/distribution/SourceRewriter.java | 34 +-
.../plan/planner/plan/node/PlanGraphPrinter.java | 10 +
.../plan/planner/plan/node/PlanNodeType.java | 6 +-
.../plan/planner/plan/node/PlanVisitor.java | 5 +
.../plan/planner/plan/node/process/TopKNode.java | 166 ++++
.../planner/plan/node/sink/ShuffleSinkNode.java | 12 +-
.../db/utils/datastructure/MergeSortHeap.java | 4 +
.../iotdb/db/utils/datastructure/MergeSortKey.java | 3 +-
.../execution/operator/MergeSortOperatorTest.java | 163 ++--
...SortOperatorTest.java => TopKOperatorTest.java} | 1009 +++++---------------
.../AlignByDeviceOrderByLimitOffsetTest.java | 278 ++++++
.../db/utils/datastructure/MergeSortHeapTest.java | 116 +++
.../iotdb/tsfile/read/common/block/TsBlock.java | 46 +
24 files changed, 1979 insertions(+), 1015 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByLimitOffsetAlignByDeviceIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByLimitOffsetAlignByDeviceIT.java
new file mode 100644
index 00000000000..6aac79065e2
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByLimitOffsetAlignByDeviceIT.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.it.alignbydevice;
+
+import org.apache.iotdb.it.env.EnvFactory;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.Statement;
+
+import static org.apache.iotdb.db.it.alignbydevice.IoTDBOrderByWithAlignByDeviceIT.insertData;
+import static org.apache.iotdb.db.it.alignbydevice.IoTDBOrderByWithAlignByDeviceIT.insertData2;
+import static org.apache.iotdb.db.it.utils.TestUtils.resultSetEqualTest;
+
+public class IoTDBOrderByLimitOffsetAlignByDeviceIT {
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().initClusterEnvironment();
+ insertData();
+ insertData2();
+ insertData3();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @Test
+ public void orderByCanNotPushLimitTest() {
+ // 1. value filter, can not push down LIMIT
+ String[] expectedHeader = new String[] {"Time,Device,s1"};
+ String[] retArray = new String[] {"3,root.db.d1,111,"};
+ resultSetEqualTest(
+ "SELECT * FROM root.db.** WHERE s1>40 ORDER BY TIME LIMIT 1 ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+
+ // 2. order by expression, can not push down LIMIT
+ retArray = new String[] {"3,root.db.d3,333,"};
+ resultSetEqualTest(
+ "SELECT * FROM root.db.** ORDER BY s1 DESC LIMIT 1 ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+
+ // 3. time filter, can push down LIMIT
+ retArray = new String[] {"2,root.db.d3,33,", "2,root.db.d2,22,"};
+ resultSetEqualTest(
+ "SELECT * FROM root.db.** WHERE time>1 and time<3 ORDER BY device DESC LIMIT 2 ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+
+ // 4. both exist OFFSET and LIMIT, can push down LIMIT as OFFSET+LIMIT
+ retArray = new String[] {"3,root.db.d2,222,"};
+ resultSetEqualTest(
+ "SELECT * FROM root.db.** ORDER BY time DESC OFFSET 1 LIMIT 1 ALIGN BY DEVICE;",
+ expectedHeader,
+ retArray);
+ }
+
+ private static final String[] SQL_LIST =
+ new String[] {
+ "CREATE DATABASE root.db;",
+ "CREATE TIMESERIES root.db.d1.s1 WITH DATATYPE=INT32, ENCODING=RLE;",
+ "CREATE TIMESERIES root.db.d2.s1 WITH DATATYPE=INT32, ENCODING=RLE;",
+ "CREATE TIMESERIES root.db.d3.s1 WITH DATATYPE=INT32, ENCODING=RLE;",
+ "INSERT INTO root.db.d1(timestamp,s1) VALUES(1, 1), (2, 11), (3, 111);",
+ "INSERT INTO root.db.d2(timestamp,s1) VALUES(1, 2), (2, 22), (3, 222);",
+ "INSERT INTO root.db.d3(timestamp,s1) VALUES(1, 3), (2, 33), (3, 333);",
+ };
+
+ protected static void insertData3() {
+ try (Connection iotDBConnection = EnvFactory.getEnv().getConnection();
+ Statement statement = iotDBConnection.createStatement()) {
+ for (String sql : SQL_LIST) {
+ statement.execute(sql);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByWithAlignByDeviceIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByWithAlignByDeviceIT.java
index 988b4954320..610c831170e 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByWithAlignByDeviceIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByWithAlignByDeviceIT.java
@@ -47,7 +47,7 @@ import static org.junit.Assert.fail;
@RunWith(IoTDBTestRunner.class)
@Category({LocalStandaloneIT.class, ClusterIT.class})
public class IoTDBOrderByWithAlignByDeviceIT {
- private static final String[] places =
+ protected static final String[] places =
new String[] {
"root.weather.London",
"root.weather.Edinburgh",
@@ -60,12 +60,12 @@ public class IoTDBOrderByWithAlignByDeviceIT {
"root.weather.Manchester",
"root.weather.Oxford"
};
- private static final long startPrecipitation = 200;
- private static final double startTemperature = 20.0;
- private static final long startTime = 1668960000000L;
- private static final int numOfPointsInDevice = 20;
- private static final long timeGap = 100L;
- private static final Map<String, Long> deviceToStartTimestamp = new HashMap<>();
+ protected static final long startPrecipitation = 200;
+ protected static final double startTemperature = 20.0;
+ protected static final long startTime = 1668960000000L;
+ protected static final int numOfPointsInDevice = 20;
+ protected static final long timeGap = 100L;
+ protected static final Map<String, Long> deviceToStartTimestamp = new HashMap<>();
public static final Map<String, double[]> deviceToMaxTemperature = new HashMap<>();
public static final Map<String, double[]> deviceToAvgTemperature = new HashMap<>();
public static final Map<String, long[]> deviceToMaxPrecipitation = new HashMap<>();
@@ -198,7 +198,8 @@ public class IoTDBOrderByWithAlignByDeviceIT {
}
}
- private void checkHeader(ResultSetMetaData resultSetMetaData, String title) throws SQLException {
+ protected static void checkHeader(ResultSetMetaData resultSetMetaData, String title)
+ throws SQLException {
String[] headers = title.split(",");
for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
assertEquals(headers[i - 1], resultSetMetaData.getColumnName(i));
@@ -363,7 +364,137 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByTimeTest1() {
- String sql = "SELECT * FROM root.weather.** ORDER BY TIME ALIGN BY DEVICE";
+ orderByTimeTest1(
+ "SELECT * FROM root.weather.** ORDER BY TIME ALIGN BY DEVICE",
+ numOfPointsInDevice * places.length);
+
+ orderByTimeTest1("SELECT * FROM root.weather.** ORDER BY TIME LIMIT 100 ALIGN BY DEVICE", 100);
+ }
+
+ @Test
+ public void orderByTimeTest2() {
+ orderByTimeTest2(
+ "SELECT * FROM root.weather.** ORDER BY TIME ASC ALIGN BY DEVICE",
+ numOfPointsInDevice * places.length);
+
+ orderByTimeTest2(
+ "SELECT * FROM root.weather.** ORDER BY TIME ASC LIMIT 100 ALIGN BY DEVICE", 100);
+ }
+
+ @Test
+ public void orderByTimeTest3() {
+ orderByTimeTest3(
+ "SELECT * FROM root.weather.** ORDER BY TIME DESC ALIGN BY DEVICE",
+ numOfPointsInDevice * places.length);
+
+ orderByTimeTest3(
+ "SELECT * FROM root.weather.** ORDER BY TIME DESC LIMIT 100 ALIGN BY DEVICE", 100);
+ }
+
+ @Test
+ public void orderByTimeExpressionTest1() {
+ orderByTimeExpressionTest1(
+ "SELECT * FROM root.weather.** ORDER BY TIME DESC, precipitation DESC ALIGN BY DEVICE",
+ numOfPointsInDevice * places.length);
+
+ orderByTimeExpressionTest1(
+ "SELECT * FROM root.weather.** ORDER BY TIME DESC, precipitation DESC LIMIT 100 ALIGN BY DEVICE",
+ 100);
+ }
+
+ @Test
+ public void orderExpressionTest1() {
+ orderByExpressionTest1(
+ "SELECT * FROM root.weather.** ORDER BY precipitation DESC, time DESC ALIGN BY DEVICE",
+ numOfPointsInDevice * places.length);
+
+ orderByExpressionTest1(
+ "SELECT * FROM root.weather.** ORDER BY precipitation DESC, time DESC LIMIT 100 ALIGN BY DEVICE",
+ 100);
+ }
+
+ @Test
+ public void orderByDeviceTimeTest1() {
+ orderByDeviceTimeTest1(
+ "SELECT * FROM root.weather.** ORDER BY DEVICE ASC,TIME DESC ALIGN BY DEVICE", 10);
+
+ orderByDeviceTimeTest1(
+ "SELECT * FROM root.weather.** ORDER BY DEVICE ASC,TIME DESC LIMIT 100 ALIGN BY DEVICE", 5);
+ }
+
+ @Test
+ public void orderByDeviceTimeTest2() {
+ orderByDeviceTimeTest2(
+ "SELECT * FROM root.weather.** ORDER BY DEVICE ASC,TIME ASC ALIGN BY DEVICE", 10);
+
+ orderByDeviceTimeTest2(
+ "SELECT * FROM root.weather.** ORDER BY DEVICE ASC,TIME ASC LIMIT 100 ALIGN BY DEVICE", 5);
+ }
+
+ @Test
+ public void orderByDeviceTimeTest3() {
+ orderByDeviceTimeTest3(
+ "SELECT * FROM root.weather.** ORDER BY DEVICE DESC,TIME DESC ALIGN BY DEVICE", 10);
+
+ orderByDeviceTimeTest3(
+ "SELECT * FROM root.weather.** ORDER BY DEVICE DESC,TIME DESC LIMIT 100 ALIGN BY DEVICE",
+ 5);
+ }
+
+ @Test
+ public void orderByDeviceTimeTest4() {
+ orderByDeviceTimeTest4(
+ "SELECT * FROM root.weather.** ORDER BY DEVICE DESC,TIME ASC ALIGN BY DEVICE", 10);
+
+ orderByDeviceTimeTest4(
+ "SELECT * FROM root.weather.** ORDER BY DEVICE DESC,TIME ASC LIMIT 100 ALIGN BY DEVICE", 5);
+ }
+
+ @Test
+ public void orderByTimeDeviceTest1() {
+ orderByTimeDeviceTest1(
+ "SELECT * FROM root.weather.** ORDER BY TIME ASC,DEVICE DESC ALIGN BY DEVICE",
+ numOfPointsInDevice * places.length);
+
+ orderByTimeDeviceTest1(
+ "SELECT * FROM root.weather.** ORDER BY TIME ASC,DEVICE DESC LIMIT 100 ALIGN BY DEVICE",
+ 100);
+ }
+
+ @Test
+ public void orderByTimeDeviceTest2() {
+ orderByTimeDeviceTest2(
+ "SELECT * FROM root.weather.** ORDER BY TIME ASC,DEVICE ASC ALIGN BY DEVICE",
+ numOfPointsInDevice * places.length);
+
+ orderByTimeDeviceTest2(
+ "SELECT * FROM root.weather.** ORDER BY TIME ASC,DEVICE ASC LIMIT 100 ALIGN BY DEVICE",
+ 100);
+ }
+
+ @Test
+ public void orderByTimeDeviceTest3() {
+ orderByTimeDeviceTest3(
+ "SELECT * FROM root.weather.** ORDER BY TIME DESC,DEVICE DESC ALIGN BY DEVICE",
+ numOfPointsInDevice * places.length);
+
+ orderByTimeDeviceTest3(
+ "SELECT * FROM root.weather.** ORDER BY TIME DESC,DEVICE DESC LIMIT 100 ALIGN BY DEVICE",
+ 100);
+ }
+
+ @Test
+ public void orderByTimeDeviceTest4() {
+ orderByTimeDeviceTest4(
+ "SELECT * FROM root.weather.** ORDER BY TIME DESC,DEVICE ASC ALIGN BY DEVICE",
+ numOfPointsInDevice * places.length);
+
+ orderByTimeDeviceTest4(
+ "SELECT * FROM root.weather.** ORDER BY TIME DESC,DEVICE ASC LIMIT 100 ALIGN BY DEVICE",
+ 100);
+ }
+
+ public static void orderByTimeTest1(String sql, int count) {
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -391,7 +522,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
< 0.00001);
total++;
}
- assertEquals(numOfPointsInDevice * places.length, total);
+ assertEquals(count, total);
}
} catch (Exception e) {
e.printStackTrace();
@@ -399,9 +530,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
}
}
- @Test
- public void orderByTimeTest2() {
- String sql = "SELECT * FROM root.weather.** ORDER BY TIME ASC ALIGN BY DEVICE";
+ public static void orderByTimeTest2(String sql, int count) {
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -429,7 +558,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
< 0.00001);
total++;
}
- assertEquals(numOfPointsInDevice * places.length, total);
+ assertEquals(count, total);
}
} catch (Exception e) {
e.printStackTrace();
@@ -437,9 +566,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
}
}
- @Test
- public void orderByTimeTest3() {
- String sql = "SELECT * FROM root.weather.** ORDER BY TIME DESC ALIGN BY DEVICE";
+ public static void orderByTimeTest3(String sql, int count) {
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -467,7 +594,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
< 0.00001);
total++;
}
- assertEquals(numOfPointsInDevice * places.length, total);
+ assertEquals(count, total);
}
} catch (Exception e) {
e.printStackTrace();
@@ -475,11 +602,85 @@ public class IoTDBOrderByWithAlignByDeviceIT {
}
}
- // ORDER BY DEVICE,TIME
+ public static void orderByTimeExpressionTest1(String sql, int count) {
+ int total = 0;
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
- @Test
- public void orderByDeviceTimeTest1() {
- String sql = "SELECT * FROM root.weather.** ORDER BY DEVICE ASC,TIME DESC ALIGN BY DEVICE";
+ try (ResultSet resultSet = statement.executeQuery(sql)) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ checkHeader(resultSetMetaData, "Time,Device,precipitation,temperature");
+ long lastTimeStamp = Long.MAX_VALUE;
+ long lastPrecipitation = Long.MAX_VALUE;
+ String lastDevice = "";
+ while (resultSet.next()) {
+ long actualTimeStamp = resultSet.getLong(1);
+ assertTrue(actualTimeStamp <= lastTimeStamp);
+ String actualDevice = resultSet.getString(2);
+
+ long actualPrecipitation = resultSet.getLong(3);
+ double actualTemperature = resultSet.getDouble(4);
+ assertEquals(
+ startPrecipitation + actualDevice.hashCode() + actualTimeStamp, actualPrecipitation);
+ assertTrue(
+ startTemperature + actualDevice.hashCode() + actualTimeStamp - actualTemperature
+ < 0.00001);
+ if (actualDevice.equals(lastDevice) && actualTimeStamp == lastTimeStamp) {
+ assertTrue(actualPrecipitation <= lastPrecipitation);
+ lastPrecipitation = actualPrecipitation;
+ }
+ lastDevice = actualDevice;
+ lastTimeStamp = actualTimeStamp;
+ total++;
+ }
+ assertEquals(count, total);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ public static void orderByExpressionTest1(String sql, int count) {
+ int total = 0;
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+
+ try (ResultSet resultSet = statement.executeQuery(sql)) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ checkHeader(resultSetMetaData, "Time,Device,precipitation,temperature");
+ long lastTimeStamp = Long.MAX_VALUE;
+ long lastPrecipitation = Long.MAX_VALUE;
+ String lastDevice = "";
+ while (resultSet.next()) {
+ long actualTimeStamp = resultSet.getLong(1);
+ String actualDevice = resultSet.getString(2);
+ long actualPrecipitation = resultSet.getLong(3);
+ double actualTemperature = resultSet.getDouble(4);
+ assertEquals(
+ startPrecipitation + actualDevice.hashCode() + actualTimeStamp, actualPrecipitation);
+ assertTrue(
+ startTemperature + actualDevice.hashCode() + actualTimeStamp - actualTemperature
+ < 0.00001);
+ assertTrue(actualPrecipitation <= lastPrecipitation);
+ if (actualPrecipitation == lastPrecipitation) {
+ assertTrue(actualTimeStamp <= lastTimeStamp);
+ }
+ lastPrecipitation = actualPrecipitation;
+ lastDevice = actualDevice;
+ lastTimeStamp = actualTimeStamp;
+ total++;
+ }
+ assertEquals(count, total);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // ORDER BY DEVICE,TIME
+ public static void orderByDeviceTimeTest1(String sql, int count) {
Object[] expectedDevice = Arrays.stream(places.clone()).sorted().toArray();
int index = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -512,7 +713,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
cnt = 0;
}
}
- assertEquals(10, index);
+ assertEquals(count, index);
}
} catch (Exception e) {
e.printStackTrace();
@@ -520,9 +721,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
}
}
- @Test
- public void orderByDeviceTimeTest2() {
- String sql = "SELECT * FROM root.weather.** ORDER BY DEVICE ASC,TIME ASC ALIGN BY DEVICE";
+ public static void orderByDeviceTimeTest2(String sql, int count) {
Object[] expectedDevice = Arrays.stream(places.clone()).sorted().toArray();
int index = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -553,7 +752,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
cnt = 0;
}
}
- assertEquals(10, index);
+ assertEquals(count, index);
}
} catch (Exception e) {
e.printStackTrace();
@@ -561,9 +760,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
}
}
- @Test
- public void orderByDeviceTimeTest3() {
- String sql = "SELECT * FROM root.weather.** ORDER BY DEVICE DESC,TIME DESC ALIGN BY DEVICE";
+ public static void orderByDeviceTimeTest3(String sql, int count) {
Object[] expectedDevice =
Arrays.stream(places.clone()).sorted(Comparator.reverseOrder()).toArray();
int index = 0;
@@ -597,7 +794,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
cnt = 0;
}
}
- assertEquals(10, index);
+ assertEquals(count, index);
}
} catch (Exception e) {
e.printStackTrace();
@@ -605,9 +802,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
}
}
- @Test
- public void orderByDeviceTimeTest4() {
- String sql = "SELECT * FROM root.weather.** ORDER BY DEVICE DESC,TIME ASC ALIGN BY DEVICE";
+ public static void orderByDeviceTimeTest4(String sql, int count) {
Object[] expectedDevice =
Arrays.stream(places.clone()).sorted(Comparator.reverseOrder()).toArray();
int index = 0;
@@ -639,7 +834,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
cnt = 0;
}
}
- assertEquals(10, index);
+ assertEquals(count, index);
}
} catch (Exception e) {
e.printStackTrace();
@@ -648,10 +843,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
}
// ORDER BY TIME,DEVICE
-
- @Test
- public void orderByTimeDeviceTest1() {
- String sql = "SELECT * FROM root.weather.** ORDER BY TIME ASC,DEVICE DESC ALIGN BY DEVICE";
+ public static void orderByTimeDeviceTest1(String sql, int count) {
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -679,7 +871,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
< 0.00001);
total++;
}
- assertEquals(numOfPointsInDevice * places.length, total);
+ assertEquals(count, total);
}
} catch (Exception e) {
e.printStackTrace();
@@ -687,9 +879,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
}
}
- @Test
- public void orderByTimeDeviceTest2() {
- String sql = "SELECT * FROM root.weather.** ORDER BY TIME ASC,DEVICE ASC ALIGN BY DEVICE";
+ public static void orderByTimeDeviceTest2(String sql, int count) {
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -717,7 +907,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
< 0.00001);
total++;
}
- assertEquals(numOfPointsInDevice * places.length, total);
+ assertEquals(count, total);
}
} catch (Exception e) {
e.printStackTrace();
@@ -725,9 +915,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
}
}
- @Test
- public void orderByTimeDeviceTest3() {
- String sql = "SELECT * FROM root.weather.** ORDER BY TIME DESC,DEVICE DESC ALIGN BY DEVICE";
+ public static void orderByTimeDeviceTest3(String sql, int count) {
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -755,7 +943,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
< 0.00001);
total++;
}
- assertEquals(numOfPointsInDevice * places.length, total);
+ assertEquals(count, total);
}
} catch (Exception e) {
e.printStackTrace();
@@ -763,9 +951,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
}
}
- @Test
- public void orderByTimeDeviceTest4() {
- String sql = "SELECT * FROM root.weather.** ORDER BY TIME DESC,DEVICE ASC ALIGN BY DEVICE";
+ public static void orderByTimeDeviceTest4(String sql, int count) {
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -793,7 +979,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
< 0.00001);
total++;
}
- assertEquals(numOfPointsInDevice * places.length, total);
+ assertEquals(count, total);
}
} catch (Exception e) {
e.printStackTrace();
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/withoutNull/IoTDBWithoutAllNullIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/withoutNull/IoTDBWithoutAllNullIT.java
index 950272766bd..7a76bc74575 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/withoutNull/IoTDBWithoutAllNullIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/withoutNull/IoTDBWithoutAllNullIT.java
@@ -41,7 +41,6 @@ import static org.junit.Assert.fail;
@RunWith(IoTDBTestRunner.class)
@Category({LocalStandaloneIT.class, ClusterIT.class})
public class IoTDBWithoutAllNullIT {
-
private static final String[] dataSet =
new String[] {
"CREATE DATABASE root.testWithoutAllNull",
@@ -221,7 +220,8 @@ public class IoTDBWithoutAllNullIT {
int cnt;
try (ResultSet resultSet =
statement.executeQuery(
- "select last_value(*) from root.testWithoutAllNull.d1 GROUP BY([1, 21), 5ms) having last_value(s1) is not null || last_value(s2) is not null || last_value(s3) is not null "
+ "select last_value(*) from root.testWithoutAllNull.d1 GROUP BY([1, 21), 5ms) "
+ + "having last_value(s1) is not null || last_value(s2) is not null || last_value(s3) is not null "
+ "ORDER BY TIME DESC LIMIT 1 OFFSET 1 ALIGN BY DEVICE")) {
cnt = 0;
while (resultSet.next()) {
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
new file mode 100644
index 00000000000..40f271e2392
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java
@@ -0,0 +1,323 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.execution.operator.process;
+
+import org.apache.iotdb.commons.exception.runtime.UnSupportedDataTypeException;
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import org.apache.iotdb.db.utils.datastructure.MergeSortHeap;
+import org.apache.iotdb.db.utils.datastructure.MergeSortKey;
+import org.apache.iotdb.db.utils.datastructure.SortKey;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.BinaryColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.BooleanColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.Column;
+import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.DoubleColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.FloatColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.IntColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.LongColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumn;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Optional;
+
+public class TopKOperator implements ProcessOperator {
+ private final OperatorContext operatorContext;
+
+ private final List<Operator> deviceOperators;
+ private int deviceIndex;
+
+ private final List<TSDataType> dataTypes;
+ private final TsBlockBuilder tsBlockBuilder;
+
+ // max heap, revered of MergeSortHeap in MergeSortOperator
+ private final MergeSortHeap mergeSortHeap;
+ private final Comparator<SortKey> comparator;
+ // value in LIMIT
+ private final int topValue;
+
+ // final query result of TopKOperator and the returned size of topKResult
+ private MergeSortKey[] topKResult;
+ private int resultReturnSize = 0;
+
+ // represent the key of mergeSortHeap, which is `TsBlock and its index`
+ private TsBlock tmpResultTsBlock;
+ private int tmpResultTsBlockIdx;
+
+ // if order by time, timeOrderPriority is highest, and no order by expression
+ // the data of every childOperator is in order
+ private final boolean childrenDataInOrder;
+
+ public TopKOperator(
+ OperatorContext operatorContext,
+ List<Operator> deviceOperators,
+ List<TSDataType> dataTypes,
+ Comparator<SortKey> comparator,
+ int topValue,
+ boolean childrenDataInOrder) {
+ this.operatorContext = operatorContext;
+ this.deviceOperators = deviceOperators;
+ this.dataTypes = dataTypes;
+ this.mergeSortHeap = new MergeSortHeap(topValue, comparator.reversed());
+ this.comparator = comparator;
+ this.tsBlockBuilder = new TsBlockBuilder(topValue, dataTypes);
+ this.topValue = topValue;
+ this.childrenDataInOrder = childrenDataInOrder;
+ initResultTsBlock();
+ }
+
+ @Override
+ public OperatorContext getOperatorContext() {
+ return operatorContext;
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ if (deviceIndex >= deviceOperators.size()) {
+ return NOT_BLOCKED;
+ }
+ ListenableFuture<?> blocked = getCurDeviceOperator().isBlocked();
+ if (!blocked.isDone()) {
+ return blocked;
+ }
+ return NOT_BLOCKED;
+ }
+
+ @Override
+ public boolean isFinished() throws Exception {
+ return !this.hasNextWithTimer();
+ }
+
+ @Override
+ public boolean hasNext() throws Exception {
+ return !(deviceIndex >= deviceOperators.size() && resultReturnSize == topKResult.length);
+ }
+
+ @Override
+ public TsBlock next() throws Exception {
+ if (deviceIndex >= deviceOperators.size() && resultReturnSize < topKResult.length) {
+ return getResultFromCachedTopKResult();
+ }
+
+ if (!getCurDeviceOperator().hasNextWithTimer()) {
+ closeCurDeviceOperator();
+ if (deviceIndex == deviceOperators.size()) {
+ return getResultFromMaxHeap(mergeSortHeap);
+ }
+ return null;
+ }
+
+ TsBlock currentTsBlock = getCurDeviceOperator().nextWithTimer();
+ if (currentTsBlock == null) {
+ return null;
+ }
+
+ boolean skipCurrentBatch = false;
+ for (int idx = 0; idx < currentTsBlock.getPositionCount(); idx++) {
+ if (mergeSortHeap.getHeapSize() < topValue) {
+ updateTsBlockValue(currentTsBlock, idx, -1);
+ } else {
+ if (comparator.compare(new MergeSortKey(currentTsBlock, idx), mergeSortHeap.peek()) < 0) {
+ MergeSortKey peek = mergeSortHeap.poll();
+ updateTsBlockValue(currentTsBlock, idx, peek.rowIndex);
+ } else if (childrenDataInOrder) {
+ skipCurrentBatch = true;
+ break;
+ }
+ }
+ }
+
+ // if current childIdx TsBlock has no value to put into heap
+ // the remaining data will also have no value to put int heap
+ if (skipCurrentBatch) {
+ closeCurDeviceOperator();
+ if (deviceIndex == deviceOperators.size()) {
+ return getResultFromMaxHeap(mergeSortHeap);
+ }
+ return null;
+ }
+
+ return null;
+ }
+
+ @Override
+ public long calculateMaxPeekMemory() {
+ // traverse each child serial,
+ // so no need to accumulate the returnSize and retainedSize of each child
+ long maxPeekMemory = calculateMaxReturnSize();
+ for (Operator operator : deviceOperators) {
+ maxPeekMemory = Math.max(maxPeekMemory, operator.calculateMaxPeekMemory());
+ }
+ return Math.max(maxPeekMemory, topValue * getMemoryUsageOfOneMergeSortKey() * 2);
+ }
+
+ @Override
+ public long calculateMaxReturnSize() {
+ return TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes();
+ }
+
+ @Override
+ public long calculateRetainedSizeAfterCallingNext() {
+ return (topValue - resultReturnSize) * getMemoryUsageOfOneMergeSortKey();
+ }
+
+ private void initResultTsBlock() {
+ int positionCount = topValue;
+ Column[] columns = new Column[dataTypes.size()];
+ for (int i = 0; i < dataTypes.size(); i++) {
+ switch (dataTypes.get(i)) {
+ case BOOLEAN:
+ columns[i] =
+ new BooleanColumn(
+ positionCount,
+ Optional.of(new boolean[positionCount]),
+ new boolean[positionCount]);
+ break;
+ case INT32:
+ columns[i] =
+ new IntColumn(
+ positionCount, Optional.of(new boolean[positionCount]), new int[positionCount]);
+ break;
+ case INT64:
+ columns[i] =
+ new LongColumn(
+ positionCount, Optional.of(new boolean[positionCount]), new long[positionCount]);
+ break;
+ case FLOAT:
+ columns[i] =
+ new FloatColumn(
+ positionCount, Optional.of(new boolean[positionCount]), new float[positionCount]);
+ break;
+ case DOUBLE:
+ columns[i] =
+ new DoubleColumn(
+ positionCount,
+ Optional.of(new boolean[positionCount]),
+ new double[positionCount]);
+ break;
+ case TEXT:
+ columns[i] =
+ new BinaryColumn(
+ positionCount,
+ Optional.of(new boolean[positionCount]),
+ new Binary[positionCount]);
+ break;
+ default:
+ throw new UnSupportedDataTypeException("Unknown datatype: " + dataTypes.get(i));
+ }
+ }
+ this.tmpResultTsBlock =
+ new TsBlock(positionCount, new TimeColumn(positionCount, new long[positionCount]), columns);
+ }
+
+ private TsBlock getResultFromMaxHeap(MergeSortHeap mergeSortHeap) {
+ int cnt = mergeSortHeap.getHeapSize();
+ topKResult = new MergeSortKey[cnt];
+ while (!mergeSortHeap.isEmpty()) {
+ topKResult[--cnt] = mergeSortHeap.poll();
+ }
+
+ return getResultFromCachedTopKResult();
+ }
+
+ private TsBlock getResultFromCachedTopKResult() {
+ tsBlockBuilder.reset();
+ ColumnBuilder[] valueColumnBuilders = tsBlockBuilder.getValueColumnBuilders();
+ for (int i = resultReturnSize; i < topKResult.length; i++) {
+ MergeSortKey mergeSortKey = topKResult[i];
+ TsBlock targetBlock = mergeSortKey.tsBlock;
+ tsBlockBuilder
+ .getTimeColumnBuilder()
+ .writeLong(targetBlock.getTimeByIndex(mergeSortKey.rowIndex));
+ for (int j = 0; j < valueColumnBuilders.length; j++) {
+ if (targetBlock.getColumn(j).isNull(mergeSortKey.rowIndex)) {
+ valueColumnBuilders[j].appendNull();
+ continue;
+ }
+ valueColumnBuilders[j].write(targetBlock.getColumn(j), mergeSortKey.rowIndex);
+ }
+ resultReturnSize += 1;
+ tsBlockBuilder.declarePosition();
+
+ if (tsBlockBuilder.isFull()) {
+ return tsBlockBuilder.build();
+ }
+ }
+
+ return tsBlockBuilder.build();
+ }
+
+ private long getMemoryUsageOfOneMergeSortKey() {
+ long memory = 0;
+ for (TSDataType dataType : dataTypes) {
+ switch (dataType) {
+ case BOOLEAN:
+ memory += 1;
+ break;
+ case INT32:
+ case FLOAT:
+ memory += 4;
+ break;
+ case INT64:
+ case DOUBLE:
+ case VECTOR:
+ memory += 8;
+ break;
+ case TEXT:
+ memory += 16;
+ break;
+ default:
+ throw new UnSupportedDataTypeException("Unknown datatype: " + dataType);
+ }
+ }
+ return memory;
+ }
+
+ private void updateTsBlockValue(TsBlock sourceTsBlock, int sourceIndex, int peekIndex) {
+ if (peekIndex < 0) {
+ tmpResultTsBlock.update(tmpResultTsBlockIdx, sourceTsBlock, sourceIndex);
+ mergeSortHeap.push(new MergeSortKey(tmpResultTsBlock, tmpResultTsBlockIdx++));
+ return;
+ }
+
+ tmpResultTsBlock.update(peekIndex, sourceTsBlock, sourceIndex);
+ mergeSortHeap.push(new MergeSortKey(tmpResultTsBlock, peekIndex));
+ }
+
+ private Operator getCurDeviceOperator() {
+ return deviceOperators.get(deviceIndex);
+ }
+
+ private void closeCurDeviceOperator() throws Exception {
+ // close finished child
+ getCurDeviceOperator().close();
+ deviceOperators.set(deviceIndex, null);
+ // increment index, move to next child
+ deviceIndex++;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
index a3e194a8e70..aee45a16193 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
@@ -271,6 +271,9 @@ public class Analysis {
// can use statistics to skip
private boolean lastLevelUseWildcard = false;
+ // if `order by limit N align by device` query use topK optimization
+ private boolean useTopKNode = false;
+
public Analysis() {
this.finishQueryAfterAnalyze = false;
}
@@ -780,6 +783,14 @@ public class Analysis {
this.lastLevelUseWildcard = lastLevelUseWildcard;
}
+ public boolean isUseTopKNode() {
+ return useTopKNode;
+ }
+
+ public void setUseTopKNode() {
+ this.useTopKNode = true;
+ }
+
public void setDeviceList(List<PartialPath> deviceList) {
this.deviceList = deviceList;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java
index b875ee70905..d4d51195607 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java
@@ -118,7 +118,9 @@ public class LocalExecutionPlanner {
if (estimatedMemorySize > freeMemoryForOperators) {
throw new MemoryNotEnoughException(
String.format(
- "There is not enough memory to execute current fragment instance, current remaining free memory is %d, estimated memory usage for current fragment instance is %d",
+ "There is not enough memory to execute current fragment instance, "
+ + "current remaining free memory is %d, "
+ + "estimated memory usage for current fragment instance is %d",
freeMemoryForOperators, estimatedMemorySize));
} else {
freeMemoryForOperators -= estimatedMemorySize;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java
index ee5b871bf93..4894f264ed1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java
@@ -64,11 +64,13 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.GroupByTag
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.IntoNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MergeSortNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MultiChildProcessNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.OffsetNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDeviceViewNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SlidingWindowAggregationNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TimeJoinNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TransformNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryTransformNode;
@@ -128,6 +130,7 @@ import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDC
import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.DEVICE;
import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.ENDTIME;
import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionTypeAnalyzer.analyzeExpression;
+import static org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode.LIMIT_USE_TOP_K_FOR_ALIGN_BY_DEVICE;
import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT_TIME;
import static org.apache.iotdb.db.utils.constant.SqlConstant.LAST_VALUE;
import static org.apache.iotdb.db.utils.constant.SqlConstant.MAX_TIME;
@@ -184,6 +187,8 @@ public class LogicalPlanBuilder {
Set<Expression> sourceExpressions,
Ordering scanOrder,
Filter timeFilter,
+ long offset,
+ long limit,
boolean lastLevelUseWildcard) {
List<PlanNode> sourceNodeList = new ArrayList<>();
List<PartialPath> selectedPaths =
@@ -192,27 +197,37 @@ public class LogicalPlanBuilder {
.collect(Collectors.toList());
List<PartialPath> groupedPaths = MetaUtils.groupAlignedSeries(selectedPaths);
for (PartialPath path : groupedPaths) {
- if (path instanceof MeasurementPath) { // non-aligned series
+ if (path instanceof MeasurementPath) {
+ // non-aligned series
+ // TODO: push down value filter
SeriesScanNode seriesScanNode =
new SeriesScanNode(
- context.getQueryId().genPlanNodeId(), (MeasurementPath) path, scanOrder);
- seriesScanNode.setTimeFilter(timeFilter);
- // TODO: push down value filter
- seriesScanNode.setValueFilter(timeFilter);
+ context.getQueryId().genPlanNodeId(),
+ (MeasurementPath) path,
+ scanOrder,
+ timeFilter,
+ timeFilter,
+ limit,
+ offset,
+ null);
sourceNodeList.add(seriesScanNode);
- } else if (path instanceof AlignedPath) { // aligned series
+ } else if (path instanceof AlignedPath) {
+ // aligned series
+ // TODO: push down value filter
AlignedSeriesScanNode alignedSeriesScanNode =
new AlignedSeriesScanNode(
context.getQueryId().genPlanNodeId(),
(AlignedPath) path,
scanOrder,
+ timeFilter,
+ timeFilter,
+ limit,
+ offset,
+ null,
lastLevelUseWildcard);
- alignedSeriesScanNode.setTimeFilter(timeFilter);
- // TODO: push down value filter
- alignedSeriesScanNode.setValueFilter(timeFilter);
sourceNodeList.add(alignedSeriesScanNode);
} else {
- throw new IllegalArgumentException("unexpected path type");
+ throw new IllegalArgumentException("Unexpected path type");
}
}
@@ -372,6 +387,8 @@ public class LogicalPlanBuilder {
sourceExpressions,
Ordering.DESC,
analysis.getGlobalTimeFilter(),
+ 0,
+ 0,
analysis.isLastLevelUseWildcard())
.planWhereAndSourceTransform(
null, sourceTransformExpressions, false, zoneId, Ordering.DESC)
@@ -757,59 +774,143 @@ public class LogicalPlanBuilder {
OrderByParameter orderByParameter = new OrderByParameter(sortItemList);
- // order by time, device can be optimized by SingleDeviceViewNode and MergeSortNode
- if (queryStatement.isOrderByBasedOnTime() && !queryStatement.hasOrderByExpression()) {
+ long limitValue =
+ queryStatement.hasOffset()
+ ? queryStatement.getRowOffset() + queryStatement.getRowLimit()
+ : queryStatement.getRowLimit();
+
+ if (!queryStatement.isAggregationQuery()
+ && queryStatement.hasLimit()
+ && queryStatement.getOrderByComponent() != null
+ && !queryStatement.isOrderByBasedOnDevice()
+ && limitValue <= LIMIT_USE_TOP_K_FOR_ALIGN_BY_DEVICE) {
+
+ // order by time and order by expression with limit, can be optimized to TopK implementation
+ TopKNode topKNode =
+ new TopKNode(
+ context.getQueryId().genPlanNodeId(),
+ (int) limitValue,
+ orderByParameter,
+ outputColumnNames);
+
+ // if value filter exists, need add a LIMIT-NODE as the child node of TopKNode
+ long valueFilterLimit = queryStatement.hasWhere() ? limitValue : -1;
+
+ if ((queryStatement.isOrderByBasedOnTime() && !queryStatement.hasOrderByExpression())) {
+ addSingleDeviceViewNodes(
+ topKNode,
+ deviceNameToSourceNodesMap,
+ outputColumnNames,
+ deviceToMeasurementIndexesMap,
+ valueFilterLimit);
+ } else {
+ topKNode.addChild(
+ addDeviceViewNode(
+ orderByParameter,
+ outputColumnNames,
+ deviceToMeasurementIndexesMap,
+ deviceNameToSourceNodesMap,
+ valueFilterLimit));
+ }
+
+ this.root = topKNode;
+ }
+ // order by time + no limit, device can be optimized by SingleDeviceViewNode and MergeSortNode
+ else if (queryStatement.isOrderByBasedOnTime() && !queryStatement.hasOrderByExpression()) {
MergeSortNode mergeSortNode =
new MergeSortNode(
context.getQueryId().genPlanNodeId(), orderByParameter, outputColumnNames);
- for (Map.Entry<String, PlanNode> entry : deviceNameToSourceNodesMap.entrySet()) {
- String deviceName = entry.getKey();
- PlanNode subPlan = entry.getValue();
- SingleDeviceViewNode singleDeviceViewNode =
- new SingleDeviceViewNode(
- context.getQueryId().genPlanNodeId(),
- outputColumnNames,
- deviceName,
- deviceToMeasurementIndexesMap.get(deviceName));
- singleDeviceViewNode.addChild(subPlan);
- mergeSortNode.addChild(singleDeviceViewNode);
- }
+ addSingleDeviceViewNodes(
+ mergeSortNode,
+ deviceNameToSourceNodesMap,
+ outputColumnNames,
+ deviceToMeasurementIndexesMap,
+ -1);
this.root = mergeSortNode;
} else {
- DeviceViewNode deviceViewNode =
- new DeviceViewNode(
- context.getQueryId().genPlanNodeId(),
+ this.root =
+ addDeviceViewNode(
orderByParameter,
outputColumnNames,
- deviceToMeasurementIndexesMap);
-
- for (Map.Entry<String, PlanNode> entry : deviceNameToSourceNodesMap.entrySet()) {
- String deviceName = entry.getKey();
- PlanNode subPlan = entry.getValue();
- deviceViewNode.addChildDeviceNode(deviceName, subPlan);
- }
- this.root = deviceViewNode;
+ deviceToMeasurementIndexesMap,
+ deviceNameToSourceNodesMap,
+ -1);
}
context.getTypeProvider().setType(DEVICE, TSDataType.TEXT);
updateTypeProvider(deviceViewOutputExpressions);
- if (queryStatement.needPushDownSort()) {
- if (selectExpression.size() != deviceViewOutputExpressions.size()) {
- this.root =
- new TransformNode(
- context.getQueryId().genPlanNodeId(),
- root,
- selectExpression.toArray(new Expression[0]),
- queryStatement.isGroupByTime(),
- queryStatement.getSelectComponent().getZoneId(),
- queryStatement.getResultTimeOrder());
- }
+ if (queryStatement.needPushDownSort()
+ && (selectExpression.size() != deviceViewOutputExpressions.size())) {
+ this.root =
+ new TransformNode(
+ context.getQueryId().genPlanNodeId(),
+ root,
+ selectExpression.toArray(new Expression[0]),
+ queryStatement.isGroupByTime(),
+ queryStatement.getSelectComponent().getZoneId(),
+ queryStatement.getResultTimeOrder());
}
return this;
}
+ private void addSingleDeviceViewNodes(
+ MultiChildProcessNode parent,
+ Map<String, PlanNode> deviceNameToSourceNodesMap,
+ List<String> outputColumnNames,
+ Map<String, List<Integer>> deviceToMeasurementIndexesMap,
+ long valueFilterLimit) {
+ for (Map.Entry<String, PlanNode> entry : deviceNameToSourceNodesMap.entrySet()) {
+ String deviceName = entry.getKey();
+ PlanNode subPlan = entry.getValue();
+ SingleDeviceViewNode singleDeviceViewNode =
+ new SingleDeviceViewNode(
+ context.getQueryId().genPlanNodeId(),
+ outputColumnNames,
+ deviceName,
+ deviceToMeasurementIndexesMap.get(deviceName));
+
+ // put LIMIT-NODE below of SingleDeviceViewNode if exists value filter
+ if (valueFilterLimit > 0) {
+ LimitNode limitNode =
+ new LimitNode(context.getQueryId().genPlanNodeId(), subPlan, valueFilterLimit);
+ singleDeviceViewNode.addChild(limitNode);
+ } else {
+ singleDeviceViewNode.addChild(subPlan);
+ }
+
+ parent.addChild(singleDeviceViewNode);
+ }
+ }
+
+ private DeviceViewNode addDeviceViewNode(
+ OrderByParameter orderByParameter,
+ List<String> outputColumnNames,
+ Map<String, List<Integer>> deviceToMeasurementIndexesMap,
+ Map<String, PlanNode> deviceNameToSourceNodesMap,
+ long valueFilterLimit) {
+ DeviceViewNode deviceViewNode =
+ new DeviceViewNode(
+ context.getQueryId().genPlanNodeId(),
+ orderByParameter,
+ outputColumnNames,
+ deviceToMeasurementIndexesMap);
+
+ for (Map.Entry<String, PlanNode> entry : deviceNameToSourceNodesMap.entrySet()) {
+ String deviceName = entry.getKey();
+ PlanNode subPlan = entry.getValue();
+ if (valueFilterLimit > 0) {
+ LimitNode limitNode =
+ new LimitNode(context.getQueryId().genPlanNodeId(), subPlan, valueFilterLimit);
+ deviceViewNode.addChildDeviceNode(deviceName, limitNode);
+ } else {
+ deviceViewNode.addChildDeviceNode(deviceName, subPlan);
+ }
+ }
+ return deviceViewNode;
+ }
+
public LogicalPlanBuilder planGroupByLevel(
Map<Expression, Set<Expression>> groupByLevelExpressions,
GroupByTimeParameter groupByTimeParameter,
@@ -1098,6 +1199,10 @@ public class LogicalPlanBuilder {
return this;
}
+ if (this.getRoot() instanceof TopKNode) {
+ return this;
+ }
+
this.root = new LimitNode(context.getQueryId().genPlanNodeId(), this.getRoot(), rowLimit);
return this;
}
@@ -1153,10 +1258,9 @@ public class LogicalPlanBuilder {
}
ColumnHeaderConstant.selectIntoAlignByDeviceColumnHeaders.forEach(
- columnHeader -> {
- updateTypeProviderWithConstantType(
- columnHeader.getColumnName(), columnHeader.getColumnType());
- });
+ columnHeader ->
+ updateTypeProviderWithConstantType(
+ columnHeader.getColumnName(), columnHeader.getColumnType()));
this.root =
new DeviceViewIntoNode(
context.getQueryId().genPlanNodeId(), this.getRoot(), deviceViewIntoPathDescriptor);
@@ -1470,6 +1574,11 @@ public class LogicalPlanBuilder {
return this;
}
+ // TopKNode is already sorted, does not need sort
+ if (root instanceof TopKNode) {
+ return this;
+ }
+
updateTypeProvider(orderByExpressions);
OrderByParameter orderByParameter = new OrderByParameter(queryStatement.getSortItemList());
if (orderByParameter.isEmpty()) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
index bd345ebf612..d7e810b2b5f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
@@ -44,6 +44,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.Int
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.InternalCreateTimeSeriesNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.MeasurementGroup;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.view.CreateLogicalViewNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
@@ -173,6 +174,7 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
}
deviceToSubPlanMap.put(deviceName, subPlanBuilder.getRoot());
}
+
// convert to ALIGN BY DEVICE view
planBuilder =
planBuilder.planDeviceView(
@@ -181,6 +183,10 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
analysis.getDeviceViewInputIndexesMap(),
analysis.getSelectExpressions(),
queryStatement);
+
+ if (planBuilder.getRoot() instanceof TopKNode) {
+ analysis.setUseTopKNode();
+ }
} else {
planBuilder =
planBuilder.withNewRoot(
@@ -255,7 +261,6 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
List<Integer> deviceViewInputIndexes,
MPPQueryContext context) {
LogicalPlanBuilder planBuilder = new LogicalPlanBuilder(analysis, context);
-
if (aggregationExpressions == null) {
// raw data query
planBuilder =
@@ -264,6 +269,8 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
sourceExpressions,
queryStatement.getResultTimeOrder(),
analysis.getGlobalTimeFilter(),
+ 0,
+ pushDownLimitToScanNode(queryStatement),
analysis.isLastLevelUseWildcard())
.planWhereAndSourceTransform(
whereExpression,
@@ -286,6 +293,8 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
sourceExpressions,
queryStatement.getResultTimeOrder(),
analysis.getGlobalTimeFilter(),
+ 0,
+ 0,
analysis.isLastLevelUseWildcard())
.planWhereAndSourceTransform(
whereExpression,
@@ -364,6 +373,26 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
return planBuilder.getRoot();
}
+ private long pushDownLimitToScanNode(QueryStatement queryStatement) {
+ // `order by time|device LIMIT N align by device` and no value filter,
+ // can push down limitValue to ScanNode
+ if (queryStatement.isAlignByDevice()
+ && queryStatement.hasLimit()
+ && !analysis.hasValueFilter()
+ && (queryStatement.isOrderByBasedOnDevice() || queryStatement.isOrderByBasedOnTime())) {
+
+ // both `offset` and `limit` exist, push `limit+offset` down as limitValue
+ if (queryStatement.hasOffset()) {
+ return queryStatement.getRowOffset() + queryStatement.getRowLimit();
+ }
+
+ // only `limit` exist, push `limit` down as limitValue
+ return queryStatement.getRowLimit();
+ }
+
+ return 0;
+ }
+
private boolean needTransform(Set<Expression> expressions) {
for (Expression expression : expressions) {
if (ExpressionAnalyzer.checkIsNeedTransform(expression)) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
index 7ffd8d446ea..6e086fac7ce 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
@@ -61,6 +61,7 @@ import org.apache.iotdb.db.queryengine.execution.operator.process.SingleDeviceVi
import org.apache.iotdb.db.queryengine.execution.operator.process.SlidingWindowAggregationOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.SortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.TagAggregationOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TopKOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.TransformOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.IFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.ILinearFill;
@@ -176,6 +177,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDevi
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SlidingWindowAggregationNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TimeJoinNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TransformNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryCollectNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryMergeNode;
@@ -843,6 +845,44 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
MergeSortComparator.getComparator(sortItemList, sortItemIndexList, sortItemDataTypeList));
}
+ @Override
+ public Operator visitTopK(TopKNode node, LocalExecutionPlanContext context) {
+ OperatorContext operatorContext =
+ context
+ .getDriverContext()
+ .addOperatorContext(
+ context.getNextOperatorId(),
+ node.getPlanNodeId(),
+ TopKOperator.class.getSimpleName());
+ List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTypeProvider());
+ context.setCachedDataTypes(dataTypes);
+ List<Operator> children = dealWithConsumeChildrenOneByOneNode(node, context);
+ List<SortItem> sortItemList = node.getMergeOrderParameter().getSortItemList();
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+
+ List<Integer> sortItemIndexList = new ArrayList<>(sortItemList.size());
+ List<TSDataType> sortItemDataTypeList = new ArrayList<>(sortItemList.size());
+ genSortInformation(
+ node.getOutputColumnNames(),
+ dataTypes,
+ sortItemList,
+ sortItemIndexList,
+ sortItemDataTypeList);
+ return new TopKOperator(
+ operatorContext,
+ children,
+ dataTypes,
+ MergeSortComparator.getComparator(sortItemList, sortItemIndexList, sortItemDataTypeList),
+ node.getTopValue(),
+ !sortItemList.isEmpty()
+ && sortItemList.get(0).getSortKey().equalsIgnoreCase(OrderByKey.TIME)
+ && sortItemList.stream()
+ .allMatch(
+ i ->
+ (i.getSortKey().equals(OrderByKey.TIME)
+ || i.getSortKey().equals(OrderByKey.DEVICE))));
+ }
+
private void genSortInformation(
List<String> outputColumnNames,
List<TSDataType> dataTypes,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java
index f0e016a5e52..de38440c127 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java
@@ -52,14 +52,12 @@ import java.util.Map;
import java.util.Set;
public class DistributionPlanner {
- private Analysis analysis;
- private MPPQueryContext context;
- private LogicalQueryPlan logicalPlan;
+ private final Analysis analysis;
+ private final MPPQueryContext context;
+ private final LogicalQueryPlan logicalPlan;
private final List<PlanOptimizer> optimizers;
- private int planFragmentIndex = 0;
-
public DistributionPlanner(Analysis analysis, LogicalQueryPlan logicalPlan) {
this.analysis = analysis;
this.logicalPlan = logicalPlan;
@@ -173,12 +171,13 @@ public class DistributionPlanner {
}
public SubPlan splitFragment(PlanNode root) {
- FragmentBuilder fragmentBuilder = new FragmentBuilder(context);
+ FragmentBuilder fragmentBuilder = new FragmentBuilder();
return fragmentBuilder.splitToSubPlan(root);
}
public DistributedQueryPlan planFragments() {
PlanNode rootAfterRewrite = rewriteSource();
+
PlanNode rootWithExchange = addExchangeNode(rootAfterRewrite);
if (analysis.getStatement() != null && analysis.getStatement().isQuery()) {
analysis
@@ -245,11 +244,6 @@ public class DistributionPlanner {
}
private class FragmentBuilder {
- private MPPQueryContext context;
-
- public FragmentBuilder(MPPQueryContext context) {
- this.context = context;
- }
public SubPlan splitToSubPlan(PlanNode root) {
SubPlan rootSubPlan = createSubPlan(root);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/ExchangeNodeAdder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/ExchangeNodeAdder.java
index 779f101a52b..6b81b6ba0db 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/ExchangeNodeAdder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/ExchangeNodeAdder.java
@@ -40,12 +40,14 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.FilterNode
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.GroupByLevelNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.GroupByTagNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.HorizontallyConcatNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MergeSortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MultiChildProcessNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDeviceViewNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SlidingWindowAggregationNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TimeJoinNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TransformNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryCollectNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryMergeNode;
@@ -59,6 +61,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesAggre
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SourceNode;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@@ -209,6 +212,11 @@ public class ExchangeNodeAdder extends PlanVisitor<PlanNode, NodeGroupContext> {
return processMultiChildNode(node, context);
}
+ @Override
+ public PlanNode visitTopK(TopKNode node, NodeGroupContext context) {
+ return processMultiChildNode(node, context);
+ }
+
@Override
public PlanNode visitLastQueryMerge(LastQueryMergeNode node, NodeGroupContext context) {
return processMultiChildNode(node, context);
@@ -275,6 +283,11 @@ public class ExchangeNodeAdder extends PlanVisitor<PlanNode, NodeGroupContext> {
return processOneChildNode(node, context);
}
+ @Override
+ public PlanNode visitLimit(LimitNode node, NodeGroupContext context) {
+ return processOneChildNode(node, context);
+ }
+
private PlanNode processMultiChildNode(MultiChildProcessNode node, NodeGroupContext context) {
if (analysis.isVirtualSource()) {
return processMultiChildNodeByLocation(node, context);
@@ -286,6 +299,7 @@ public class ExchangeNodeAdder extends PlanVisitor<PlanNode, NodeGroupContext> {
.map(child -> visit(child, context))
.collect(Collectors.toList());
+ // DataRegion which node locates
TRegionReplicaSet dataRegion;
boolean isChildrenDistributionSame = nodeDistributionIsSame(visitedChildren, context);
NodeDistributionType distributionType =
@@ -303,7 +317,6 @@ public class ExchangeNodeAdder extends PlanVisitor<PlanNode, NodeGroupContext> {
context.putNodeDistribution(
newNode.getPlanNodeId(), new NodeDistribution(distributionType, dataRegion));
} else {
- // TODO For align by time, we keep old logic for now
dataRegion = calculateDataRegionByChildren(visitedChildren, context);
context.putNodeDistribution(
newNode.getPlanNodeId(), new NodeDistribution(distributionType, dataRegion));
@@ -315,24 +328,69 @@ public class ExchangeNodeAdder extends PlanVisitor<PlanNode, NodeGroupContext> {
return newNode;
}
+ // optimize `order by time limit N align by device` query,
+ // to ensure that the number of ExchangeNode equals to DataRegionNum but not equals to DeviceNum
+ if (node instanceof TopKNode) {
+ TopKNode rootNode = (TopKNode) node;
+ Map<TRegionReplicaSet, TopKNode> regionTopKNodeMap = new HashMap<>();
+ for (PlanNode child : visitedChildren) {
+ if (child instanceof SingleDeviceViewNode) {
+ ((SingleDeviceViewNode) child).setCacheOutputColumnNames(true);
+ }
+ TRegionReplicaSet region = context.getNodeDistribution(child.getPlanNodeId()).region;
+ regionTopKNodeMap
+ .computeIfAbsent(
+ region,
+ k -> {
+ TopKNode childTopKNode =
+ new TopKNode(
+ context.queryContext.getQueryId().genPlanNodeId(),
+ rootNode.getTopValue(),
+ rootNode.getMergeOrderParameter(),
+ rootNode.getOutputColumnNames());
+ context.putNodeDistribution(
+ childTopKNode.getPlanNodeId(),
+ new NodeDistribution(NodeDistributionType.SAME_WITH_ALL_CHILDREN, region));
+ return childTopKNode;
+ })
+ .addChild(child);
+ }
+
+ for (Map.Entry<TRegionReplicaSet, TopKNode> entry : regionTopKNodeMap.entrySet()) {
+ TRegionReplicaSet topKNodeLocatedRegion = entry.getKey();
+ TopKNode topKNode = entry.getValue();
+
+ if (!dataRegion.equals(topKNodeLocatedRegion)) {
+ ExchangeNode exchangeNode =
+ new ExchangeNode(context.queryContext.getQueryId().genPlanNodeId());
+ exchangeNode.setChild(topKNode);
+ exchangeNode.setOutputColumnNames(topKNode.getOutputColumnNames());
+ context.hasExchangeNode = true;
+ newNode.addChild(exchangeNode);
+ } else {
+ newNode.addChild(topKNode);
+ }
+ }
+ return newNode;
+ }
+
// Otherwise, we need to add ExchangeNode for the child whose DataRegion is different from the
// parent.
- visitedChildren.forEach(
- child -> {
- if (!dataRegion.equals(context.getNodeDistribution(child.getPlanNodeId()).region)) {
- if (child instanceof SingleDeviceViewNode) {
- ((SingleDeviceViewNode) child).setCacheOutputColumnNames(true);
- }
- ExchangeNode exchangeNode =
- new ExchangeNode(context.queryContext.getQueryId().genPlanNodeId());
- exchangeNode.setChild(child);
- exchangeNode.setOutputColumnNames(child.getOutputColumnNames());
- context.hasExchangeNode = true;
- newNode.addChild(exchangeNode);
- } else {
- newNode.addChild(child);
- }
- });
+ for (PlanNode child : visitedChildren) {
+ if (!dataRegion.equals(context.getNodeDistribution(child.getPlanNodeId()).region)) {
+ if (child instanceof SingleDeviceViewNode) {
+ ((SingleDeviceViewNode) child).setCacheOutputColumnNames(true);
+ }
+ ExchangeNode exchangeNode =
+ new ExchangeNode(context.queryContext.getQueryId().genPlanNodeId());
+ exchangeNode.setChild(child);
+ exchangeNode.setOutputColumnNames(child.getOutputColumnNames());
+ context.hasExchangeNode = true;
+ newNode.addChild(exchangeNode);
+ } else {
+ newNode.addChild(child);
+ }
+ }
return newNode;
}
@@ -395,22 +453,24 @@ public class ExchangeNodeAdder extends PlanVisitor<PlanNode, NodeGroupContext> {
return groupByRegion.keySet().iterator().next();
}
- // Step 2: return the RegionReplicaSet with max count
- long maxRegionCount = -1;
+ // Step 2: return the RegionReplicaSet with max node count
+ long maxCount = -1;
TRegionReplicaSet result = null;
for (Map.Entry<TRegionReplicaSet, Long> entry : groupByRegion.entrySet()) {
- if (DataPartition.NOT_ASSIGNED.equals(entry.getKey())) {
+ TRegionReplicaSet region = entry.getKey();
+ long planNodeCount = entry.getValue();
+ if (DataPartition.NOT_ASSIGNED.equals(region)) {
continue;
}
- if (entry.getKey().equals(context.getMostlyUsedDataRegion())) {
- return entry.getKey();
+ if (region.equals(context.getMostlyUsedDataRegion())) {
+ return region;
}
- if (entry.getValue() > maxRegionCount) {
- maxRegionCount = entry.getValue();
- result = entry.getKey();
+ if (planNodeCount > maxCount) {
+ maxCount = planNodeCount;
+ result = region;
}
}
- return result;
+ return result == null ? context.getMostlyUsedDataRegion() : result;
}
private TRegionReplicaSet calculateSchemaRegionByChildren(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
index 14108d74e3c..f19d036822f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
@@ -41,6 +41,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.DeviceView
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.GroupByLevelNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.GroupByTagNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.HorizontallyConcatNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MergeSortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MultiChildProcessNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDeviceViewNode;
@@ -194,11 +195,7 @@ public class SourceRewriter extends SimplePlanNodeRewriter<DistributionPlanConte
deviceViewNodeList.add(regionDeviceViewNode);
}
- if (deviceViewNodeList.size() == 1) {
- return deviceViewNodeList;
- }
-
- if (analysis.isHasSort()) {
+ if (deviceViewNodeList.size() == 1 || analysis.isHasSort() || analysis.isUseTopKNode()) {
return deviceViewNodeList;
}
@@ -954,6 +951,18 @@ public class SourceRewriter extends SimplePlanNodeRewriter<DistributionPlanConte
: groupSourcesForGroupByTag(root, sourceGroup, context));
}
+ @Override
+ public List<PlanNode> visitLimit(LimitNode node, DistributionPlanContext context) {
+ List<PlanNode> result = new ArrayList<>();
+ for (PlanNode planNode : rewrite(node.getChild(), context)) {
+ LimitNode newNode =
+ new LimitNode(
+ context.queryContext.getQueryId().genPlanNodeId(), planNode, node.getLimit());
+ result.add(newNode);
+ }
+ return result;
+ }
+
// If the Aggregation Query contains value filter, we need to use the naive query plan
// for it. That is, do the raw data query and then do the aggregation operation.
// Currently, the method to judge whether the query should use naive query plan is whether
@@ -1122,14 +1131,13 @@ public class SourceRewriter extends SimplePlanNodeRewriter<DistributionPlanConte
n ->
n.getAggregationDescriptorList()
.forEach(
- v -> {
- childDescriptors.add(
- new AggregationDescriptor(
- v.getAggregationFuncName(),
- AggregationStep.INTERMEDIATE,
- v.getInputExpressions(),
- v.getInputAttributes()));
- }));
+ v ->
+ childDescriptors.add(
+ new AggregationDescriptor(
+ v.getAggregationFuncName(),
+ AggregationStep.INTERMEDIATE,
+ v.getInputExpressions(),
+ v.getInputAttributes()))));
parentOfGroup.setAggregationDescriptorList(childDescriptors);
if (sourceNodes.size() == 1) {
parentOfGroup.addChild(sourceNodes.get(0));
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
index 5e6f4089b1c..3de01571cb4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
@@ -41,6 +41,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDevi
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SlidingWindowAggregationNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TimeJoinNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TransformNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryCollectNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryMergeNode;
@@ -193,6 +194,15 @@ public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter
return render(node, boxValue, context);
}
+ @Override
+ public List<String> visitTopK(TopKNode node, GraphContext context) {
+ List<String> boxValue = new ArrayList<>();
+ boxValue.add(String.format("TopK-%s", node.getPlanNodeId().getId()));
+ boxValue.add(String.format("LimitValue: %d", node.getTopValue()));
+ boxValue.add(node.getMergeOrderParameter().toString());
+ return render(node, boxValue, context);
+ }
+
@Override
public List<String> visitDeviceMerge(DeviceMergeNode node, GraphContext context) {
List<String> boxValue = new ArrayList<>();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
index d3ed441e563..e732129f68c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
@@ -74,6 +74,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDevi
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SlidingWindowAggregationNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TimeJoinNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TransformNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryCollectNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryMergeNode;
@@ -183,7 +184,8 @@ public enum PlanNodeType {
ALTER_LOGICAL_VIEW((short) 78),
PIPE_ENRICHED_INSERT((short) 79),
FORECAST((short) 80),
- LAST_QUERY_TRANSFORM((short) 81);
+ LAST_QUERY_TRANSFORM((short) 81),
+ TOP_K((short) 82);
public static final int BYTES = Short.BYTES;
@@ -392,6 +394,8 @@ public enum PlanNodeType {
return ForecastNode.deserialize(buffer);
case 81:
return LastQueryTransformNode.deserialize(buffer);
+ case 82:
+ return TopKNode.deserialize(buffer);
default:
throw new IllegalArgumentException("Invalid node type: " + nodeType);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
index 8859fbca46f..7589b2fe7f3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
@@ -74,6 +74,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDevi
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SlidingWindowAggregationNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TimeJoinNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TransformNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryCollectNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryMergeNode;
@@ -248,6 +249,10 @@ public abstract class PlanVisitor<R, C> {
return visitMultiChildProcess(node, context);
}
+ public R visitTopK(TopKNode node, C context) {
+ return visitMultiChildProcess(node, context);
+ }
+
public R visitHorizontallyConcat(HorizontallyConcatNode node, C context) {
return visitMultiChildProcess(node, context);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/TopKNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/TopKNode.java
new file mode 100644
index 00000000000..8b5845d0467
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/TopKNode.java
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.planner.plan.node.process;
+
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.OrderByParameter;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+/** TopNode is optimized for `order by time|expression limit N align by device` query. */
+public class TopKNode extends MultiChildProcessNode {
+
+ // when LIMIT value in `order by time|expression LIMIT N align by device` query is less this
+ // value,
+ // use TopKNode, otherwise use MergeSortNode
+ public static final int LIMIT_USE_TOP_K_FOR_ALIGN_BY_DEVICE = 1000000;
+
+ private final int topValue;
+
+ private final OrderByParameter mergeOrderParameter;
+
+ private final List<String> outputColumns;
+
+ public TopKNode(
+ PlanNodeId id,
+ int topValue,
+ OrderByParameter mergeOrderParameter,
+ List<String> outputColumns) {
+ super(id);
+ this.topValue = topValue;
+ this.mergeOrderParameter = mergeOrderParameter;
+ this.outputColumns = outputColumns;
+ }
+
+ public TopKNode(
+ PlanNodeId id,
+ int topValue,
+ List<PlanNode> children,
+ OrderByParameter mergeOrderParameter,
+ List<String> outputColumns) {
+ super(id, children);
+ this.topValue = topValue;
+ this.mergeOrderParameter = mergeOrderParameter;
+ this.outputColumns = outputColumns;
+ }
+
+ public OrderByParameter getMergeOrderParameter() {
+ return mergeOrderParameter;
+ }
+
+ @Override
+ public PlanNode clone() {
+ return new TopKNode(getPlanNodeId(), getTopValue(), getMergeOrderParameter(), outputColumns);
+ }
+
+ @Override
+ public PlanNode createSubNode(int subNodeId, int startIndex, int endIndex) {
+ return new TopKNode(
+ new PlanNodeId(String.format("%s-%s", getPlanNodeId(), subNodeId)),
+ getTopValue(),
+ new ArrayList<>(children.subList(startIndex, endIndex)),
+ getMergeOrderParameter(),
+ outputColumns);
+ }
+
+ @Override
+ public List<String> getOutputColumnNames() {
+ return outputColumns;
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitTopK(this, context);
+ }
+
+ @Override
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.TOP_K.serialize(byteBuffer);
+ ReadWriteIOUtils.write(topValue, byteBuffer);
+ mergeOrderParameter.serializeAttributes(byteBuffer);
+ ReadWriteIOUtils.write(outputColumns.size(), byteBuffer);
+ for (String column : outputColumns) {
+ ReadWriteIOUtils.write(column, byteBuffer);
+ }
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.TOP_K.serialize(stream);
+ ReadWriteIOUtils.write(topValue, stream);
+ mergeOrderParameter.serializeAttributes(stream);
+ ReadWriteIOUtils.write(outputColumns.size(), stream);
+ for (String column : outputColumns) {
+ ReadWriteIOUtils.write(column, stream);
+ }
+ }
+
+ public static TopKNode deserialize(ByteBuffer byteBuffer) {
+ int topValue = ReadWriteIOUtils.readInt(byteBuffer);
+ OrderByParameter orderByParameter = OrderByParameter.deserialize(byteBuffer);
+ int columnSize = ReadWriteIOUtils.readInt(byteBuffer);
+ List<String> outputColumns = new ArrayList<>();
+ while (columnSize > 0) {
+ outputColumns.add(ReadWriteIOUtils.readString(byteBuffer));
+ columnSize--;
+ }
+ PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ return new TopKNode(planNodeId, topValue, orderByParameter, outputColumns);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ TopKNode that = (TopKNode) o;
+ return topValue == that.getTopValue()
+ && Objects.equals(mergeOrderParameter, that.getMergeOrderParameter());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), topValue, mergeOrderParameter);
+ }
+
+ @Override
+ public String toString() {
+ return String.format("TopK-%s", this.getPlanNodeId());
+ }
+
+ public int getTopValue() {
+ return this.topValue;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/sink/ShuffleSinkNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/sink/ShuffleSinkNode.java
index e44722a5881..9f4a82820ce 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/sink/ShuffleSinkNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/sink/ShuffleSinkNode.java
@@ -45,13 +45,6 @@ public class ShuffleSinkNode extends MultiChildrenSinkNode {
super(id, downStreamChannelLocationList);
}
- public ShuffleSinkNode(
- PlanNodeId id,
- List<PlanNode> children,
- List<DownStreamChannelLocation> downStreamChannelLocationList) {
- super(id, children, downStreamChannelLocationList);
- }
-
@Override
public PlanNode clone() {
return new ShuffleSinkNode(getPlanNodeId(), getDownStreamChannelLocationList());
@@ -93,6 +86,11 @@ public class ShuffleSinkNode extends MultiChildrenSinkNode {
}
}
+ @Override
+ public String toString() {
+ return String.format("ShuffleSinkNode-%s", this.getPlanNodeId());
+ }
+
public static ShuffleSinkNode deserialize(ByteBuffer byteBuffer) {
int size = ReadWriteIOUtils.readInt(byteBuffer);
List<DownStreamChannelLocation> downStreamChannelLocationList = new ArrayList<>();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortHeap.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortHeap.java
index 7cab3e5e736..4d6fd026744 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortHeap.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortHeap.java
@@ -110,4 +110,8 @@ public class MergeSortHeap {
heap[childIndex] = child;
}
}
+
+ public int getHeapSize() {
+ return this.heapSize;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortKey.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortKey.java
index 5cc23eeb219..021e6b990f6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortKey.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortKey.java
@@ -23,7 +23,8 @@ import org.apache.iotdb.tsfile.read.common.block.TsBlock;
public class MergeSortKey extends SortKey {
- // This filed only used in operation as a intermediate tool, which is used to locate the origin of
+ // This filed only used in operation as an intermediate tool, which is used to locate the origin
+ // of
// sortKey when there are more than one channels as input of mergeSort.
// It was initialized during the calculation in operator.
public int inputChannelIndex;
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java
index e846dcceb96..814517086c9 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java
@@ -72,7 +72,6 @@ import org.junit.Before;
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.Mockito;
-import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import java.io.IOException;
@@ -379,7 +378,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -387,7 +386,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -395,7 +394,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(3).isNull(i));
assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -403,7 +402,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(1500, count);
}
@Test
@@ -426,7 +425,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(3).isNull(i));
assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -434,7 +433,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
assertTrue(tsBlock.getColumn(2).isNull(i));
@@ -443,7 +442,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
// make sure the device column is by desc
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -451,7 +450,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(1500, count);
}
@Test
@@ -474,7 +473,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(3).isNull(i));
assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -482,7 +481,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
assertTrue(tsBlock.getColumn(2).isNull(i));
@@ -491,7 +490,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
// make sure the device column is by desc
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -499,7 +498,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(1500, count);
}
@Test
@@ -523,7 +522,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -531,7 +530,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -539,7 +538,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(3).isNull(i));
assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -547,7 +546,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(1500, count);
}
// ------------------------------------------------------------------------------------------------
@@ -883,22 +882,22 @@ public class MergeSortOperatorTest {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 3);
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -906,7 +905,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -927,23 +926,23 @@ public class MergeSortOperatorTest {
if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by desc
- assertEquals(checkDevice, 3);
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -951,7 +950,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -971,23 +970,23 @@ public class MergeSortOperatorTest {
if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by desc
- assertEquals(checkDevice, 3);
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -995,7 +994,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -1016,22 +1015,22 @@ public class MergeSortOperatorTest {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 3);
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -1039,8 +1038,9 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
+
// ------------------------------------------------------------------------------------------------
// order by device
// ------------------------------------------------------------------------------------------------
@@ -1368,7 +1368,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -1421,7 +1421,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -1474,7 +1474,7 @@ public class MergeSortOperatorTest {
}
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -1526,7 +1526,7 @@ public class MergeSortOperatorTest {
}
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
// ------------------------------------------------------------------------------------------------
@@ -1653,7 +1653,6 @@ public class MergeSortOperatorTest {
@Mock Operator childOperator1 = Mockito.mock(Operator.class);
@Mock Operator childOperator2 = Mockito.mock(Operator.class);
- @Mock ListenableFuture<?> listenableFuture = Mockito.mock(ListenableFuture.class);
private TsBlock[] constructOutput(long[] time) {
TsBlockBuilder tsBlockBuilder1 =
@@ -1662,8 +1661,8 @@ public class MergeSortOperatorTest {
ColumnBuilder[] columnBuilders = tsBlockBuilder1.getValueColumnBuilders();
for (int i = 0; i < time.length / 2; i++) {
timeColumnBuilder.writeLong(time[i]);
- for (int j = 0; j < columnBuilders.length; j++) {
- columnBuilders[j].writeLong(time[i]);
+ for (ColumnBuilder columnBuilder : columnBuilders) {
+ columnBuilder.writeLong(time[i]);
}
tsBlockBuilder1.declarePosition();
}
@@ -1674,8 +1673,8 @@ public class MergeSortOperatorTest {
ColumnBuilder[] columnBuilders2 = tsBlockBuilder2.getValueColumnBuilders();
for (int i = time.length / 2; i < time.length; i++) {
timeColumnBuilder.writeLong(time[i]);
- for (int j = 0; j < columnBuilders2.length; j++) {
- columnBuilders2[j].writeLong(time[i]);
+ for (ColumnBuilder columnBuilder : columnBuilders2) {
+ columnBuilder.writeLong(time[i]);
}
tsBlockBuilder2.declarePosition();
}
@@ -1700,54 +1699,46 @@ public class MergeSortOperatorTest {
when(mockFuture.isDone()).thenReturn(true);
when(childOperator1.nextWithTimer())
.thenAnswer(
- new Answer<TsBlock>() {
- @Override
- public TsBlock answer(InvocationOnMock invocationOnMock) throws Throwable {
- int count = count1.getAndIncrement();
- if (count == 0) {
- return tsBlocks1[0];
- } else if (count == 1) {
- return tsBlocks1[1];
- } else {
- return null;
- }
- }
- });
+ (Answer<TsBlock>)
+ invocationOnMock -> {
+ int count = count1.getAndIncrement();
+ if (count == 0) {
+ return tsBlocks1[0];
+ } else if (count == 1) {
+ return tsBlocks1[1];
+ } else {
+ return null;
+ }
+ });
when(childOperator2.nextWithTimer())
.thenAnswer(
- new Answer<TsBlock>() {
- @Override
- public TsBlock answer(InvocationOnMock invocationOnMock) throws Throwable {
- int count = count2.getAndIncrement();
- if (count == 0) {
- return tsBlocks2[0];
- } else if (count == 1) {
- return tsBlocks2[1];
- } else {
- return null;
- }
- }
- });
+ (Answer<TsBlock>)
+ invocationOnMock -> {
+ int count = count2.getAndIncrement();
+ if (count == 0) {
+ return tsBlocks2[0];
+ } else if (count == 1) {
+ return tsBlocks2[1];
+ } else {
+ return null;
+ }
+ });
when(childOperator1.hasNextWithTimer())
.thenAnswer(
- new Answer<Boolean>() {
- @Override
- public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable {
- int count = count1.get();
- return count < 2;
- }
- });
+ (Answer<Boolean>)
+ invocationOnMock -> {
+ int count = count1.get();
+ return count < 2;
+ });
when(childOperator2.hasNextWithTimer())
.thenAnswer(
- new Answer<Boolean>() {
- @Override
- public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable {
- int count = count2.get();
- return count < 2;
- }
- });
+ (Answer<Boolean>)
+ invocationOnMock -> {
+ int count = count2.get();
+ return count < 2;
+ });
QueryId queryId = new QueryId("stub_query");
ExecutorService instanceNotificationExecutor =
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java
similarity index 60%
copy from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java
copy to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java
index e846dcceb96..55b7f8f536c 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java
@@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.db.queryengine.execution.operator;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
@@ -26,36 +27,29 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
import org.apache.iotdb.db.queryengine.common.QueryId;
-import org.apache.iotdb.db.queryengine.common.header.DatasetHeader;
-import org.apache.iotdb.db.queryengine.common.header.DatasetHeaderFactory;
import org.apache.iotdb.db.queryengine.execution.driver.DriverContext;
import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateMachine;
import org.apache.iotdb.db.queryengine.execution.operator.process.DeviceViewOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.MergeSortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.SingleDeviceViewOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.process.SortOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TopKOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.RowBasedTimeJoinOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.DescTimeComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.MergeSortComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.SingleColumnMerger;
import org.apache.iotdb.db.queryengine.execution.operator.source.SeriesScanOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.source.ShowQueriesOperator;
-import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
-import org.apache.iotdb.db.queryengine.plan.execution.IQueryExecution;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
-import org.apache.iotdb.db.queryengine.plan.statement.Statement;
import org.apache.iotdb.db.queryengine.plan.statement.component.OrderByKey;
import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
import org.apache.iotdb.db.queryengine.plan.statement.component.SortItem;
import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
import org.apache.iotdb.db.storageengine.dataregion.read.reader.series.SeriesReaderTestUtil;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
-import org.apache.iotdb.db.utils.datastructure.SortKey;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
@@ -64,7 +58,6 @@ import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ListenableFuture;
import io.airlift.units.Duration;
import org.junit.After;
@@ -72,19 +65,15 @@ import org.junit.Before;
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.Mockito;
-import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
-import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -96,28 +85,34 @@ import static org.junit.Assert.fail;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.when;
-public class MergeSortOperatorTest {
-
- private static final String MERGE_SORT_OPERATOR_TEST_SG = "root.MergeSortOperatorTest";
+/**
+ * The original data and structure of this test is same as {@link
+ * org.apache.iotdb.db.queryengine.execution.operator.MergeSortOperatorTest}.
+ */
+public class TopKOperatorTest {
+ private static final String TOP_K_OPERATOR_TEST_SG = "root.TopKOperatorTest";
private final List<String> deviceIds = new ArrayList<>();
private final List<MeasurementSchema> measurementSchemas = new ArrayList<>();
private final List<TsFileResource> seqResources = new ArrayList<>();
private final List<TsFileResource> unSeqResources = new ArrayList<>();
- private static final String DEVICE0 = MERGE_SORT_OPERATOR_TEST_SG + ".device0";
- private static final String DEVICE1 = MERGE_SORT_OPERATOR_TEST_SG + ".device1";
- private static final String DEVICE2 = MERGE_SORT_OPERATOR_TEST_SG + ".device2";
- private static final String DEVICE3 = MERGE_SORT_OPERATOR_TEST_SG + ".device3";
+ private static final String DEVICE0 = TOP_K_OPERATOR_TEST_SG + ".device0";
+ private static final String DEVICE1 = TOP_K_OPERATOR_TEST_SG + ".device1";
+ private static final String DEVICE2 = TOP_K_OPERATOR_TEST_SG + ".device2";
+ private static final String DEVICE3 = TOP_K_OPERATOR_TEST_SG + ".device3";
private int dataNodeId;
+ private final int limitValue = 100;
+
@Before
public void setUp() throws MetadataException, IOException, WriteProcessException {
dataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0);
+ TSFileDescriptor.getInstance().getConfig().setMaxTsBlockLineNumber(3);
SeriesReaderTestUtil.setUp(
- measurementSchemas, deviceIds, seqResources, unSeqResources, MERGE_SORT_OPERATOR_TEST_SG);
+ measurementSchemas, deviceIds, seqResources, unSeqResources, TOP_K_OPERATOR_TEST_SG);
}
@After
@@ -138,10 +133,10 @@ public class MergeSortOperatorTest {
}
}
- // ------------------------------------------------------------------------------------------------
+ // ----------------------------------------------------------------------------------------------
// order by time - 1
- // ------------------------------------------------------------------------------------------------
- // MergeSortOperator
+ // ----------------------------------------------------------------------------------------------
+ // TopKOperator
// ____________|_______________
// / | \
// SingleDeviceViewOperator SingleDeviceViewOperator SingleDeviceViewOperator
@@ -149,8 +144,9 @@ public class MergeSortOperatorTest {
// SeriesScanOperator TimeJoinOperator TimeJoinOperator
// / \ / \
// SeriesScanOperator SeriesScanOperator SeriesScanOperator SeriesScanOperator
- // ------------------------------------------------------------------------------------------------
- public MergeSortOperator mergeSortOperatorTest(Ordering timeOrdering, Ordering deviceOrdering) {
+ // ----------------------------------------------------------------------------------------------
+ public TopKOperator topKOperatorTest(
+ Ordering timeOrdering, Ordering deviceOrdering, int limitValue) {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
try {
@@ -184,21 +180,11 @@ public class MergeSortOperatorTest {
9, new PlanNodeId("9"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
10, new PlanNodeId("10"), SingleDeviceViewOperator.class.getSimpleName());
-
driverContext.addOperatorContext(
- 11, new PlanNodeId("11"), MergeSortOperator.class.getSimpleName());
+ 11, new PlanNodeId("11"), TopKOperator.class.getSimpleName());
MeasurementPath measurementPath1 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
- MeasurementPath measurementPath2 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
- MeasurementPath measurementPath3 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
- MeasurementPath measurementPath4 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
- MeasurementPath measurementPath5 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
-
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
SeriesScanOperator seriesScanOperator1 =
new SeriesScanOperator(
driverContext.getOperatorContexts().get(0),
@@ -211,6 +197,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -223,6 +211,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -235,6 +225,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -247,6 +239,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -297,6 +291,7 @@ public class MergeSortOperatorTest {
timeJoinOperator1
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+
SingleDeviceViewOperator singleDeviceViewOperator2 =
new SingleDeviceViewOperator(
driverContext.getOperatorContexts().get(7),
@@ -335,8 +330,8 @@ public class MergeSortOperatorTest {
Arrays.asList(4, 5),
tsDataTypes);
- MergeSortOperator mergeSortOperator =
- new MergeSortOperator(
+ TopKOperator topKOperator =
+ new TopKOperator(
driverContext.getOperatorContexts().get(10),
Arrays.asList(
singleDeviceViewOperator1, singleDeviceViewOperator2, singleDeviceViewOperator3),
@@ -346,11 +341,11 @@ public class MergeSortOperatorTest {
new SortItem(OrderByKey.TIME, timeOrdering),
new SortItem(OrderByKey.DEVICE, deviceOrdering)),
Arrays.asList(-1, 0),
- Arrays.asList(TSDataType.INT64, TSDataType.TEXT)));
- mergeSortOperator
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- return mergeSortOperator;
+ Arrays.asList(TSDataType.INT64, TSDataType.TEXT)),
+ limitValue,
+ true);
+ topKOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ return topKOperator;
} catch (IllegalPathException e) {
e.printStackTrace();
fail();
@@ -360,13 +355,15 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByTime1() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.ASC, Ordering.ASC);
+ TopKOperator topKOperator = topKOperatorTest(Ordering.ASC, Ordering.ASC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
+ if (tsBlock == null) {
+ continue;
+ }
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
for (int i = 0; i < tsBlock.getPositionCount(); i++) {
@@ -377,25 +374,25 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(3).isNull(i));
assertTrue(tsBlock.getColumn(4).isNull(i));
assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(1).getInt(i));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
assertTrue(tsBlock.getColumn(4).isNull(i));
assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(getValue(lastTime), tsBlock.getColumn(2).getInt(i));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(3).getInt(i));
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
assertTrue(tsBlock.getColumn(2).isNull(i));
assertTrue(tsBlock.getColumn(3).isNull(i));
- assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(getValue(lastTime), tsBlock.getColumn(4).getInt(i));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(5).getInt(i));
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -403,143 +400,49 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(limitValue, count);
}
@Test
public void testOrderByTime2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.ASC, Ordering.DESC);
+ TopKOperator topKOperator = topKOperatorTest(Ordering.ASC, Ordering.DESC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(6, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) >= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertTrue(tsBlock.getColumn(1).isNull(i));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- assertTrue(tsBlock.getColumn(3).isNull(i));
- assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertTrue(tsBlock.getColumn(1).isNull(i));
- assertTrue(tsBlock.getColumn(4).isNull(i));
- assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertTrue(tsBlock.getColumn(2).isNull(i));
- assertTrue(tsBlock.getColumn(3).isNull(i));
- assertTrue(tsBlock.getColumn(4).isNull(i));
- assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- // make sure the device column is by desc
- assertEquals(checkDevice, 2);
- checkDevice = 0;
- } else {
- fail();
- }
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
+ if (tsBlock == null) {
+ continue;
}
- }
-
- assertEquals(count, 1500);
- }
-
- @Test
- public void testOrderByTime3() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.DESC, Ordering.DESC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
+ assertTrue(tsBlock.getTimeByIndex(i) >= lastTime);
lastTime = tsBlock.getTimeByIndex(i);
if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
assertTrue(tsBlock.getColumn(2).isNull(i));
assertTrue(tsBlock.getColumn(3).isNull(i));
- assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(getValue(lastTime), tsBlock.getColumn(4).getInt(i));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(5).getInt(i));
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
assertTrue(tsBlock.getColumn(4).isNull(i));
assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(getValue(lastTime), tsBlock.getColumn(2).getInt(i));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(3).getInt(i));
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
assertTrue(tsBlock.getColumn(2).isNull(i));
assertTrue(tsBlock.getColumn(3).isNull(i));
assertTrue(tsBlock.getColumn(4).isNull(i));
assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(1).getInt(i));
// make sure the device column is by desc
- assertEquals(checkDevice, 2);
- checkDevice = 0;
- } else {
- fail();
- }
- }
- }
-
- assertEquals(count, 1500);
- }
-
- @Test
- public void testOrderByTime4() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.DESC, Ordering.ASC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(6, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertTrue(tsBlock.getColumn(2).isNull(i));
- assertTrue(tsBlock.getColumn(3).isNull(i));
- assertTrue(tsBlock.getColumn(4).isNull(i));
- assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- // make sure the device column is by asc
- assertEquals(checkDevice, 0);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertTrue(tsBlock.getColumn(1).isNull(i));
- assertTrue(tsBlock.getColumn(4).isNull(i));
- assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertTrue(tsBlock.getColumn(1).isNull(i));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- assertTrue(tsBlock.getColumn(3).isNull(i));
- assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -547,27 +450,11 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(limitValue, count);
}
- // ------------------------------------------------------------------------------------------------
- // order by time - 2
- // ------------------------------------------------------------------------------------------------
- // [SSO]:SeriesScanOperator
- // [SDO]:SingleDeviceOperator
- // MergeSortOperator
- // ___________|___________
- // / \
- // MergeSortOperator MergeSortOperator
- // / | / \
- // [SDO] [SDO] [SDO] [SDO]
- // | | | |
- // [SSO] RowBasedTimeJoinOperator RowBasedTimeJoinOperator
- // RowBasedTimeJoinOperator
- // / \ / \ / \
- // [SSO] [SSO] [SSO] [SSO] [SSO] [SSO]
- // ------------------------------------------------------------------------------------------------
- public MergeSortOperator mergeSortOperatorTest2(Ordering timeOrdering, Ordering deviceOrdering) {
+ public TopKOperator topKOperatorTest2(
+ Ordering timeOrdering, Ordering deviceOrdering, int limitValue) {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
try {
@@ -611,27 +498,14 @@ public class MergeSortOperatorTest {
driverContext.addOperatorContext(
14, new PlanNodeId("14"), SingleDeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 15, new PlanNodeId("15"), MergeSortOperator.class.getSimpleName());
+ 15, new PlanNodeId("15"), TopKOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 16, new PlanNodeId("16"), MergeSortOperator.class.getSimpleName());
+ 16, new PlanNodeId("16"), TopKOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 17, new PlanNodeId("17"), MergeSortOperator.class.getSimpleName());
+ 17, new PlanNodeId("17"), TopKOperator.class.getSimpleName());
MeasurementPath measurementPath1 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
- MeasurementPath measurementPath2 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
- MeasurementPath measurementPath3 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
- MeasurementPath measurementPath4 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
- MeasurementPath measurementPath5 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
- MeasurementPath measurementPath6 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device3.sensor0", TSDataType.INT32);
- MeasurementPath measurementPath7 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device3.sensor1", TSDataType.INT32);
-
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
SeriesScanOperator seriesScanOperator1 =
new SeriesScanOperator(
driverContext.getOperatorContexts().get(0),
@@ -644,6 +518,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -656,6 +532,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -668,6 +546,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -680,6 +560,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -692,6 +574,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ MeasurementPath measurementPath6 =
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device3.sensor0", TSDataType.INT32);
SeriesScanOperator seriesScanOperator6 =
new SeriesScanOperator(
driverContext.getOperatorContexts().get(5),
@@ -704,6 +588,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ MeasurementPath measurementPath7 =
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device3.sensor1", TSDataType.INT32);
SeriesScanOperator seriesScanOperator7 =
new SeriesScanOperator(
driverContext.getOperatorContexts().get(6),
@@ -716,9 +602,6 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- List<TSDataType> tsDataTypes =
- new LinkedList<>(Arrays.asList(TSDataType.TEXT, TSDataType.INT32, TSDataType.INT32));
-
RowBasedTimeJoinOperator timeJoinOperator1 =
new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(7),
@@ -785,6 +668,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ List<TSDataType> tsDataTypes =
+ new LinkedList<>(Arrays.asList(TSDataType.TEXT, TSDataType.INT32, TSDataType.INT32));
SingleDeviceViewOperator singleDeviceViewOperator1 =
new SingleDeviceViewOperator(
driverContext.getOperatorContexts().get(10),
@@ -814,8 +699,8 @@ public class MergeSortOperatorTest {
Arrays.asList(1, 2),
tsDataTypes);
- MergeSortOperator mergeSortOperator1 =
- new MergeSortOperator(
+ TopKOperator topKOperator1 =
+ new TopKOperator(
driverContext.getOperatorContexts().get(14),
Arrays.asList(singleDeviceViewOperator1, singleDeviceViewOperator2),
tsDataTypes,
@@ -824,12 +709,12 @@ public class MergeSortOperatorTest {
new SortItem(OrderByKey.TIME, timeOrdering),
new SortItem(OrderByKey.DEVICE, deviceOrdering)),
Arrays.asList(-1, 0),
- Arrays.asList(TSDataType.INT64, TSDataType.TEXT)));
- mergeSortOperator1
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- MergeSortOperator mergeSortOperator2 =
- new MergeSortOperator(
+ Arrays.asList(TSDataType.INT64, TSDataType.TEXT)),
+ limitValue,
+ true);
+ topKOperator1.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ TopKOperator topKOperator2 =
+ new TopKOperator(
driverContext.getOperatorContexts().get(15),
Arrays.asList(singleDeviceViewOperator3, singleDeviceViewOperator4),
tsDataTypes,
@@ -838,26 +723,26 @@ public class MergeSortOperatorTest {
new SortItem(OrderByKey.TIME, timeOrdering),
new SortItem(OrderByKey.DEVICE, deviceOrdering)),
Arrays.asList(-1, 0),
- Arrays.asList(TSDataType.INT64, TSDataType.TEXT)));
- mergeSortOperator2
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ Arrays.asList(TSDataType.INT64, TSDataType.TEXT)),
+ limitValue,
+ true);
+ topKOperator2.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- MergeSortOperator mergeSortOperator =
- new MergeSortOperator(
+ TopKOperator topKOperator =
+ new TopKOperator(
driverContext.getOperatorContexts().get(16),
- Arrays.asList(mergeSortOperator1, mergeSortOperator2),
+ Arrays.asList(topKOperator1, topKOperator2),
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)));
- mergeSortOperator
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- return mergeSortOperator;
+ Arrays.asList(TSDataType.INT64, TSDataType.TEXT)),
+ limitValue,
+ true);
+ topKOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ return topKOperator;
} catch (IllegalPathException e) {
e.printStackTrace();
fail();
@@ -867,38 +752,40 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByTime1_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.ASC, Ordering.ASC);
+ TopKOperator topKOperator = topKOperatorTest2(Ordering.ASC, Ordering.ASC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
+ if (tsBlock == null) {
+ continue;
+ }
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
for (int i = 0; i < tsBlock.getPositionCount(); i++) {
assertTrue(tsBlock.getTimeByIndex(i) >= lastTime);
lastTime = tsBlock.getTimeByIndex(i);
if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(1).getInt(i));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(getValue(lastTime), tsBlock.getColumn(1).getInt(i));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(2).getInt(i));
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(getValue(lastTime), tsBlock.getColumn(1).getInt(i));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(2).getInt(i));
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 3);
+ assertEquals(getValue(lastTime), tsBlock.getColumn(1).getInt(i));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(2).getInt(i));
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -906,88 +793,46 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(limitValue, count);
}
@Test
public void testOrderByTime2_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.ASC, Ordering.DESC);
+ TopKOperator topKOperator = topKOperatorTest2(Ordering.ASC, Ordering.DESC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(3, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) >= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- // make sure the device column is by desc
- assertEquals(checkDevice, 3);
- checkDevice = 0;
- } else {
- fail();
- }
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
+ if (tsBlock == null) {
+ continue;
}
- }
-
- assertEquals(count, 2000);
- }
-
- @Test
- public void testOrderByTime3_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.DESC, Ordering.DESC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
+ assertTrue(tsBlock.getTimeByIndex(i) >= lastTime);
lastTime = tsBlock.getTimeByIndex(i);
if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(getValue(lastTime), tsBlock.getColumn(1).getInt(i));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(2).getInt(i));
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(getValue(lastTime), tsBlock.getColumn(1).getInt(i));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(2).getInt(i));
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(getValue(lastTime), tsBlock.getColumn(1).getInt(i));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(2).getInt(i));
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
+ assertEquals(getValue(lastTime), tsBlock.getColumn(1).getInt(i));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by desc
- assertEquals(checkDevice, 3);
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -995,57 +840,14 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(limitValue, count);
}
- @Test
- public void testOrderByTime4_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.DESC, Ordering.ASC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(3, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- // make sure the device column is by asc
- assertEquals(checkDevice, 0);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 3);
- checkDevice = 0;
- } else {
- fail();
- }
- }
- }
-
- assertEquals(count, 2000);
- }
- // ------------------------------------------------------------------------------------------------
+ // -----------------------------------------------------------------------------------
// order by device
- // ------------------------------------------------------------------------------------------------
+ // -----------------------------------------------------------------------------------
// [SSO]:SeriesScanOperator
- // MergeSortOperator
+ // TopKOperator
// ___________|___________
// / \
// DeviceViewOperator DeviceViewOperator
@@ -1054,14 +856,14 @@ public class MergeSortOperatorTest {
// RowBasedTimeJoinOperator
// / \ / \ / \
// [SSO] [SSO] [SSO] [SSO] [SSO] [SSO]
- // ------------------------------------------------------------------------------------------------
- public MergeSortOperator mergeSortOperatorTest3(Ordering timeOrdering, Ordering deviceOrdering) {
+ // ------------------------------------------------------------------------------------
+ public TopKOperator topKOperatorTest3(
+ 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 =
@@ -1095,23 +897,10 @@ public class MergeSortOperatorTest {
driverContext.addOperatorContext(
12, new PlanNodeId("12"), DeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 13, new PlanNodeId("13"), MergeSortOperator.class.getSimpleName());
+ 13, new PlanNodeId("13"), TopKOperator.class.getSimpleName());
MeasurementPath measurementPath1 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
- MeasurementPath measurementPath2 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
- MeasurementPath measurementPath3 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
- MeasurementPath measurementPath4 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
- MeasurementPath measurementPath5 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
- MeasurementPath measurementPath6 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device3.sensor0", TSDataType.INT32);
- MeasurementPath measurementPath7 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device3.sensor1", TSDataType.INT32);
-
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
SeriesScanOperator seriesScanOperator1 =
new SeriesScanOperator(
driverContext.getOperatorContexts().get(0),
@@ -1124,6 +913,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -1136,6 +927,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -1148,6 +941,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -1160,6 +955,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.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),
@@ -1172,6 +969,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ MeasurementPath measurementPath6 =
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device3.sensor0", TSDataType.INT32);
SeriesScanOperator seriesScanOperator6 =
new SeriesScanOperator(
driverContext.getOperatorContexts().get(5),
@@ -1184,6 +983,8 @@ public class MergeSortOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ MeasurementPath measurementPath7 =
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device3.sensor1", TSDataType.INT32);
SeriesScanOperator seriesScanOperator7 =
new SeriesScanOperator(
driverContext.getOperatorContexts().get(6),
@@ -1198,7 +999,6 @@ public class MergeSortOperatorTest {
List<TSDataType> tsDataTypes =
new LinkedList<>(Arrays.asList(TSDataType.TEXT, TSDataType.INT32, TSDataType.INT32));
-
RowBasedTimeJoinOperator timeJoinOperator1 =
new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(7),
@@ -1220,7 +1020,6 @@ public class MergeSortOperatorTest {
timeJoinOperator1
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
-
RowBasedTimeJoinOperator timeJoinOperator2 =
new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(8),
@@ -1242,7 +1041,6 @@ public class MergeSortOperatorTest {
timeJoinOperator2
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
-
RowBasedTimeJoinOperator timeJoinOperator3 =
new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(9),
@@ -1264,13 +1062,15 @@ public class MergeSortOperatorTest {
timeJoinOperator3
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
-
List<String> devices = new ArrayList<>(Arrays.asList(DEVICE0, DEVICE1, DEVICE2, DEVICE3));
- if (deviceOrdering == Ordering.DESC) Collections.reverse(devices);
- List<List<Integer>> deviceColumnIndex = new ArrayList<>();
- deviceColumnIndex.add(Collections.singletonList(1));
- deviceColumnIndex.add(Arrays.asList(1, 2));
- if (deviceOrdering == Ordering.DESC) Collections.reverse(deviceColumnIndex);
+ if (deviceOrdering == Ordering.DESC) {
+ Collections.reverse(devices);
+ }
+ List<List<Integer>> deviceColumnIndex =
+ Arrays.asList(Collections.singletonList(1), Arrays.asList(1, 2));
+ if (deviceOrdering == Ordering.DESC) {
+ Collections.reverse(deviceColumnIndex);
+ }
DeviceViewOperator deviceViewOperator1 =
new DeviceViewOperator(
driverContext.getOperatorContexts().get(10),
@@ -1282,9 +1082,7 @@ public class MergeSortOperatorTest {
: Arrays.asList(timeJoinOperator1, seriesScanOperator1),
deviceColumnIndex,
tsDataTypes);
- deviceColumnIndex = new ArrayList<>();
- deviceColumnIndex.add(Arrays.asList(1, 2));
- deviceColumnIndex.add(Arrays.asList(1, 2));
+ deviceColumnIndex = Arrays.asList(Arrays.asList(1, 2), Arrays.asList(1, 2));
DeviceViewOperator deviceViewOperator2 =
new DeviceViewOperator(
driverContext.getOperatorContexts().get(11),
@@ -1296,8 +1094,8 @@ public class MergeSortOperatorTest {
: Arrays.asList(timeJoinOperator3, timeJoinOperator2),
deviceColumnIndex,
tsDataTypes);
- MergeSortOperator mergeSortOperator =
- new MergeSortOperator(
+ TopKOperator topKOperator =
+ new TopKOperator(
driverContext.getOperatorContexts().get(12),
Arrays.asList(deviceViewOperator1, deviceViewOperator2),
tsDataTypes,
@@ -1306,11 +1104,11 @@ public class MergeSortOperatorTest {
new SortItem(OrderByKey.DEVICE, deviceOrdering),
new SortItem(OrderByKey.TIME, timeOrdering)),
Arrays.asList(0, -1),
- Arrays.asList(TSDataType.TEXT, TSDataType.INT64)));
- mergeSortOperator
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- return mergeSortOperator;
+ Arrays.asList(TSDataType.TEXT, TSDataType.INT64)),
+ limitValue,
+ false);
+ topKOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ return topKOperator;
} catch (IllegalPathException e) {
e.printStackTrace();
fail();
@@ -1320,13 +1118,15 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByDevice1() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.ASC, Ordering.ASC);
+ TopKOperator topKOperator = topKOperatorTest3(Ordering.ASC, Ordering.ASC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
+ if (tsBlock == null) {
+ continue;
+ }
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
for (int i = 0; i < tsBlock.getPositionCount(); i++) {
@@ -1368,18 +1168,20 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(limitValue, count);
}
@Test
public void testOrderByDevice2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.ASC, Ordering.DESC);
+ TopKOperator topKOperator = topKOperatorTest3(Ordering.ASC, Ordering.DESC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
+ if (tsBlock == null) {
+ continue;
+ }
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
for (int i = 0; i < tsBlock.getPositionCount(); i++) {
@@ -1421,239 +1223,11 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
- }
-
- @Test
- public void testOrderByDevice3() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.DESC, Ordering.ASC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
-
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(3, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- // make sure the device column is by asc
- assertTrue(checkDevice < 500);
- checkDevice++;
- if (checkDevice == 500) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice >= 500 && checkDevice < 1000);
- checkDevice++;
- if (checkDevice == 1000) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice >= 1000 && checkDevice < 1500);
- checkDevice++;
- if (checkDevice == 1500) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice >= 1500 && checkDevice < 2000);
- checkDevice++;
- } else {
- fail();
- }
- }
- }
- assertEquals(count, 2000);
- }
-
- @Test
- public void testOrderByDevice4() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.DESC, Ordering.DESC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(3, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice < 500);
- checkDevice++;
- if (checkDevice == 500) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice >= 500 && checkDevice < 1000);
- checkDevice++;
- if (checkDevice == 1000) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice >= 1000 && checkDevice < 1500);
- checkDevice++;
- if (checkDevice == 1500) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- // make sure the device column is by desc
- assertTrue(checkDevice >= 1500 && checkDevice < 2000);
- checkDevice++;
- } else {
- fail();
- }
- }
- }
- assertEquals(count, 2000);
- }
-
- // ------------------------------------------------------------------------------------------------
- // order by Time, DataNodeID
- // ------------------------------------------------------------------------------------------------
- //
- // MergeSortOperator
- // ___________|__________
- // / \
- // SortOperator SortOperator
- // | |
- // ShowQueriesOperator ShowQueriesOperator
- // ------------------------------------------------------------------------------------------------
- @Test
- public void mergeSortWithSortOperatorTest() throws Exception {
- 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 planNodeId0 = new PlanNodeId("0");
- driverContext.addOperatorContext(0, planNodeId0, ShowQueriesOperator.class.getSimpleName());
- PlanNodeId planNodeId1 = new PlanNodeId("1");
- driverContext.addOperatorContext(1, planNodeId1, ShowQueriesOperator.class.getSimpleName());
- PlanNodeId planNodeId2 = new PlanNodeId("2");
- driverContext.addOperatorContext(2, planNodeId2, SortOperator.class.getSimpleName());
- PlanNodeId planNodeId3 = new PlanNodeId("3");
- driverContext.addOperatorContext(3, planNodeId3, SortOperator.class.getSimpleName());
- PlanNodeId planNodeId4 = new PlanNodeId("4");
- driverContext.addOperatorContext(4, planNodeId4, MergeSortOperator.class.getSimpleName());
-
- List<OperatorContext> operatorContexts = driverContext.getOperatorContexts();
- List<TSDataType> dataTypes = DatasetHeaderFactory.getShowQueriesHeader().getRespDataTypes();
- Comparator<SortKey> comparator =
- MergeSortComparator.getComparator(
- Arrays.asList(
- new SortItem(OrderByKey.TIME, Ordering.ASC),
- new SortItem(OrderByKey.DATANODEID, Ordering.DESC)),
- ImmutableList.of(-1, 1),
- ImmutableList.of(TSDataType.INT64, TSDataType.INT32));
-
- Coordinator coordinator1 = Mockito.mock(Coordinator.class);
- when(coordinator1.getAllQueryExecutions())
- .thenReturn(
- ImmutableList.of(
- new FakeQueryExecution(3, "20221229_000000_00003_1", "sql3_node1"),
- new FakeQueryExecution(1, "20221229_000000_00001_1", "sql1_node1"),
- new FakeQueryExecution(2, "20221229_000000_00002_1", "sql2_node1")));
- Coordinator coordinator2 = Mockito.mock(Coordinator.class);
- when(coordinator2.getAllQueryExecutions())
- .thenReturn(
- ImmutableList.of(
- new FakeQueryExecution(3, "20221229_000000_00003_2", "sql3_node2"),
- new FakeQueryExecution(2, "20221229_000000_00002_2", "sql2_node2"),
- new FakeQueryExecution(1, "20221229_000000_00001_2", "sql1_node2")));
-
- ShowQueriesOperator showQueriesOperator1 =
- new ShowQueriesOperator(operatorContexts.get(0), planNodeId0, coordinator1);
- ShowQueriesOperator showQueriesOperator2 =
- new ShowQueriesOperator(operatorContexts.get(1), planNodeId1, coordinator2);
- SortOperator sortOperator1 =
- new SortOperator(
- operatorContexts.get(2), showQueriesOperator1, dataTypes, "", comparator);
- SortOperator sortOperator2 =
- new SortOperator(
- operatorContexts.get(3), showQueriesOperator2, dataTypes, "", comparator);
- Operator root =
- new MergeSortOperator(
- operatorContexts.get(4),
- new ArrayList<Operator>() {
- {
- add(sortOperator1);
- add(sortOperator2);
- }
- },
- dataTypes,
- comparator);
- root.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
-
- int index = 0;
- // Time ASC
- long[] expectedTime = new long[] {1, 1, 2, 2, 3, 3};
- // DataNodeId DESC if Times are equal
- String[] expectedQueryId =
- new String[] {
- "20221229_000000_00001_2",
- "20221229_000000_00001_1",
- "20221229_000000_00002_2",
- "20221229_000000_00002_1",
- "20221229_000000_00003_2",
- "20221229_000000_00003_1"
- };
- int[] expectedDataNodeId = new int[] {2, 1, 2, 1, 2, 1};
- String[] expectedStatement =
- new String[] {
- "sql1_node2", "sql1_node1", "sql2_node2", "sql2_node1", "sql3_node2", "sql3_node1"
- };
- while (root.isBlocked().isDone() && root.hasNext()) {
- TsBlock result = root.next();
- if (result == null) {
- continue;
- }
-
- for (int i = 0; i < result.getPositionCount(); i++, index++) {
- assertEquals(expectedTime[index], result.getTimeColumn().getLong(i));
- assertEquals(expectedQueryId[index], result.getColumn(0).getBinary(i).toString());
- assertEquals(expectedDataNodeId[index], result.getColumn(1).getInt(i));
- assertEquals(expectedStatement[index], result.getColumn(3).getBinary(i).toString());
- }
- }
- } finally {
- instanceNotificationExecutor.shutdown();
- }
+ assertEquals(limitValue, count);
}
@Mock Operator childOperator1 = Mockito.mock(Operator.class);
@Mock Operator childOperator2 = Mockito.mock(Operator.class);
- @Mock ListenableFuture<?> listenableFuture = Mockito.mock(ListenableFuture.class);
private TsBlock[] constructOutput(long[] time) {
TsBlockBuilder tsBlockBuilder1 =
@@ -1662,8 +1236,8 @@ public class MergeSortOperatorTest {
ColumnBuilder[] columnBuilders = tsBlockBuilder1.getValueColumnBuilders();
for (int i = 0; i < time.length / 2; i++) {
timeColumnBuilder.writeLong(time[i]);
- for (int j = 0; j < columnBuilders.length; j++) {
- columnBuilders[j].writeLong(time[i]);
+ for (ColumnBuilder columnBuilder : columnBuilders) {
+ columnBuilder.writeLong(time[i]);
}
tsBlockBuilder1.declarePosition();
}
@@ -1674,8 +1248,8 @@ public class MergeSortOperatorTest {
ColumnBuilder[] columnBuilders2 = tsBlockBuilder2.getValueColumnBuilders();
for (int i = time.length / 2; i < time.length; i++) {
timeColumnBuilder.writeLong(time[i]);
- for (int j = 0; j < columnBuilders2.length; j++) {
- columnBuilders2[j].writeLong(time[i]);
+ for (ColumnBuilder columnBuilder : columnBuilders2) {
+ columnBuilder.writeLong(time[i]);
}
tsBlockBuilder2.declarePosition();
}
@@ -1684,7 +1258,7 @@ public class MergeSortOperatorTest {
}
@Test
- public void mergeSortTest() throws Exception {
+ public void topKOperatorSortTest() throws Exception {
AtomicInteger count1 = new AtomicInteger(0);
AtomicInteger count2 = new AtomicInteger(0);
long[] time1 = new long[] {1, 2, 3, 4, 5, 6};
@@ -1700,54 +1274,46 @@ public class MergeSortOperatorTest {
when(mockFuture.isDone()).thenReturn(true);
when(childOperator1.nextWithTimer())
.thenAnswer(
- new Answer<TsBlock>() {
- @Override
- public TsBlock answer(InvocationOnMock invocationOnMock) throws Throwable {
- int count = count1.getAndIncrement();
- if (count == 0) {
- return tsBlocks1[0];
- } else if (count == 1) {
- return tsBlocks1[1];
- } else {
- return null;
- }
- }
- });
+ (Answer<TsBlock>)
+ invocationOnMock -> {
+ int count = count1.getAndIncrement();
+ if (count == 0) {
+ return tsBlocks1[0];
+ } else if (count == 1) {
+ return tsBlocks1[1];
+ } else {
+ return null;
+ }
+ });
when(childOperator2.nextWithTimer())
.thenAnswer(
- new Answer<TsBlock>() {
- @Override
- public TsBlock answer(InvocationOnMock invocationOnMock) throws Throwable {
- int count = count2.getAndIncrement();
- if (count == 0) {
- return tsBlocks2[0];
- } else if (count == 1) {
- return tsBlocks2[1];
- } else {
- return null;
- }
- }
- });
+ (Answer<TsBlock>)
+ invocationOnMock -> {
+ int count = count2.getAndIncrement();
+ if (count == 0) {
+ return tsBlocks2[0];
+ } else if (count == 1) {
+ return tsBlocks2[1];
+ } else {
+ return null;
+ }
+ });
when(childOperator1.hasNextWithTimer())
.thenAnswer(
- new Answer<Boolean>() {
- @Override
- public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable {
- int count = count1.get();
- return count < 2;
- }
- });
+ (Answer<Boolean>)
+ invocationOnMock -> {
+ int count = count1.get();
+ return count < 2;
+ });
when(childOperator2.hasNextWithTimer())
.thenAnswer(
- new Answer<Boolean>() {
- @Override
- public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable {
- int count = count2.get();
- return count < 2;
- }
- });
+ (Answer<Boolean>)
+ invocationOnMock -> {
+ int count = count2.get();
+ return count < 2;
+ });
QueryId queryId = new QueryId("stub_query");
ExecutorService instanceNotificationExecutor =
@@ -1762,20 +1328,25 @@ public class MergeSortOperatorTest {
PlanNodeId planNodeId1 = new PlanNodeId("1");
driverContext.addOperatorContext(1, planNodeId1, MergeSortOperator.class.getSimpleName());
- MergeSortOperator mergeSortOperator =
- new MergeSortOperator(
+ TopKOperator topKOperator =
+ new TopKOperator(
driverContext.getOperatorContexts().get(0),
Arrays.asList(childOperator1, childOperator2),
Collections.singletonList(TSDataType.INT64),
MergeSortComparator.getComparator(
Collections.singletonList(new SortItem(OrderByKey.TIME, Ordering.ASC)),
Collections.singletonList(-1),
- Collections.singletonList(TSDataType.INT64)));
- mergeSortOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ Collections.singletonList(TSDataType.INT64)),
+ 20,
+ true);
+ topKOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
int index = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock result = mergeSortOperator.next();
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock result = topKOperator.next();
+ if (result == null) {
+ continue;
+ }
for (int i = 0; i < result.getPositionCount(); i++) {
long time = result.getTimeByIndex(i);
assertEquals(time, ans[index++]);
@@ -1783,94 +1354,4 @@ public class MergeSortOperatorTest {
}
assertEquals(index, ans.length);
}
-
- static class FakeQueryExecution implements IQueryExecution {
- private final long startTime;
- private final String queryId;
- private final String sql;
-
- FakeQueryExecution(long startTime, String queryId, String sql) {
- this.startTime = startTime;
- this.queryId = queryId;
- this.sql = sql;
- }
-
- @Override
- public String getQueryId() {
- return queryId;
- }
-
- @Override
- public long getStartExecutionTime() {
- return startTime;
- }
-
- @Override
- public void recordExecutionTime(long executionTime) {}
-
- @Override
- public long getTotalExecutionTime() {
- return 0;
- }
-
- @Override
- public Optional<String> getExecuteSQL() {
- return Optional.of(sql);
- }
-
- @Override
- public Statement getStatement() {
- return null;
- }
-
- @Override
- public void start() {}
-
- @Override
- public void stop(Throwable t) {}
-
- @Override
- public void stopAndCleanup() {}
-
- @Override
- public void stopAndCleanup(Throwable t) {}
-
- @Override
- public void cancel() {}
-
- @Override
- public ExecutionResult getStatus() {
- return null;
- }
-
- @Override
- public Optional<TsBlock> getBatchResult() {
- return Optional.empty();
- }
-
- @Override
- public Optional<ByteBuffer> getByteBufferBatchResult() {
- return Optional.empty();
- }
-
- @Override
- public boolean hasNextResult() {
- return false;
- }
-
- @Override
- public int getOutputValueColumnCount() {
- return 0;
- }
-
- @Override
- public DatasetHeader getDatasetHeader() {
- return null;
- }
-
- @Override
- public boolean isQuery() {
- return false;
- }
- }
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/plan/distribution/AlignByDeviceOrderByLimitOffsetTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/plan/distribution/AlignByDeviceOrderByLimitOffsetTest.java
new file mode 100644
index 00000000000..18953d210d5
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/plan/distribution/AlignByDeviceOrderByLimitOffsetTest.java
@@ -0,0 +1,278 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.plan.distribution;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
+import org.apache.iotdb.db.queryengine.plan.planner.distribution.DistributionPlanner;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.DeviceViewNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.FilterNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.LimitNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDeviceViewNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TimeJoinNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.sink.IdentitySinkNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.AlignedSeriesScanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesScanNode;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class AlignByDeviceOrderByLimitOffsetTest {
+
+ private static final long LIMIT_VALUE = 10;
+
+ @Test
+ public void orderByTimeNoValueFilterTest() {
+ QueryId queryId = new QueryId("test");
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+
+ // 1. order by time, no value filter
+ String sql =
+ String.format(
+ "select * from root.sg.** ORDER BY TIME DESC LIMIT %s align by device", LIMIT_VALUE);
+ Analysis analysis = Util.analyze(sql, context);
+ PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ PlanNode firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ assertTrue(firstFiRoot instanceof IdentitySinkNode);
+ assertEquals(4, firstFiRoot.getChildren().get(0).getChildren().size());
+ PlanNode firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TopKNode);
+ for (PlanNode node : firstFiTopNode.getChildren().get(0).getChildren()) {
+ assertTrue(node instanceof SingleDeviceViewNode);
+ }
+ assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(3) instanceof ExchangeNode);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(0).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(1).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(2).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(3).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ }
+
+ @Test
+ public void orderByTimeWithValueFilterTest() {
+ QueryId queryId = new QueryId("test");
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+
+ // 2. order by time, has value filter
+ // put LIMIT-NODE below of SingleDeviceViewNode
+ String sql =
+ String.format(
+ "select * from root.sg.** where s1>1 ORDER BY TIME DESC LIMIT %s align by device",
+ LIMIT_VALUE);
+ Analysis analysis = Util.analyze(sql, context);
+ PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ PlanNode firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ PlanNode firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TopKNode);
+ for (PlanNode node : firstFiTopNode.getChildren().get(0).getChildren()) {
+ assertTrue(node instanceof SingleDeviceViewNode);
+ assertTrue(node.getChildren().get(0) instanceof LimitNode);
+ assertTrue(node.getChildren().get(0).getChildren().get(0) instanceof FilterNode);
+ }
+ assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(3) instanceof ExchangeNode);
+ assertScanNodeLimitValue(plan.getInstances().get(0).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan.getInstances().get(1).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan.getInstances().get(2).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan.getInstances().get(3).getFragment().getPlanNodeTree(), 0);
+ }
+
+ @Test
+ public void orderByTimeAndExpressionNoValueFilterTest() {
+ QueryId queryId = new QueryId("test");
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+
+ // 3. order by time, expression
+ // need read all data, use DeviceViewNode instead of SingleDeviceViewNode
+ // no LIMIT-NODE below DeviceViewNode
+ // can push down LIMIT value to ScanNode
+ String sql =
+ String.format(
+ "select * from root.sg.** ORDER BY TIME DESC, s1 DESC LIMIT %s align by device",
+ LIMIT_VALUE);
+ Analysis analysis = Util.analyze(sql, context);
+ PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ PlanNode firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ PlanNode firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TopKNode);
+ for (PlanNode node : firstFiTopNode.getChildren().get(0).getChildren()) {
+ assertTrue(node instanceof DeviceViewNode);
+ }
+ assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(3) instanceof ExchangeNode);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(0).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(1).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(2).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(3).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ }
+
+ @Test
+ public void orderByTimeAndExpressionWithValueFilterTest() {
+ QueryId queryId = new QueryId("test");
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+
+ // 4. order by time, expression, has value filter
+ // need read all data, use DeviceViewNode instead of SingleDeviceViewNode
+ // no LIMIT-NODE below DeviceViewNode
+ String sql =
+ String.format(
+ "select * from root.sg.** where s1>1 ORDER BY TIME DESC, s1 DESC LIMIT %s align by device",
+ LIMIT_VALUE);
+ Analysis analysis = Util.analyze(sql, context);
+ PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ DistributedQueryPlan plan4 = planner.planFragments();
+ assertEquals(4, plan4.getInstances().size());
+ PlanNode firstFiRoot = plan4.getInstances().get(0).getFragment().getPlanNodeTree();
+ PlanNode firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TopKNode);
+ for (PlanNode node : firstFiTopNode.getChildren().get(0).getChildren()) {
+ assertTrue(node instanceof DeviceViewNode);
+ assertTrue(node.getChildren().get(0) instanceof LimitNode);
+ assertTrue(node.getChildren().get(0).getChildren().get(0) instanceof FilterNode);
+ }
+ assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(3) instanceof ExchangeNode);
+ assertScanNodeLimitValue(plan4.getInstances().get(0).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan4.getInstances().get(1).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan4.getInstances().get(2).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan4.getInstances().get(3).getFragment().getPlanNodeTree(), 0);
+ }
+
+ @Test
+ public void orderByExpressionTest() {
+ QueryId queryId = new QueryId("test");
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+
+ // 5. only order by expression
+ String sql =
+ String.format(
+ "select * from root.sg.** ORDER BY s1 DESC LIMIT %s align by device", LIMIT_VALUE);
+ Analysis analysis = Util.analyze(sql, context);
+ PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ DistributedQueryPlan plan5 = planner.planFragments();
+ assertEquals(4, plan5.getInstances().size());
+ PlanNode firstFiRoot = plan5.getInstances().get(0).getFragment().getPlanNodeTree();
+ PlanNode firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TopKNode);
+ for (PlanNode node : firstFiTopNode.getChildren().get(0).getChildren()) {
+ assertTrue(node instanceof DeviceViewNode);
+ assertTrue(node.getChildren().get(0) instanceof TimeJoinNode);
+ }
+ assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(3) instanceof ExchangeNode);
+ assertScanNodeLimitValue(plan5.getInstances().get(0).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan5.getInstances().get(1).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan5.getInstances().get(2).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan5.getInstances().get(3).getFragment().getPlanNodeTree(), 0);
+ }
+
+ @Test
+ public void orderByTimeWithOffsetTest() {
+ QueryId queryId = new QueryId("test");
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+
+ // 6. order by time, offset + limit
+ // on top of TOP-K NODE, LIMIT-NODE is necessary
+ String sql =
+ String.format(
+ "select * from root.sg.** ORDER BY time DESC OFFSET %s LIMIT %s align by device",
+ LIMIT_VALUE, LIMIT_VALUE);
+ Analysis analysis = Util.analyze(sql, context);
+ PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ DistributedQueryPlan plan6 = planner.planFragments();
+ assertEquals(4, plan6.getInstances().size());
+ PlanNode firstFiRoot = plan6.getInstances().get(0).getFragment().getPlanNodeTree();
+ PlanNode firstFIFirstNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFIFirstNode instanceof LimitNode);
+ PlanNode firstFiTopNode = ((LimitNode) firstFIFirstNode).getChild().getChildren().get(0);
+ for (PlanNode node : firstFiTopNode.getChildren().get(0).getChildren()) {
+ assertTrue(node instanceof SingleDeviceViewNode);
+ }
+ assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(3) instanceof ExchangeNode);
+ assertScanNodeLimitValue(
+ plan6.getInstances().get(0).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
+ assertScanNodeLimitValue(
+ plan6.getInstances().get(1).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
+ assertScanNodeLimitValue(
+ plan6.getInstances().get(2).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
+ assertScanNodeLimitValue(
+ plan6.getInstances().get(3).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
+ }
+
+ private void assertScanNodeLimitValue(PlanNode root, long limitValue) {
+ for (PlanNode node : root.getChildren()) {
+ if (node instanceof SeriesScanNode) {
+ assertEquals(limitValue, ((SeriesScanNode) node).getLimit());
+ } else if (node instanceof AlignedSeriesScanNode) {
+ assertEquals(limitValue, ((AlignedSeriesScanNode) node).getLimit());
+ } else {
+ assertScanNodeLimitValue(node, limitValue);
+ }
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/MergeSortHeapTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/MergeSortHeapTest.java
new file mode 100644
index 00000000000..4e94b01087f
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/MergeSortHeapTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.utils.datastructure;
+
+import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.MergeSortComparator;
+import org.apache.iotdb.db.queryengine.plan.statement.component.OrderByKey;
+import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
+import org.apache.iotdb.db.queryengine.plan.statement.component.SortItem;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Comparator;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class MergeSortHeapTest {
+
+ @Test
+ public void minHeapTest() {
+ String device0 = "device_0";
+ TsBlockBuilder inputBuilder1 =
+ new TsBlockBuilder(Arrays.asList(TSDataType.TEXT, TSDataType.INT32));
+ inputBuilder1.getTimeColumnBuilder().writeLong(2);
+ inputBuilder1.getColumnBuilder(0).writeBinary(new Binary(device0));
+ inputBuilder1.getColumnBuilder(1).writeInt(20);
+ inputBuilder1.declarePosition();
+ inputBuilder1.getTimeColumnBuilder().writeLong(4);
+ inputBuilder1.getColumnBuilder(0).writeBinary(new Binary(device0));
+ inputBuilder1.getColumnBuilder(1).writeInt(40);
+ inputBuilder1.declarePosition();
+ inputBuilder1.getTimeColumnBuilder().writeLong(5);
+ inputBuilder1.getColumnBuilder(0).writeBinary(new Binary(device0));
+ inputBuilder1.getColumnBuilder(1).appendNull();
+ inputBuilder1.declarePosition();
+ inputBuilder1.getTimeColumnBuilder().writeLong(6);
+ inputBuilder1.getColumnBuilder(0).writeBinary(new Binary(device0));
+ inputBuilder1.getColumnBuilder(1).writeInt(60);
+ inputBuilder1.declarePosition();
+
+ String device1 = "device_1";
+ TsBlockBuilder inputBuilder2 =
+ new TsBlockBuilder(Arrays.asList(TSDataType.TEXT, TSDataType.INT32));
+ inputBuilder2.getTimeColumnBuilder().writeLong(1);
+ inputBuilder2.getColumnBuilder(0).writeBinary(new Binary(device1));
+ inputBuilder2.getColumnBuilder(1).writeInt(10);
+ inputBuilder2.declarePosition();
+ inputBuilder2.getTimeColumnBuilder().writeLong(3);
+ inputBuilder2.getColumnBuilder(0).writeBinary(new Binary(device1));
+ inputBuilder2.getColumnBuilder(1).writeInt(30);
+ inputBuilder2.declarePosition();
+ inputBuilder2.getTimeColumnBuilder().writeLong(7);
+ inputBuilder2.getColumnBuilder(0).writeBinary(new Binary(device1));
+ inputBuilder2.getColumnBuilder(1).appendNull();
+ inputBuilder2.declarePosition();
+ inputBuilder2.getTimeColumnBuilder().writeLong(8);
+ inputBuilder2.getColumnBuilder(0).writeBinary(new Binary(device1));
+ inputBuilder2.getColumnBuilder(1).writeInt(80);
+ inputBuilder2.declarePosition();
+
+ Comparator<SortKey> comparator =
+ MergeSortComparator.getComparator(
+ Arrays.asList(
+ new SortItem(OrderByKey.TIME, Ordering.ASC),
+ new SortItem(OrderByKey.DEVICE, Ordering.ASC)),
+ Arrays.asList(-1, 0),
+ Arrays.asList(TSDataType.INT64, TSDataType.TEXT));
+ MergeSortHeap minHeap = new MergeSortHeap(2, comparator);
+ minHeap.push(new MergeSortKey(inputBuilder1.build(), 0));
+ minHeap.push(new MergeSortKey(inputBuilder2.build(), 0));
+
+ MergeSortKey k = minHeap.poll();
+ assertEquals(1, k.tsBlock.getTimeByIndex(k.rowIndex));
+ assertEquals(device1, k.tsBlock.getColumn(0).getBinary(k.rowIndex).toString());
+ k = minHeap.poll();
+ assertEquals(2, k.tsBlock.getTimeByIndex(k.rowIndex));
+ assertEquals(device0, k.tsBlock.getColumn(0).getBinary(k.rowIndex).toString());
+
+ MergeSortHeap maxHeap = new MergeSortHeap(2, comparator.reversed());
+ TsBlock tsBlock1 = inputBuilder1.build();
+ TsBlock tsBlock2 = inputBuilder2.build();
+ maxHeap.push(new MergeSortKey(tsBlock1, 0));
+ maxHeap.push(new MergeSortKey(tsBlock2, 0));
+
+ assertTrue(comparator.compare(new MergeSortKey(tsBlock2, 1), maxHeap.peek()) > 0);
+
+ k = maxHeap.poll();
+ assertEquals(2, k.tsBlock.getTimeByIndex(k.rowIndex));
+ assertEquals(device0, k.tsBlock.getColumn(0).getBinary(k.rowIndex).toString());
+ k = maxHeap.poll();
+ assertEquals(1, k.tsBlock.getTimeByIndex(k.rowIndex));
+ assertEquals(device1, k.tsBlock.getColumn(0).getBinary(k.rowIndex).toString());
+ }
+}
diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlock.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlock.java
index 677e9aea89d..f21b779439e 100644
--- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlock.java
+++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlock.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.tsfile.read.common.block;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.common.IBatchDataIterator;
import org.apache.iotdb.tsfile.read.common.block.column.Column;
@@ -507,4 +508,49 @@ public class TsBlock {
return columns[0].getPositionCount();
}
+
+ public void update(int updateIdx, TsBlock sourceTsBlock, int sourceIndex) {
+ timeColumn.getTimes()[updateIdx] = sourceTsBlock.getTimeByIndex(sourceIndex);
+ for (int i = 0; i < getValueColumnCount(); i++) {
+ if (sourceTsBlock.getValueColumns()[i].isNull(sourceIndex)) {
+ valueColumns[i].isNull()[updateIdx] = true;
+ continue;
+ }
+ switch (valueColumns[i].getDataType()) {
+ case BOOLEAN:
+ valueColumns[i].isNull()[updateIdx] = false;
+ valueColumns[i].getBooleans()[updateIdx] =
+ sourceTsBlock.getValueColumns()[i].getBoolean(sourceIndex);
+ break;
+ case INT32:
+ valueColumns[i].isNull()[updateIdx] = false;
+ valueColumns[i].getInts()[updateIdx] =
+ sourceTsBlock.getValueColumns()[i].getInt(sourceIndex);
+ break;
+ case INT64:
+ valueColumns[i].isNull()[updateIdx] = false;
+ valueColumns[i].getLongs()[updateIdx] =
+ sourceTsBlock.getValueColumns()[i].getLong(sourceIndex);
+ break;
+ case FLOAT:
+ valueColumns[i].isNull()[updateIdx] = false;
+ valueColumns[i].getFloats()[updateIdx] =
+ sourceTsBlock.getValueColumns()[i].getFloat(sourceIndex);
+ break;
+ case DOUBLE:
+ valueColumns[i].isNull()[updateIdx] = false;
+ valueColumns[i].getDoubles()[updateIdx] =
+ sourceTsBlock.getValueColumns()[i].getDouble(sourceIndex);
+ break;
+ case TEXT:
+ valueColumns[i].isNull()[updateIdx] = false;
+ valueColumns[i].getBinaries()[updateIdx] =
+ sourceTsBlock.getValueColumns()[i].getBinary(sourceIndex);
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ "Unknown datatype: " + valueColumns[i].getDataType());
+ }
+ }
+ }
}