You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/11/21 17:20:58 UTC
[GitHub] [iotdb] ycycse opened a new pull request, #8082: [IOTDB-4572] [IOTDB-3580] support order by in align by device
ycycse opened a new pull request, #8082:
URL: https://github.com/apache/iotdb/pull/8082
This PR add the support of order time and order by device in align by device.
It designs a general PlanOperator called mergeSortOperatpr to support order by time and replace the orginal deviceMergeOperator in order by device.
Design document: https://apache-iotdb.feishu.cn/docx/NNQVdbYTtoOY31x7FvAcDT61nPb
Test document: https://apache-iotdb.feishu.cn/docx/PsbqduXBYod16kxENhAcvevMn9d
UserGuide document: https://apache-iotdb.feishu.cn/docx/PBjMdhKORouRU5xJpCLcRIeYnbe
In response time test(ns), MergeSortOperator shows similar performance comparing with DeviceMergeOperator.
<html xmlns:v="urn:schemas-microsoft-com:vml"
xmlns:o="urn:schemas-microsoft-com:office:office"
xmlns:x="urn:schemas-microsoft-com:office:excel"
xmlns="http://www.w3.org/TR/REC-html40">
<head>
<meta name=ProgId content=Excel.Sheet>
<meta name=Generator content="Microsoft Excel 15">
<link id=Main-File rel=Main-File href="file:///D:/temp/msohtmlclip1/01/clip.htm">
<link rel=File-List href="file:///D:/temp/msohtmlclip1/01/clip_filelist.xml">
</head>
<body link="#0563C1" vlink="#954F72">
| MergeSortOperator | DeviceMergeOperator
-- | -- | --
1 | 173463900 | 184936500
2 | 162205200 | 184653000
3 | 164549900 | 180927200
4 | 165452100 | 172936700
5 | 170124600 | 184039800
6 | 166756500 | 170618800
7 | 167813900 | 180634000
8 | 178506200 | 202682700
9 | 170717500 | 181681900
10 | 175692700 | 179206100
SUM | 1695282500 | 1822316700
</body>
</html>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] JackieTien97 commented on a diff in pull request #8082: [IOTDB-4572] [IOTDB-3580] support order by in align by device
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on code in PR #8082:
URL: https://github.com/apache/iotdb/pull/8082#discussion_r1039384657
##########
integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByWithAlignByDeviceIT.java:
##########
@@ -0,0 +1,1201 @@
+/*
+ * 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.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
Review Comment:
```suggestion
@RunWith(IoTDBTestRunner.class)
@Category({LocalStandaloneIT.class, ClusterIT.class})
```
##########
integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByWithAlignByDeviceIT.java:
##########
@@ -0,0 +1,1201 @@
+/*
+ * 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.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+public class IoTDBOrderByWithAlignByDeviceIT {
+ private static final String[] places =
+ new String[] {
+ "root.weather.London",
+ "root.weather.Edinburgh",
+ "root.weather.Belfast",
+ "root.weather.Birmingham",
+ "root.weather.Liverpool",
+ "root.weather.Derby",
+ "root.weather.Durham",
+ "root.weather.Hereford",
+ "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 numOfTSInDevice = 20;
+ private static final long timeGap = 100L;
+ private 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<>();
+ public static final Map<String, Double[]> deviceToAvgPrecipitation = new HashMap<>();
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().initBeforeClass();
+ insertData();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanAfterClass();
+ }
+
+ private static void insertData() {
+ try (Connection iotDBConnection = EnvFactory.getEnv().getConnection();
+ Statement statement = iotDBConnection.createStatement()) {
+ // create TimeSeries
+ for (String place : places) {
+ String PRE_PRECIPITATION = place + ".precipitation";
+ String PRE_TEMPERATURE = place + ".temperature";
+ String createPrecipitationSql =
+ "CREATE TIMESERIES " + PRE_PRECIPITATION + " WITH DATATYPE=INT64, ENCODING=RLE";
+ String createTemperatureSql =
+ "CREATE TIMESERIES " + PRE_TEMPERATURE + " WITH DATATYPE=DOUBLE, ENCODING=RLE";
+ statement.execute(createPrecipitationSql);
+ statement.execute(createTemperatureSql);
+ }
+ // insert data
+ long start = startTime;
+ double[][] temperatures = new double[places.length][29];
+ long[][] precipitations = new long[places.length][29];
Review Comment:
29 or 20?
##########
integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByWithAlignByDeviceIT.java:
##########
@@ -0,0 +1,1201 @@
+/*
+ * 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.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+public class IoTDBOrderByWithAlignByDeviceIT {
+ private static final String[] places =
+ new String[] {
+ "root.weather.London",
+ "root.weather.Edinburgh",
+ "root.weather.Belfast",
+ "root.weather.Birmingham",
+ "root.weather.Liverpool",
+ "root.weather.Derby",
+ "root.weather.Durham",
+ "root.weather.Hereford",
+ "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 numOfTSInDevice = 20;
+ private static final long timeGap = 100L;
+ private 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<>();
+ public static final Map<String, Double[]> deviceToAvgPrecipitation = new HashMap<>();
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().initBeforeClass();
+ insertData();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanAfterClass();
+ }
+
+ private static void insertData() {
+ try (Connection iotDBConnection = EnvFactory.getEnv().getConnection();
+ Statement statement = iotDBConnection.createStatement()) {
+ // create TimeSeries
+ for (String place : places) {
+ String PRE_PRECIPITATION = place + ".precipitation";
+ String PRE_TEMPERATURE = place + ".temperature";
+ String createPrecipitationSql =
+ "CREATE TIMESERIES " + PRE_PRECIPITATION + " WITH DATATYPE=INT64, ENCODING=RLE";
+ String createTemperatureSql =
+ "CREATE TIMESERIES " + PRE_TEMPERATURE + " WITH DATATYPE=DOUBLE, ENCODING=RLE";
+ statement.execute(createPrecipitationSql);
+ statement.execute(createTemperatureSql);
+ }
+ // insert data
+ long start = startTime;
+ double[][] temperatures = new double[places.length][29];
+ long[][] precipitations = new long[places.length][29];
+ for (int index = 0; index < places.length; index++) {
+ String place = places[index];
+
+ for (int i = 0; i < numOfTSInDevice; i++) {
+ long precipitation = startPrecipitation + place.hashCode() + (start + i * timeGap);
+ double temperature = startTemperature + place.hashCode() + (start + i * timeGap);
+ precipitations[index][(int) ((start - startTime) / timeGap) + i] = precipitation;
Review Comment:
`i` or `((start - startTime) / timeGap) + i`?
##########
server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SingleDeviceViewNode.java:
##########
@@ -0,0 +1,148 @@
+/*
+ * 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.planner.plan.node.process;
+
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+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;
+
+public class SingleDeviceViewNode extends SingleChildProcessNode {
+
+ private final String device;
+ private final List<String> outputColumnNames;
Review Comment:
avoid putting this field in each `SingleDeviceViewNode`
##########
server/src/main/java/org/apache/iotdb/db/utils/MergeSortUtils.java:
##########
@@ -0,0 +1,196 @@
+/*
+ * 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;
+
+import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock.TsBlockSingleColumnIterator;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public abstract class MergeSortUtils {
Review Comment:
change it according to last night's discussion.
##########
server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java:
##########
@@ -639,6 +645,24 @@ public Operator visitNodePathsCount(NodePathsCountNode node, LocalExecutionPlanC
return new NodePathsCountOperator(operatorContext, child);
}
+ @Override
+ public Operator visitSingleDeviceView(
+ SingleDeviceViewNode node, LocalExecutionPlanContext context) {
+ OperatorContext operatorContext =
+ context
+ .getInstanceContext()
+ .addOperatorContext(
+ context.getNextOperatorId(),
+ node.getPlanNodeId(),
+ SingleDeviceViewOperator.class.getSimpleName());
+ Operator child = node.getChild().accept(this, context);
+ List<Integer> deviceColumnIndex = node.getDeviceToMeasurementIndexes();
+ List<TSDataType> outputColumnTypes = getOutputColumnTypes(node, context.getTypeProvider());
Review Comment:
This will cause each `SingleDeviceView` create a new `List<TsDataType>`, you should put `List<TsDataType>` in `LocalExecutionPlanContext` while visiting MergeSortNode and then reuse that in each `SingleDeviceView`
##########
integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByWithAlignByDeviceIT.java:
##########
@@ -0,0 +1,1201 @@
+/*
+ * 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.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
Review Comment:
Without these annotations, your added ITs won't be run.
##########
integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByWithAlignByDeviceIT.java:
##########
@@ -0,0 +1,1201 @@
+/*
+ * 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.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+public class IoTDBOrderByWithAlignByDeviceIT {
+ private static final String[] places =
+ new String[] {
+ "root.weather.London",
+ "root.weather.Edinburgh",
+ "root.weather.Belfast",
+ "root.weather.Birmingham",
+ "root.weather.Liverpool",
+ "root.weather.Derby",
+ "root.weather.Durham",
+ "root.weather.Hereford",
+ "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 numOfTSInDevice = 20;
+ private static final long timeGap = 100L;
+ private 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<>();
+ public static final Map<String, Double[]> deviceToAvgPrecipitation = new HashMap<>();
Review Comment:
check all your changed files about the primitive type array.
##########
server/src/main/java/org/apache/iotdb/db/utils/DeviceMergeUtils.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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;
+
+import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
+import org.apache.iotdb.db.mpp.plan.statement.component.SortItem;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+public class DeviceMergeUtils extends MergeSortUtils {
+
+ String[] startKey;
+ String[] endKey;
+
+ public DeviceMergeUtils(List<SortItem> sortItemList, int childNum) {
+ this.deviceOrdering = sortItemList.get(0).getOrdering();
+ this.timeOrdering = sortItemList.get(1).getOrdering();
+ this.tsBlockCount = childNum;
+ this.startKey = new String[tsBlockCount];
+ this.endKey = new String[tsBlockCount];
+ this.tsBlocksExist = new boolean[tsBlockCount];
+ this.tsBlocks = new TsBlock[tsBlockCount];
+ this.keyValueSelector = new KeyValueSelector(tsBlockCount);
+ }
+
+ @Override
+ public void addTsBlock(TsBlock tsBlock, int index) {
+ this.tsBlocks[index] = tsBlock;
+ startKey[index] = tsBlock.getColumn(0).getBinary(0).toString();
+ endKey[index] = startKey[index];
+ tsBlocksExist[index] = true;
+ }
+
+ @Override
+ public void updateTsBlock(int index, TsBlock tsBlock) {
+ if (tsBlock == null) {
+ tsBlocks[index] = null;
+ tsBlocksExist[index] = false;
+ } else {
Review Comment:
Should never use nullness judgement to do different things. It's better to add a new method like `reset(int index)`.
##########
server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanGraphPrinter.java:
##########
@@ -152,6 +162,21 @@ public List<String> visitDeviceView(DeviceViewNode node, GraphContext context) {
return render(node, boxValue, context);
}
+ @Override
+ public List<String> visitMergeSort(MergeSortNode node, GraphContext context) {
+ List<String> boxValue = new ArrayList<>();
+ boxValue.add(String.format("MergeSort-%s", node.getPlanNodeId().getId()));
+ boxValue.add(String.format("ChildrenCount: %d", node.getChildren().size()));
+ boxValue.add(
+ String.format(
+ "Order: %s %s,%s %s",
+ node.getMergeOrderParameter().getSortItemList().get(0).getSortKey(),
+ node.getMergeOrderParameter().getSortItemList().get(0).getOrdering(),
+ node.getMergeOrderParameter().getSortItemList().get(1).getSortKey(),
+ node.getMergeOrderParameter().getSortItemList().get(1).getOrdering()));
Review Comment:
may not only contain these two, try to print all sort items.
##########
integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByWithAlignByDeviceIT.java:
##########
@@ -0,0 +1,1201 @@
+/*
+ * 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.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+public class IoTDBOrderByWithAlignByDeviceIT {
+ private static final String[] places =
+ new String[] {
+ "root.weather.London",
+ "root.weather.Edinburgh",
+ "root.weather.Belfast",
+ "root.weather.Birmingham",
+ "root.weather.Liverpool",
+ "root.weather.Derby",
+ "root.weather.Durham",
+ "root.weather.Hereford",
+ "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 numOfTSInDevice = 20;
+ private static final long timeGap = 100L;
+ private 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<>();
+ public static final Map<String, Double[]> deviceToAvgPrecipitation = new HashMap<>();
Review Comment:
```suggestion
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<>();
public static final Map<String, double[]> deviceToAvgPrecipitation = new HashMap<>();
```
##########
server/src/main/java/org/apache/iotdb/db/utils/DeviceMergeUtils.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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;
+
+import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
+import org.apache.iotdb.db.mpp.plan.statement.component.SortItem;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+public class DeviceMergeUtils extends MergeSortUtils {
Review Comment:
delete.
##########
server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/MergeSortNode.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.planner.plan.node.process;
+
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.mpp.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;
+import java.util.stream.Collectors;
+
+public class MergeSortNode extends MultiChildProcessNode {
+
+ private final OrderByParameter mergeOrderParameter;
+
+ private final List<String> devices;
Review Comment:
this field should not appear in `MergeSortNode`.
##########
integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByWithAlignByDeviceIT.java:
##########
@@ -0,0 +1,1201 @@
+/*
+ * 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.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+public class IoTDBOrderByWithAlignByDeviceIT {
Review Comment:
Add a google doc for this IT to describe the data it generates, like the following:
```
/**
* This class generates data for test cases in aligned time series scenarios.
*
* <p>You can comprehensively view the generated data in the following online doc:
*
* <p>https://docs.google.com/spreadsheets/d/1kfrSR1_paSd9B1Z0jnPBD3WQIMDslDuNm4R0mpWx9Ms/edit?usp=sharing
*/
public class AlignedWriteUtil {
```
##########
server/src/main/java/org/apache/iotdb/db/utils/DeviceMergeUtils.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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;
+
+import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
+import org.apache.iotdb.db.mpp.plan.statement.component.SortItem;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+public class DeviceMergeUtils extends MergeSortUtils {
+
+ String[] startKey;
+ String[] endKey;
+
+ public DeviceMergeUtils(List<SortItem> sortItemList, int childNum) {
+ this.deviceOrdering = sortItemList.get(0).getOrdering();
+ this.timeOrdering = sortItemList.get(1).getOrdering();
+ this.tsBlockCount = childNum;
+ this.startKey = new String[tsBlockCount];
+ this.endKey = new String[tsBlockCount];
+ this.tsBlocksExist = new boolean[tsBlockCount];
+ this.tsBlocks = new TsBlock[tsBlockCount];
+ this.keyValueSelector = new KeyValueSelector(tsBlockCount);
+ }
+
+ @Override
+ public void addTsBlock(TsBlock tsBlock, int index) {
+ this.tsBlocks[index] = tsBlock;
+ startKey[index] = tsBlock.getColumn(0).getBinary(0).toString();
+ endKey[index] = startKey[index];
+ tsBlocksExist[index] = true;
+ }
+
+ @Override
+ public void updateTsBlock(int index, TsBlock tsBlock) {
+ if (tsBlock == null) {
+ tsBlocks[index] = null;
+ tsBlocksExist[index] = false;
+ } else {
+ tsBlocks[index] = tsBlock;
+ startKey[index] = tsBlocks[index].getColumn(0).getBinary(0).toString();
Review Comment:
each time call toString() is not effective, actually `Binary` class has already implemented `Comparable<Binary>`
##########
server/src/main/java/org/apache/iotdb/db/utils/TimeMergeUtils.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * 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;
+
+import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
+import org.apache.iotdb.db.mpp.plan.statement.component.SortItem;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+
+import java.util.Collections;
+import java.util.List;
+
+public class TimeMergeUtils extends MergeSortUtils {
Review Comment:
delete
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] JackieTien97 merged pull request #8082: [IOTDB-4572] [IOTDB-3580] support order by in align by device
Posted by GitBox <gi...@apache.org>.
JackieTien97 merged PR #8082:
URL: https://github.com/apache/iotdb/pull/8082
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] JackieTien97 commented on a diff in pull request #8082: [IOTDB-4572] [IOTDB-3580] support order by in align by device
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on code in PR #8082:
URL: https://github.com/apache/iotdb/pull/8082#discussion_r1044057253
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/merge/MergeSortComparator.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.execution.operator.process.join.merge;
+
+import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
+import org.apache.iotdb.db.mpp.plan.statement.component.SortItem;
+import org.apache.iotdb.db.mpp.plan.statement.component.SortKey;
+import org.apache.iotdb.db.utils.datastructure.MergeSortKey;
+
+import java.util.Comparator;
+import java.util.List;
+
+public class MergeSortComparator {
+
+ public static final Comparator<MergeSortKey> ASC_TIME_ASC_DEVICE =
+ getTimeComparator(Ordering.ASC, Ordering.ASC);
+ public static final Comparator<MergeSortKey> ASC_TIME_DESC_DEVICE =
+ getTimeComparator(Ordering.ASC, Ordering.DESC);
+ public static final Comparator<MergeSortKey> DESC_TIME_ASC_DEVICE =
+ getTimeComparator(Ordering.DESC, Ordering.ASC);
+ public static final Comparator<MergeSortKey> DESC_TIME_DESC_DEVICE =
+ getTimeComparator(Ordering.DESC, Ordering.DESC);
+
+ public static final Comparator<MergeSortKey> ASC_DEVICE_ASC_TIME =
+ getDeviceComparator(Ordering.ASC, Ordering.ASC);
+ public static final Comparator<MergeSortKey> ASC_DEVICE_DESC_TIME =
+ getDeviceComparator(Ordering.ASC, Ordering.DESC);
+ public static final Comparator<MergeSortKey> DESC_DEVICE_ASC_TIME =
+ getDeviceComparator(Ordering.DESC, Ordering.ASC);
+ public static final Comparator<MergeSortKey> DESC_DEVICE_DESC_TIME =
+ getDeviceComparator(Ordering.DESC, Ordering.DESC);
+
+ private static Comparator<MergeSortKey> getTimeComparator(
+ Ordering timeOrdering, Ordering deviceOrdering) {
+ return (MergeSortKey o1, MergeSortKey o2) ->
+ o1.tsBlock.getTimeByIndex(o1.rowIndex) == o2.tsBlock.getTimeByIndex(o2.rowIndex)
+ ? (deviceOrdering == Ordering.ASC
Review Comment:
we shouldn't do this check in each compare operation, because the ordering won't be changed after compiling.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/merge/MergeSortComparator.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.execution.operator.process.join.merge;
+
+import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
+import org.apache.iotdb.db.mpp.plan.statement.component.SortItem;
+import org.apache.iotdb.db.mpp.plan.statement.component.SortKey;
+import org.apache.iotdb.db.utils.datastructure.MergeSortKey;
+
+import java.util.Comparator;
+import java.util.List;
+
+public class MergeSortComparator {
+
+ public static final Comparator<MergeSortKey> ASC_TIME_ASC_DEVICE =
+ getTimeComparator(Ordering.ASC, Ordering.ASC);
+ public static final Comparator<MergeSortKey> ASC_TIME_DESC_DEVICE =
+ getTimeComparator(Ordering.ASC, Ordering.DESC);
+ public static final Comparator<MergeSortKey> DESC_TIME_ASC_DEVICE =
+ getTimeComparator(Ordering.DESC, Ordering.ASC);
+ public static final Comparator<MergeSortKey> DESC_TIME_DESC_DEVICE =
+ getTimeComparator(Ordering.DESC, Ordering.DESC);
+
+ public static final Comparator<MergeSortKey> ASC_DEVICE_ASC_TIME =
+ getDeviceComparator(Ordering.ASC, Ordering.ASC);
+ public static final Comparator<MergeSortKey> ASC_DEVICE_DESC_TIME =
+ getDeviceComparator(Ordering.ASC, Ordering.DESC);
+ public static final Comparator<MergeSortKey> DESC_DEVICE_ASC_TIME =
+ getDeviceComparator(Ordering.DESC, Ordering.ASC);
+ public static final Comparator<MergeSortKey> DESC_DEVICE_DESC_TIME =
+ getDeviceComparator(Ordering.DESC, Ordering.DESC);
+
+ private static Comparator<MergeSortKey> getTimeComparator(
+ Ordering timeOrdering, Ordering deviceOrdering) {
+ return (MergeSortKey o1, MergeSortKey o2) ->
+ o1.tsBlock.getTimeByIndex(o1.rowIndex) == o2.tsBlock.getTimeByIndex(o2.rowIndex)
+ ? (deviceOrdering == Ordering.ASC
Review Comment:
check all the remaing places in this class by yourself
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/MergeSortOperator.java:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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.execution.operator.process;
+
+import org.apache.iotdb.db.mpp.execution.operator.Operator;
+import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
+import org.apache.iotdb.db.utils.datastructure.MergeSortHeap;
+import org.apache.iotdb.db.utils.datastructure.MergeSortKey;
+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.ColumnBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+
+import static com.google.common.util.concurrent.Futures.successfulAsList;
+
+public class MergeSortOperator implements ProcessOperator {
+
+ private final OperatorContext operatorContext;
+ private final List<Operator> inputOperators;
+ private final List<TSDataType> dataTypes;
+ private final TsBlockBuilder tsBlockBuilder;
+ private final int inputOperatorsCount;
+ private final TsBlock[] inputTsBlocks;
+ private final boolean[] noMoreTsBlocks;
+ private final MergeSortHeap mergeSortHeap;
+ private final Comparator<MergeSortKey> comparator;
+
+ private boolean finished;
+
+ public MergeSortOperator(
+ OperatorContext operatorContext,
+ List<Operator> inputOperators,
+ List<TSDataType> dataTypes,
+ Comparator<MergeSortKey> comparator) {
+ this.operatorContext = operatorContext;
+ this.inputOperators = inputOperators;
+ this.dataTypes = dataTypes;
+ this.inputOperatorsCount = inputOperators.size();
+ this.mergeSortHeap = new MergeSortHeap(inputOperatorsCount, comparator);
+ this.comparator = comparator;
+ this.inputTsBlocks = new TsBlock[inputOperatorsCount];
+ this.noMoreTsBlocks = new boolean[inputOperatorsCount];
+ this.tsBlockBuilder = new TsBlockBuilder(dataTypes);
+ }
+
+ @Override
+ public OperatorContext getOperatorContext() {
+ return operatorContext;
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ List<ListenableFuture<?>> listenableFutures = new ArrayList<>();
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ if (!noMoreTsBlocks[i] && isTsBlockEmpty(i)) {
+ ListenableFuture<?> blocked = inputOperators.get(i).isBlocked();
+ if (!blocked.isDone()) {
+ listenableFutures.add(blocked);
+ }
+ }
+ }
+ return listenableFutures.isEmpty() ? NOT_BLOCKED : successfulAsList(listenableFutures);
+ }
+
+ /** If the tsBlock is null or has no more data in the tsBlock, return true; else return false; */
+ private boolean isTsBlockEmpty(int tsBlockIndex) {
+ return inputTsBlocks[tsBlockIndex] == null
+ || inputTsBlocks[tsBlockIndex].getPositionCount() == 0;
+ }
+
+ @Override
+ public TsBlock next() {
+ // 1. fill consumed up TsBlock
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ if (!noMoreTsBlocks[i] && isTsBlockEmpty(i) && inputOperators.get(i).hasNext()) {
+ inputTsBlocks[i] = inputOperators.get(i).next();
+ if (inputTsBlocks[i] == null || inputTsBlocks[i].isEmpty()) {
+ return null;
+ }
+ mergeSortHeap.push(new MergeSortKey(inputTsBlocks[i], 0, i));
+ }
+ }
+
+ // 2. check if we can directly return the original TsBlock instead of merging way
+ MergeSortKey minMergeSortKey = mergeSortHeap.poll();
+ if (mergeSortHeap.isEmpty()
+ || comparator.compare(
+ new MergeSortKey(
+ minMergeSortKey.tsBlock, minMergeSortKey.tsBlock.getPositionCount() - 1),
+ mergeSortHeap.peek())
+ < 0) {
+ inputTsBlocks[minMergeSortKey.columnIndex] = null;
+ return minMergeSortKey.tsBlock.subTsBlock(minMergeSortKey.rowIndex);
Review Comment:
```suggestion
return minMergeSortKey.columnIndex == 0 ? minMergeSortKey.tsBlock : minMergeSortKey.tsBlock.subTsBlock(minMergeSortKey.rowIndex);
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java:
##########
@@ -207,6 +210,90 @@ public PlanNode visitDeviceMerge(DeviceMergeNode node, NodeGroupContext context)
return processMultiChildNode(node, context);
}
+ @Override
+ public PlanNode visitSingleDeviceView(SingleDeviceViewNode node, NodeGroupContext context) {
+ return processOneChildNode(node, context);
+ }
+
+ @Override
+ public PlanNode visitMergeSort(MergeSortNode node, NodeGroupContext context) {
+ // 1. Group children by dataRegion
+ Map<TRegionReplicaSet, List<PlanNode>> childrenGroupMap = new HashMap<>();
+ for (int i = 0; i < node.getChildren().size(); i++) {
+ PlanNode rawChildNode = node.getChildren().get(i);
+ PlanNode visitedChild = visit(rawChildNode, context);
+ TRegionReplicaSet region = context.getNodeDistribution(visitedChild.getPlanNodeId()).region;
+ if (childrenGroupMap.containsKey(region)) {
+ List<PlanNode> group = childrenGroupMap.get(region);
+ group.add(visitedChild);
+ } else {
+ List<PlanNode> group = new ArrayList<>();
+ group.add(visitedChild);
+ childrenGroupMap.put(region, group);
+ }
Review Comment:
```suggestion
childrenGroupMap.computeIfAbsent(region, k -> new ArrayList<>()).add(visitedChild);
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SingleDeviceViewOperator.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.execution.operator.process;
+
+import org.apache.iotdb.db.mpp.execution.operator.Operator;
+import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
+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.column.BinaryColumnBuilder;
+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.NullColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.List;
+
+/**
+ * The SingleDeviceViewOperator plays a similar role with DeviceViewOperator of adding a device
+ * column to current resultSet.
+ *
+ * <p>Different from DeviceViewOperator which merge the resultSet from different devices,
+ * SingleDeviceViewOperator only focuses on one single device, the goal of it is to add a device
+ * view. It's just a transition and won't change the way data flows.
+ */
+public class SingleDeviceViewOperator implements ProcessOperator {
+
+ private final OperatorContext operatorContext;
+ private final String device;
+ private final Operator deviceOperator;
+ // Used to fill columns and leave null columns which doesn't exist in some devices.
+ private final List<Integer> deviceColumnIndex;
+ // Column dataTypes that includes device column
+ private final List<TSDataType> dataTypes;
+
+ public SingleDeviceViewOperator(
+ OperatorContext operatorContext,
+ String device,
+ Operator deviceOperator,
+ List<Integer> deviceColumnIndex,
+ List<TSDataType> dataTypes) {
+ this.operatorContext = operatorContext;
+ this.device = device;
+ this.deviceOperator = deviceOperator;
+ this.deviceColumnIndex = deviceColumnIndex;
+ this.dataTypes = dataTypes;
+ }
+
+ @Override
+ public OperatorContext getOperatorContext() {
+ return operatorContext;
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ ListenableFuture<?> blocked = deviceOperator.isBlocked();
+ if (!blocked.isDone()) {
+ return blocked;
+ }
+ return NOT_BLOCKED;
+ }
+
+ @Override
+ public TsBlock next() {
+ TsBlock tsBlock = deviceOperator.next();
+ if (tsBlock == null) {
+ return null;
+ }
+ // fill existing columns
+ Column[] newValueColumns = new Column[dataTypes.size()];
+ for (int i = 0; i < deviceColumnIndex.size(); i++) {
+ newValueColumns[deviceColumnIndex.get(i)] = tsBlock.getColumn(i);
+ }
+ // construct device column
+ ColumnBuilder deviceColumnBuilder = new BinaryColumnBuilder(null, 1);
+ deviceColumnBuilder.writeBinary(new Binary(device));
Review Comment:
this `new Binary(device)` should be a constant value, no need to new one in each next call
##########
server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SingleDeviceViewNode.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.planner.plan.node.process;
+
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+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;
+
+public class SingleDeviceViewNode extends SingleChildProcessNode {
+
+ private final String device;
+ private List<String> outputColumnNames;
Review Comment:
add some comments about this field indicating that this field won't be serialized and deserialized, we need to rebuild it using infos from its parent node
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SingleDeviceViewOperator.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.execution.operator.process;
+
+import org.apache.iotdb.db.mpp.execution.operator.Operator;
+import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
+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.column.BinaryColumnBuilder;
+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.NullColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.List;
+
+/**
+ * The SingleDeviceViewOperator plays a similar role with DeviceViewOperator of adding a device
+ * column to current resultSet.
+ *
+ * <p>Different from DeviceViewOperator which merge the resultSet from different devices,
+ * SingleDeviceViewOperator only focuses on one single device, the goal of it is to add a device
+ * view. It's just a transition and won't change the way data flows.
+ */
+public class SingleDeviceViewOperator implements ProcessOperator {
+
+ private final OperatorContext operatorContext;
+ private final String device;
+ private final Operator deviceOperator;
+ // Used to fill columns and leave null columns which doesn't exist in some devices.
+ private final List<Integer> deviceColumnIndex;
+ // Column dataTypes that includes device column
+ private final List<TSDataType> dataTypes;
+
+ public SingleDeviceViewOperator(
+ OperatorContext operatorContext,
+ String device,
+ Operator deviceOperator,
+ List<Integer> deviceColumnIndex,
+ List<TSDataType> dataTypes) {
+ this.operatorContext = operatorContext;
+ this.device = device;
+ this.deviceOperator = deviceOperator;
+ this.deviceColumnIndex = deviceColumnIndex;
+ this.dataTypes = dataTypes;
+ }
+
+ @Override
+ public OperatorContext getOperatorContext() {
+ return operatorContext;
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ ListenableFuture<?> blocked = deviceOperator.isBlocked();
+ if (!blocked.isDone()) {
+ return blocked;
+ }
+ return NOT_BLOCKED;
+ }
+
+ @Override
+ public TsBlock next() {
+ TsBlock tsBlock = deviceOperator.next();
+ if (tsBlock == null) {
+ return null;
+ }
+ // fill existing columns
+ Column[] newValueColumns = new Column[dataTypes.size()];
+ for (int i = 0; i < deviceColumnIndex.size(); i++) {
+ newValueColumns[deviceColumnIndex.get(i)] = tsBlock.getColumn(i);
+ }
+ // construct device column
+ ColumnBuilder deviceColumnBuilder = new BinaryColumnBuilder(null, 1);
+ deviceColumnBuilder.writeBinary(new Binary(device));
Review Comment:
actually, we can store a constant BinaryColumn directly, no need to build each time using ColumnBuilder
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] JackieTien97 commented on a diff in pull request #8082: [IOTDB-4572] [IOTDB-3580] support order by in align by device
Posted by GitBox <gi...@apache.org>.
JackieTien97 commented on code in PR #8082:
URL: https://github.com/apache/iotdb/pull/8082#discussion_r1044291819
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/SingleDeviceViewOperator.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.execution.operator.process;
+
+import org.apache.iotdb.db.mpp.execution.operator.Operator;
+import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
+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.column.BinaryColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.Column;
+import org.apache.iotdb.tsfile.read.common.block.column.NullColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * The SingleDeviceViewOperator plays a similar role with DeviceViewOperator of adding a device
+ * column to current resultSet.
+ *
+ * <p>Different from DeviceViewOperator which merge the resultSet from different devices,
+ * SingleDeviceViewOperator only focuses on one single device, the goal of it is to add a device
+ * view. It's just a transition and won't change the way data flows.
+ */
+public class SingleDeviceViewOperator implements ProcessOperator {
+
+ private final OperatorContext operatorContext;
+ private final Operator deviceOperator;
+ // Used to fill columns and leave null columns which doesn't exist in some devices.
+ private final List<Integer> deviceColumnIndex;
+ // Column dataTypes that includes device column
+ private final List<TSDataType> dataTypes;
+ private final BinaryColumn binaryColumn;
+
+ public SingleDeviceViewOperator(
+ OperatorContext operatorContext,
+ String device,
+ Operator deviceOperator,
+ List<Integer> deviceColumnIndex,
+ List<TSDataType> dataTypes) {
+ this.operatorContext = operatorContext;
+ this.deviceOperator = deviceOperator;
+ this.deviceColumnIndex = deviceColumnIndex;
+ this.dataTypes = dataTypes;
+ this.binaryColumn =
+ new BinaryColumn(1, Optional.of(new boolean[] {false}), new Binary[] {new Binary(device)});
Review Comment:
```suggestion
new BinaryColumn(1, Optional.empty(), new Binary[] {new Binary(device)});
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/merge/MergeSortComparator.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.execution.operator.process.join.merge;
+
+import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
+import org.apache.iotdb.db.mpp.plan.statement.component.SortItem;
+import org.apache.iotdb.db.mpp.plan.statement.component.SortKey;
+import org.apache.iotdb.db.utils.datastructure.MergeSortKey;
+
+import java.util.Comparator;
+import java.util.List;
+
+public class MergeSortComparator {
+
+ public static final Comparator<MergeSortKey> ASC_TIME_ASC_DEVICE =
+ (MergeSortKey o1, MergeSortKey o2) ->
+ o1.tsBlock.getTimeByIndex(o1.rowIndex) == o2.tsBlock.getTimeByIndex(o2.rowIndex)
+ ? o1.tsBlock
+ .getColumn(0)
+ .getBinary(o1.rowIndex)
+ .compareTo(o2.tsBlock.getColumn(0).getBinary(o2.rowIndex))
+ : (int)
+ (o1.tsBlock.getTimeByIndex(o1.rowIndex) - o2.tsBlock.getTimeByIndex(o2.rowIndex));
+ public static final Comparator<MergeSortKey> ASC_TIME_DESC_DEVICE =
+ (MergeSortKey o1, MergeSortKey o2) ->
+ o1.tsBlock.getTimeByIndex(o1.rowIndex) == o2.tsBlock.getTimeByIndex(o2.rowIndex)
+ ? o2.tsBlock
+ .getColumn(0)
+ .getBinary(o2.rowIndex)
+ .compareTo(o1.tsBlock.getColumn(0).getBinary(o1.rowIndex))
+ : (int)
+ (o1.tsBlock.getTimeByIndex(o1.rowIndex) - o2.tsBlock.getTimeByIndex(o2.rowIndex));
+ public static final Comparator<MergeSortKey> DESC_TIME_ASC_DEVICE =
+ (MergeSortKey o1, MergeSortKey o2) ->
+ o1.tsBlock.getTimeByIndex(o1.rowIndex) == o2.tsBlock.getTimeByIndex(o2.rowIndex)
+ ? o1.tsBlock
+ .getColumn(0)
+ .getBinary(o1.rowIndex)
+ .compareTo(o2.tsBlock.getColumn(0).getBinary(o2.rowIndex))
+ : (int)
+ (o2.tsBlock.getTimeByIndex(o2.rowIndex) - o1.tsBlock.getTimeByIndex(o1.rowIndex));
+ public static final Comparator<MergeSortKey> DESC_TIME_DESC_DEVICE =
+ (MergeSortKey o1, MergeSortKey o2) ->
+ o1.tsBlock.getTimeByIndex(o1.rowIndex) == o2.tsBlock.getTimeByIndex(o2.rowIndex)
+ ? o2.tsBlock
+ .getColumn(0)
+ .getBinary(o2.rowIndex)
+ .compareTo(o1.tsBlock.getColumn(0).getBinary(o1.rowIndex))
+ : (int)
+ (o2.tsBlock.getTimeByIndex(o2.rowIndex) - o1.tsBlock.getTimeByIndex(o1.rowIndex));
+
+ public static final Comparator<MergeSortKey> ASC_DEVICE_ASC_TIME =
+ (MergeSortKey o1, MergeSortKey o2) ->
+ o1.tsBlock
+ .getColumn(0)
+ .getBinary(o1.rowIndex)
+ .compareTo(o2.tsBlock.getColumn(0).getBinary(o2.rowIndex))
Review Comment:
reuse this result.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org