You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2021/07/14 02:25:25 UTC
[iotdb] branch alignbydevice updated: remove meta data methods in
Cluster
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
The following commit(s) were added to refs/heads/alignbydevice by this push:
new 902faba remove meta data methods in Cluster
902faba is described below
commit 902fabaa275129293cd8a4663d2aaad21d00ca0f
Author: Alima777 <wx...@gmail.com>
AuthorDate: Wed Jul 14 10:24:46 2021 +0800
remove meta data methods in Cluster
---
.../apache/iotdb/cluster/metadata/CMManager.java | 56 ----------------------
.../apache/iotdb/cluster/server/ClientServer.java | 22 ---------
.../iotdb/db/qp/logical/crud/QueryOperator.java | 32 ++++++-------
.../iotdb/db/qp/physical/crud/MeasurementInfo.java | 2 +-
.../org/apache/iotdb/db/service/TSServiceImpl.java | 7 +--
.../org/apache/iotdb/db/utils/SchemaUtils.java | 18 -------
6 files changed, 17 insertions(+), 120 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 73d27d2..4d9d258 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
@@ -66,7 +66,6 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.dataset.ShowDevicesResult;
import org.apache.iotdb.db.query.dataset.ShowTimeSeriesResult;
import org.apache.iotdb.db.service.IoTDB;
-import org.apache.iotdb.db.utils.SchemaUtils;
import org.apache.iotdb.db.utils.TypeInferenceUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
@@ -943,61 +942,6 @@ public class CMManager extends MManager {
}
/**
- * 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.
- *
- * @param aggregations nullable, when not null, correspond to "paths" one-to-one.
- * @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>> getSeriesTypesByPath(
- List<PartialPath> paths, List<String> aggregations) throws MetadataException {
- try {
- return getSeriesTypesByPathLocally(paths, aggregations);
- } catch (PathNotExistException e) {
- return getSeriesTypesByPathRemotely(paths, aggregations);
- }
- }
-
- /**
- * get data types of the given paths considering the aggregations from CMManger.
- *
- * @param aggregations nullable, when not null, correspond to "paths" one-to-one.
- * @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)
- */
- private Pair<List<TSDataType>, List<TSDataType>> getSeriesTypesByPathLocally(
- List<PartialPath> paths, List<String> aggregations) throws MetadataException {
- List<TSDataType> measurementDataTypes = SchemaUtils.getSeriesTypesByPaths(paths);
- // if the aggregation function is null, the type of column in result set
- // is equal to the real type of the measurement
- if (aggregations == 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, aggregations);
- return new Pair<>(columnDataTypes, measurementDataTypes);
- }
- }
-
- /**
- * pull schemas from remote nodes and cache them, then get data types of the given paths
- * considering the aggregations from CMManger.
- *
- * @param aggregations nullable, when not null, correspond to "paths" one-to-one.
- * @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)
- */
- private Pair<List<TSDataType>, List<TSDataType>> getSeriesTypesByPathRemotely(
- List<PartialPath> paths, List<String> aggregations) throws MetadataException {
- // pull schemas remotely and cache them
- metaPuller.pullTimeSeriesSchemas(paths, null);
- return getSeriesTypesByPathLocally(paths, aggregations);
- }
-
- /**
* Get all devices after removing wildcards in the path
*
* @param originPath a path potentially with wildcard
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 2dda983..ad0f443 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
@@ -24,7 +24,6 @@ import org.apache.iotdb.cluster.client.sync.SyncDataClient;
import org.apache.iotdb.cluster.config.ClusterConfig;
import org.apache.iotdb.cluster.config.ClusterDescriptor;
import org.apache.iotdb.cluster.coordinator.Coordinator;
-import org.apache.iotdb.cluster.metadata.CMManager;
import org.apache.iotdb.cluster.query.ClusterPlanExecutor;
import org.apache.iotdb.cluster.query.ClusterPlanner;
import org.apache.iotdb.cluster.query.RemoteQueryContext;
@@ -34,12 +33,9 @@ import org.apache.iotdb.cluster.server.handlers.caller.GenericHandler;
import org.apache.iotdb.cluster.server.member.MetaGroupMember;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StorageEngineException;
-import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.query.context.QueryContext;
-import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.service.TSServiceImpl;
import org.apache.iotdb.db.utils.CommonUtils;
import org.apache.iotdb.rpc.RpcTransportFactory;
@@ -47,7 +43,6 @@ import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.TSIService.Processor;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
@@ -67,7 +62,6 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.InetSocketAddress;
-import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
@@ -246,22 +240,6 @@ public class ClientServer extends TSServiceImpl {
}
/**
- * Get the data types of each path in “paths”. If "aggregations" is not null, then it should be
- * corresponding to "paths" one to one and the data type will be the type of the aggregation over
- * the corresponding path.
- *
- * @param paths full timeseries paths
- * @param aggregations if not null, it should be the same size as "paths"
- * @return the data types of "paths" (using the aggregations)
- * @throws MetadataException
- */
- @Override
- protected List<TSDataType> getSeriesTypesByPaths(
- List<PartialPath> paths, List<String> aggregations) throws MetadataException {
- return ((CMManager) IoTDB.metaManager).getSeriesTypesByPath(paths, aggregations).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 1de305c..0813832 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
@@ -264,16 +264,11 @@ public class QueryOperator extends Operator {
nonExistMeasurement, new MeasurementInfo(MeasurementType.NonExist));
}
} else {
- for (int pathIdx = 0; pathIdx < actualPaths.size(); pathIdx++) {
- PartialPath path = actualPaths.get(pathIdx);
+ for (PartialPath path : actualPaths) {
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);
+ TSDataType columnDataType = getAggregationType(aggregation);
+ columnDataType = columnDataType == null ? measurementDataType : columnDataType;
MeasurementInfo measurementInfo =
measurementInfoMap.getOrDefault(measurementName, new MeasurementInfo());
@@ -315,21 +310,13 @@ public class QueryOperator extends Operator {
measurements.addAll(measurementSetOfGivenSuffix);
}
- convertSpecialClauseValues(alignByDevicePlan);
- // sLimit trim on the measurementColumnList
- if (specialClauseComponent.hasSlimit()) {
- int seriesSLimit = specialClauseComponent.getSeriesLimit();
- int seriesOffset = specialClauseComponent.getSeriesOffset();
- measurements = slimitTrimColumn(measurements, seriesSLimit, seriesOffset);
- }
-
+ convertSpecialClauseValues(alignByDevicePlan, measurements);
// assigns to alignByDevicePlan
alignByDevicePlan.setMeasurements(measurements);
alignByDevicePlan.setMeasurementInfoMap(measurementInfoMap);
alignByDevicePlan.setDevices(devices);
alignByDevicePlan.setPaths(paths);
- // get deviceToFilterMap
if (whereComponent != null) {
alignByDevicePlan.setDeviceToFilterMap(
concatFilterByDevice(generator, devices, whereComponent.getFilterOperator()));
@@ -349,6 +336,17 @@ public class QueryOperator extends Operator {
}
}
+ private void convertSpecialClauseValues(QueryPlan queryPlan, List<String> measurements)
+ throws QueryProcessException {
+ convertSpecialClauseValues(queryPlan);
+ // sLimit trim on the measurementColumnList
+ if (specialClauseComponent.hasSlimit()) {
+ int seriesSLimit = specialClauseComponent.getSeriesLimit();
+ int seriesOffset = specialClauseComponent.getSeriesOffset();
+ slimitTrimColumn(measurements, seriesSLimit, seriesOffset);
+ }
+ }
+
private List<PartialPath> removeStarsInDeviceWithUnique(List<PartialPath> paths)
throws LogicalOptimizeException {
List<PartialPath> retDevices;
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
index 453ad36..7c6dd17 100644
--- 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
@@ -35,7 +35,7 @@ public class MeasurementInfo {
// to record different kinds of measurement
private MeasurementType measurementType;
- // to record the real type of the measurement
+ // to record the real type of the measurement, used for actual query
private TSDataType measurementDataType;
// to record the datatype of the column in the result set
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 cdad7fe..8089042 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
@@ -979,7 +979,7 @@ public class TSServiceImpl implements TSIService.Iface {
for (ResultColumn resultColumn : resultColumns) {
respColumns.add(resultColumn.getResultColumnName());
}
- seriesTypes = getSeriesTypesByPaths(paths, aggregations);
+ seriesTypes = SchemaUtils.getSeriesTypesByPaths(paths, aggregations);
break;
case UDTF:
seriesTypes = new ArrayList<>();
@@ -2007,11 +2007,6 @@ public class TSServiceImpl implements TSIService.Iface {
: RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR);
}
- protected List<TSDataType> getSeriesTypesByPaths(
- List<PartialPath> paths, List<String> aggregations) throws MetadataException {
- return SchemaUtils.getSeriesTypesByPaths(paths, aggregations);
- }
-
protected TSDataType getSeriesTypeByPath(PartialPath path) throws MetadataException {
return SchemaUtils.getSeriesTypeByPath(path);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
index 828c83b..e071504 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
@@ -134,24 +134,6 @@ public class SchemaUtils {
}
/**
- * @param paths time series paths
- * @param aggregation aggregation function, may be null
- * @return The data type of aggregation or (data type of paths if aggregation is null)
- */
- public static List<TSDataType> getSeriesTypesByPaths(
- Collection<PartialPath> paths, String aggregation) throws MetadataException {
- TSDataType dataType = getAggregationType(aggregation);
- if (dataType != null) {
- return Collections.nCopies(paths.size(), dataType);
- }
- List<TSDataType> dataTypes = new ArrayList<>();
- for (PartialPath path : paths) {
- dataTypes.add(path == null ? null : IoTDB.metaManager.getSeriesType(path));
- }
- return dataTypes;
- }
-
- /**
* If the datatype of 'aggregation' depends on 'measurementDataType' (min_value, max_value),
* return 'measurementDataType' directly, or return a list whose elements are all the datatype of
* 'aggregation' and its length is the same as 'measurementDataType'.