[GitHub] [iotdb] mychaow commented on a change in pull request #2875: [IOTDB-1117][Distributed]Batched creation and fetch of RemoteSeriesRe…

[GitBox <gi...@apache.org>]

mychaow commented on a change in pull request #2875:
URL: https://github.com/apache/iotdb/pull/2875#discussion_r601008516



##########
File path: cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutor.java
##########
@@ -58,6 +64,78 @@
     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 {
+    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) {

Review comment:
       I think AbstractMultPointReader -> AbstractMultiPointReader is better.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org