This is an automated email from the ASF dual-hosted git repository.
neuyilan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 910b184 [IOTDB-1117][Distributed]Batched creation and fetch of
RemoteSeriesReader(#2875)
910b184 is described below
commit 910b184f2a157c28adf6e863d48b6184bb49e6c0
Author: wangchao316 <[email protected]>
AuthorDate: Thu Mar 25 20:18:06 2021 +0800
[IOTDB-1117][Distributed]Batched creation and fetch of
RemoteSeriesReader(#2875)
---
.../cluster/query/ClusterDataQueryExecutor.java | 80 ++++++
.../iotdb/cluster/query/LocalQueryExecutor.java | 143 ++++++++++
.../cluster/query/reader/ClusterReaderFactory.java | 295 +++++++++++++++++++++
.../query/reader/mult/AbstractMultPointReader.java | 70 +++++
.../reader/mult/AssignPathManagedMergeReader.java | 92 +++++++
.../reader/mult/AssignPathPriorityMergeReader.java | 66 +++++
.../query/reader/mult/IMultBatchReader.java | 31 +++
.../cluster/query/reader/mult/MultBatchReader.java | 73 +++++
.../query/reader/mult/MultDataSourceInfo.java | 264 ++++++++++++++++++
.../cluster/query/reader/mult/MultEmptyReader.java | 52 ++++
.../reader/mult/MultSeriesRawDataPointReader.java | 55 ++++
.../query/reader/mult/RemoteMultSeriesReader.java | 222 ++++++++++++++++
.../iotdb/cluster/server/DataClusterServer.java | 37 +++
.../cluster/server/service/DataAsyncService.java | 27 ++
.../cluster/server/service/DataSyncService.java | 21 ++
.../iotdb/cluster/common/TestAsyncDataClient.java | 35 +++
.../mult/AssignPathManagedMergeReaderTest.java | 190 +++++++++++++
.../mult/MultSeriesRawDataPointReaderTest.java | 67 +++++
.../reader/mult/RemoteMultSeriesReaderTest.java | 286 ++++++++++++++++++++
.../universal/CachedPriorityMergeReader.java | 3 +-
.../reader/universal/DescPriorityMergeReader.java | 4 +-
.../iotdb/db/query/reader/universal/Element.java | 72 +++++
.../reader/universal/PriorityMergeReader.java | 39 +--
thrift/src/main/thrift/cluster.thrift | 27 ++
24 files changed, 2212 insertions(+), 39 deletions(-)
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutor.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutor.java
index da371d9..9ebde27 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutor.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutor.java
@@ -24,6 +24,8 @@ import
org.apache.iotdb.cluster.exception.EmptyIntervalException;
import org.apache.iotdb.cluster.partition.PartitionGroup;
import org.apache.iotdb.cluster.query.reader.ClusterReaderFactory;
import org.apache.iotdb.cluster.query.reader.ClusterTimeGenerator;
+import org.apache.iotdb.cluster.query.reader.mult.AbstractMultPointReader;
+import org.apache.iotdb.cluster.query.reader.mult.AssignPathManagedMergeReader;
import org.apache.iotdb.cluster.server.member.DataGroupMember;
import org.apache.iotdb.cluster.server.member.MetaGroupMember;
import org.apache.iotdb.db.exception.StorageEngineException;
@@ -43,9 +45,11 @@ import
org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import org.apache.iotdb.tsfile.read.query.timegenerator.TimeGenerator;
import org.apache.iotdb.tsfile.read.reader.IPointReader;
+import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -65,6 +69,82 @@ public class ClusterDataQueryExecutor extends
RawDataQueryExecutor {
this.readerFactory = new ClusterReaderFactory(metaGroupMember);
}
+ /**
+ * use mult batch query for without value filter
+ *
+ * @param context query context
+ * @return query data set
+ * @throws StorageEngineException
+ */
+ @Override
+ public QueryDataSet executeWithoutValueFilter(QueryContext context)
+ throws StorageEngineException {
+ QueryDataSet dataSet = needRedirect(context, false);
+ if (dataSet != null) {
+ return dataSet;
+ }
+ try {
+ List<ManagedSeriesReader> readersOfSelectedSeries =
initMultSeriesReader(context);
+ return new RawQueryDataSetWithoutValueFilter(
+ context.getQueryId(),
+ queryPlan.getDeduplicatedPaths(),
+ queryPlan.getDeduplicatedDataTypes(),
+ readersOfSelectedSeries,
+ queryPlan.isAscending());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new StorageEngineException(e.getMessage());
+ } catch (IOException | EmptyIntervalException | QueryProcessException e) {
+ throw new StorageEngineException(e.getMessage());
+ }
+ }
+
+ private List<ManagedSeriesReader> initMultSeriesReader(QueryContext context)
+ throws StorageEngineException, IOException, EmptyIntervalException,
QueryProcessException {
+ Filter timeFilter = null;
+ if (queryPlan.getExpression() != null) {
+ timeFilter = ((GlobalTimeExpression)
queryPlan.getExpression()).getFilter();
+ }
+
+ // make sure the partition table is new
+ try {
+ metaGroupMember.syncLeaderWithConsistencyCheck(false);
+ } catch (CheckConsistencyException e) {
+ throw new StorageEngineException(e);
+ }
+ List<ManagedSeriesReader> readersOfSelectedSeries = Lists.newArrayList();
+ List<AbstractMultPointReader> multPointReaders = Lists.newArrayList();
+
+ multPointReaders =
+ readerFactory.getMultSeriesReader(
+ queryPlan.getDeduplicatedPaths(),
+ queryPlan.getDeviceToMeasurements(),
+ queryPlan.getDeduplicatedDataTypes(),
+ timeFilter,
+ null,
+ context,
+ queryPlan.isAscending());
+
+ // combine reader of different partition group of the same path
+ // into a MultManagedMergeReader
+ for (int i = 0; i < queryPlan.getDeduplicatedPaths().size(); i++) {
+ PartialPath partialPath = queryPlan.getDeduplicatedPaths().get(i);
+ TSDataType dataType = queryPlan.getDeduplicatedDataTypes().get(i);
+ AssignPathManagedMergeReader assignPathManagedMergeReader =
+ new AssignPathManagedMergeReader(partialPath.getFullPath(),
dataType);
+ for (AbstractMultPointReader multPointReader : multPointReaders) {
+ if (multPointReader.getAllPaths().contains(partialPath.getFullPath()))
{
+ assignPathManagedMergeReader.addReader(multPointReader, 0);
+ }
+ }
+ readersOfSelectedSeries.add(assignPathManagedMergeReader);
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initialized {} readers for {}",
readersOfSelectedSeries.size(), queryPlan);
+ }
+ return readersOfSelectedSeries;
+ }
+
@Override
protected List<ManagedSeriesReader> initManagedSeriesReader(QueryContext
context)
throws StorageEngineException {
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/LocalQueryExecutor.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/LocalQueryExecutor.java
index 24c7847..f078519 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/query/LocalQueryExecutor.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/LocalQueryExecutor.java
@@ -26,9 +26,11 @@ import
org.apache.iotdb.cluster.partition.slot.SlotPartitionTable;
import org.apache.iotdb.cluster.query.filter.SlotTsFileFilter;
import org.apache.iotdb.cluster.query.manage.ClusterQueryManager;
import org.apache.iotdb.cluster.query.reader.ClusterReaderFactory;
+import org.apache.iotdb.cluster.query.reader.mult.IMultBatchReader;
import org.apache.iotdb.cluster.rpc.thrift.GetAggrResultRequest;
import org.apache.iotdb.cluster.rpc.thrift.GroupByRequest;
import org.apache.iotdb.cluster.rpc.thrift.LastQueryRequest;
+import org.apache.iotdb.cluster.rpc.thrift.MultSeriesQueryRequest;
import org.apache.iotdb.cluster.rpc.thrift.Node;
import org.apache.iotdb.cluster.rpc.thrift.PreviousFillRequest;
import org.apache.iotdb.cluster.rpc.thrift.PullSchemaRequest;
@@ -70,6 +72,8 @@ import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.schema.TimeseriesSchema;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,6 +83,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import static org.apache.iotdb.session.Config.DEFAULT_FETCH_SIZE;
@@ -151,6 +156,45 @@ public class LocalQueryExecutor {
}
/**
+ * Fetch a batch from the reader whose id is "readerId".
+ *
+ * @param readerId reader id
+ * @param paths mult series path
+ */
+ public Map<String, ByteBuffer> fetchMultSeries(long readerId, List<String>
paths)
+ throws ReaderNotFoundException, IOException {
+ IMultBatchReader reader =
+ (IMultBatchReader)
dataGroupMember.getQueryManager().getReader(readerId);
+ if (reader == null) {
+ throw new ReaderNotFoundException(readerId);
+ }
+
+ Map<String, ByteBuffer> pathByteBuffers = Maps.newHashMap();
+
+ for (String path : paths) {
+ ByteBuffer byteBuffer = null;
+ if (reader.hasNextBatch(path)) {
+ BatchData batchData = reader.nextBatch(path);
+
+ ByteArrayOutputStream byteArrayOutputStream = new
ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new
DataOutputStream(byteArrayOutputStream);
+
+ SerializeUtils.serializeBatchData(batchData, dataOutputStream);
+ logger.debug(
+ "{}: Send results of reader {}, size:{}",
+ dataGroupMember.getName(),
+ readerId,
+ batchData.length());
+ byteBuffer = ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+ } else {
+ byteBuffer = ByteBuffer.allocate(0);
+ }
+ pathByteBuffers.put(path, byteBuffer);
+ }
+ return pathByteBuffers;
+ }
+
+ /**
* Create an IBatchReader of a path, register it in the query manager to get
a reader id for it
* and send the id back to the requester. If the reader does not have any
data, an id of -1 will
* be returned.
@@ -237,6 +281,105 @@ public class LocalQueryExecutor {
}
/**
+ * Create an IBatchReader of a path, register it in the query manager to get
a reader id for it
+ * and send the id back to the requester. If the reader does not have any
data, an id of -1 will
+ * be returned.
+ *
+ * @param request
+ */
+ public long queryMultSeries(MultSeriesQueryRequest request)
+ throws CheckConsistencyException, QueryProcessException,
StorageEngineException, IOException {
+ logger.debug(
+ "{}: {} is querying {}, queryId: {}",
+ name,
+ request.getRequester(),
+ request.getPath(),
+ request.getQueryId());
+ dataGroupMember.syncLeaderWithConsistencyCheck(false);
+
+ List<PartialPath> paths = Lists.newArrayList();
+ request
+ .getPath()
+ .forEach(
+ fullPath -> {
+ try {
+ paths.add(new PartialPath(fullPath));
+ } catch (IllegalPathException e) {
+ logger.warn("Failed to create partial path, fullPath is {}.",
fullPath, e);
+ }
+ });
+
+ List<TSDataType> dataTypes = Lists.newArrayList();
+ request
+ .getDataTypeOrdinal()
+ .forEach(
+ dataType -> {
+ dataTypes.add(TSDataType.values()[dataType]);
+ });
+
+ Filter timeFilter = null;
+ Filter valueFilter = null;
+ if (request.isSetTimeFilterBytes()) {
+ timeFilter = FilterFactory.deserialize(request.timeFilterBytes);
+ }
+ if (request.isSetValueFilterBytes()) {
+ valueFilter = FilterFactory.deserialize(request.valueFilterBytes);
+ }
+ Map<String, Set<String>> deviceMeasurements =
request.getDeviceMeasurements();
+
+ // the same query from a requester correspond to a context here
+ RemoteQueryContext queryContext =
+ queryManager.getQueryContext(
+ request.getRequester(),
+ request.getQueryId(),
+ request.getFetchSize(),
+ request.getDeduplicatedPathNum());
+ logger.debug(
+ "{}: local queryId for {}#{} is {}",
+ name,
+ request.getQueryId(),
+ request.getPath(),
+ queryContext.getQueryId());
+ IBatchReader batchReader =
+ readerFactory.getMultSeriesBatchReader(
+ paths,
+ deviceMeasurements,
+ dataTypes,
+ timeFilter,
+ valueFilter,
+ queryContext,
+ dataGroupMember,
+ request.ascending);
+
+ // if the reader contains no data, send a special id of -1 to prevent the
requester from
+ // meaninglessly fetching data
+ if (batchReader != null && batchReader.hasNextBatch()) {
+ long readerId = queryManager.registerReader(batchReader);
+ queryContext.registerLocalReader(readerId);
+ logger.debug(
+ "{}: Build a reader of {} for {}#{}, readerId: {}",
+ name,
+ paths,
+ request.getRequester(),
+ request.getQueryId(),
+ readerId);
+ return readerId;
+ } else {
+ logger.debug(
+ "{}: There is no data of {} for {}#{}",
+ name,
+ paths,
+ request.getRequester(),
+ request.getQueryId());
+
+ if (batchReader != null) {
+ batchReader.close();
+ }
+ return -1;
+ }
+ }
+
+ /**
* Send the timeseries schemas of some prefix paths to the requester. The
schemas will be sent in
* the form of a list of MeasurementSchema, but notice the measurements in
them are the full
* paths.
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactory.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactory.java
index 77ffc01..35fabe3 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactory.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactory.java
@@ -33,7 +33,14 @@ 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;
@@ -64,6 +71,9 @@ 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;
@@ -72,6 +82,7 @@ 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")
@@ -202,6 +213,140 @@ public class ClusterReaderFactory {
}
/**
+ * 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
@@ -395,6 +540,68 @@ public class ClusterReaderFactory {
* @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,
@@ -441,6 +648,45 @@ public class ClusterReaderFactory {
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 -> {
+ 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,
@@ -711,6 +957,55 @@ public class ClusterReaderFactory {
}
/**
+ * 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
*
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/AbstractMultPointReader.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/AbstractMultPointReader.java
new file mode 100644
index 0000000..0cd4d89
--- /dev/null
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/AbstractMultPointReader.java
@@ -0,0 +1,70 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.reader.IPointReader;
+
+import java.io.IOException;
+import java.util.Set;
+
+public abstract class AbstractMultPointReader implements IPointReader {
+
+ public abstract boolean hasNextTimeValuePair(String fullPath) throws
IOException;
+
+ public abstract TimeValuePair nextTimeValuePair(String fullPath) throws
IOException;
+
+ public abstract Set<String> getAllPaths();
+
+ /**
+ * do not support this method
+ *
+ * @return only false
+ * @throws IOException
+ */
+ @Override
+ @Deprecated
+ public boolean hasNextTimeValuePair() throws IOException {
+ return false;
+ }
+
+ /**
+ * do not support this method
+ *
+ * @return only null
+ * @throws IOException
+ */
+ @Override
+ @Deprecated
+ public TimeValuePair nextTimeValuePair() throws IOException {
+ return null;
+ }
+
+ /**
+ * do not support this method
+ *
+ * @return only null
+ * @throws IOException
+ */
+ @Override
+ @Deprecated
+ public TimeValuePair currentTimeValuePair() throws IOException {
+ return null;
+ }
+}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/AssignPathManagedMergeReader.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/AssignPathManagedMergeReader.java
new file mode 100644
index 0000000..34ecc13
--- /dev/null
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/AssignPathManagedMergeReader.java
@@ -0,0 +1,92 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.db.query.reader.series.ManagedSeriesReader;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+
+public class AssignPathManagedMergeReader extends AssignPathPriorityMergeReader
+ implements ManagedSeriesReader {
+
+ private static final int BATCH_SIZE = 4096;
+ private volatile boolean managedByPool;
+ private volatile boolean hasRemaining;
+
+ private BatchData batchData;
+ private TSDataType dataType;
+
+ public AssignPathManagedMergeReader(String fullPath, TSDataType dataType) {
+ super(fullPath);
+ this.dataType = dataType;
+ }
+
+ @Override
+ public boolean isManagedByQueryManager() {
+ return managedByPool;
+ }
+
+ @Override
+ public void setManagedByQueryManager(boolean managedByQueryManager) {
+ this.managedByPool = managedByQueryManager;
+ }
+
+ @Override
+ public boolean hasRemaining() {
+ return hasRemaining;
+ }
+
+ @Override
+ public void setHasRemaining(boolean hasRemaining) {
+ this.hasRemaining = hasRemaining;
+ }
+
+ @Override
+ public boolean hasNextBatch() throws IOException {
+ if (batchData != null) {
+ return true;
+ }
+ constructBatch();
+ return batchData != null;
+ }
+
+ private void constructBatch() throws IOException {
+ if (hasNextTimeValuePair()) {
+ batchData = new BatchData(dataType);
+ while (hasNextTimeValuePair() && batchData.length() < BATCH_SIZE) {
+ TimeValuePair next = nextTimeValuePair();
+ batchData.putAnObject(next.getTimestamp(), next.getValue().getValue());
+ }
+ }
+ }
+
+ @Override
+ public BatchData nextBatch() throws IOException {
+ if (!hasNextBatch()) {
+ throw new NoSuchElementException();
+ }
+ BatchData ret = batchData;
+ batchData = null;
+ return ret;
+ }
+}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/AssignPathPriorityMergeReader.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/AssignPathPriorityMergeReader.java
new file mode 100644
index 0000000..30c6f70
--- /dev/null
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/AssignPathPriorityMergeReader.java
@@ -0,0 +1,66 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.db.query.reader.universal.Element;
+import org.apache.iotdb.db.query.reader.universal.PriorityMergeReader;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+
+import java.io.IOException;
+
+/**
+ * This class extends {@link extends PriorityMergeReader} for data sources
with different
+ * priorities.
+ */
+public class AssignPathPriorityMergeReader extends PriorityMergeReader {
+
+ private String fullPath;
+
+ public AssignPathPriorityMergeReader(String fullPath) {
+ super();
+ this.fullPath = fullPath;
+ }
+
+ public void addReader(AbstractMultPointReader reader, long priority) throws
IOException {
+ if (reader.hasNextTimeValuePair(fullPath)) {
+ heap.add(
+ new MultElement(
+ reader, reader.nextTimeValuePair(fullPath), new
MergeReaderPriority(priority, 0)));
+ } else {
+ reader.close();
+ }
+ }
+
+ public class MultElement extends Element {
+ public MultElement(
+ AbstractMultPointReader reader, TimeValuePair timeValuePair,
MergeReaderPriority priority) {
+ super(reader, timeValuePair, priority);
+ }
+
+ @Override
+ public boolean hasNext() throws IOException {
+ return ((AbstractMultPointReader) reader).hasNextTimeValuePair(fullPath);
+ }
+
+ @Override
+ public void next() throws IOException {
+ timeValuePair = ((AbstractMultPointReader)
reader).nextTimeValuePair(fullPath);
+ }
+ }
+}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/IMultBatchReader.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/IMultBatchReader.java
new file mode 100644
index 0000000..07c63cf
--- /dev/null
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/IMultBatchReader.java
@@ -0,0 +1,31 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.reader.IBatchReader;
+
+import java.io.IOException;
+
+public interface IMultBatchReader extends IBatchReader {
+
+ boolean hasNextBatch(String fullPath) throws IOException;
+
+ BatchData nextBatch(String fullPath) throws IOException;
+}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultBatchReader.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultBatchReader.java
new file mode 100644
index 0000000..dd7f21b
--- /dev/null
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultBatchReader.java
@@ -0,0 +1,73 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.reader.IBatchReader;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class MultBatchReader implements IMultBatchReader {
+
+ private Map<String, IBatchReader> pathBatchReaders;
+
+ public MultBatchReader(Map<String, IBatchReader> pathBatchReaders) {
+ this.pathBatchReaders = pathBatchReaders;
+ }
+
+ /**
+ * reader has next batch data
+ *
+ * @return true if only one reader has next batch data, otherwise false
+ * @throws IOException
+ */
+ @Override
+ public boolean hasNextBatch() throws IOException {
+ for (IBatchReader reader : pathBatchReaders.values()) {
+ if (reader.hasNextBatch()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public boolean hasNextBatch(String fullPath) throws IOException {
+ return pathBatchReaders.get(fullPath).hasNextBatch();
+ }
+
+ @Override
+ public BatchData nextBatch(String fullPath) throws IOException {
+ return pathBatchReaders.get(fullPath).nextBatch();
+ }
+
+ @Override
+ public BatchData nextBatch() throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * close in query resource
+ *
+ * @throws IOException
+ */
+ @Override
+ public void close() throws IOException {}
+}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultDataSourceInfo.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultDataSourceInfo.java
new file mode 100644
index 0000000..27c9f59
--- /dev/null
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultDataSourceInfo.java
@@ -0,0 +1,264 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.cluster.client.async.AsyncDataClient;
+import org.apache.iotdb.cluster.client.sync.SyncDataClient;
+import org.apache.iotdb.cluster.config.ClusterDescriptor;
+import org.apache.iotdb.cluster.partition.PartitionGroup;
+import org.apache.iotdb.cluster.query.RemoteQueryContext;
+import org.apache.iotdb.cluster.rpc.thrift.MultSeriesQueryRequest;
+import org.apache.iotdb.cluster.rpc.thrift.Node;
+import org.apache.iotdb.cluster.server.RaftServer;
+import org.apache.iotdb.cluster.server.handlers.caller.GenericHandler;
+import org.apache.iotdb.cluster.server.member.MetaGroupMember;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.utils.SerializeUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.filter.TimeFilter;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.filter.factory.FilterFactory;
+import org.apache.iotdb.tsfile.read.filter.operator.AndFilter;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * provide client which could connect to all nodes of the partitionGroup, and
mult reader Notice:
+ * methods like getter should be called only after nextDataClient() has been
called
+ */
+public class MultDataSourceInfo {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MultDataSourceInfo.class);
+
+ private long readerId;
+ private Node curSource;
+ private PartitionGroup partitionGroup;
+ private List<PartialPath> partialPaths;
+ private List<TSDataType> dataTypes;
+ private MultSeriesQueryRequest request;
+ private RemoteQueryContext context;
+ private MetaGroupMember metaGroupMember;
+ private List<Node> nodes;
+ private int curPos;
+ private boolean isNoData = false;
+ private boolean isNoClient = false;
+
+ public MultDataSourceInfo(
+ PartitionGroup group,
+ List<PartialPath> partialPaths,
+ List<TSDataType> dataTypes,
+ MultSeriesQueryRequest request,
+ RemoteQueryContext context,
+ MetaGroupMember metaGroupMember,
+ List<Node> nodes) {
+ this.readerId = -1;
+ this.partitionGroup = group;
+ this.partialPaths = partialPaths;
+ this.dataTypes = dataTypes;
+ this.request = request;
+ this.context = context;
+ this.metaGroupMember = metaGroupMember;
+ this.nodes = nodes;
+ // set to the last node so after nextDataClient() is called it will scan
from the first node
+ this.curPos = nodes.size() - 1;
+ this.curSource = nodes.get(curPos);
+ }
+
+ public boolean hasNextDataClient(long timestamp) {
+ if (this.nodes.isEmpty()) {
+ this.isNoData = false;
+ return false;
+ }
+
+ int nextNodePos = (this.curPos + 1) % this.nodes.size();
+ while (true) {
+ Node node = nodes.get(nextNodePos);
+ logger.debug("querying {} from {} of {}", request.path, node,
partitionGroup.getHeader());
+ try {
+ Long newReaderId = getReaderId(node, timestamp);
+ if (newReaderId != null) {
+ logger.debug("get a readerId {} for {} from {}", newReaderId,
request.path, node);
+ if (newReaderId != -1) {
+ // register the node so the remote resources can be released
+ context.registerRemoteNode(node, partitionGroup.getHeader());
+ this.readerId = newReaderId;
+ this.curSource = node;
+ this.curPos = nextNodePos;
+ return true;
+ } else {
+ // the id being -1 means there is no satisfying data on the remote
node, create an
+ // empty reader to reduce further communication
+ this.isNoClient = true;
+ this.isNoData = true;
+ return false;
+ }
+ }
+ } catch (TException | IOException e) {
+ logger.error("Cannot query {} from {}", this.request.path, node, e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ logger.error("Cannot query {} from {}", this.request.path, node, e);
+ }
+ nextNodePos = (nextNodePos + 1) % this.nodes.size();
+ if (nextNodePos == this.curPos) {
+ // has iterate over all nodes
+ isNoClient = true;
+ break;
+ }
+ }
+ // all nodes are failed
+ this.isNoData = false;
+ return false;
+ }
+
+ public List<PartialPath> getPartialPaths() {
+ return partialPaths;
+ }
+
+ private Long getReaderId(Node node, long timestamp)
+ throws TException, InterruptedException, IOException {
+ if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) {
+ return applyForReaderIdAsync(node, timestamp);
+ }
+ return applyForReaderIdSync(node, timestamp);
+ }
+
+ private Long applyForReaderIdAsync(Node node, long timestamp)
+ throws TException, InterruptedException, IOException {
+ AsyncDataClient client =
+ this.metaGroupMember
+ .getClientProvider()
+ .getAsyncDataClient(node, RaftServer.getReadOperationTimeoutMS());
+ AtomicReference<Long> result = new AtomicReference<>();
+ GenericHandler<Long> handler = new GenericHandler<>(client.getNode(),
result);
+ Filter newFilter;
+ // add timestamp to as a timeFilter to skip the data which has been read
+ if (request.isSetTimeFilterBytes()) {
+ Filter timeFilter = FilterFactory.deserialize(request.timeFilterBytes);
+ newFilter = new AndFilter(timeFilter, TimeFilter.gt(timestamp));
+ } else {
+ newFilter = TimeFilter.gt(timestamp);
+ }
+ request.setTimeFilterBytes(SerializeUtils.serializeFilter(newFilter));
+ client.queryMultSeries(request, handler);
+ synchronized (result) {
+ if (result.get() == null && handler.getException() == null) {
+ result.wait(RaftServer.getReadOperationTimeoutMS());
+ }
+ }
+ return result.get();
+ }
+
+ private Long applyForReaderIdSync(Node node, long timestamp) throws
TException {
+
+ Long newReaderId;
+ try (SyncDataClient client =
+ this.metaGroupMember
+ .getClientProvider()
+ .getSyncDataClient(node, RaftServer.getReadOperationTimeoutMS())) {
+
+ Filter newFilter;
+ // add timestamp to as a timeFilter to skip the data which has been read
+ if (request.isSetTimeFilterBytes()) {
+ Filter timeFilter = FilterFactory.deserialize(request.timeFilterBytes);
+ newFilter = new AndFilter(timeFilter, TimeFilter.gt(timestamp));
+ } else {
+ newFilter = TimeFilter.gt(timestamp);
+ }
+ request.setTimeFilterBytes(SerializeUtils.serializeFilter(newFilter));
+ newReaderId = client.queryMultSeries(request);
+ return newReaderId;
+ }
+ }
+
+ public long getReaderId() {
+ return this.readerId;
+ }
+
+ public List<TSDataType> getDataTypes() {
+ return this.dataTypes;
+ }
+
+ public Node getHeader() {
+ return partitionGroup.getHeader();
+ }
+
+ AsyncDataClient getCurAsyncClient(int timeout) throws IOException {
+ return isNoClient
+ ? null
+ :
metaGroupMember.getClientProvider().getAsyncDataClient(this.curSource, timeout);
+ }
+
+ SyncDataClient getCurSyncClient(int timeout) throws TException {
+ return isNoClient
+ ? null
+ :
metaGroupMember.getClientProvider().getSyncDataClient(this.curSource, timeout);
+ }
+
+ public boolean isNoData() {
+ return this.isNoData;
+ }
+
+ private boolean isNoClient() {
+ return this.isNoClient;
+ }
+
+ @Override
+ public String toString() {
+ return "DataSourceInfo{"
+ + "readerId="
+ + readerId
+ + ", curSource="
+ + curSource
+ + ", partitionGroup="
+ + partitionGroup
+ + ", request="
+ + request
+ + '}';
+ }
+
+ /**
+ * Check if there is still any available client and there is still any left
data.
+ *
+ * @return true if there is an available client and data to read, false all
data has been read.
+ * @throws IOException if all clients are unavailable.
+ */
+ boolean checkCurClient() throws IOException {
+ if (isNoClient()) {
+ if (!isNoData()) {
+ throw new IOException("no available client.");
+ } else {
+ // no data
+ return false;
+ }
+ }
+ return true;
+ }
+
+ Node getCurrentNode() {
+ return this.curSource;
+ }
+}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultEmptyReader.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultEmptyReader.java
new file mode 100644
index 0000000..2bb463c
--- /dev/null
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultEmptyReader.java
@@ -0,0 +1,52 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+
+import java.io.IOException;
+import java.util.Set;
+
+/** empty mult reader */
+public class MultEmptyReader extends AbstractMultPointReader {
+
+ private Set<String> fullPaths;
+
+ public MultEmptyReader(Set<String> fullPaths) {
+ this.fullPaths = fullPaths;
+ }
+
+ @Override
+ public boolean hasNextTimeValuePair(String fullPath) throws IOException {
+ return false;
+ }
+
+ @Override
+ public TimeValuePair nextTimeValuePair(String fullPath) throws IOException {
+ return null;
+ }
+
+ @Override
+ public Set<String> getAllPaths() {
+ return fullPaths;
+ }
+
+ @Override
+ public void close() {}
+}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultSeriesRawDataPointReader.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultSeriesRawDataPointReader.java
new file mode 100644
index 0000000..0561cfe
--- /dev/null
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultSeriesRawDataPointReader.java
@@ -0,0 +1,55 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.reader.IPointReader;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Set;
+
+/** mult reader of local partition group */
+public class MultSeriesRawDataPointReader extends AbstractMultPointReader {
+ private Map<String, IPointReader> partitalPathReaders;
+
+ public MultSeriesRawDataPointReader(Map<String, IPointReader>
partitalPathReaders) {
+ this.partitalPathReaders = partitalPathReaders;
+ }
+
+ @Override
+ public boolean hasNextTimeValuePair(String fullPath) throws IOException {
+ IPointReader seriesRawDataPointReader = partitalPathReaders.get(fullPath);
+ return seriesRawDataPointReader.hasNextTimeValuePair();
+ }
+
+ @Override
+ public TimeValuePair nextTimeValuePair(String fullPath) throws IOException {
+ IPointReader seriesRawDataPointReader = partitalPathReaders.get(fullPath);
+ return seriesRawDataPointReader.nextTimeValuePair();
+ }
+
+ @Override
+ public Set<String> getAllPaths() {
+ return partitalPathReaders.keySet();
+ }
+
+ @Override
+ public void close() {}
+}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/RemoteMultSeriesReader.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/RemoteMultSeriesReader.java
new file mode 100644
index 0000000..9c09bcb
--- /dev/null
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/RemoteMultSeriesReader.java
@@ -0,0 +1,222 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.cluster.client.sync.SyncDataClient;
+import org.apache.iotdb.cluster.config.ClusterDescriptor;
+import org.apache.iotdb.cluster.server.RaftServer;
+import org.apache.iotdb.cluster.server.handlers.caller.GenericHandler;
+import org.apache.iotdb.db.utils.SerializeUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicReference;
+
+/** mult reader without value filter that reads points from a remote side. */
+public class RemoteMultSeriesReader extends AbstractMultPointReader {
+
+ private static final Logger logger =
LoggerFactory.getLogger(RemoteMultSeriesReader.class);
+ private static final int FETCH_BATCH_DATA_SIZE = 10;
+
+ private MultDataSourceInfo sourceInfo;
+
+ private Map<String, Queue<BatchData>> cachedBatchs;
+
+ private AtomicReference<Map<String, ByteBuffer>> fetchResult = new
AtomicReference<>();
+ private GenericHandler<Map<String, ByteBuffer>> handler;
+
+ private BatchStrategy batchStrategy;
+
+ private Map<String, BatchData> currentBatchDatas;
+
+ private Map<String, TSDataType> pathToDataType;
+
+ public RemoteMultSeriesReader(MultDataSourceInfo sourceInfo) {
+ this.sourceInfo = sourceInfo;
+ this.handler = new GenericHandler<>(sourceInfo.getCurrentNode(),
fetchResult);
+ this.currentBatchDatas = Maps.newHashMap();
+ this.batchStrategy = new DefaultBatchStrategy();
+
+ this.cachedBatchs = Maps.newHashMap();
+ this.pathToDataType = Maps.newHashMap();
+ for (int i = 0; i < sourceInfo.getPartialPaths().size(); i++) {
+ String fullPath = sourceInfo.getPartialPaths().get(i).getFullPath();
+ this.cachedBatchs.put(fullPath, new ConcurrentLinkedQueue<>());
+ this.pathToDataType.put(fullPath, sourceInfo.getDataTypes().get(i));
+ }
+ }
+
+ @Override
+ public boolean hasNextTimeValuePair(String fullPath) throws IOException {
+ BatchData batchData = currentBatchDatas.get(fullPath);
+ if (batchData != null && batchData.hasCurrent()) {
+ return true;
+ }
+ fetchBatch();
+ return checkPathBatchData(fullPath);
+ }
+
+ private boolean checkPathBatchData(String fullPath) {
+ BatchData batchData = cachedBatchs.get(fullPath).peek();
+ if (batchData != null && !batchData.isEmpty()) {
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public TimeValuePair nextTimeValuePair(String fullPath) throws IOException {
+ BatchData batchData = currentBatchDatas.get(fullPath);
+ if ((batchData == null || !batchData.hasCurrent()) &&
checkPathBatchData(fullPath)) {
+ batchData = cachedBatchs.get(fullPath).poll();
+ currentBatchDatas.put(fullPath, batchData);
+ }
+
+ if (!hasNextTimeValuePair(fullPath)) {
+ throw new NoSuchElementException();
+ }
+
+ TimeValuePair timeValuePair =
+ new TimeValuePair(
+ batchData.currentTime(),
+ TsPrimitiveType.getByType(pathToDataType.get(fullPath),
batchData.currentValue()));
+ batchData.next();
+ return timeValuePair;
+ }
+
+ @Override
+ public Set<String> getAllPaths() {
+ return cachedBatchs.keySet();
+ }
+
+ /** query resource deal close there is not dealing. */
+ @Override
+ public void close() {}
+
+ private void fetchBatch() throws IOException {
+ if (!sourceInfo.checkCurClient()) {
+ cachedBatchs = null;
+ return;
+ }
+ List<String> paths = batchStrategy.selectBatchPaths(this.cachedBatchs);
+ if (paths.isEmpty()) return;
+
+ Map<String, ByteBuffer> result;
+ if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) {
+ result = fetchResultAsync(paths);
+ } else {
+ result = fetchResultSync(paths);
+ }
+
+ if (result == null) return;
+
+ for (String path : result.keySet()) {
+
+ BatchData batchData =
SerializeUtils.deserializeBatchData(result.get(path));
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Fetched a batch from {}, size:{}",
+ sourceInfo.getCurrentNode(),
+ batchData == null ? 0 : batchData.length());
+ }
+ // if data query end, batchData is null,
+ // will create empty BatchData, and add queue.
+ if (batchData == null) {
+ batchData = new BatchData();
+ }
+ cachedBatchs
+ .computeIfAbsent(path, n -> new ConcurrentLinkedQueue<BatchData>())
+ .add(batchData);
+ }
+ }
+
+ @SuppressWarnings("java:S2274") // enable timeout
+ private Map<String, ByteBuffer> fetchResultAsync(List<String> paths) throws
IOException {
+ synchronized (fetchResult) {
+ fetchResult.set(null);
+ try {
+ sourceInfo
+ .getCurAsyncClient(RaftServer.getReadOperationTimeoutMS())
+ .fetchMultSeries(sourceInfo.getHeader(), sourceInfo.getReaderId(),
paths, handler);
+ fetchResult.wait(RaftServer.getReadOperationTimeoutMS());
+ } catch (TException | InterruptedException e) {
+ logger.error("Failed to fetch result async, connect to {}",
sourceInfo, e);
+ return null;
+ }
+ }
+ return fetchResult.get();
+ }
+
+ private Map<String, ByteBuffer> fetchResultSync(List<String> paths) throws
IOException {
+
+ try (SyncDataClient curSyncClient =
+ sourceInfo.getCurSyncClient(RaftServer.getReadOperationTimeoutMS()); )
{
+
+ return curSyncClient.fetchMultSeries(sourceInfo.getHeader(),
sourceInfo.getReaderId(), paths);
+ } catch (TException e) {
+ logger.error("Failed to fetch result sync, connect to {}", sourceInfo,
e);
+ return null;
+ }
+ }
+
+ /** select path, which could batch-fetch result */
+ interface BatchStrategy {
+ List<String> selectBatchPaths(Map<String, Queue<BatchData>> cacheBatchs);
+ }
+
+ static class DefaultBatchStrategy implements BatchStrategy {
+
+ @Override
+ public List<String> selectBatchPaths(Map<String, Queue<BatchData>>
cacheBatchs) {
+ List<String> paths = Lists.newArrayList();
+
+ for (String path : cacheBatchs.keySet()) {
+ Queue<BatchData> batchDataQueue = cacheBatchs.get(path);
+ BatchData batchData = batchDataQueue.peek();
+
+ // data read finished, so can not batch get data
+ if (batchData != null && batchData.isEmpty()) {
+ continue;
+ }
+
+ if (batchDataQueue.size() < FETCH_BATCH_DATA_SIZE) {
+ paths.add(path);
+ }
+ }
+ return paths;
+ }
+ }
+}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/DataClusterServer.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/DataClusterServer.java
index de815b3..a54de23 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/server/DataClusterServer.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/server/DataClusterServer.java
@@ -39,6 +39,7 @@ import org.apache.iotdb.cluster.rpc.thrift.GroupByRequest;
import org.apache.iotdb.cluster.rpc.thrift.HeartBeatRequest;
import org.apache.iotdb.cluster.rpc.thrift.HeartBeatResponse;
import org.apache.iotdb.cluster.rpc.thrift.LastQueryRequest;
+import org.apache.iotdb.cluster.rpc.thrift.MultSeriesQueryRequest;
import org.apache.iotdb.cluster.rpc.thrift.Node;
import org.apache.iotdb.cluster.rpc.thrift.PreviousFillRequest;
import org.apache.iotdb.cluster.rpc.thrift.PullSchemaRequest;
@@ -335,6 +336,17 @@ public class DataClusterServer extends RaftServer
}
@Override
+ public void queryMultSeries(
+ MultSeriesQueryRequest request, AsyncMethodCallback<Long> resultHandler)
throws TException {
+ DataAsyncService service =
+ getDataAsyncService(
+ request.getHeader(), resultHandler, "Query series:" +
request.getPath());
+ if (service != null) {
+ service.queryMultSeries(request, resultHandler);
+ }
+ }
+
+ @Override
public void fetchSingleSeries(
Node header, long readerId, AsyncMethodCallback<ByteBuffer>
resultHandler) {
DataAsyncService service =
@@ -345,6 +357,20 @@ public class DataClusterServer extends RaftServer
}
@Override
+ public void fetchMultSeries(
+ Node header,
+ long readerId,
+ List<String> paths,
+ AsyncMethodCallback<Map<String, ByteBuffer>> resultHandler)
+ throws TException {
+ DataAsyncService service =
+ getDataAsyncService(header, resultHandler, "Fetch reader:" + readerId);
+ if (service != null) {
+ service.fetchMultSeries(header, readerId, paths, resultHandler);
+ }
+ }
+
+ @Override
public void getAllPaths(
Node header,
List<String> paths,
@@ -738,11 +764,22 @@ public class DataClusterServer extends RaftServer
}
@Override
+ public long queryMultSeries(MultSeriesQueryRequest request) throws
TException {
+ return getDataSyncService(request.getHeader()).queryMultSeries(request);
+ }
+
+ @Override
public ByteBuffer fetchSingleSeries(Node header, long readerId) throws
TException {
return getDataSyncService(header).fetchSingleSeries(header, readerId);
}
@Override
+ public Map<String, ByteBuffer> fetchMultSeries(Node header, long readerId,
List<String> paths)
+ throws TException {
+ return getDataSyncService(header).fetchMultSeries(header, readerId, paths);
+ }
+
+ @Override
public long querySingleSeriesByTimestamp(SingleSeriesQueryRequest request)
throws TException {
return
getDataSyncService(request.getHeader()).querySingleSeriesByTimestamp(request);
}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/service/DataAsyncService.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/service/DataAsyncService.java
index 256feb2..49c5a1e 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/server/service/DataAsyncService.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/server/service/DataAsyncService.java
@@ -28,6 +28,7 @@ import
org.apache.iotdb.cluster.rpc.thrift.GetAggrResultRequest;
import org.apache.iotdb.cluster.rpc.thrift.GetAllPathsResult;
import org.apache.iotdb.cluster.rpc.thrift.GroupByRequest;
import org.apache.iotdb.cluster.rpc.thrift.LastQueryRequest;
+import org.apache.iotdb.cluster.rpc.thrift.MultSeriesQueryRequest;
import org.apache.iotdb.cluster.rpc.thrift.Node;
import org.apache.iotdb.cluster.rpc.thrift.PreviousFillRequest;
import org.apache.iotdb.cluster.rpc.thrift.PullSchemaRequest;
@@ -52,6 +53,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
+import java.util.Map;
import java.util.Set;
public class DataAsyncService extends BaseAsyncService implements
TSDataService.AsyncIface {
@@ -175,6 +177,16 @@ public class DataAsyncService extends BaseAsyncService
implements TSDataService.
}
@Override
+ public void queryMultSeries(
+ MultSeriesQueryRequest request, AsyncMethodCallback<Long> resultHandler)
throws TException {
+ try {
+
resultHandler.onComplete(dataGroupMember.getLocalQueryExecutor().queryMultSeries(request));
+ } catch (Exception e) {
+ resultHandler.onError(e);
+ }
+ }
+
+ @Override
public void querySingleSeriesByTimestamp(
SingleSeriesQueryRequest request, AsyncMethodCallback<Long>
resultHandler) {
try {
@@ -207,6 +219,21 @@ public class DataAsyncService extends BaseAsyncService
implements TSDataService.
}
@Override
+ public void fetchMultSeries(
+ Node header,
+ long readerId,
+ List<String> paths,
+ AsyncMethodCallback<Map<String, ByteBuffer>> resultHandler)
+ throws TException {
+ try {
+ resultHandler.onComplete(
+ dataGroupMember.getLocalQueryExecutor().fetchMultSeries(readerId,
paths));
+ } catch (ReaderNotFoundException | IOException e) {
+ resultHandler.onError(e);
+ }
+ }
+
+ @Override
public void fetchSingleSeriesByTimestamps(
Node header,
long readerId,
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/service/DataSyncService.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/service/DataSyncService.java
index 77ba4db..94ac92f 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/server/service/DataSyncService.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/server/service/DataSyncService.java
@@ -28,6 +28,7 @@ import
org.apache.iotdb.cluster.rpc.thrift.GetAggrResultRequest;
import org.apache.iotdb.cluster.rpc.thrift.GetAllPathsResult;
import org.apache.iotdb.cluster.rpc.thrift.GroupByRequest;
import org.apache.iotdb.cluster.rpc.thrift.LastQueryRequest;
+import org.apache.iotdb.cluster.rpc.thrift.MultSeriesQueryRequest;
import org.apache.iotdb.cluster.rpc.thrift.Node;
import org.apache.iotdb.cluster.rpc.thrift.PreviousFillRequest;
import org.apache.iotdb.cluster.rpc.thrift.PullSchemaRequest;
@@ -52,6 +53,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
+import java.util.Map;
import java.util.Set;
public class DataSyncService extends BaseSyncService implements
TSDataService.Iface {
@@ -197,6 +199,15 @@ public class DataSyncService extends BaseSyncService
implements TSDataService.If
}
@Override
+ public long queryMultSeries(MultSeriesQueryRequest request) throws
TException {
+ try {
+ return dataGroupMember.getLocalQueryExecutor().queryMultSeries(request);
+ } catch (Exception e) {
+ throw new TException(e);
+ }
+ }
+
+ @Override
public long querySingleSeriesByTimestamp(SingleSeriesQueryRequest request)
throws TException {
try {
return
dataGroupMember.getLocalQueryExecutor().querySingleSeriesByTimestamp(request);
@@ -224,6 +235,16 @@ public class DataSyncService extends BaseSyncService
implements TSDataService.If
}
@Override
+ public Map<String, ByteBuffer> fetchMultSeries(Node header, long readerId,
List<String> paths)
+ throws TException {
+ try {
+ return dataGroupMember.getLocalQueryExecutor().fetchMultSeries(readerId,
paths);
+ } catch (ReaderNotFoundException | IOException e) {
+ throw new TException(e);
+ }
+ }
+
+ @Override
public ByteBuffer fetchSingleSeriesByTimestamps(Node header, long readerId,
List<Long> timestamps)
throws TException {
try {
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/common/TestAsyncDataClient.java
b/cluster/src/test/java/org/apache/iotdb/cluster/common/TestAsyncDataClient.java
index 65d4847..c5461dc 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/common/TestAsyncDataClient.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/common/TestAsyncDataClient.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.cluster.rpc.thrift.ExecutNonQueryReq;
import org.apache.iotdb.cluster.rpc.thrift.GetAggrResultRequest;
import org.apache.iotdb.cluster.rpc.thrift.GetAllPathsResult;
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.PreviousFillRequest;
import org.apache.iotdb.cluster.rpc.thrift.PullSchemaRequest;
@@ -44,6 +45,7 @@ import org.apache.iotdb.db.qp.executor.PlanExecutor;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
+import org.apache.thrift.TException;
import org.apache.thrift.async.AsyncMethodCallback;
import java.io.File;
@@ -79,6 +81,24 @@ public class TestAsyncDataClient extends AsyncDataClient {
}
@Override
+ public void fetchMultSeries(
+ Node header,
+ long readerId,
+ List<String> paths,
+ AsyncMethodCallback<Map<String, ByteBuffer>> resultHandler) {
+ new Thread(
+ () -> {
+ try {
+ new DataAsyncService(dataGroupMemberMap.get(header))
+ .fetchMultSeries(header, readerId, paths, resultHandler);
+ } catch (TException e) {
+ e.printStackTrace();
+ }
+ })
+ .start();
+ }
+
+ @Override
public void getAggrResult(
GetAggrResultRequest request, AsyncMethodCallback<List<ByteBuffer>>
resultHandler) {
new Thread(
@@ -99,6 +119,21 @@ public class TestAsyncDataClient extends AsyncDataClient {
}
@Override
+ public void queryMultSeries(
+ MultSeriesQueryRequest request, AsyncMethodCallback<Long> resultHandler)
{
+ new Thread(
+ () -> {
+ try {
+ new
DataAsyncService(dataGroupMemberMap.get(request.getHeader()))
+ .queryMultSeries(request, resultHandler);
+ } catch (TException e) {
+ e.printStackTrace();
+ }
+ })
+ .start();
+ }
+
+ @Override
public void fetchSingleSeriesByTimestamps(
Node header,
long readerId,
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/mult/AssignPathManagedMergeReaderTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/mult/AssignPathManagedMergeReaderTest.java
new file mode 100644
index 0000000..5864a99
--- /dev/null
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/mult/AssignPathManagedMergeReaderTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.cluster.client.DataClientProvider;
+import org.apache.iotdb.cluster.client.async.AsyncDataClient;
+import org.apache.iotdb.cluster.common.TestMetaGroupMember;
+import org.apache.iotdb.cluster.common.TestUtils;
+import org.apache.iotdb.cluster.config.ClusterDescriptor;
+import org.apache.iotdb.cluster.partition.PartitionGroup;
+import org.apache.iotdb.cluster.query.RemoteQueryContext;
+import org.apache.iotdb.cluster.rpc.thrift.MultSeriesQueryRequest;
+import org.apache.iotdb.cluster.rpc.thrift.Node;
+import org.apache.iotdb.cluster.server.member.MetaGroupMember;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.utils.SerializeUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.thrift.TException;
+import org.apache.thrift.async.AsyncMethodCallback;
+import org.apache.thrift.protocol.TBinaryProtocol.Factory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentSkipListSet;
+
+import static junit.framework.TestCase.assertEquals;
+
+public class AssignPathManagedMergeReaderTest {
+
+ private AssignPathManagedMergeReader assignPathManagedMergeReader;
+ private RemoteMultSeriesReader reader;
+ private List<BatchData> batchData;
+ private boolean batchUsed;
+ private MetaGroupMember metaGroupMember;
+ private Set<Node> failedNodes = new ConcurrentSkipListSet<>();
+ private boolean prevUseAsyncServer;
+ private List<PartialPath> paths;
+ private List<TSDataType> dataTypes;
+
+ @Before
+ public void setUp() throws IllegalPathException {
+ paths = Lists.newArrayList();
+ dataTypes = Lists.newArrayList();
+ PartialPath partialPath = new PartialPath("root.a.b");
+ paths.add(partialPath);
+ partialPath = new PartialPath("root.a.c");
+ paths.add(partialPath);
+ dataTypes.add(TSDataType.DOUBLE);
+ dataTypes.add(TSDataType.INT32);
+ prevUseAsyncServer =
ClusterDescriptor.getInstance().getConfig().isUseAsyncServer();
+ batchData = Lists.newArrayList();
+ batchData.add(TestUtils.genBatchData(TSDataType.DOUBLE, 0, 100));
+ batchData.add(TestUtils.genBatchData(TSDataType.INT32, 0, 100));
+ batchUsed = false;
+ metaGroupMember = new TestMetaGroupMember();
+ assignPathManagedMergeReader = new
AssignPathManagedMergeReader("root.a.b", TSDataType.DOUBLE);
+ }
+
+ @After
+ public void tearDown() {
+
ClusterDescriptor.getInstance().getConfig().setUseAsyncServer(prevUseAsyncServer);
+ }
+
+ @Test
+ public void testMultManagerMergeRemoteSeriesReader() throws IOException,
StorageEngineException {
+ ClusterDescriptor.getInstance().getConfig().setUseAsyncServer(true);
+ PartitionGroup group = new PartitionGroup();
+ setAsyncDataClient();
+ group.add(TestUtils.getNode(0));
+ group.add(TestUtils.getNode(1));
+ group.add(TestUtils.getNode(2));
+
+ MultSeriesQueryRequest request = new MultSeriesQueryRequest();
+ RemoteQueryContext context = new RemoteQueryContext(1);
+
+ try {
+ MultDataSourceInfo sourceInfo =
+ new MultDataSourceInfo(group, paths, dataTypes, request, context,
metaGroupMember, group);
+ sourceInfo.hasNextDataClient(Long.MIN_VALUE);
+
+ reader = new RemoteMultSeriesReader(sourceInfo);
+ assignPathManagedMergeReader.addReader(reader, 0);
+
+ for (int i = 0; i < 100; i++) {
+ assertEquals(true,
assignPathManagedMergeReader.hasNextTimeValuePair());
+ TimeValuePair pair = assignPathManagedMergeReader.nextTimeValuePair();
+ assertEquals(i, pair.getTimestamp());
+ assertEquals(i * 1.0, pair.getValue().getDouble(), 0.00001);
+ }
+ assertEquals(false, assignPathManagedMergeReader.hasNextTimeValuePair());
+
+ } finally {
+ QueryResourceManager.getInstance().endQuery(context.getQueryId());
+ }
+ }
+
+ private void setAsyncDataClient() {
+ metaGroupMember.setClientProvider(
+ new DataClientProvider(new Factory()) {
+ @Override
+ public AsyncDataClient getAsyncDataClient(Node node, int timeout)
throws IOException {
+ return new AsyncDataClient(null, null, node, null) {
+ @Override
+ public void fetchMultSeries(
+ Node header,
+ long readerId,
+ List<String> paths,
+ AsyncMethodCallback<Map<String, ByteBuffer>> resultHandler)
+ throws TException {
+ if (failedNodes.contains(node)) {
+ throw new TException("Node down.");
+ }
+
+ new Thread(
+ () -> {
+ Map<String, ByteBuffer> stringByteBufferMap =
Maps.newHashMap();
+ if (batchUsed) {
+ paths.forEach(
+ path -> {
+ stringByteBufferMap.put(path,
ByteBuffer.allocate(0));
+ });
+ } else {
+ batchUsed = true;
+
+ for (int i = 0; i < batchData.size(); i++) {
+ stringByteBufferMap.put(
+ paths.get(i),
generateByteBuffer(batchData.get(i)));
+ }
+
+ resultHandler.onComplete(stringByteBufferMap);
+ }
+ })
+ .start();
+ }
+
+ @Override
+ public void queryMultSeries(
+ MultSeriesQueryRequest request, AsyncMethodCallback<Long>
resultHandler)
+ throws TException {
+ if (failedNodes.contains(node)) {
+ throw new TException("Node down.");
+ }
+
+ new Thread(() -> resultHandler.onComplete(1L)).start();
+ }
+ };
+ }
+ });
+ }
+
+ private ByteBuffer generateByteBuffer(BatchData batchData) {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new
DataOutputStream(byteArrayOutputStream);
+ SerializeUtils.serializeBatchData(batchData, dataOutputStream);
+ ByteBuffer byteBuffer =
ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+ return byteBuffer;
+ }
+}
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/mult/MultSeriesRawDataPointReaderTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/mult/MultSeriesRawDataPointReaderTest.java
new file mode 100644
index 0000000..9d30d4c
--- /dev/null
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/mult/MultSeriesRawDataPointReaderTest.java
@@ -0,0 +1,67 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.cluster.common.TestUtils;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.query.reader.series.SeriesRawDataPointReader;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.reader.IPointReader;
+
+import com.google.common.collect.Maps;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static junit.framework.TestCase.assertEquals;
+
+public class MultSeriesRawDataPointReaderTest {
+
+ private MultSeriesRawDataPointReader reader;
+
+ @Before
+ public void setUp() throws IllegalPathException, IOException {
+ BatchData batchData = TestUtils.genBatchData(TSDataType.DOUBLE, 0, 100);
+ Map<String, IPointReader> pointReaderMap = Maps.newHashMap();
+ SeriesRawDataPointReader seriesRawDataBatchReader =
+ Mockito.mock(SeriesRawDataPointReader.class);
+
Mockito.when(seriesRawDataBatchReader.hasNextTimeValuePair()).thenReturn(true);
+ TimeValuePair timeValuePair =
+ new TimeValuePair(batchData.currentTime(),
batchData.currentTsPrimitiveType());
+
Mockito.when(seriesRawDataBatchReader.nextTimeValuePair()).thenReturn(timeValuePair);
+ pointReaderMap.put("root.a.b", seriesRawDataBatchReader);
+ pointReaderMap.put("root.a.c", seriesRawDataBatchReader);
+ reader = new MultSeriesRawDataPointReader(pointReaderMap);
+ }
+
+ @Test
+ public void testMultSeriesReader() throws IOException,
StorageEngineException {
+ boolean hasNext = this.reader.hasNextTimeValuePair("root.a.b");
+ assertEquals(true, hasNext);
+ TimeValuePair timeValuePair = this.reader.nextTimeValuePair("root.a.b");
+ assertEquals(0, timeValuePair.getTimestamp());
+ assertEquals(0 * 1.0, timeValuePair.getValue().getDouble(), 0.0001);
+ }
+}
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/mult/RemoteMultSeriesReaderTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/mult/RemoteMultSeriesReaderTest.java
new file mode 100644
index 0000000..73e1a5d
--- /dev/null
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/mult/RemoteMultSeriesReaderTest.java
@@ -0,0 +1,286 @@
+/*
+ * 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.mult;
+
+import org.apache.iotdb.cluster.client.DataClientProvider;
+import org.apache.iotdb.cluster.client.async.AsyncDataClient;
+import org.apache.iotdb.cluster.client.sync.SyncDataClient;
+import org.apache.iotdb.cluster.common.TestMetaGroupMember;
+import org.apache.iotdb.cluster.common.TestUtils;
+import org.apache.iotdb.cluster.config.ClusterDescriptor;
+import org.apache.iotdb.cluster.partition.PartitionGroup;
+import org.apache.iotdb.cluster.query.RemoteQueryContext;
+import org.apache.iotdb.cluster.rpc.thrift.MultSeriesQueryRequest;
+import org.apache.iotdb.cluster.rpc.thrift.Node;
+import org.apache.iotdb.cluster.server.member.MetaGroupMember;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.utils.SerializeUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.thrift.TException;
+import org.apache.thrift.async.AsyncMethodCallback;
+import org.apache.thrift.protocol.TBinaryProtocol.Factory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentSkipListSet;
+
+import static junit.framework.TestCase.assertEquals;
+import static junit.framework.TestCase.assertFalse;
+import static junit.framework.TestCase.assertTrue;
+
+public class RemoteMultSeriesReaderTest {
+
+ private RemoteMultSeriesReader reader;
+ private List<BatchData> batchData;
+ private boolean batchUsed;
+ private MetaGroupMember metaGroupMember;
+ private Set<Node> failedNodes = new ConcurrentSkipListSet<>();
+ private boolean prevUseAsyncServer;
+ private List<PartialPath> paths;
+ private List<TSDataType> dataTypes;
+
+ @Before
+ public void setUp() throws IllegalPathException {
+ paths = Lists.newArrayList();
+ dataTypes = Lists.newArrayList();
+ PartialPath partialPath = new PartialPath("root.a.b");
+ paths.add(partialPath);
+ partialPath = new PartialPath("root.a.c");
+ paths.add(partialPath);
+ dataTypes.add(TSDataType.DOUBLE);
+ dataTypes.add(TSDataType.INT32);
+ prevUseAsyncServer =
ClusterDescriptor.getInstance().getConfig().isUseAsyncServer();
+ batchData = Lists.newArrayList();
+ batchData.add(TestUtils.genBatchData(TSDataType.DOUBLE, 0, 100));
+ batchData.add(TestUtils.genBatchData(TSDataType.INT32, 0, 100));
+ batchUsed = false;
+ metaGroupMember = new TestMetaGroupMember();
+ }
+
+ @After
+ public void tearDown() {
+
ClusterDescriptor.getInstance().getConfig().setUseAsyncServer(prevUseAsyncServer);
+ }
+
+ @Test
+ public void testAsyncMultSeriesReader() throws IOException,
StorageEngineException {
+ ClusterDescriptor.getInstance().getConfig().setUseAsyncServer(true);
+ PartitionGroup group = new PartitionGroup();
+ setAsyncDataClient();
+ group.add(TestUtils.getNode(0));
+ group.add(TestUtils.getNode(1));
+ group.add(TestUtils.getNode(2));
+
+ MultSeriesQueryRequest request = new MultSeriesQueryRequest();
+ RemoteQueryContext context = new RemoteQueryContext(1);
+
+ try {
+ MultDataSourceInfo sourceInfo =
+ new MultDataSourceInfo(group, paths, dataTypes, request, context,
metaGroupMember, group);
+ sourceInfo.hasNextDataClient(Long.MIN_VALUE);
+
+ reader = new RemoteMultSeriesReader(sourceInfo);
+
+ for (int i = 0; i < 100; i++) {
+ assertTrue(reader.hasNextTimeValuePair(paths.get(0).getFullPath()));
+ TimeValuePair pair =
reader.nextTimeValuePair(paths.get(0).getFullPath());
+ assertEquals(i, pair.getTimestamp());
+ assertEquals(i * 1.0, pair.getValue().getDouble(), 0.00001);
+ }
+ assertFalse(reader.hasNextTimeValuePair(paths.get(0).getFullPath()));
+
+ } finally {
+ QueryResourceManager.getInstance().endQuery(context.getQueryId());
+ }
+ }
+
+ @Test
+ public void testSyncMultSeriesReader() throws IOException,
StorageEngineException {
+ ClusterDescriptor.getInstance().getConfig().setUseAsyncServer(false);
+ setSyncDataClient();
+ PartitionGroup group = new PartitionGroup();
+ group.add(TestUtils.getNode(0));
+ group.add(TestUtils.getNode(1));
+ group.add(TestUtils.getNode(2));
+
+ MultSeriesQueryRequest request = new MultSeriesQueryRequest();
+ RemoteQueryContext context = new RemoteQueryContext(1);
+
+ try {
+ MultDataSourceInfo sourceInfo =
+ new MultDataSourceInfo(group, paths, dataTypes, request, context,
metaGroupMember, group);
+ sourceInfo.hasNextDataClient(Long.MIN_VALUE);
+
+ reader = new RemoteMultSeriesReader(sourceInfo);
+
+ for (int i = 0; i < 100; i++) {
+ assertTrue(reader.hasNextTimeValuePair(paths.get(0).getFullPath()));
+ TimeValuePair pair =
reader.nextTimeValuePair(paths.get(0).getFullPath());
+ assertEquals(i, pair.getTimestamp());
+ assertEquals(i * 1.0, pair.getValue().getDouble(), 0.00001);
+ }
+ assertFalse(reader.hasNextTimeValuePair(paths.get(0).getFullPath()));
+
+ for (int i = 0; i < 100; i++) {
+ assertTrue(reader.hasNextTimeValuePair(paths.get(1).getFullPath()));
+ TimeValuePair pair =
reader.nextTimeValuePair(paths.get(1).getFullPath());
+ assertEquals(i, pair.getTimestamp());
+ assertEquals(i * 1.0, pair.getValue().getInt(), 0.00001);
+ }
+ assertFalse(reader.hasNextTimeValuePair(paths.get(1).getFullPath()));
+ } finally {
+ QueryResourceManager.getInstance().endQuery(context.getQueryId());
+ }
+ }
+
+ @Test
+ public void testDefaultBatchStrategySelect() {
+ RemoteMultSeriesReader.DefaultBatchStrategy defaultBatchStrategy =
+ new RemoteMultSeriesReader.DefaultBatchStrategy();
+ Map cachedBatches = Maps.newHashMap();
+ Queue queue = new ConcurrentLinkedQueue<BatchData>();
+ batchData.forEach(
+ data -> {
+ queue.add(data);
+ });
+ cachedBatches.put("root.a.b", queue);
+ assertEquals(1,
defaultBatchStrategy.selectBatchPaths(cachedBatches).size());
+ }
+
+ private void setAsyncDataClient() {
+ metaGroupMember.setClientProvider(
+ new DataClientProvider(new Factory()) {
+ @Override
+ public AsyncDataClient getAsyncDataClient(Node node, int timeout)
throws IOException {
+ return new AsyncDataClient(null, null, node, null) {
+ @Override
+ public void fetchMultSeries(
+ Node header,
+ long readerId,
+ List<String> paths,
+ AsyncMethodCallback<Map<String, ByteBuffer>> resultHandler)
+ throws TException {
+ if (failedNodes.contains(node)) {
+ throw new TException("Node down.");
+ }
+
+ new Thread(
+ () -> {
+ Map<String, ByteBuffer> stringByteBufferMap =
Maps.newHashMap();
+ if (batchUsed) {
+ paths.forEach(
+ path -> {
+ stringByteBufferMap.put(path,
ByteBuffer.allocate(0));
+ });
+ } else {
+ batchUsed = true;
+
+ for (int i = 0; i < batchData.size(); i++) {
+ stringByteBufferMap.put(
+ paths.get(i),
generateByteBuffer(batchData.get(i)));
+ }
+
+ resultHandler.onComplete(stringByteBufferMap);
+ }
+ })
+ .start();
+ }
+
+ @Override
+ public void queryMultSeries(
+ MultSeriesQueryRequest request, AsyncMethodCallback<Long>
resultHandler)
+ throws TException {
+ if (failedNodes.contains(node)) {
+ throw new TException("Node down.");
+ }
+
+ new Thread(() -> resultHandler.onComplete(1L)).start();
+ }
+ };
+ }
+ });
+ }
+
+ private void setSyncDataClient() {
+ metaGroupMember.setClientProvider(
+ new DataClientProvider(new Factory()) {
+ @Override
+ public SyncDataClient getSyncDataClient(Node node, int timeout) {
+ return new SyncDataClient(null) {
+ @Override
+ public Map<String, ByteBuffer> fetchMultSeries(
+ Node header, long readerId, List<String> paths) throws
TException {
+ if (failedNodes.contains(node)) {
+ throw new TException("Node down.");
+ }
+
+ Map<String, ByteBuffer> stringByteBufferMap =
Maps.newHashMap();
+ if (batchUsed) {
+ paths.forEach(
+ path -> {
+ stringByteBufferMap.put(path, ByteBuffer.allocate(0));
+ });
+ } else {
+ batchUsed = true;
+ for (int i = 0; i < batchData.size(); i++) {
+ stringByteBufferMap.put(paths.get(i),
generateByteBuffer(batchData.get(i)));
+ }
+ }
+ return stringByteBufferMap;
+ }
+
+ @Override
+ public long queryMultSeries(MultSeriesQueryRequest request)
throws TException {
+ if (failedNodes.contains(node)) {
+ throw new TException("Node down.");
+ }
+
+ return 1L;
+ }
+ };
+ }
+ });
+ }
+
+ private ByteBuffer generateByteBuffer(BatchData batchData) {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new
DataOutputStream(byteArrayOutputStream);
+ SerializeUtils.serializeBatchData(batchData, dataOutputStream);
+ ByteBuffer byteBuffer =
ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+ return byteBuffer;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/CachedPriorityMergeReader.java
b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/CachedPriorityMergeReader.java
index 69c67c4..2f5a956 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/CachedPriorityMergeReader.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/CachedPriorityMergeReader.java
@@ -55,7 +55,8 @@ public class CachedPriorityMergeReader extends
PriorityMergeReader {
while (!heap.isEmpty() && cacheLimit < CACHE_SIZE) {
Element top = heap.peek();
if (lastTimestamp == null || top.currTime() != lastTimestamp) {
- TimeValuePairUtils.setTimeValuePair(top.timeValuePair,
timeValuePairCache[cacheLimit++]);
+ TimeValuePairUtils.setTimeValuePair(
+ top.getTimeValuePair(), timeValuePairCache[cacheLimit++]);
lastTimestamp = top.currTime();
}
// remove duplicates
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/DescPriorityMergeReader.java
b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/DescPriorityMergeReader.java
index d73e72b..d73fc14 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/DescPriorityMergeReader.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/DescPriorityMergeReader.java
@@ -31,8 +31,8 @@ public class DescPriorityMergeReader extends
PriorityMergeReader {
new PriorityQueue<>(
(o1, o2) -> {
int timeCompare =
- Long.compare(o2.timeValuePair.getTimestamp(),
o1.timeValuePair.getTimestamp());
- return timeCompare != 0 ? timeCompare :
o2.priority.compareTo(o1.priority);
+ Long.compare(o2.currPair().getTimestamp(),
o1.currPair().getTimestamp());
+ return timeCompare != 0 ? timeCompare :
o2.getPriority().compareTo(o1.getPriority());
});
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/Element.java
b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/Element.java
new file mode 100644
index 0000000..c5b3ad5
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/Element.java
@@ -0,0 +1,72 @@
+/*
+ * 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.query.reader.universal;
+
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.reader.IPointReader;
+
+import java.io.IOException;
+
+public class Element {
+
+ public PriorityMergeReader.MergeReaderPriority priority;
+ protected IPointReader reader;
+ public TimeValuePair timeValuePair;
+
+ public Element(
+ IPointReader reader,
+ TimeValuePair timeValuePair,
+ PriorityMergeReader.MergeReaderPriority priority) {
+ this.reader = reader;
+ this.timeValuePair = timeValuePair;
+ this.priority = priority;
+ }
+
+ public long currTime() {
+ return timeValuePair.getTimestamp();
+ }
+
+ public TimeValuePair currPair() {
+ return timeValuePair;
+ }
+
+ public boolean hasNext() throws IOException {
+ return reader.hasNextTimeValuePair();
+ }
+
+ public void next() throws IOException {
+ timeValuePair = reader.nextTimeValuePair();
+ }
+
+ public void close() throws IOException {
+ reader.close();
+ }
+
+ public IPointReader getReader() {
+ return reader;
+ }
+
+ public TimeValuePair getTimeValuePair() {
+ return timeValuePair;
+ }
+
+ public PriorityMergeReader.MergeReaderPriority getPriority() {
+ return priority;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java
b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java
index de377e4..0982f6d 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java
@@ -91,7 +91,7 @@ public class PriorityMergeReader implements IPointReader {
@Override
public TimeValuePair nextTimeValuePair() throws IOException {
Element top = heap.poll();
- TimeValuePair ret = top.timeValuePair;
+ TimeValuePair ret = top.getTimeValuePair();
TimeValuePair topNext = null;
if (top.hasNext()) {
top.next();
@@ -108,10 +108,10 @@ public class PriorityMergeReader implements IPointReader {
@Override
public TimeValuePair currentTimeValuePair() throws IOException {
- return heap.peek().timeValuePair;
+ return heap.peek().getTimeValuePair();
}
- private void updateHeap(long topTime, long topNextTime) throws IOException {
+ protected void updateHeap(long topTime, long topNextTime) throws IOException
{
while (!heap.isEmpty() && heap.peek().currTime() == topTime) {
Element e = heap.poll();
if (!e.hasNext()) {
@@ -143,39 +143,6 @@ public class PriorityMergeReader implements IPointReader {
}
}
- static class Element {
-
- IPointReader reader;
- TimeValuePair timeValuePair;
- MergeReaderPriority priority;
-
- Element(IPointReader reader, TimeValuePair timeValuePair,
MergeReaderPriority priority) {
- this.reader = reader;
- this.timeValuePair = timeValuePair;
- this.priority = priority;
- }
-
- long currTime() {
- return timeValuePair.getTimestamp();
- }
-
- TimeValuePair currPair() {
- return timeValuePair;
- }
-
- boolean hasNext() throws IOException {
- return reader.hasNextTimeValuePair();
- }
-
- void next() throws IOException {
- timeValuePair = reader.nextTimeValuePair();
- }
-
- void close() throws IOException {
- reader.close();
- }
- }
-
public static class MergeReaderPriority implements
Comparable<MergeReaderPriority> {
long version;
long offset;
diff --git a/thrift/src/main/thrift/cluster.thrift
b/thrift/src/main/thrift/cluster.thrift
index 1c2b95a..c8edbe3 100644
--- a/thrift/src/main/thrift/cluster.thrift
+++ b/thrift/src/main/thrift/cluster.thrift
@@ -186,6 +186,20 @@ struct SingleSeriesQueryRequest {
11: required int deduplicatedPathNum
}
+struct MultSeriesQueryRequest {
+ 1: required list<string> path
+ 2: optional binary timeFilterBytes
+ 3: optional binary valueFilterBytes
+ 4: required long queryId
+ 5: required Node requester
+ 6: required Node header
+ 7: required list<int> dataTypeOrdinal
+ 8: required map<string,set<string>> deviceMeasurements
+ 9: required bool ascending
+ 10: required int fetchSize
+ 11: required int deduplicatedPathNum
+}
+
struct PreviousFillRequest {
1: required string path
2: required long queryTime
@@ -331,12 +345,25 @@ service TSDataService extends RaftService {
long querySingleSeries(1:SingleSeriesQueryRequest request)
/**
+ * Query mult time series without value filter.
+ * @return a readerId >= 0 if the query succeeds, otherwise the query fails
+ **/
+ long queryMultSeries(1:MultSeriesQueryRequest request)
+
+ /**
* Fetch at max fetchSize time-value pairs using the resultSetId generated by
querySingleSeries.
* @return a ByteBuffer containing the serialized time-value pairs or an
empty buffer if there
* are not more results.
**/
binary fetchSingleSeries(1:Node header, 2:long readerId)
+ /**
+ * Fetch mult series at max fetchSize time-value pairs using the
resultSetId generated by querySingleSeries.
+ * @return a map containing key-value,the serialized time-value pairs or an
empty buffer if there
+ * are not more results.
+ **/
+ map<string,binary> fetchMultSeries(1:Node header, 2:long readerId,
list<string> paths)
+
/**
* Query a time series and generate an IReaderByTimestamp.
* @return a readerId >= 0 if the query succeeds, otherwise the query fails
