| /* |
| * 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.cluster.query.reader; |
| |
| import org.apache.iotdb.cluster.client.async.AsyncDataClient; |
| import org.apache.iotdb.cluster.client.sync.SyncClientAdaptor; |
| import org.apache.iotdb.cluster.client.sync.SyncDataClient; |
| import org.apache.iotdb.cluster.config.ClusterDescriptor; |
| import org.apache.iotdb.cluster.exception.CheckConsistencyException; |
| import org.apache.iotdb.cluster.exception.EmptyIntervalException; |
| import org.apache.iotdb.cluster.exception.RequestTimeOutException; |
| import org.apache.iotdb.cluster.partition.PartitionGroup; |
| import org.apache.iotdb.cluster.partition.slot.SlotPartitionTable; |
| import org.apache.iotdb.cluster.query.LocalQueryExecutor; |
| import org.apache.iotdb.cluster.query.RemoteQueryContext; |
| import org.apache.iotdb.cluster.query.filter.SlotTsFileFilter; |
| import org.apache.iotdb.cluster.query.groupby.RemoteGroupByExecutor; |
| import org.apache.iotdb.cluster.query.manage.QueryCoordinator; |
| import org.apache.iotdb.cluster.query.reader.mult.AbstractMultPointReader; |
| import org.apache.iotdb.cluster.query.reader.mult.MultBatchReader; |
| import org.apache.iotdb.cluster.query.reader.mult.MultDataSourceInfo; |
| import org.apache.iotdb.cluster.query.reader.mult.MultEmptyReader; |
| import org.apache.iotdb.cluster.query.reader.mult.MultSeriesRawDataPointReader; |
| import org.apache.iotdb.cluster.query.reader.mult.RemoteMultSeriesReader; |
| import org.apache.iotdb.cluster.rpc.thrift.GroupByRequest; |
| import org.apache.iotdb.cluster.rpc.thrift.MultSeriesQueryRequest; |
| import org.apache.iotdb.cluster.rpc.thrift.Node; |
| import org.apache.iotdb.cluster.rpc.thrift.SingleSeriesQueryRequest; |
| import org.apache.iotdb.cluster.server.RaftServer; |
| import org.apache.iotdb.cluster.server.member.DataGroupMember; |
| import org.apache.iotdb.cluster.server.member.MetaGroupMember; |
| import org.apache.iotdb.cluster.utils.ClusterQueryUtils; |
| import org.apache.iotdb.db.engine.querycontext.QueryDataSource; |
| import org.apache.iotdb.db.exception.StorageEngineException; |
| import org.apache.iotdb.db.exception.query.QueryProcessException; |
| import org.apache.iotdb.db.metadata.PartialPath; |
| import org.apache.iotdb.db.metadata.VectorPartialPath; |
| import org.apache.iotdb.db.query.aggregation.AggregationType; |
| import org.apache.iotdb.db.query.context.QueryContext; |
| import org.apache.iotdb.db.query.control.QueryResourceManager; |
| import org.apache.iotdb.db.query.dataset.groupby.GroupByExecutor; |
| import org.apache.iotdb.db.query.factory.AggregateResultFactory; |
| import org.apache.iotdb.db.query.reader.series.IReaderByTimestamp; |
| import org.apache.iotdb.db.query.reader.series.ManagedSeriesReader; |
| import org.apache.iotdb.db.query.reader.series.SeriesRawDataBatchReader; |
| import org.apache.iotdb.db.query.reader.series.SeriesRawDataPointReader; |
| import org.apache.iotdb.db.query.reader.series.SeriesReader; |
| import org.apache.iotdb.db.query.reader.series.SeriesReaderByTimestamp; |
| import org.apache.iotdb.db.query.reader.series.SeriesReaderFactory; |
| import org.apache.iotdb.db.utils.SerializeUtils; |
| import org.apache.iotdb.rpc.TSStatusCode; |
| import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; |
| import org.apache.iotdb.tsfile.read.common.Path; |
| import org.apache.iotdb.tsfile.read.filter.TimeFilter; |
| import org.apache.iotdb.tsfile.read.filter.basic.Filter; |
| import org.apache.iotdb.tsfile.read.reader.IBatchReader; |
| import org.apache.iotdb.tsfile.read.reader.IPointReader; |
| |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Sets; |
| import org.apache.thrift.TException; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| |
| @SuppressWarnings("java:S107") |
| public class ClusterReaderFactory { |
| |
| private static final Logger logger = LoggerFactory.getLogger(ClusterReaderFactory.class); |
| private final MetaGroupMember metaGroupMember; |
| |
| public ClusterReaderFactory(MetaGroupMember metaGroupMember) { |
| this.metaGroupMember = metaGroupMember; |
| } |
| |
| public void syncMetaGroup() throws CheckConsistencyException { |
| metaGroupMember.syncLeaderWithConsistencyCheck(false); |
| } |
| |
| /** |
| * Create an IReaderByTimestamp that can read the data of "path" by timestamp in the whole |
| * cluster. This will query every group and merge the result from them. |
| */ |
| public IReaderByTimestamp getReaderByTimestamp( |
| PartialPath path, |
| Set<String> deviceMeasurements, |
| TSDataType dataType, |
| QueryContext context, |
| boolean ascending) |
| throws StorageEngineException, QueryProcessException { |
| // get all data groups |
| List<PartitionGroup> partitionGroups; |
| try { |
| partitionGroups = metaGroupMember.routeFilter(null, path); |
| } catch (EmptyIntervalException e) { |
| logger.warn(e.getMessage()); |
| partitionGroups = Collections.emptyList(); |
| } |
| logger.debug( |
| "{}: Sending query of {} to {} groups", |
| metaGroupMember.getName(), |
| path, |
| partitionGroups.size()); |
| List<IReaderByTimestamp> readers = new ArrayList<>(partitionGroups.size()); |
| for (PartitionGroup partitionGroup : partitionGroups) { |
| // query each group to get a reader in that group |
| IReaderByTimestamp readerByTimestamp = |
| getSeriesReaderByTime( |
| partitionGroup, path, deviceMeasurements, context, dataType, ascending); |
| readers.add(readerByTimestamp); |
| } |
| return new MergedReaderByTime(readers); |
| } |
| |
| /** |
| * Create a IReaderByTimestamp that read data of "path" by timestamp in the given group. If the |
| * local node is a member of that group, query locally. Otherwise create a remote reader pointing |
| * to one node in that group. |
| */ |
| private IReaderByTimestamp getSeriesReaderByTime( |
| PartitionGroup partitionGroup, |
| PartialPath path, |
| Set<String> deviceMeasurements, |
| QueryContext context, |
| TSDataType dataType, |
| boolean ascending) |
| throws StorageEngineException, QueryProcessException { |
| if (partitionGroup.contains(metaGroupMember.getThisNode())) { |
| // the target storage group contains this node, perform a local query |
| DataGroupMember dataGroupMember = |
| metaGroupMember.getLocalDataMember(partitionGroup.getHeader()); |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: creating a local reader for {}#{}", |
| metaGroupMember.getName(), |
| path.getFullPath(), |
| context.getQueryId()); |
| } |
| return getReaderByTimestamp( |
| path, deviceMeasurements, dataType, context, dataGroupMember, ascending); |
| } else { |
| return getRemoteReaderByTimestamp( |
| path, deviceMeasurements, dataType, partitionGroup, context, ascending); |
| } |
| } |
| |
| /** |
| * Create a IReaderByTimestamp that read data of "path" by timestamp in the given group that does |
| * not contain the local node. Send a request to one node in that group to build a reader and use |
| * that reader's id to build a remote reader. |
| */ |
| private IReaderByTimestamp getRemoteReaderByTimestamp( |
| Path path, |
| Set<String> deviceMeasurements, |
| TSDataType dataType, |
| PartitionGroup partitionGroup, |
| QueryContext context, |
| boolean ascending) |
| throws StorageEngineException { |
| SingleSeriesQueryRequest request = |
| constructSingleQueryRequest( |
| null, null, dataType, path, deviceMeasurements, partitionGroup, context, ascending); |
| |
| // reorder the nodes by their communication delays |
| List<Node> reorderedNodes = QueryCoordinator.getINSTANCE().reorderNodes(partitionGroup); |
| DataSourceInfo dataSourceInfo = |
| new DataSourceInfo( |
| partitionGroup, |
| dataType, |
| request, |
| (RemoteQueryContext) context, |
| metaGroupMember, |
| reorderedNodes); |
| |
| // try building a reader from one of the nodes |
| boolean hasClient = dataSourceInfo.hasNextDataClient(true, Long.MIN_VALUE); |
| if (hasClient) { |
| return new RemoteSeriesReaderByTimestamp(dataSourceInfo); |
| } else if (dataSourceInfo.isNoData()) { |
| return new EmptyReader(); |
| } |
| |
| throw new StorageEngineException( |
| new RequestTimeOutException("Query by timestamp: " + path + " in " + partitionGroup)); |
| } |
| |
| /** |
| * Create a MultSeriesReader that can read the data of "path" with filters in the whole cluster. |
| * The data groups that should be queried will be determined by the timeFilter, then for each |
| * group a series reader will be created, and finally all such readers will be merged into one. |
| * |
| * @param paths all path |
| * @param deviceMeasurements device to measurements |
| * @param dataTypes data type |
| * @param timeFilter time filter |
| * @param valueFilter value filter |
| * @param context query context |
| * @param ascending asc or aesc |
| * @return |
| * @throws StorageEngineException |
| * @throws EmptyIntervalException |
| */ |
| public List<AbstractMultPointReader> getMultSeriesReader( |
| List<PartialPath> paths, |
| Map<String, Set<String>> deviceMeasurements, |
| List<TSDataType> dataTypes, |
| Filter timeFilter, |
| Filter valueFilter, |
| QueryContext context, |
| boolean ascending) |
| throws StorageEngineException, EmptyIntervalException, QueryProcessException { |
| |
| Map<PartitionGroup, List<PartialPath>> partitionGroupListMap = Maps.newHashMap(); |
| for (PartialPath partialPath : paths) { |
| List<PartitionGroup> partitionGroups = metaGroupMember.routeFilter(timeFilter, partialPath); |
| partitionGroups.forEach( |
| partitionGroup -> { |
| partitionGroupListMap |
| .computeIfAbsent(partitionGroup, n -> new ArrayList<>()) |
| .add(partialPath); |
| }); |
| } |
| |
| List<AbstractMultPointReader> multPointReaders = Lists.newArrayList(); |
| |
| // different path of the same partition group are constructed as a AbstractMultPointReader |
| // if be local partition, constructed a MultBatchReader |
| // if be a remote partition, constructed a RemoteMultSeriesReader |
| for (Map.Entry<PartitionGroup, List<PartialPath>> entityPartitionGroup : |
| partitionGroupListMap.entrySet()) { |
| List<PartialPath> partialPaths = entityPartitionGroup.getValue(); |
| Map<String, Set<String>> partitionGroupDeviceMeasurements = Maps.newHashMap(); |
| List<TSDataType> partitionGroupTSDataType = Lists.newArrayList(); |
| partialPaths.forEach( |
| partialPath -> { |
| Set<String> measurements = |
| deviceMeasurements.getOrDefault(partialPath.getDevice(), Collections.emptySet()); |
| partitionGroupDeviceMeasurements.put(partialPath.getFullPath(), measurements); |
| partitionGroupTSDataType.add(dataTypes.get(paths.lastIndexOf(partialPath))); |
| }); |
| |
| AbstractMultPointReader abstractMultPointReader = |
| getMultSeriesReader( |
| entityPartitionGroup.getKey(), |
| partialPaths, |
| partitionGroupTSDataType, |
| partitionGroupDeviceMeasurements, |
| timeFilter, |
| valueFilter, |
| context, |
| ascending); |
| multPointReaders.add(abstractMultPointReader); |
| } |
| return multPointReaders; |
| } |
| |
| /** |
| * Query one node in "partitionGroup" for data of "path" with "timeFilter" and "valueFilter". If |
| * "partitionGroup" contains the local node, a local reader will be returned. Otherwise a remote |
| * reader will be returned. |
| * |
| * @param timeFilter nullable |
| * @param valueFilter nullable |
| */ |
| private AbstractMultPointReader getMultSeriesReader( |
| PartitionGroup partitionGroup, |
| List<PartialPath> partialPaths, |
| List<TSDataType> dataTypes, |
| Map<String, Set<String>> deviceMeasurements, |
| Filter timeFilter, |
| Filter valueFilter, |
| QueryContext context, |
| boolean ascending) |
| throws StorageEngineException, QueryProcessException { |
| if (partitionGroup.contains(metaGroupMember.getThisNode())) { |
| // the target storage group contains this node, perform a local query |
| DataGroupMember dataGroupMember = |
| metaGroupMember.getLocalDataMember( |
| partitionGroup.getHeader(), |
| String.format( |
| "Query: %s, time filter: %s, queryId: %d", |
| partialPaths, timeFilter, context.getQueryId())); |
| Map<String, IPointReader> partialPathPointReaderMap = Maps.newHashMap(); |
| for (int i = 0; i < partialPaths.size(); i++) { |
| PartialPath partialPath = partialPaths.get(i); |
| IPointReader seriesPointReader = |
| getSeriesPointReader( |
| partialPath, |
| deviceMeasurements.get(partialPath.getFullPath()), |
| dataTypes.get(i), |
| timeFilter, |
| valueFilter, |
| context, |
| dataGroupMember, |
| ascending); |
| partialPathPointReaderMap.put(partialPath.getFullPath(), seriesPointReader); |
| } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: creating a local reader for {}#{} of {}", |
| metaGroupMember.getName(), |
| partialPaths, |
| context.getQueryId(), |
| partitionGroup.getHeader()); |
| } |
| return new MultSeriesRawDataPointReader(partialPathPointReaderMap); |
| } else { |
| return getRemoteMultSeriesPointReader( |
| timeFilter, |
| valueFilter, |
| dataTypes, |
| partialPaths, |
| deviceMeasurements, |
| partitionGroup, |
| context, |
| ascending); |
| } |
| } |
| |
| /** |
| * Create a ManagedSeriesReader that can read the data of "path" with filters in the whole |
| * cluster. The data groups that should be queried will be determined by the timeFilter, then for |
| * each group a series reader will be created, and finally all such readers will be merged into |
| * one. |
| * |
| * @param timeFilter nullable, when null, all data groups will be queried |
| * @param valueFilter nullable |
| */ |
| public ManagedSeriesReader getSeriesReader( |
| PartialPath path, |
| Set<String> deviceMeasurements, |
| TSDataType dataType, |
| Filter timeFilter, |
| Filter valueFilter, |
| QueryContext context, |
| boolean ascending) |
| throws StorageEngineException, EmptyIntervalException { |
| // find the groups that should be queried using the timeFilter |
| List<PartitionGroup> partitionGroups = metaGroupMember.routeFilter(timeFilter, path); |
| logger.debug( |
| "{}: Sending data query of {} to {} groups", |
| metaGroupMember.getName(), |
| path, |
| partitionGroups.size()); |
| ManagedMergeReader mergeReader = new ManagedMergeReader(dataType); |
| try { |
| // build a reader for each group and merge them |
| for (PartitionGroup partitionGroup : partitionGroups) { |
| IPointReader seriesReader = |
| getSeriesReader( |
| partitionGroup, |
| path, |
| deviceMeasurements, |
| timeFilter, |
| valueFilter, |
| context, |
| dataType, |
| ascending); |
| mergeReader.addReader(seriesReader, 0); |
| } |
| } catch (IOException | QueryProcessException e) { |
| throw new StorageEngineException(e); |
| } |
| return mergeReader; |
| } |
| |
| /** |
| * Query one node in "partitionGroup" for data of "path" with "timeFilter" and "valueFilter". If |
| * "partitionGroup" contains the local node, a local reader will be returned. Otherwise a remote |
| * reader will be returned. |
| * |
| * @param timeFilter nullable |
| * @param valueFilter nullable |
| */ |
| private IPointReader getSeriesReader( |
| PartitionGroup partitionGroup, |
| PartialPath path, |
| Set<String> deviceMeasurements, |
| Filter timeFilter, |
| Filter valueFilter, |
| QueryContext context, |
| TSDataType dataType, |
| boolean ascending) |
| throws IOException, StorageEngineException, QueryProcessException { |
| if (partitionGroup.contains(metaGroupMember.getThisNode())) { |
| // the target storage group contains this node, perform a local query |
| DataGroupMember dataGroupMember = |
| metaGroupMember.getLocalDataMember( |
| partitionGroup.getHeader(), |
| String.format( |
| "Query: %s, time filter: %s, queryId: %d", |
| path, timeFilter, context.getQueryId())); |
| IPointReader seriesPointReader = |
| getSeriesPointReader( |
| path, |
| deviceMeasurements, |
| dataType, |
| timeFilter, |
| valueFilter, |
| context, |
| dataGroupMember, |
| ascending); |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: creating a local reader for {}#{} of {}, empty: {}", |
| metaGroupMember.getName(), |
| path.getFullPath(), |
| context.getQueryId(), |
| partitionGroup.getHeader(), |
| !seriesPointReader.hasNextTimeValuePair()); |
| } |
| return seriesPointReader; |
| } else { |
| return getRemoteSeriesPointReader( |
| timeFilter, |
| valueFilter, |
| dataType, |
| path, |
| deviceMeasurements, |
| partitionGroup, |
| context, |
| ascending); |
| } |
| } |
| |
| /** |
| * Create an IPointReader of "path" with “timeFilter” and "valueFilter". A synchronization with |
| * the leader will be performed according to consistency level |
| * |
| * @param path series path |
| * @param dataType data type |
| * @param timeFilter nullable |
| * @param valueFilter nullable |
| * @param context query context |
| * @return reader |
| * @throws StorageEngineException encounter exception |
| */ |
| public IPointReader getSeriesPointReader( |
| PartialPath path, |
| Set<String> allSensors, |
| TSDataType dataType, |
| Filter timeFilter, |
| Filter valueFilter, |
| QueryContext context, |
| DataGroupMember dataGroupMember, |
| boolean ascending) |
| throws StorageEngineException, QueryProcessException { |
| // pull the newest data |
| try { |
| dataGroupMember.syncLeaderWithConsistencyCheck(false); |
| } catch (CheckConsistencyException e) { |
| throw new StorageEngineException(e); |
| } |
| return new SeriesRawDataPointReader( |
| getSeriesReader( |
| path, |
| allSensors, |
| dataType, |
| timeFilter, |
| valueFilter, |
| context, |
| dataGroupMember.getHeader(), |
| ascending)); |
| } |
| |
| /** |
| * Create a SeriesReader of "path" with “timeFilter” and "valueFilter". The consistency is not |
| * guaranteed here and only data slots managed by the member will be queried. |
| * |
| * @param path series path |
| * @param dataType data type |
| * @param timeFilter nullable |
| * @param valueFilter nullable |
| * @param context query context |
| * @return reader for series |
| * @throws StorageEngineException encounter exception |
| */ |
| private SeriesReader getSeriesReader( |
| PartialPath path, |
| Set<String> allSensors, |
| TSDataType dataType, |
| Filter timeFilter, |
| Filter valueFilter, |
| QueryContext context, |
| Node header, |
| boolean ascending) |
| throws StorageEngineException, QueryProcessException { |
| ClusterQueryUtils.checkPathExistence(path); |
| List<Integer> nodeSlots = |
| ((SlotPartitionTable) metaGroupMember.getPartitionTable()).getNodeSlots(header); |
| QueryDataSource queryDataSource = |
| QueryResourceManager.getInstance().getQueryDataSource(path, context, timeFilter); |
| return SeriesReaderFactory.createSeriesReader( |
| path, |
| allSensors, |
| dataType, |
| context, |
| queryDataSource, |
| timeFilter, |
| valueFilter, |
| new SlotTsFileFilter(nodeSlots), |
| ascending); |
| } |
| |
| /** |
| * Query a remote node in "partitionGroup" to get the reader of "path" with "timeFilter" and |
| * "valueFilter". Firstly, a request will be sent to that node to construct a reader there, then |
| * the id of the reader will be returned so that we can fetch data from that node using the reader |
| * id. |
| * |
| * @param timeFilter nullable |
| * @param valueFilter nullable |
| */ |
| private AbstractMultPointReader getRemoteMultSeriesPointReader( |
| Filter timeFilter, |
| Filter valueFilter, |
| List<TSDataType> dataType, |
| List<PartialPath> paths, |
| Map<String, Set<String>> deviceMeasurements, |
| PartitionGroup partitionGroup, |
| QueryContext context, |
| boolean ascending) |
| throws StorageEngineException { |
| MultSeriesQueryRequest request = |
| constructMultQueryRequest( |
| timeFilter, |
| valueFilter, |
| dataType, |
| paths, |
| deviceMeasurements, |
| partitionGroup, |
| context, |
| ascending); |
| |
| // reorder the nodes such that the nodes that suit the query best (have lowest latenct or |
| // highest throughput) will be put to the front |
| List<Node> orderedNodes = QueryCoordinator.getINSTANCE().reorderNodes(partitionGroup); |
| |
| MultDataSourceInfo dataSourceInfo = |
| new MultDataSourceInfo( |
| partitionGroup, |
| paths, |
| dataType, |
| request, |
| (RemoteQueryContext) context, |
| metaGroupMember, |
| orderedNodes); |
| |
| boolean hasClient = dataSourceInfo.hasNextDataClient(Long.MIN_VALUE); |
| if (hasClient) { |
| return new RemoteMultSeriesReader(dataSourceInfo); |
| } else if (dataSourceInfo.isNoData()) { |
| // there is no satisfying data on the remote node |
| Set<String> fullPaths = Sets.newHashSet(); |
| dataSourceInfo |
| .getPartialPaths() |
| .forEach( |
| partialPath -> { |
| fullPaths.add(partialPath.getFullPath()); |
| }); |
| return new MultEmptyReader(fullPaths); |
| } |
| throw new StorageEngineException( |
| new RequestTimeOutException("Query " + paths + " in " + partitionGroup)); |
| } |
| |
| /** |
| * Query a remote node in "partitionGroup" to get the reader of "path" with "timeFilter" and |
| * "valueFilter". Firstly, a request will be sent to that node to construct a reader there, then |
| * the id of the reader will be returned so that we can fetch data from that node using the reader |
| * id. |
| * |
| * @param timeFilter nullable |
| * @param valueFilter nullable |
| */ |
| private IPointReader getRemoteSeriesPointReader( |
| Filter timeFilter, |
| Filter valueFilter, |
| TSDataType dataType, |
| Path path, |
| Set<String> deviceMeasurements, |
| PartitionGroup partitionGroup, |
| QueryContext context, |
| boolean ascending) |
| throws StorageEngineException { |
| SingleSeriesQueryRequest request = |
| constructSingleQueryRequest( |
| timeFilter, |
| valueFilter, |
| dataType, |
| path, |
| deviceMeasurements, |
| partitionGroup, |
| context, |
| ascending); |
| |
| // reorder the nodes such that the nodes that suit the query best (have lowest latenct or |
| // highest throughput) will be put to the front |
| List<Node> orderedNodes = QueryCoordinator.getINSTANCE().reorderNodes(partitionGroup); |
| |
| DataSourceInfo dataSourceInfo = |
| new DataSourceInfo( |
| partitionGroup, |
| dataType, |
| request, |
| (RemoteQueryContext) context, |
| metaGroupMember, |
| orderedNodes); |
| |
| boolean hasClient = dataSourceInfo.hasNextDataClient(false, Long.MIN_VALUE); |
| if (hasClient) { |
| return new RemoteSimpleSeriesReader(dataSourceInfo); |
| } else if (dataSourceInfo.isNoData()) { |
| // there is no satisfying data on the remote node |
| return new EmptyReader(); |
| } |
| |
| throw new StorageEngineException( |
| new RequestTimeOutException("Query " + path + " in " + partitionGroup)); |
| } |
| |
| private MultSeriesQueryRequest constructMultQueryRequest( |
| Filter timeFilter, |
| Filter valueFilter, |
| List<TSDataType> dataTypes, |
| List<PartialPath> paths, |
| Map<String, Set<String>> deviceMeasurements, |
| PartitionGroup partitionGroup, |
| QueryContext context, |
| boolean ascending) { |
| MultSeriesQueryRequest request = new MultSeriesQueryRequest(); |
| if (timeFilter != null) { |
| request.setTimeFilterBytes(SerializeUtils.serializeFilter(timeFilter)); |
| } |
| if (valueFilter != null) { |
| request.setValueFilterBytes(SerializeUtils.serializeFilter(valueFilter)); |
| } |
| |
| List<String> fullPaths = Lists.newArrayList(); |
| paths.forEach( |
| path -> { |
| if (path instanceof VectorPartialPath) { |
| StringBuilder builder = new StringBuilder(path.getFullPath()); |
| List<PartialPath> pathList = ((VectorPartialPath) path).getSubSensorsPathList(); |
| for (int i = 0; i < pathList.size(); i++) { |
| builder.append(":"); |
| builder.append(pathList.get(i).getFullPath()); |
| } |
| fullPaths.add(builder.toString()); |
| } else { |
| fullPaths.add(path.getFullPath()); |
| } |
| }); |
| |
| List<Integer> dataTypeOrdinals = Lists.newArrayList(); |
| dataTypes.forEach( |
| dataType -> { |
| dataTypeOrdinals.add(dataType.ordinal()); |
| }); |
| |
| request.setPath(fullPaths); |
| request.setHeader(partitionGroup.getHeader()); |
| request.setQueryId(context.getQueryId()); |
| request.setRequester(metaGroupMember.getThisNode()); |
| request.setDataTypeOrdinal(dataTypeOrdinals); |
| request.setDeviceMeasurements(deviceMeasurements); |
| request.setAscending(ascending); |
| return request; |
| } |
| |
| private SingleSeriesQueryRequest constructSingleQueryRequest( |
| Filter timeFilter, |
| Filter valueFilter, |
| TSDataType dataType, |
| Path path, |
| Set<String> deviceMeasurements, |
| PartitionGroup partitionGroup, |
| QueryContext context, |
| boolean ascending) { |
| SingleSeriesQueryRequest request = new SingleSeriesQueryRequest(); |
| if (timeFilter != null) { |
| request.setTimeFilterBytes(SerializeUtils.serializeFilter(timeFilter)); |
| } |
| if (valueFilter != null) { |
| request.setValueFilterBytes(SerializeUtils.serializeFilter(valueFilter)); |
| } |
| request.setPath(path.getFullPath()); |
| request.setHeader(partitionGroup.getHeader()); |
| request.setQueryId(context.getQueryId()); |
| request.setRequester(metaGroupMember.getThisNode()); |
| request.setDataTypeOrdinal(dataType.ordinal()); |
| request.setDeviceMeasurements(deviceMeasurements); |
| request.setAscending(ascending); |
| return request; |
| } |
| |
| /** |
| * Get GroupByExecutors the will executor the aggregations of "aggregationTypes" over "path". |
| * First, the groups to be queried will be determined by the timeFilter. Then for group, a local |
| * or remote GroupByExecutor will be created and finally all such executors will be returned. |
| * |
| * @param timeFilter nullable |
| */ |
| public List<GroupByExecutor> getGroupByExecutors( |
| PartialPath path, |
| Set<String> deviceMeasurements, |
| TSDataType dataType, |
| QueryContext context, |
| Filter timeFilter, |
| List<Integer> aggregationTypes, |
| boolean ascending) |
| throws StorageEngineException, QueryProcessException { |
| // make sure the partition table is new |
| try { |
| metaGroupMember.syncLeaderWithConsistencyCheck(false); |
| } catch (CheckConsistencyException e) { |
| throw new QueryProcessException(e.getMessage()); |
| } |
| // find out the groups that should be queried |
| List<PartitionGroup> partitionGroups; |
| try { |
| partitionGroups = metaGroupMember.routeFilter(timeFilter, path); |
| } catch (EmptyIntervalException e) { |
| logger.info(e.getMessage()); |
| partitionGroups = Collections.emptyList(); |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: Sending group by query of {} to {} groups", |
| metaGroupMember.getName(), |
| path, |
| partitionGroups.size()); |
| } |
| // create an executor for each group |
| List<GroupByExecutor> executors = new ArrayList<>(); |
| for (PartitionGroup partitionGroup : partitionGroups) { |
| GroupByExecutor groupByExecutor = |
| getGroupByExecutor( |
| path, |
| deviceMeasurements, |
| partitionGroup, |
| timeFilter, |
| context, |
| dataType, |
| aggregationTypes, |
| ascending); |
| executors.add(groupByExecutor); |
| } |
| return executors; |
| } |
| |
| /** |
| * Get a GroupByExecutor that will run "aggregationTypes" over "path" within "partitionGroup". If |
| * the local node is a member of the group, a local executor will be created. Otherwise a remote |
| * executor will be created. |
| * |
| * @param timeFilter nullable |
| */ |
| private GroupByExecutor getGroupByExecutor( |
| PartialPath path, |
| Set<String> deviceMeasurements, |
| PartitionGroup partitionGroup, |
| Filter timeFilter, |
| QueryContext context, |
| TSDataType dataType, |
| List<Integer> aggregationTypes, |
| boolean ascending) |
| throws StorageEngineException, QueryProcessException { |
| if (partitionGroup.contains(metaGroupMember.getThisNode())) { |
| // the target storage group contains this node, perform a local query |
| DataGroupMember dataGroupMember = |
| metaGroupMember.getLocalDataMember(partitionGroup.getHeader()); |
| LocalQueryExecutor localQueryExecutor = new LocalQueryExecutor(dataGroupMember); |
| logger.debug( |
| "{}: creating a local group by executor for {}#{}", |
| metaGroupMember.getName(), |
| path.getFullPath(), |
| context.getQueryId()); |
| return localQueryExecutor.getGroupByExecutor( |
| path, deviceMeasurements, dataType, timeFilter, aggregationTypes, context, ascending); |
| } else { |
| return getRemoteGroupByExecutor( |
| timeFilter, |
| aggregationTypes, |
| dataType, |
| path, |
| deviceMeasurements, |
| partitionGroup, |
| context, |
| ascending); |
| } |
| } |
| |
| /** |
| * Get a GroupByExecutor that will run "aggregationTypes" over "path" within a remote group |
| * "partitionGroup". Send a request to one node in the group to create an executor there and use |
| * the return executor id to fetch result later. |
| * |
| * @param timeFilter nullable |
| */ |
| private GroupByExecutor getRemoteGroupByExecutor( |
| Filter timeFilter, |
| List<Integer> aggregationTypes, |
| TSDataType dataType, |
| Path path, |
| Set<String> deviceMeasurements, |
| PartitionGroup partitionGroup, |
| QueryContext context, |
| boolean ascending) |
| throws StorageEngineException { |
| GroupByRequest request = new GroupByRequest(); |
| if (timeFilter != null) { |
| request.setTimeFilterBytes(SerializeUtils.serializeFilter(timeFilter)); |
| } |
| request.setPath(path.getFullPath()); |
| request.setHeader(partitionGroup.getHeader()); |
| request.setQueryId(context.getQueryId()); |
| request.setAggregationTypeOrdinals(aggregationTypes); |
| request.setDataTypeOrdinal(dataType.ordinal()); |
| request.setRequestor(metaGroupMember.getThisNode()); |
| request.setDeviceMeasurements(deviceMeasurements); |
| request.setAscending(ascending); |
| |
| // select a node with lowest latency or highest throughput with high priority |
| List<Node> orderedNodes = QueryCoordinator.getINSTANCE().reorderNodes(partitionGroup); |
| for (Node node : orderedNodes) { |
| // query a remote node |
| logger.debug("{}: querying group by {} from {}", metaGroupMember.getName(), path, node); |
| |
| try { |
| Long executorId = getRemoteGroupByExecutorId(node, request); |
| |
| if (executorId == null) { |
| continue; |
| } |
| |
| if (executorId != -1) { |
| // record the queried node to release resources later |
| ((RemoteQueryContext) context).registerRemoteNode(node, partitionGroup.getHeader()); |
| logger.debug( |
| "{}: get an executorId {} for {}@{} from {}", |
| metaGroupMember.getName(), |
| executorId, |
| aggregationTypes, |
| path, |
| node); |
| // create a remote executor with the return id |
| RemoteGroupByExecutor remoteGroupByExecutor = |
| new RemoteGroupByExecutor( |
| executorId, metaGroupMember, node, partitionGroup.getHeader()); |
| for (Integer aggregationType : aggregationTypes) { |
| remoteGroupByExecutor.addAggregateResult( |
| AggregateResultFactory.getAggrResultByType( |
| AggregationType.values()[aggregationType], dataType, ascending)); |
| } |
| return remoteGroupByExecutor; |
| } else { |
| // an id of -1 means there is no satisfying data on the remote node, create an empty |
| // reader tp reduce further communication |
| logger.debug("{}: no data for {} from {}", metaGroupMember.getName(), path, node); |
| return new EmptyReader(); |
| } |
| } catch (TException | IOException e) { |
| logger.error("{}: Cannot query {} from {}", metaGroupMember.getName(), path, node, e); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.error("{}: Cannot query {} from {}", metaGroupMember.getName(), path, node, e); |
| } |
| } |
| throw new StorageEngineException( |
| new RequestTimeOutException("Query " + path + " in " + partitionGroup)); |
| } |
| |
| private Long getRemoteGroupByExecutorId(Node node, GroupByRequest request) |
| throws IOException, TException, InterruptedException { |
| Long executorId; |
| if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { |
| AsyncDataClient client = |
| metaGroupMember |
| .getClientProvider() |
| .getAsyncDataClient(node, RaftServer.getReadOperationTimeoutMS()); |
| executorId = SyncClientAdaptor.getGroupByExecutor(client, request); |
| } else { |
| try (SyncDataClient syncDataClient = |
| metaGroupMember |
| .getClientProvider() |
| .getSyncDataClient(node, RaftServer.getReadOperationTimeoutMS())) { |
| |
| executorId = syncDataClient.getGroupByExecutor(request); |
| } |
| } |
| return executorId; |
| } |
| |
| /** |
| * Create an IBatchReader of "path" with “timeFilter” and "valueFilter". A synchronization with |
| * the leader will be performed according to consistency level |
| * |
| * @param path |
| * @param dataType |
| * @param timeFilter nullable |
| * @param valueFilter nullable |
| * @param context |
| * @return an IBatchReader or null if there is no satisfying data |
| * @throws StorageEngineException |
| */ |
| public IBatchReader getSeriesBatchReader( |
| PartialPath path, |
| Set<String> allSensors, |
| TSDataType dataType, |
| Filter timeFilter, |
| Filter valueFilter, |
| QueryContext context, |
| DataGroupMember dataGroupMember, |
| boolean ascending) |
| throws StorageEngineException, QueryProcessException, IOException { |
| // pull the newest data |
| try { |
| dataGroupMember.syncLeaderWithConsistencyCheck(false); |
| } catch (CheckConsistencyException e) { |
| throw new StorageEngineException(e); |
| } |
| |
| SeriesReader seriesReader = |
| getSeriesReader( |
| path, |
| allSensors, |
| dataType, |
| timeFilter, |
| valueFilter, |
| context, |
| dataGroupMember.getHeader(), |
| ascending); |
| if (seriesReader.isEmpty()) { |
| return null; |
| } |
| return new SeriesRawDataBatchReader(seriesReader); |
| } |
| |
| /** |
| * Create an IBatchReader of "path" with “timeFilter” and "valueFilter". A synchronization with |
| * the leader will be performed according to consistency level |
| * |
| * @param paths |
| * @param dataTypes |
| * @param timeFilter nullable |
| * @param valueFilter nullable |
| * @param context |
| * @return an IBatchReader or null if there is no satisfying data |
| * @throws StorageEngineException |
| */ |
| public IBatchReader getMultSeriesBatchReader( |
| List<PartialPath> paths, |
| Map<String, Set<String>> allSensors, |
| List<TSDataType> dataTypes, |
| Filter timeFilter, |
| Filter valueFilter, |
| QueryContext context, |
| DataGroupMember dataGroupMember, |
| boolean ascending) |
| throws StorageEngineException, QueryProcessException, IOException { |
| // pull the newest data |
| try { |
| dataGroupMember.syncLeaderWithConsistencyCheck(false); |
| } catch (CheckConsistencyException e) { |
| throw new StorageEngineException(e); |
| } |
| |
| Map<String, IBatchReader> partialPathBatchReaderMap = Maps.newHashMap(); |
| |
| for (int i = 0; i < paths.size(); i++) { |
| PartialPath partialPath = paths.get(i); |
| SeriesReader seriesReader = |
| getSeriesReader( |
| partialPath, |
| allSensors.get(partialPath.getFullPath()), |
| dataTypes.get(i), |
| timeFilter, |
| valueFilter, |
| context, |
| dataGroupMember.getHeader(), |
| ascending); |
| partialPathBatchReaderMap.put( |
| partialPath.getFullPath(), new SeriesRawDataBatchReader(seriesReader)); |
| } |
| return new MultBatchReader(partialPathBatchReaderMap); |
| } |
| |
| /** |
| * Create an IReaderByTimestamp of "path". A synchronization with the leader will be performed |
| * according to consistency level |
| * |
| * @param path |
| * @param dataType |
| * @param context |
| * @return an IReaderByTimestamp or null if there is no satisfying data |
| * @throws StorageEngineException |
| */ |
| public IReaderByTimestamp getReaderByTimestamp( |
| PartialPath path, |
| Set<String> allSensors, |
| TSDataType dataType, |
| QueryContext context, |
| DataGroupMember dataGroupMember, |
| boolean ascending) |
| throws StorageEngineException, QueryProcessException { |
| try { |
| dataGroupMember.syncLeaderWithConsistencyCheck(false); |
| } catch (CheckConsistencyException e) { |
| throw new StorageEngineException(e); |
| } |
| SeriesReader seriesReader = |
| getSeriesReader( |
| path, |
| allSensors, |
| dataType, |
| TimeFilter.gtEq(Long.MIN_VALUE), |
| null, |
| context, |
| dataGroupMember.getHeader(), |
| ascending); |
| try { |
| if (seriesReader.isEmpty()) { |
| return null; |
| } |
| } catch (IOException e) { |
| throw new QueryProcessException(e, TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); |
| } |
| return new SeriesReaderByTimestamp(seriesReader, ascending); |
| } |
| } |
