This is an automated email from the ASF dual-hosted git repository. xiangweiwei pushed a commit to branch alignbydevice in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c2e555d3c0e7fb7a3e40edea6d415cab22590e38 Author: Alima777 <[email protected]> AuthorDate: Wed Jul 14 09:28:24 2021 +0800 Optimize align by device logic --- .../apache/iotdb/cluster/metadata/CMManager.java | 36 ----- .../cluster/query/ClusterPhysicalGenerator.java | 13 +- .../cluster/query/reader/ClusterTimeGenerator.java | 14 +- .../apache/iotdb/cluster/server/ClientServer.java | 14 -- .../iotdb/db/qp/logical/crud/QueryOperator.java | 175 +++++++++------------ .../db/qp/physical/crud/AlignByDevicePlan.java | 57 ++----- .../iotdb/db/qp/physical/crud/MeasurementInfo.java | 75 +++++++++ .../iotdb/db/qp/strategy/PhysicalGenerator.java | 21 --- .../db/query/dataset/AlignByDeviceDataSet.java | 20 ++- .../org/apache/iotdb/db/service/TSServiceImpl.java | 14 +- .../iotdb/db/integration/IoTDBAlignByDeviceIT.java | 3 +- .../org/apache/iotdb/db/integration/IoTDBAsIT.java | 2 +- .../db/integration/IoTDBQueryMemoryControlIT.java | 20 +-- 13 files changed, 196 insertions(+), 268 deletions(-) diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java b/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java index f01870c..73d27d2 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java @@ -943,42 +943,6 @@ public class CMManager extends MManager { } /** - * Get the data types of "paths". If "aggregation" is not null, every path will use the - * aggregation. First get types locally and if some paths does not exists, pull them from other - * nodes. - * - * @return the left one of the pair is the column types (considering aggregation), the right one - * of the pair is the measurement types (not considering aggregation) - */ - public Pair<List<TSDataType>, List<TSDataType>> getSeriesTypesByPaths( - List<PartialPath> pathStrs, String aggregation) throws MetadataException { - try { - return getSeriesTypesByPathsLocally(pathStrs, aggregation); - } catch (PathNotExistException e) { - // pull schemas remotely and cache them - metaPuller.pullTimeSeriesSchemas(pathStrs, null); - return getSeriesTypesByPathsLocally(pathStrs, aggregation); - } - } - - private Pair<List<TSDataType>, List<TSDataType>> getSeriesTypesByPathsLocally( - List<PartialPath> pathStrs, String aggregation) throws MetadataException { - List<TSDataType> measurementDataTypes = - SchemaUtils.getSeriesTypesByPaths(pathStrs, (String) null); - // if the aggregation function is null, the type of column in result set - // is equal to the real type of the measurement - if (aggregation == null) { - return new Pair<>(measurementDataTypes, measurementDataTypes); - } else { - // if the aggregation function is not null, - // we should recalculate the type of column in result set - List<TSDataType> columnDataTypes = - SchemaUtils.getAggregatedDataTypes(measurementDataTypes, aggregation); - return new Pair<>(columnDataTypes, measurementDataTypes); - } - } - - /** * Get the data types of "paths". If "aggregations" is not null, each one of it correspond to one * in "paths". First get types locally and if some paths does not exists, pull them from other * nodes. diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPhysicalGenerator.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPhysicalGenerator.java index 3141d18..c237ca3a 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPhysicalGenerator.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPhysicalGenerator.java @@ -44,6 +44,7 @@ import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; import java.net.URL; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Properties; @@ -57,14 +58,12 @@ public class ClusterPhysicalGenerator extends PhysicalGenerator { } @Override - public Pair<List<TSDataType>, List<TSDataType>> getSeriesTypes( - List<PartialPath> paths, String aggregation) throws MetadataException { - return getCMManager().getSeriesTypesByPaths(paths, aggregation); - } - - @Override public List<TSDataType> getSeriesTypes(List<PartialPath> paths) throws MetadataException { - return getCMManager().getSeriesTypesByPaths(paths, null).left; + List<TSDataType> dataTypes = new ArrayList<>(); + for (PartialPath path : paths) { + dataTypes.add(path == null ? null : IoTDB.metaManager.getSeriesType(path)); + } + return dataTypes; } @Override diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterTimeGenerator.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterTimeGenerator.java index e0ea6ab..d1b38e7 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterTimeGenerator.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterTimeGenerator.java @@ -20,7 +20,6 @@ package org.apache.iotdb.cluster.query.reader; import org.apache.iotdb.cluster.exception.CheckConsistencyException; -import org.apache.iotdb.cluster.metadata.CMManager; import org.apache.iotdb.cluster.partition.PartitionGroup; import org.apache.iotdb.cluster.server.member.DataGroupMember; import org.apache.iotdb.cluster.server.member.MetaGroupMember; @@ -48,7 +47,6 @@ import org.apache.iotdb.tsfile.read.reader.IBatchReader; import org.apache.iotdb.tsfile.read.reader.IPointReader; import java.io.IOException; -import java.util.Collections; import java.util.List; public class ClusterTimeGenerator extends ServerTimeGenerator { @@ -113,11 +111,7 @@ public class ClusterTimeGenerator extends ServerTimeGenerator { TSDataType dataType; ManagedSeriesReader mergeReader; try { - dataType = - ((CMManager) IoTDB.metaManager) - .getSeriesTypesByPaths(Collections.singletonList(path), null) - .left - .get(0); + dataType = IoTDB.metaManager.getSeriesType(path); mergeReader = readerFactory.getSeriesReader( path, @@ -179,11 +173,7 @@ public class ClusterTimeGenerator extends ServerTimeGenerator { PartialPath path = (PartialPath) expression.getSeriesPath(); TSDataType dataType; try { - dataType = - ((CMManager) IoTDB.metaManager) - .getSeriesTypesByPaths(Collections.singletonList(path), null) - .left - .get(0); + dataType = IoTDB.metaManager.getSeriesType(path); List<PartitionGroup> partitionGroups = metaGroupMember.routeFilter(null, path); for (PartitionGroup partitionGroup : partitionGroups) { diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/server/ClientServer.java b/cluster/src/main/java/org/apache/iotdb/cluster/server/ClientServer.java index 1200c2d..2dda983 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/server/ClientServer.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/ClientServer.java @@ -262,20 +262,6 @@ public class ClientServer extends TSServiceImpl { } /** - * Get the data types of each path in “paths”. If "aggregation" is not null, all "paths" will use - * this aggregation. - * - * @param paths full timeseries paths - * @param aggregation if not null, it means "paths" all use this aggregation - * @return the data types of "paths" (using the aggregation) - * @throws MetadataException - */ - protected List<TSDataType> getSeriesTypesByString(List<PartialPath> paths, String aggregation) - throws MetadataException { - return ((CMManager) IoTDB.metaManager).getSeriesTypesByPaths(paths, aggregation).left; - } - - /** * Generate and cache a QueryContext using "queryId". In the distributed version, the QueryContext * is a RemoteQueryContext. * diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java index f2668d3..1de305c 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/QueryOperator.java @@ -29,6 +29,7 @@ import org.apache.iotdb.db.qp.logical.Operator; import org.apache.iotdb.db.qp.physical.PhysicalPlan; import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan; import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType; +import org.apache.iotdb.db.qp.physical.crud.MeasurementInfo; import org.apache.iotdb.db.qp.physical.crud.QueryIndexPlan; import org.apache.iotdb.db.qp.physical.crud.QueryPlan; import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan; @@ -40,7 +41,6 @@ import org.apache.iotdb.db.query.expression.unary.TimeSeriesOperand; import org.apache.iotdb.db.service.IoTDB; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.read.expression.IExpression; -import org.apache.iotdb.tsfile.utils.Pair; import java.util.ArrayList; import java.util.HashMap; @@ -50,6 +50,8 @@ import java.util.List; import java.util.Map; import java.util.Set; +import static org.apache.iotdb.db.utils.SchemaUtils.getAggregationType; + public class QueryOperator extends Operator { protected SelectComponent selectComponent; @@ -221,130 +223,89 @@ public class QueryOperator extends Operator { // remove stars in fromPaths and get deviceId with deduplication List<PartialPath> devices = removeStarsInDeviceWithUnique(prefixPaths); List<ResultColumn> resultColumns = selectComponent.getResultColumns(); - List<String> originAggregations = selectComponent.getAggregationFunctions(); + List<String> aggregationFuncs = selectComponent.getAggregationFunctions(); // to record result measurement columns List<String> measurements = new ArrayList<>(); - Map<String, String> measurementAliasMap = new HashMap<>(); - // to check the same measurement of different devices having the same datatype - // record the data type of each column of result set - Map<String, TSDataType> columnDataTypeMap = new HashMap<>(); - Map<String, MeasurementType> measurementTypeMap = new HashMap<>(); - - // to record the real type of the corresponding measurement - Map<String, TSDataType> measurementDataTypeMap = new HashMap<>(); + Map<String, MeasurementInfo> measurementInfoMap = new HashMap<>(); List<PartialPath> paths = new ArrayList<>(); for (int i = 0; i < resultColumns.size(); i++) { // per suffix in SELECT ResultColumn resultColumn = resultColumns.get(i); Expression suffixExpression = resultColumn.getExpression(); - PartialPath suffixPath = - suffixExpression instanceof TimeSeriesOperand - ? ((TimeSeriesOperand) suffixExpression).getPath() - : (((FunctionExpression) suffixExpression).getPaths().get(0)); - - // to record measurements in the loop of a suffix path - Set<String> measurementSetOfGivenSuffix = new LinkedHashSet<>(); + PartialPath suffixPath = getSuffixPathFromExpression(suffixExpression); + String aggregation = aggregationFuncs != null ? aggregationFuncs.get(i) : null; // if const measurement if (suffixPath.getMeasurement().startsWith("'")) { - measurements.add(suffixPath.getMeasurement()); - measurementTypeMap.put(suffixPath.getMeasurement(), MeasurementType.Constant); + String measurementName = suffixPath.getMeasurement(); + measurements.add(measurementName); + measurementInfoMap.put(measurementName, new MeasurementInfo(MeasurementType.Constant)); continue; } - for (PartialPath device : devices) { // per device in FROM after deduplication + // to record measurements in the loop of a suffix path + Set<String> measurementSetOfGivenSuffix = new LinkedHashSet<>(); + for (PartialPath device : devices) { PartialPath fullPath = device.concatPath(suffixPath); try { // remove stars in SELECT to get actual paths List<PartialPath> actualPaths = getMatchedTimeseries(fullPath); - if (resultColumn.hasAlias()) { - if (actualPaths.size() == 1) { - String columnName = actualPaths.get(0).getMeasurement(); - if (originAggregations != null && !originAggregations.isEmpty()) { - measurementAliasMap.put( - originAggregations.get(i) + "(" + columnName + ")", resultColumn.getAlias()); - } else { - measurementAliasMap.put(columnName, resultColumn.getAlias()); - } - } else if (actualPaths.size() >= 2) { - throw new QueryProcessException( - "alias '" - + resultColumn.getAlias() - + "' can only be matched with one time series"); - } + if (resultColumn.hasAlias() && actualPaths.size() >= 2) { + throw new QueryProcessException( + String.format( + "alias %s can only be matched with one time series", resultColumn.getAlias())); } - // for actual non exist path - if (originAggregations != null && actualPaths.isEmpty() && originAggregations.isEmpty()) { - String nonExistMeasurement = fullPath.getMeasurement(); - if (measurementSetOfGivenSuffix.add(nonExistMeasurement) - && measurementTypeMap.get(nonExistMeasurement) != MeasurementType.Exist) { - measurementTypeMap.put(fullPath.getMeasurement(), MeasurementType.NonExist); + if (actualPaths.isEmpty()) { + String nonExistMeasurement = getMeasurementName(fullPath.getMeasurement(), aggregation); + if (measurementSetOfGivenSuffix.add(nonExistMeasurement)) { + measurementInfoMap.putIfAbsent( + nonExistMeasurement, new MeasurementInfo(MeasurementType.NonExist)); } - } + } else { + for (int pathIdx = 0; pathIdx < actualPaths.size(); pathIdx++) { + PartialPath path = actualPaths.get(pathIdx); + String measurementName = getMeasurementName(path.getMeasurement(), aggregation); + // Data type with aggregation function `columnDataTypes` is used for: + // 1. Data type consistency check 2. Header calculation, output result set + // The actual data type of the time series `measurementDataTypes` is used for + // the actual query in the AlignByDeviceDataSet + TSDataType measurementDataType = IoTDB.metaManager.getSeriesType(path); + TSDataType columnDataType = + aggregation == null ? measurementDataType : getAggregationType(aggregation); + MeasurementInfo measurementInfo = + measurementInfoMap.getOrDefault(measurementName, new MeasurementInfo()); + + if (resultColumn.hasAlias()) { + measurementInfo.setMeasurementAlias(resultColumn.getAlias()); + } - // Get data types with and without aggregate functions (actual time series) respectively - // Data type with aggregation function `columnDataTypes` is used for: - // 1. Data type consistency check 2. Header calculation, output result set - // The actual data type of the time series `measurementDataTypes` is used for - // the actual query in the AlignByDeviceDataSet - String aggregation = - originAggregations != null && !originAggregations.isEmpty() - ? originAggregations.get(i) - : null; - - Pair<List<TSDataType>, List<TSDataType>> pair = - generator.getSeriesTypes(actualPaths, aggregation); - List<TSDataType> columnDataTypes = pair.left; - List<TSDataType> measurementDataTypes = pair.right; - for (int pathIdx = 0; pathIdx < actualPaths.size(); pathIdx++) { - PartialPath path = new PartialPath(actualPaths.get(pathIdx).getNodes()); - - // check datatype consistency - // a example of inconsistency: select s0 from root.sg1.d1, root.sg1.d2 align by - // device, - // while root.sg1.d1.s0 is INT32 and root.sg1.d2.s0 is FLOAT. - String measurementChecked; - if (originAggregations != null && !originAggregations.isEmpty()) { - measurementChecked = originAggregations.get(i) + "(" + path.getMeasurement() + ")"; - } else { - measurementChecked = path.getMeasurement(); - } - TSDataType columnDataType = columnDataTypes.get(pathIdx); - if (columnDataTypeMap.containsKey(measurementChecked)) { - if (!columnDataType.equals(columnDataTypeMap.get(measurementChecked))) { - throw new QueryProcessException( - "The data types of the same measurement column should be the same across " - + "devices in ALIGN_BY_DEVICE sql. For more details please refer to the " - + "SQL document."); + // check datatype consistency + // an inconsistent example: select s0 from root.sg1.d1, root.sg1.d2 align by device + // while root.sg1.d1.s0 is INT32 and root.sg1.d2.s0 is FLOAT. + if (measurementInfo.getColumnDataType() != null) { + if (!columnDataType.equals(measurementInfo.getColumnDataType())) { + throw new QueryProcessException( + "The data types of the same measurement column should be the same across devices."); + } + } else { + measurementInfo.setColumnDataType(columnDataType); + measurementInfo.setMeasurementDataType(measurementDataType); } - } else { - columnDataTypeMap.put(measurementChecked, columnDataType); - measurementDataTypeMap.put(measurementChecked, measurementDataTypes.get(pathIdx)); - } - // This step indicates that the measurement exists under the device and is correct, - // First, update measurementSetOfGivenSuffix which is distinct - // Then if this measurement is recognized as NonExist before，update it to Exist - if (measurementSetOfGivenSuffix.add(measurementChecked) - || measurementTypeMap.get(measurementChecked) != MeasurementType.Exist) { - measurementTypeMap.put(measurementChecked, MeasurementType.Exist); + measurementSetOfGivenSuffix.add(measurementName); + measurementInfo.setMeasurementType(MeasurementType.Exist); + measurementInfoMap.put(measurementName, measurementInfo); + // update paths + paths.add(path); } - - // update paths - paths.add(path); } - } catch (MetadataException | QueryProcessException e) { - throw new LogicalOptimizeException( - String.format( - "Error when getting all paths of a full path: %s", fullPath.getFullPath()) - + e.getMessage()); + throw new QueryProcessException(e.getMessage()); } } - // update measurements // Note that in the loop of a suffix path, set is used. // And across the loops of suffix paths, list is used. // e.g. select *,s1 from root.sg.d0, root.sg.d1 @@ -355,20 +316,17 @@ public class QueryOperator extends Operator { } convertSpecialClauseValues(alignByDevicePlan); - // slimit trim on the measurementColumnList + // sLimit trim on the measurementColumnList if (specialClauseComponent.hasSlimit()) { - int seriesSlimit = specialClauseComponent.getSeriesLimit(); + int seriesSLimit = specialClauseComponent.getSeriesLimit(); int seriesOffset = specialClauseComponent.getSeriesOffset(); - measurements = slimitTrimColumn(measurements, seriesSlimit, seriesOffset); + measurements = slimitTrimColumn(measurements, seriesSLimit, seriesOffset); } // assigns to alignByDevicePlan alignByDevicePlan.setMeasurements(measurements); - alignByDevicePlan.setMeasurementAliasMap(measurementAliasMap); + alignByDevicePlan.setMeasurementInfoMap(measurementInfoMap); alignByDevicePlan.setDevices(devices); - alignByDevicePlan.setColumnDataTypeMap(columnDataTypeMap); - alignByDevicePlan.setMeasurementTypeMap(measurementTypeMap); - alignByDevicePlan.setMeasurementDataTypeMap(measurementDataTypeMap); alignByDevicePlan.setPaths(paths); // get deviceToFilterMap @@ -407,6 +365,19 @@ public class QueryOperator extends Operator { return retDevices; } + private PartialPath getSuffixPathFromExpression(Expression expression) { + return expression instanceof TimeSeriesOperand + ? ((TimeSeriesOperand) expression).getPath() + : (((FunctionExpression) expression).getPaths().get(0)); + } + + private String getMeasurementName(String initialMeasurement, String aggregation) { + if (aggregation != null) { + initialMeasurement = aggregation + "(" + initialMeasurement + ")"; + } + return initialMeasurement; + } + private List<String> slimitTrimColumn(List<String> columnList, int seriesLimit, int seriesOffset) throws QueryProcessException { int size = columnList.size(); diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/AlignByDevicePlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/AlignByDevicePlan.java index 55a75eb..fd1bebd 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/AlignByDevicePlan.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/AlignByDevicePlan.java @@ -22,7 +22,6 @@ import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.qp.logical.Operator; import org.apache.iotdb.db.qp.logical.Operator.OperatorType; import org.apache.iotdb.db.qp.strategy.PhysicalGenerator; -import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.read.expression.IExpression; import java.util.List; @@ -30,23 +29,15 @@ import java.util.Map; public class AlignByDevicePlan extends QueryPlan { - private List<String> - measurements; // to record result measurement columns, e.g. temperature, status, speed - private Map<String, String> - measurementAliasMap; // select s1, s2 as speed from root, then s2 -> speed + // to record result measurement columns, e.g. temperature, status, speed + private List<String> measurements; + private Map<String, MeasurementInfo> measurementInfoMap; + // to check data type consistency for the same name sensor of different devices private List<PartialPath> devices; - // to record the datatype of the column in the result set - private Map<String, TSDataType> columnDataTypeMap; private Map<String, IExpression> deviceToFilterMap; - // to record different kinds of measurement - private Map<String, MeasurementType> measurementTypeMap; - - // to record the real type of the measurement - private Map<String, TSDataType> measurementDataTypeMap; private GroupByTimePlan groupByTimePlan; - private FillQueryPlan fillQueryPlan; private AggregationPlan aggregationPlan; @@ -67,14 +58,6 @@ public class AlignByDevicePlan extends QueryPlan { return measurements; } - public void setMeasurementAliasMap(Map<String, String> measurementAliasMap) { - this.measurementAliasMap = measurementAliasMap; - } - - public Map<String, String> getMeasurementAliasMap() { - return measurementAliasMap; - } - public void setDevices(List<PartialPath> devices) { this.devices = devices; } @@ -83,14 +66,6 @@ public class AlignByDevicePlan extends QueryPlan { return devices; } - public void setColumnDataTypeMap(Map<String, TSDataType> columnDataTypeMap) { - this.columnDataTypeMap = columnDataTypeMap; - } - - public Map<String, TSDataType> getColumnDataTypeMap() { - return columnDataTypeMap; - } - public Map<String, IExpression> getDeviceToFilterMap() { return deviceToFilterMap; } @@ -99,22 +74,6 @@ public class AlignByDevicePlan extends QueryPlan { this.deviceToFilterMap = deviceToFilterMap; } - public Map<String, MeasurementType> getMeasurementTypeMap() { - return measurementTypeMap; - } - - public void setMeasurementTypeMap(Map<String, MeasurementType> measurementTypeMap) { - this.measurementTypeMap = measurementTypeMap; - } - - public Map<String, TSDataType> getMeasurementDataTypeMap() { - return measurementDataTypeMap; - } - - public void setMeasurementDataTypeMap(Map<String, TSDataType> measurementDataTypeMap) { - this.measurementDataTypeMap = measurementDataTypeMap; - } - public GroupByTimePlan getGroupByTimePlan() { return groupByTimePlan; } @@ -142,6 +101,14 @@ public class AlignByDevicePlan extends QueryPlan { this.setOperatorType(Operator.OperatorType.AGGREGATION); } + public void setMeasurementInfoMap(Map<String, MeasurementInfo> measurementInfoMap) { + this.measurementInfoMap = measurementInfoMap; + } + + public Map<String, MeasurementInfo> getMeasurementInfoMap() { + return measurementInfoMap; + } + /** * Exist: the measurements which don't belong to NonExist and Constant. NonExist: the measurements * that do not exist in any device, data type is considered as String. The value is considered as diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/MeasurementInfo.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/MeasurementInfo.java new file mode 100644 index 0000000..453ad36 --- /dev/null +++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/MeasurementInfo.java @@ -0,0 +1,75 @@ +/* + * 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.qp.physical.crud; + +import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; + +public class MeasurementInfo { + + public MeasurementInfo() {} + + public MeasurementInfo(MeasurementType measurementType) { + this.measurementType = measurementType; + } + + // select s1, s2 as speed from root, then s2 -> speed + private String measurementAlias; + + // to record different kinds of measurement + private MeasurementType measurementType; + + // to record the real type of the measurement + private TSDataType measurementDataType; + + // to record the datatype of the column in the result set + private TSDataType columnDataType; + + public void setMeasurementAlias(String measurementAlias) { + this.measurementAlias = measurementAlias; + } + + public String getMeasurementAlias() { + return measurementAlias; + } + + public void setMeasurementType(MeasurementType measurementType) { + this.measurementType = measurementType; + } + + public MeasurementType getMeasurementType() { + return measurementType; + } + + public void setMeasurementDataType(TSDataType measurementDataType) { + this.measurementDataType = measurementDataType; + } + + public TSDataType getMeasurementDataType() { + return measurementDataType; + } + + public void setColumnDataType(TSDataType columnDataType) { + this.columnDataType = columnDataType; + } + + public TSDataType getColumnDataType() { + return columnDataType; + } +} diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java index e2be172..242810b 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java @@ -57,27 +57,6 @@ public class PhysicalGenerator { } } - /** - * get types for path list - * - * @return pair.left is the type of column in result set, pair.right is the real type of the - * measurement - */ - public Pair<List<TSDataType>, List<TSDataType>> getSeriesTypes( - List<PartialPath> paths, String aggregation) throws MetadataException { - List<TSDataType> measurementDataTypes = SchemaUtils.getSeriesTypesByPaths(paths, (String) null); - // if the aggregation function is null, the type of column in result set - // is equal to the real type of the measurement - if (aggregation == null) { - return new Pair<>(measurementDataTypes, measurementDataTypes); - } else { - // if the aggregation function is not null, - // we should recalculate the type of column in result set - List<TSDataType> columnDataTypes = SchemaUtils.getSeriesTypesByPaths(paths, aggregation); - return new Pair<>(columnDataTypes, measurementDataTypes); - } - } - public List<TSDataType> getSeriesTypes(List<PartialPath> paths) throws MetadataException { return SchemaUtils.getSeriesTypesByPaths(paths); } diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java index 829f3f1..a34dd0b 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java +++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java @@ -30,6 +30,7 @@ import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan; import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType; import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan; import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan; +import org.apache.iotdb.db.qp.physical.crud.MeasurementInfo; import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan; import org.apache.iotdb.db.query.context.QueryContext; import org.apache.iotdb.db.query.executor.IQueryRouter; @@ -50,6 +51,7 @@ import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Map.Entry; import java.util.Set; /** This QueryDataSet is used for ALIGN_BY_DEVICE query result. */ @@ -63,9 +65,7 @@ public class AlignByDeviceDataSet extends QueryDataSet { private List<String> measurements; private List<PartialPath> devices; private Map<String, IExpression> deviceToFilterMap; - private Map<String, MeasurementType> measurementTypeMap; - // record the real type of the corresponding measurement - private Map<String, TSDataType> measurementDataTypeMap; + private Map<String, MeasurementInfo> measurementInfoMap; private GroupByTimePlan groupByTimePlan; private FillQueryPlan fillQueryPlan; @@ -85,11 +85,10 @@ public class AlignByDeviceDataSet extends QueryDataSet { this.measurements = alignByDevicePlan.getMeasurements(); this.devices = alignByDevicePlan.getDevices(); - this.measurementDataTypeMap = alignByDevicePlan.getMeasurementDataTypeMap(); + this.measurementInfoMap = alignByDevicePlan.getMeasurementInfoMap(); this.queryRouter = queryRouter; this.context = context; this.deviceToFilterMap = alignByDevicePlan.getDeviceToFilterMap(); - this.measurementTypeMap = alignByDevicePlan.getMeasurementTypeMap(); switch (alignByDevicePlan.getOperatorType()) { case GROUP_BY_TIME: @@ -143,7 +142,11 @@ public class AlignByDeviceDataSet extends QueryDataSet { List<PartialPath> executePaths = new ArrayList<>(); List<TSDataType> tsDataTypes = new ArrayList<>(); List<String> executeAggregations = new ArrayList<>(); - for (String column : measurementDataTypeMap.keySet()) { + for (Entry<String, MeasurementInfo> entry : measurementInfoMap.entrySet()) { + if (entry.getValue().getMeasurementType() != MeasurementType.Exist) { + continue; + } + String column = entry.getKey(); String measurement = column; if (dataSetType == DataSetType.GROUPBYTIME || dataSetType == DataSetType.AGGREGATE) { measurement = column.substring(column.indexOf('(') + 1, column.indexOf(')')); @@ -154,7 +157,7 @@ public class AlignByDeviceDataSet extends QueryDataSet { if (measurementOfGivenDevice.contains(measurement)) { executeColumns.add(column); executePaths.add(currentDevice.concatNode(measurement)); - tsDataTypes.add(measurementDataTypeMap.get(column)); + tsDataTypes.add(measurementInfoMap.get(column).getMeasurementDataType()); } } @@ -263,7 +266,8 @@ public class AlignByDeviceDataSet extends QueryDataSet { } for (String measurement : measurements) { - switch (measurementTypeMap.get(measurement)) { + MeasurementInfo measurementInfo = measurementInfoMap.get(measurement); + switch (measurementInfo.getMeasurementType()) { case Exist: if (currentColumnMap.get(measurement) != null) { rowRecord.addField(currentColumnMap.get(measurement)); diff --git a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java index 0616f61..cdad7fe 100644 --- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java @@ -47,7 +47,6 @@ import org.apache.iotdb.db.qp.logical.Operator.OperatorType; import org.apache.iotdb.db.qp.physical.PhysicalPlan; import org.apache.iotdb.db.qp.physical.crud.AggregationPlan; import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan; -import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType; import org.apache.iotdb.db.qp.physical.crud.CreateTemplatePlan; import org.apache.iotdb.db.qp.physical.crud.DeletePlan; import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan; @@ -57,6 +56,7 @@ import org.apache.iotdb.db.qp.physical.crud.InsertRowsOfOneDevicePlan; import org.apache.iotdb.db.qp.physical.crud.InsertRowsPlan; import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan; import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan; +import org.apache.iotdb.db.qp.physical.crud.MeasurementInfo; import org.apache.iotdb.db.qp.physical.crud.QueryPlan; import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan; import org.apache.iotdb.db.qp.physical.crud.SetDeviceTemplatePlan; @@ -1009,23 +1009,23 @@ public class TSServiceImpl implements TSIService.Iface { deduplicatedColumnsType.add(TSDataType.TEXT); // the DEVICE column of ALIGN_BY_DEVICE result Set<String> deduplicatedMeasurements = new LinkedHashSet<>(); - Map<String, TSDataType> measurementDataTypeMap = plan.getColumnDataTypeMap(); + Map<String, MeasurementInfo> measurementInfoMap = plan.getMeasurementInfoMap(); // build column header with constant and non exist column and deduplication List<String> measurements = plan.getMeasurements(); - Map<String, String> measurementAliasMap = plan.getMeasurementAliasMap(); - Map<String, MeasurementType> measurementTypeMap = plan.getMeasurementTypeMap(); for (String measurement : measurements) { + MeasurementInfo measurementInfo = measurementInfoMap.get(measurement); TSDataType type = TSDataType.TEXT; - switch (measurementTypeMap.get(measurement)) { + switch (measurementInfo.getMeasurementType()) { case Exist: - type = measurementDataTypeMap.get(measurement); + type = measurementInfo.getColumnDataType(); break; case NonExist: case Constant: type = TSDataType.TEXT; } - respColumns.add(measurementAliasMap.getOrDefault(measurement, measurement)); + String measurementAlias = measurementInfo.getMeasurementAlias(); + respColumns.add(measurementAlias != null ? measurementAlias : measurement); columnTypes.add(type.toString()); if (!deduplicatedMeasurements.contains(measurement)) { diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAlignByDeviceIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAlignByDeviceIT.java index a57d6d3..f901b97 100644 --- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAlignByDeviceIT.java +++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAlignByDeviceIT.java @@ -698,8 +698,7 @@ public class IoTDBAlignByDeviceIT { Assert.assertTrue( e.getMessage() .contains( - "The data types of the same measurement column should be the same across devices in " - + "ALIGN_BY_DEVICE sql.")); + "The data types of the same measurement column should be the same across devices.")); } } diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAsIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAsIT.java index 2ab39c9..5116c63 100644 --- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAsIT.java +++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAsIT.java @@ -424,7 +424,7 @@ public class IoTDBAsIT { fail(); } catch (Exception e) { Assert.assertTrue( - e.getMessage().contains("alias 'speed' can only be matched with one time series")); + e.getMessage().contains("alias speed can only be matched with one time series")); } } diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBQueryMemoryControlIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBQueryMemoryControlIT.java index 05e3f14..5ba8a8e 100644 --- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBQueryMemoryControlIT.java +++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBQueryMemoryControlIT.java @@ -23,8 +23,8 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.jdbc.Config; -import org.junit.After; -import org.junit.Before; +import org.junit.AfterClass; +import org.junit.BeforeClass; import org.junit.Test; import java.sql.Connection; @@ -39,8 +39,6 @@ import static org.junit.Assert.fail; public class IoTDBQueryMemoryControlIT { - private int defaultMaxQueryDeduplicatedPathNum; - private static final String[] sqls = new String[] { "set storage group to root.ln", @@ -61,10 +59,8 @@ public class IoTDBQueryMemoryControlIT { "create timeseries root.ln.wf03.wt05 with datatype=TEXT,encoding=PLAIN", }; - @Before - public void setUp() throws Exception { - defaultMaxQueryDeduplicatedPathNum = - IoTDBDescriptor.getInstance().getConfig().getMaxQueryDeduplicatedPathNum(); + @BeforeClass + public static void setUp() throws Exception { IoTDBDescriptor.getInstance().getConfig().setMaxQueryDeduplicatedPathNum(10); EnvironmentUtils.envSetUp(); Class.forName(Config.JDBC_DRIVER_NAME); @@ -84,12 +80,10 @@ public class IoTDBQueryMemoryControlIT { } } - @After - public void tearDown() throws Exception { + @AfterClass + public static void tearDown() throws Exception { EnvironmentUtils.cleanEnv(); - IoTDBDescriptor.getInstance() - .getConfig() - .setMaxQueryDeduplicatedPathNum(defaultMaxQueryDeduplicatedPathNum); + IoTDBDescriptor.getInstance().getConfig().setMaxQueryDeduplicatedPathNum(1000); } @Test
