[incubator-iotdb] 09/19: update

[lt...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lta pushed a commit to branch cluster_read
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git

commit 2134f33f64a86ece98d34caf5091a06e67b80ce3
Author: lta <li...@163.com>
AuthorDate: Mon Apr 15 18:05:13 2019 +0800

    update
---
 .../manager/ClusterRpcQueryManager.java            | 20 ++++++++-
 .../manager/IClusterQueryManager.java              | 49 +++++++++++++++++++++-
 .../cluster/service/TSServiceClusterImpl.java      | 43 +++++++++++++++++++
 .../org/apache/iotdb/db/service/TSServiceImpl.java |  4 +-
 4 files changed, 111 insertions(+), 5 deletions(-)

diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/coordinatornode/manager/ClusterRpcQueryManager.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/coordinatornode/manager/ClusterRpcQueryManager.java
index c289796..3df1ef6 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/query/coordinatornode/manager/ClusterRpcQueryManager.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/coordinatornode/manager/ClusterRpcQueryManager.java
@@ -22,16 +22,34 @@ import java.util.HashMap;
 import java.util.Map;
 import org.apache.iotdb.db.qp.physical.PhysicalPlan;
 
-public class ClusterRpcQueryManager {
+public class ClusterRpcQueryManager implements IClusterQueryManager{
 
   Map<Long, Map<String, PhysicalPlan>> selectPathPlan = new HashMap<>();
   Map<Long, Map<String, PhysicalPlan>> filterPathPlan = new HashMap<>();
 
   private ClusterRpcQueryManager(){
+  }
+
+  @Override
+  public void registerQuery(Long jobId, PhysicalPlan plan) {
+
+  }
 
+  @Override
+  public PhysicalPlan getSelectPathPhysicalPlan(Long jobId, String fullPath) {
+    return selectPathPlan.get(jobId).get(fullPath);
   }
 
+  @Override
+  public PhysicalPlan getFilterPathPhysicalPlan(Long jobId, String fullPath) {
+    return filterPathPlan.get(jobId).get(fullPath);
+  }
 
+  @Override
+  public void remove(Long jobId) {
+    selectPathPlan.remove(jobId);
+    filterPathPlan.remove(jobId);
+  }
 
   public static final ClusterRpcQueryManager getInstance() {
     return ClusterRpcQueryManagerHolder.INSTANCE;
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/coordinatornode/manager/IClusterQueryManager.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/coordinatornode/manager/IClusterQueryManager.java
index 6ce56f0..a7a1418 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/query/coordinatornode/manager/IClusterQueryManager.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/coordinatornode/manager/IClusterQueryManager.java
@@ -1,10 +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.coordinatornode.manager;
 
 import org.apache.iotdb.db.qp.physical.PhysicalPlan;
 
 public interface IClusterQueryManager {
 
-  void registerQuery(Long);
+  /**
+   * Register a query, divide physical plan into several sub physical plans according to timeseries
+   * full path.
+   *
+   * @param jobId Query job id assigned by QueryResourceManager.
+   * @param plan Physical plan parsed by QueryProcessor
+   */
+  void registerQuery(Long jobId, PhysicalPlan plan);
 
-  PhysicalPlan getPhysicalPlan();
+  /**
+   * Get physical plan of select path
+   *
+   * @param jobId Query job id assigned by QueryResourceManager.
+   * @param path Timeseries full path in select paths
+   */
+  PhysicalPlan getSelectPathPhysicalPlan(Long jobId, String path);
+
+  /**
+   * Get physical plan of filter path
+   *
+   * @param jobId Query job id assigned by QueryResourceManager.
+   * @param path Timeseries full path in filter
+   */
+  PhysicalPlan getFilterPathPhysicalPlan(Long jobId, String path);
+
+
+  /**
+   * Remove resource of a job id
+   */
+  void remove(Long jobId);
 }
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/service/TSServiceClusterImpl.java b/cluster/src/main/java/org/apache/iotdb/cluster/service/TSServiceClusterImpl.java
index 4013fa5..d33b759 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/service/TSServiceClusterImpl.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/service/TSServiceClusterImpl.java
@@ -25,6 +25,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -35,6 +36,7 @@ import org.apache.iotdb.cluster.qp.executor.QueryMetadataExecutor;
 import org.apache.iotdb.cluster.query.coordinatornode.executor.QueryProcessorExecutor;
 import org.apache.iotdb.db.auth.AuthException;
 import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.exception.FileNodeManagerException;
 import org.apache.iotdb.db.exception.PathErrorException;
 import org.apache.iotdb.db.exception.ProcessorException;
 import org.apache.iotdb.db.metadata.MManager;
@@ -43,15 +45,19 @@ import org.apache.iotdb.db.qp.QueryProcessor;
 import org.apache.iotdb.db.qp.logical.Operator;
 import org.apache.iotdb.db.qp.physical.PhysicalPlan;
 import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
+import org.apache.iotdb.db.query.context.QueryContext;
 import org.apache.iotdb.db.query.control.QueryResourceManager;
 import org.apache.iotdb.db.service.TSServiceImpl;
+import org.apache.iotdb.service.rpc.thrift.TSCloseOperationReq;
 import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementReq;
 import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementResp;
 import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq;
 import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp;
+import org.apache.iotdb.service.rpc.thrift.TSFetchResultsReq;
 import org.apache.iotdb.service.rpc.thrift.TSHandleIdentifier;
 import org.apache.iotdb.service.rpc.thrift.TSOperationHandle;
 import org.apache.iotdb.service.rpc.thrift.TS_StatusCode;
+import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
 import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
 import org.apache.iotdb.tsfile.read.common.Path;
 import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
@@ -291,6 +297,43 @@ public class TSServiceClusterImpl extends TSServiceImpl {
     }
   }
 
+  @Override
+  public void releaseQueryResource(TSCloseOperationReq req) throws FileNodeManagerException {
+    Map<String, Long> queryStatus =
+    if (contextMap == null) {
+      return;
+    }
+    if (req == null || req.queryId == -1) {
+      // end query for all the query tokens created by current thread
+      for (QueryContext context : contextMap.values()) {
+        QueryResourceManager.getInstance().endQueryForGivenJob(context.getJobId());
+      }
+    } else {
+      QueryResourceManager.getInstance()
+          .endQueryForGivenJob(contextMap.remove(req.queryId).getJobId());
+    }
+  }
+
+  @Override
+  public QueryDataSet createNewDataSet(String statement, int fetchSize, TSFetchResultsReq req)
+      throws PathErrorException, QueryFilterOptimizationException, FileNodeManagerException,
+      ProcessorException, IOException {
+    PhysicalPlan physicalPlan = queryStatus.get().get(statement);
+    processor.getExecutor().setFetchSize(fetchSize);
+
+    QueryContext context = new QueryContext(QueryResourceManager.getInstance().assignJobId());
+    Map<Long, QueryContext> contextMap = contextMapLocal.get();
+    if (contextMap == null) {
+      contextMap = new HashMap<>();
+      contextMapLocal.set(contextMap);
+    }
+    contextMap.put(req.queryId, context);
+
+    QueryDataSet queryDataSet = processor.getExecutor().processQuery((QueryPlan) physicalPlan,
+        context);
+    queryRet.get().put(statement, queryDataSet);
+    return queryDataSet;
+  }
   /**
    * Close cluster service
    */
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 298db9c..f48a7d8 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -198,7 +198,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
     return new TSCloseOperationResp(new TS_Status(TS_StatusCode.SUCCESS_STATUS));
   }
 
-  private void releaseQueryResource(TSCloseOperationReq req) throws FileNodeManagerException {
+  public void releaseQueryResource(TSCloseOperationReq req) throws FileNodeManagerException {
     Map<Long, QueryContext> contextMap = contextMapLocal.get();
     if (contextMap == null) {
       return;
@@ -691,7 +691,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
     }
   }
 
-  private QueryDataSet createNewDataSet(String statement, int fetchSize, TSFetchResultsReq req)
+  public QueryDataSet createNewDataSet(String statement, int fetchSize, TSFetchResultsReq req)
       throws PathErrorException, QueryFilterOptimizationException, FileNodeManagerException,
       ProcessorException, IOException {
     PhysicalPlan physicalPlan = queryStatus.get().get(statement);