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 2022/12/12 06:34:27 UTC
[iotdb] branch master updated: [IOTDB-5172] Fix SingleDeviceViewNode can't get outputColumnTypes when it's isolated
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 a28c982a6b [IOTDB-5172] Fix SingleDeviceViewNode can't get outputColumnTypes when it's isolated
a28c982a6b is described below
commit a28c982a6b522008fab651744964ffc6a8accacc
Author: YangCaiyin <yc...@gmail.com>
AuthorDate: Mon Dec 12 14:34:22 2022 +0800
[IOTDB-5172] Fix SingleDeviceViewNode can't get outputColumnTypes when it's isolated
---
.../IoTDBOrderByWithAlignByDeviceIT.java | 90 ++++++++++++++++++----
.../db/mpp/plan/planner/OperatorTreeGenerator.java | 5 +-
.../planner/distribution/ExchangeNodeAdder.java | 6 +-
.../plan/node/process/SingleDeviceViewNode.java | 59 ++++++++++++--
.../process/SingleDeviceViewNodeSerdeTest.java | 61 +++++++++++++++
5 files changed, 198 insertions(+), 23 deletions(-)
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 e6bbd3633a..dce4c76c66 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
@@ -163,6 +163,11 @@ public class IoTDBOrderByWithAlignByDeviceIT {
deviceToAvgTemperature.put(places[i], aT);
deviceToAvgPrecipitation.put(places[i], aP);
}
+
+ for (String sql : optimizedSQL) {
+ statement.execute(sql);
+ }
+
} catch (Exception e) {
e.printStackTrace();
}
@@ -178,7 +183,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
// ORDER BY DEVICE
@Test
public void orderByDeviceTest1() {
- String sql = "SELECT * FROM root.** ORDER BY DEVICE ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY DEVICE ALIGN BY DEVICE";
Object[] expectedDevice = Arrays.stream(places.clone()).sorted().toArray();
int index = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -219,7 +224,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByDeviceTest2() {
- String sql = "SELECT * FROM root.** ORDER BY DEVICE ASC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY DEVICE ASC ALIGN BY DEVICE";
Object[] expectedDevice = Arrays.stream(places.clone()).sorted().toArray();
int index = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -260,7 +265,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByDeviceTest3() {
- String sql = "SELECT * FROM root.** ORDER BY DEVICE DESC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY DEVICE DESC ALIGN BY DEVICE";
Object[] expectedDevice =
Arrays.stream(places.clone()).sorted(Comparator.reverseOrder()).toArray();
int index = 0;
@@ -304,7 +309,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByTimeTest1() {
- String sql = "SELECT * FROM root.** ORDER BY TIME ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY TIME ALIGN BY DEVICE";
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -342,7 +347,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByTimeTest2() {
- String sql = "SELECT * FROM root.** ORDER BY TIME ASC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY TIME ASC ALIGN BY DEVICE";
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -380,7 +385,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByTimeTest3() {
- String sql = "SELECT * FROM root.** ORDER BY TIME DESC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY TIME DESC ALIGN BY DEVICE";
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -420,7 +425,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByDeviceTimeTest1() {
- String sql = "SELECT * FROM root.** ORDER BY DEVICE ASC,TIME DESC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY DEVICE ASC,TIME DESC ALIGN BY DEVICE";
Object[] expectedDevice = Arrays.stream(places.clone()).sorted().toArray();
int index = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -463,7 +468,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByDeviceTimeTest2() {
- String sql = "SELECT * FROM root.** ORDER BY DEVICE ASC,TIME ASC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY DEVICE ASC,TIME ASC ALIGN BY DEVICE";
Object[] expectedDevice = Arrays.stream(places.clone()).sorted().toArray();
int index = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -504,7 +509,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByDeviceTimeTest3() {
- String sql = "SELECT * FROM root.** ORDER BY DEVICE DESC,TIME DESC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY DEVICE DESC,TIME DESC ALIGN BY DEVICE";
Object[] expectedDevice =
Arrays.stream(places.clone()).sorted(Comparator.reverseOrder()).toArray();
int index = 0;
@@ -548,7 +553,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByDeviceTimeTest4() {
- String sql = "SELECT * FROM root.** ORDER BY DEVICE DESC,TIME ASC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY DEVICE DESC,TIME ASC ALIGN BY DEVICE";
Object[] expectedDevice =
Arrays.stream(places.clone()).sorted(Comparator.reverseOrder()).toArray();
int index = 0;
@@ -592,7 +597,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByTimeDeviceTest1() {
- String sql = "SELECT * FROM root.** ORDER BY TIME ASC,DEVICE DESC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY TIME ASC,DEVICE DESC ALIGN BY DEVICE";
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -630,7 +635,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByTimeDeviceTest2() {
- String sql = "SELECT * FROM root.** ORDER BY TIME ASC,DEVICE ASC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY TIME ASC,DEVICE ASC ALIGN BY DEVICE";
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -668,7 +673,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByTimeDeviceTest3() {
- String sql = "SELECT * FROM root.** ORDER BY TIME DESC,DEVICE DESC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY TIME DESC,DEVICE DESC ALIGN BY DEVICE";
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -706,7 +711,7 @@ public class IoTDBOrderByWithAlignByDeviceIT {
@Test
public void orderByTimeDeviceTest4() {
- String sql = "SELECT * FROM root.** ORDER BY TIME DESC,DEVICE ASC ALIGN BY DEVICE";
+ String sql = "SELECT * FROM root.weather.** ORDER BY TIME DESC,DEVICE ASC ALIGN BY DEVICE";
int total = 0;
try (Connection connection = EnvFactory.getEnv().getConnection();
Statement statement = connection.createStatement()) {
@@ -1175,4 +1180,61 @@ public class IoTDBOrderByWithAlignByDeviceIT {
fail(e.getMessage());
}
}
+
+ // test the optimized plan
+ public static String[] optimizedSQL =
+ new String[] {
+ "insert into root.ln.wf01.wt01(timestamp,temperature,status) values(2017-11-01T00:00:00.000+08:00,25.96,true)",
+ "insert into root.ln.wf01.wt01(timestamp,temperature,status) values(2017-11-01T00:01:00.000+08:00,24.36,true)",
+ "insert into root.ln.wf02.wt02(timestamp,status,hardware) values(1970-01-01T08:00:00.001+08:00,true,'v1')",
+ "insert into root.ln.wf02.wt02(timestamp,status,hardware) values(1970-01-01T08:00:00.002+08:00,false,'v2')",
+ "insert into root.ln.wf02.wt02(timestamp,status,hardware) values(2017-11-01T00:00:00.000+08:00,true,'v2')",
+ "insert into root.ln.wf02.wt02(timestamp,status,hardware) values(2017-11-01T00:01:00.000+08:00,true,'v2')"
+ };
+
+ String[] res =
+ new String[] {
+ "v1,true,null",
+ "v2,false,null",
+ "null,true,25.96",
+ "v2,true,null",
+ "null,true,24.36",
+ "v2,true,null"
+ };
+
+ @Test
+ public void optimizedPlanTest() {
+ String sql = "SELECT * FROM root.ln.** ORDER BY TIME DESC,DEVICE DESC ALIGN BY DEVICE";
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+
+ try (ResultSet resultSet = statement.executeQuery(sql)) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ checkHeader(resultSetMetaData, "Time,Device,hardware,status,temperature");
+ long lastTimeStamp = Long.MAX_VALUE;
+ String lastDevice = "";
+ int i = 0;
+ while (resultSet.next()) {
+ long actualTimeStamp = resultSet.getLong(1);
+ assertTrue(actualTimeStamp <= lastTimeStamp);
+ String actualDevice = resultSet.getString(2);
+ if (!lastDevice.equals("") && actualTimeStamp == lastTimeStamp) {
+ assertTrue(actualDevice.compareTo(lastDevice) <= 0);
+ }
+ lastDevice = actualDevice;
+ lastTimeStamp = actualTimeStamp;
+ String actualHardware = resultSet.getString(3);
+ boolean actualStatus = resultSet.getBoolean(4);
+ double actualTemperature = resultSet.getDouble(5);
+ String stringBuilder = actualHardware + "," + actualStatus + "," + actualTemperature;
+ assertEquals(stringBuilder, res[i]);
+ i++;
+ }
+ assertEquals(i, 6);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
index 2af0bd0708..c61cb93e26 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
@@ -663,7 +663,10 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
SingleDeviceViewOperator.class.getSimpleName());
Operator child = node.getChild().accept(this, context);
List<Integer> deviceColumnIndex = node.getDeviceToMeasurementIndexes();
- List<TSDataType> outputColumnTypes = context.getCachedDataTypes();
+ List<TSDataType> outputColumnTypes =
+ node.isCacheOutputColumnNames()
+ ? getOutputColumnTypes(node, context.getTypeProvider())
+ : context.getCachedDataTypes();
if (outputColumnTypes == null || outputColumnTypes.size() == 0) {
throw new IllegalStateException("OutputColumTypes should not be null/empty");
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java
index 82b56565bb..cc24dd0c97 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java
@@ -230,7 +230,11 @@ public class ExchangeNodeAdder extends PlanVisitor<PlanNode, NodeGroupContext> {
List<PlanNode> mergeSortNodeList = new ArrayList<>();
for (List<PlanNode> group : childrenGroupMap.values()) {
if (group.size() == 1) {
- mergeSortNodeList.add(group.get(0));
+ PlanNode planNode = group.get(0);
+ if (planNode instanceof SingleDeviceViewNode) {
+ ((SingleDeviceViewNode) planNode).setCacheOutputColumnNames(true);
+ }
+ mergeSortNodeList.add(planNode);
continue;
}
MergeSortNode mergeSortNode =
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SingleDeviceViewNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SingleDeviceViewNode.java
index caefda0455..b5c2fecf7d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SingleDeviceViewNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SingleDeviceViewNode.java
@@ -36,10 +36,12 @@ public class SingleDeviceViewNode extends SingleChildProcessNode {
private final String device;
// To reduce memory cost, SingleDeviceViewNode doesn't serialize and deserialize
- // outputColumnNames.
- // It just rebuilds using the infos from parent node.
- private List<String> outputColumnNames;
+ // outputColumnNames.It just rebuilds using the infos from parent node.
+ private final List<String> outputColumnNames;
private final List<Integer> deviceToMeasurementIndexes;
+ // For some isolated SingleDeviceViewNodes without parent node, they need to cache
+ // the outputColumnNames themselves.
+ private boolean cacheOutputColumnNames = false;
public SingleDeviceViewNode(
PlanNodeId id,
@@ -53,27 +55,45 @@ public class SingleDeviceViewNode extends SingleChildProcessNode {
}
public SingleDeviceViewNode(
- PlanNodeId id, String device, List<Integer> deviceToMeasurementIndexes) {
+ PlanNodeId id,
+ boolean cacheOutputColumnNames,
+ List<String> outputColumnNames,
+ String device,
+ List<Integer> deviceToMeasurementIndexes) {
super(id);
this.device = device;
+ this.cacheOutputColumnNames = cacheOutputColumnNames;
+ this.outputColumnNames = outputColumnNames;
this.deviceToMeasurementIndexes = deviceToMeasurementIndexes;
}
@Override
public PlanNode clone() {
return new SingleDeviceViewNode(
- getPlanNodeId(), outputColumnNames, device, deviceToMeasurementIndexes);
+ getPlanNodeId(),
+ cacheOutputColumnNames,
+ outputColumnNames,
+ device,
+ deviceToMeasurementIndexes);
}
@Override
public List<String> getOutputColumnNames() {
- return outputColumnNames;
+ return this.outputColumnNames;
+ }
+
+ public void setCacheOutputColumnNames(boolean cacheOutputColumnNames) {
+ this.cacheOutputColumnNames = cacheOutputColumnNames;
}
public String getDevice() {
return device;
}
+ public boolean isCacheOutputColumnNames() {
+ return cacheOutputColumnNames;
+ }
+
public List<Integer> getDeviceToMeasurementIndexes() {
return deviceToMeasurementIndexes;
}
@@ -87,33 +107,58 @@ public class SingleDeviceViewNode extends SingleChildProcessNode {
protected void serializeAttributes(ByteBuffer byteBuffer) {
PlanNodeType.SINGLE_DEVICE_VIEW.serialize(byteBuffer);
ReadWriteIOUtils.write(device, byteBuffer);
+ ReadWriteIOUtils.write(cacheOutputColumnNames, byteBuffer);
ReadWriteIOUtils.write(deviceToMeasurementIndexes.size(), byteBuffer);
for (Integer index : deviceToMeasurementIndexes) {
ReadWriteIOUtils.write(index, byteBuffer);
}
+ if (cacheOutputColumnNames) {
+ ReadWriteIOUtils.write(outputColumnNames.size(), byteBuffer);
+ for (String column : outputColumnNames) {
+ ReadWriteIOUtils.write(column, byteBuffer);
+ }
+ }
}
@Override
protected void serializeAttributes(DataOutputStream stream) throws IOException {
PlanNodeType.SINGLE_DEVICE_VIEW.serialize(stream);
ReadWriteIOUtils.write(device, stream);
+ ReadWriteIOUtils.write(cacheOutputColumnNames, stream);
ReadWriteIOUtils.write(deviceToMeasurementIndexes.size(), stream);
for (Integer index : deviceToMeasurementIndexes) {
ReadWriteIOUtils.write(index, stream);
}
+ if (cacheOutputColumnNames) {
+ ReadWriteIOUtils.write(outputColumnNames.size(), stream);
+ for (String column : outputColumnNames) {
+ ReadWriteIOUtils.write(column, stream);
+ }
+ }
}
public static SingleDeviceViewNode deserialize(ByteBuffer byteBuffer) {
String device = ReadWriteIOUtils.readString(byteBuffer);
+ boolean cacheOutputColumnNames = ReadWriteIOUtils.readBool(byteBuffer);
int listSize = ReadWriteIOUtils.readInt(byteBuffer);
List<Integer> deviceToMeasurementIndexes = new ArrayList<>(listSize);
while (listSize > 0) {
deviceToMeasurementIndexes.add(ReadWriteIOUtils.readInt(byteBuffer));
listSize--;
}
+ List<String> outputColumnNames = null;
+ if (cacheOutputColumnNames) {
+ int columnSize = ReadWriteIOUtils.readInt(byteBuffer);
+ outputColumnNames = new ArrayList<>();
+ while (columnSize > 0) {
+ outputColumnNames.add(ReadWriteIOUtils.readString(byteBuffer));
+ columnSize--;
+ }
+ }
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
- return new SingleDeviceViewNode(planNodeId, device, deviceToMeasurementIndexes);
+ return new SingleDeviceViewNode(
+ planNodeId, cacheOutputColumnNames, outputColumnNames, device, deviceToMeasurementIndexes);
}
@Override
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/process/SingleDeviceViewNodeSerdeTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/process/SingleDeviceViewNodeSerdeTest.java
new file mode 100644
index 0000000000..76415259e1
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/process/SingleDeviceViewNodeSerdeTest.java
@@ -0,0 +1,61 @@
+/*
+ * 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.mpp.plan.plan.node.process;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.db.mpp.plan.plan.node.PlanNodeDeserializeHelper;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.SingleDeviceViewNode;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
+
+public class SingleDeviceViewNodeSerdeTest {
+
+ private static final List<String> outputColumnNames = Arrays.asList("Time", "Device", "value");
+
+ @Test
+ public void testCacheOutputColumnNames() throws IllegalPathException {
+ SingleDeviceViewNode singleDeviceViewNode =
+ new SingleDeviceViewNode(
+ new PlanNodeId("TestSingleDeviceViewNode"),
+ outputColumnNames,
+ "TestDevice",
+ Arrays.asList(1, 2));
+
+ // when cache is false
+ ByteBuffer byteBufferWithoutCache = ByteBuffer.allocate(2048);
+ singleDeviceViewNode.serialize(byteBufferWithoutCache);
+ byteBufferWithoutCache.flip();
+ Assert.assertNull(
+ PlanNodeDeserializeHelper.deserialize(byteBufferWithoutCache).getOutputColumnNames());
+
+ // when cache is true
+ singleDeviceViewNode.setCacheOutputColumnNames(true);
+ ByteBuffer byteBufferWithCache = ByteBuffer.allocate(2048);
+ singleDeviceViewNode.serialize(byteBufferWithCache);
+ byteBufferWithCache.flip();
+ Assert.assertEquals(
+ PlanNodeDeserializeHelper.deserialize(byteBufferWithCache), singleDeviceViewNode);
+ }
+}