| /* |
| * 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.coordinator; |
| |
| import org.apache.iotdb.cluster.client.async.AsyncDataClient; |
| import org.apache.iotdb.cluster.client.sync.SyncDataClient; |
| import org.apache.iotdb.cluster.config.ClusterConstant; |
| import org.apache.iotdb.cluster.config.ClusterDescriptor; |
| import org.apache.iotdb.cluster.exception.ChangeMembershipException; |
| import org.apache.iotdb.cluster.exception.CheckConsistencyException; |
| import org.apache.iotdb.cluster.exception.UnknownLogTypeException; |
| import org.apache.iotdb.cluster.exception.UnsupportedPlanException; |
| import org.apache.iotdb.cluster.log.Log; |
| import org.apache.iotdb.cluster.metadata.CMManager; |
| import org.apache.iotdb.cluster.partition.PartitionGroup; |
| import org.apache.iotdb.cluster.query.ClusterPlanRouter; |
| import org.apache.iotdb.cluster.rpc.thrift.Node; |
| import org.apache.iotdb.cluster.rpc.thrift.RaftNode; |
| import org.apache.iotdb.cluster.rpc.thrift.RaftService; |
| import org.apache.iotdb.cluster.server.RaftServer; |
| import org.apache.iotdb.cluster.server.member.MetaGroupMember; |
| import org.apache.iotdb.cluster.server.monitor.Timer; |
| import org.apache.iotdb.cluster.utils.PartitionUtils; |
| import org.apache.iotdb.cluster.utils.StatusUtils; |
| import org.apache.iotdb.db.conf.IoTDBConstant; |
| import org.apache.iotdb.db.exception.metadata.IllegalPathException; |
| import org.apache.iotdb.db.exception.metadata.MetadataException; |
| import org.apache.iotdb.db.exception.metadata.PathNotExistException; |
| import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException; |
| import org.apache.iotdb.db.exception.query.QueryProcessException; |
| import org.apache.iotdb.db.metadata.PartialPath; |
| import org.apache.iotdb.db.qp.physical.BatchPlan; |
| import org.apache.iotdb.db.qp.physical.PhysicalPlan; |
| import org.apache.iotdb.db.qp.physical.crud.InsertMultiTabletPlan; |
| import org.apache.iotdb.db.qp.physical.crud.InsertPlan; |
| import org.apache.iotdb.db.qp.physical.crud.InsertRowsPlan; |
| import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan; |
| import org.apache.iotdb.db.qp.physical.crud.SetSchemaTemplatePlan; |
| import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan; |
| import org.apache.iotdb.db.qp.physical.sys.CreateMultiTimeSeriesPlan; |
| import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan; |
| import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan; |
| import org.apache.iotdb.db.service.IoTDB; |
| import org.apache.iotdb.rpc.RpcUtils; |
| import org.apache.iotdb.rpc.TSStatusCode; |
| import org.apache.iotdb.service.rpc.thrift.EndPoint; |
| import org.apache.iotdb.service.rpc.thrift.TSStatus; |
| |
| import org.apache.thrift.TException; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.CopyOnWriteArrayList; |
| import java.util.concurrent.CountDownLatch; |
| |
| /** Coordinator of client non-query request */ |
| public class Coordinator { |
| |
| private static final Logger logger = LoggerFactory.getLogger(Coordinator.class); |
| |
| private MetaGroupMember metaGroupMember; |
| |
| private String name; |
| private Node thisNode; |
| /** router calculates the partition groups that a partitioned plan should be sent to */ |
| private ClusterPlanRouter router; |
| |
| private static final String MSG_MULTIPLE_ERROR = |
| "The following errors occurred when executing " |
| + "the query, please retry or contact the DBA: "; |
| |
| public Coordinator(MetaGroupMember metaGroupMember) { |
| this.metaGroupMember = metaGroupMember; |
| this.name = metaGroupMember.getName(); |
| this.thisNode = metaGroupMember.getThisNode(); |
| } |
| |
| public Coordinator() {} |
| |
| public void setMetaGroupMember(MetaGroupMember metaGroupMember) { |
| this.metaGroupMember = metaGroupMember; |
| this.name = metaGroupMember.getName(); |
| this.thisNode = metaGroupMember.getThisNode(); |
| } |
| |
| public void setRouter(ClusterPlanRouter router) { |
| this.router = router; |
| } |
| |
| /** |
| * Execute a non-query plan. According to the type of the plan, the plan will be executed on all |
| * nodes (like timeseries deletion) or the nodes that belong to certain groups (like data |
| * ingestion). |
| * |
| * @param plan a non-query plan. |
| */ |
| public TSStatus executeNonQueryPlan(PhysicalPlan plan) { |
| TSStatus result; |
| long startTime = Timer.Statistic.COORDINATOR_EXECUTE_NON_QUERY.getOperationStartTime(); |
| if (PartitionUtils.isLocalNonQueryPlan(plan)) { |
| // run locally |
| result = executeNonQueryLocally(plan); |
| } else if (PartitionUtils.isGlobalMetaPlan(plan)) { |
| // forward the plan to all meta group nodes |
| result = metaGroupMember.processNonPartitionedMetaPlan(plan); |
| } else if (PartitionUtils.isGlobalDataPlan(plan)) { |
| // forward the plan to all data group nodes |
| result = processNonPartitionedDataPlan(plan); |
| } else { |
| // split the plan and forward them to some PartitionGroups |
| try { |
| result = processPartitionedPlan(plan); |
| } catch (UnsupportedPlanException e) { |
| return StatusUtils.getStatus(StatusUtils.UNSUPPORTED_OPERATION, e.getMessage()); |
| } |
| } |
| Timer.Statistic.COORDINATOR_EXECUTE_NON_QUERY.calOperationCostTimeFromStart(startTime); |
| return result; |
| } |
| |
| /** execute a non-query plan that is not necessary to be executed on other nodes. */ |
| private TSStatus executeNonQueryLocally(PhysicalPlan plan) { |
| boolean execRet; |
| try { |
| execRet = metaGroupMember.getLocalExecutor().processNonQuery(plan); |
| } catch (QueryProcessException e) { |
| if (e.getErrorCode() != TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()) { |
| logger.debug("meet error while processing non-query. ", e); |
| } else { |
| logger.warn("meet error while processing non-query. ", e); |
| } |
| return RpcUtils.getStatus(e.getErrorCode(), e.getMessage()); |
| } catch (Exception e) { |
| logger.error("{}: server Internal Error: ", IoTDBConstant.GLOBAL_DB_NAME, e); |
| return RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage()); |
| } |
| |
| return execRet |
| ? RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "Execute successfully") |
| : RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR); |
| } |
| |
| /** |
| * A non-partitioned plan (like DeleteData) should be executed on all data group nodes, so the |
| * DataGroupLeader should take the responsible to make sure that every node receives the plan. |
| * Thus the plan will be processed locally only by the DataGroupLeader and forwarded by non-leader |
| * nodes. |
| */ |
| private TSStatus processNonPartitionedDataPlan(PhysicalPlan plan) { |
| try { |
| if (plan instanceof DeleteTimeSeriesPlan) { |
| // as delete related plans may have abstract paths (paths with wildcards), we convert |
| // them to full paths so the executor nodes will not need to query the metadata holders, |
| // eliminating the risk that when they are querying the metadata holders, the timeseries |
| // has already been deleted |
| ((CMManager) IoTDB.metaManager).convertToFullPaths(plan); |
| } else { |
| // function convertToFullPaths has already sync leader |
| metaGroupMember.syncLeaderWithConsistencyCheck(true); |
| } |
| } catch (PathNotExistException e) { |
| if (plan.getPaths().isEmpty()) { |
| // only reports an error when there is no matching path |
| return StatusUtils.getStatus(StatusUtils.TIMESERIES_NOT_EXIST_ERROR, e.getMessage()); |
| } |
| } catch (CheckConsistencyException e) { |
| logger.debug( |
| "Forwarding global data plan {} to meta leader {}", plan, metaGroupMember.getLeader()); |
| metaGroupMember.waitLeader(); |
| return metaGroupMember.forwardPlan(plan, metaGroupMember.getLeader(), null); |
| } |
| try { |
| createSchemaIfNecessary(plan); |
| } catch (MetadataException | CheckConsistencyException e) { |
| logger.error("{}: Cannot find storage groups for {}", name, plan); |
| return StatusUtils.NO_STORAGE_GROUP; |
| } |
| List<PartitionGroup> globalGroups = metaGroupMember.getPartitionTable().getGlobalGroups(); |
| logger.debug("Forwarding global data plan {} to {} groups", plan, globalGroups.size()); |
| return forwardPlan(globalGroups, plan); |
| } |
| |
| public void createSchemaIfNecessary(PhysicalPlan plan) |
| throws MetadataException, CheckConsistencyException { |
| if (plan instanceof SetSchemaTemplatePlan) { |
| try { |
| IoTDB.metaManager.getStorageGroupPath( |
| new PartialPath(((SetSchemaTemplatePlan) plan).getPrefixPath())); |
| } catch (IllegalPathException e) { |
| // the plan has been checked |
| } catch (StorageGroupNotSetException e) { |
| ((CMManager) IoTDB.metaManager).createSchema(plan); |
| } |
| } |
| } |
| |
| /** |
| * A partitioned plan (like batch insertion) will be split into several sub-plans, each belongs to |
| * a data group. And these sub-plans will be sent to and executed on the corresponding groups |
| * separately. |
| */ |
| public TSStatus processPartitionedPlan(PhysicalPlan plan) throws UnsupportedPlanException { |
| logger.debug("{}: Received a partitioned plan {}", name, plan); |
| if (metaGroupMember.getPartitionTable() == null) { |
| logger.debug("{}: Partition table is not ready", name); |
| return StatusUtils.PARTITION_TABLE_NOT_READY; |
| } |
| |
| if (!checkPrivilegeForBatchExecution(plan)) { |
| return concludeFinalStatus( |
| plan, plan.getPaths().size(), true, null, false, null, Collections.emptyList()); |
| } |
| |
| // split the plan into sub-plans that each only involve one data group |
| Map<PhysicalPlan, PartitionGroup> planGroupMap; |
| try { |
| planGroupMap = splitPlan(plan); |
| } catch (CheckConsistencyException checkConsistencyException) { |
| return StatusUtils.getStatus( |
| StatusUtils.CONSISTENCY_FAILURE, checkConsistencyException.getMessage()); |
| } |
| |
| // the storage group is not found locally |
| if (planGroupMap == null || planGroupMap.isEmpty()) { |
| if ((plan instanceof InsertPlan |
| || plan instanceof CreateTimeSeriesPlan |
| || plan instanceof CreateAlignedTimeSeriesPlan |
| || plan instanceof CreateMultiTimeSeriesPlan) |
| && ClusterDescriptor.getInstance().getConfig().isEnableAutoCreateSchema()) { |
| |
| logger.debug("{}: No associated storage group found for {}, auto-creating", name, plan); |
| try { |
| ((CMManager) IoTDB.metaManager).createSchema(plan); |
| return processPartitionedPlan(plan); |
| } catch (MetadataException | CheckConsistencyException e) { |
| logger.error( |
| String.format("Failed to set storage group or create timeseries, because %s", e)); |
| } |
| } |
| logger.error("{}: Cannot find storage groups for {}", name, plan); |
| return StatusUtils.NO_STORAGE_GROUP; |
| } |
| logger.debug("{}: The data groups of {} are {}", name, plan, planGroupMap); |
| return forwardPlan(planGroupMap, plan); |
| } |
| |
| /** |
| * check if batch execution plan has privilege on any sg |
| * |
| * @param plan |
| * @return |
| */ |
| private boolean checkPrivilegeForBatchExecution(PhysicalPlan plan) { |
| if (plan instanceof BatchPlan) { |
| return ((BatchPlan) plan).getResults().size() != plan.getPaths().size(); |
| } else { |
| return true; |
| } |
| } |
| |
| /** |
| * Forward a plan to all DataGroupMember groups. Only when all nodes time out, will a TIME_OUT be |
| * returned. The error messages from each group (if any) will be compacted into one string. |
| * |
| * @param partitionGroups |
| * @param plan |
| */ |
| private TSStatus forwardPlan(List<PartitionGroup> partitionGroups, PhysicalPlan plan) { |
| // the error codes from the groups that cannot execute the plan |
| TSStatus status; |
| List<String> errorCodePartitionGroups = new ArrayList<>(); |
| for (PartitionGroup partitionGroup : partitionGroups) { |
| if (partitionGroup.contains(thisNode)) { |
| // the query should be handled by a group the local node is in, handle it with in the group |
| status = |
| metaGroupMember |
| .getLocalDataMember(partitionGroup.getHeader()) |
| .executeNonQueryPlan(plan); |
| logger.debug( |
| "Execute {} in a local group of {} with status {}", |
| plan, |
| partitionGroup.getHeader(), |
| status); |
| } else { |
| // forward the query to the group that should handle it |
| status = forwardPlan(plan, partitionGroup); |
| logger.debug( |
| "Forward {} to a remote group of {} with status {}", |
| plan, |
| partitionGroup.getHeader(), |
| status); |
| } |
| if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() |
| && !(plan instanceof SetSchemaTemplatePlan |
| && status.getCode() == TSStatusCode.DUPLICATED_TEMPLATE.getStatusCode()) |
| && !(plan instanceof DeleteTimeSeriesPlan |
| && status.getCode() == TSStatusCode.TIMESERIES_NOT_EXIST.getStatusCode())) { |
| // execution failed, record the error message |
| errorCodePartitionGroups.add( |
| String.format( |
| "[%s@%s:%s]", status.getCode(), partitionGroup.getHeader(), status.getMessage())); |
| } |
| } |
| if (errorCodePartitionGroups.isEmpty()) { |
| status = StatusUtils.OK; |
| } else { |
| status = |
| StatusUtils.getStatus( |
| StatusUtils.EXECUTE_STATEMENT_ERROR, MSG_MULTIPLE_ERROR + errorCodePartitionGroups); |
| } |
| logger.debug("{}: executed {} with answer {}", name, plan, status); |
| return status; |
| } |
| |
| public void sendLogToAllDataGroups(Log log) throws ChangeMembershipException { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Send log {} to all data groups: start", log); |
| } |
| |
| Map<PhysicalPlan, PartitionGroup> planGroupMap = router.splitAndRouteChangeMembershipLog(log); |
| List<String> errorCodePartitionGroups = new CopyOnWriteArrayList<>(); |
| CountDownLatch counter = new CountDownLatch(planGroupMap.size()); |
| for (Map.Entry<PhysicalPlan, PartitionGroup> entry : planGroupMap.entrySet()) { |
| metaGroupMember |
| .getAppendLogThreadPool() |
| .submit(() -> forwardChangeMembershipPlan(log, entry, errorCodePartitionGroups, counter)); |
| } |
| try { |
| counter.await(); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| throw new ChangeMembershipException( |
| String.format("Can not wait all data groups to apply %s", log)); |
| } |
| if (!errorCodePartitionGroups.isEmpty()) { |
| throw new ChangeMembershipException( |
| String.format("Apply %s failed with status {%s}", log, errorCodePartitionGroups)); |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("Send log {} to all data groups: end", log); |
| } |
| } |
| |
| private void forwardChangeMembershipPlan( |
| Log log, |
| Map.Entry<PhysicalPlan, PartitionGroup> entry, |
| List<String> errorCodePartitionGroups, |
| CountDownLatch counter) { |
| int retryTime = 0; |
| long startTime = System.currentTimeMillis(); |
| try { |
| while (true) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Send change membership log {} to data group {}, retry time: {}", |
| log, |
| entry.getValue(), |
| retryTime); |
| } |
| try { |
| TSStatus status = forwardToSingleGroup(entry); |
| if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Success to send change membership log {} to data group {}", |
| log, |
| entry.getValue()); |
| } |
| return; |
| } |
| long cost = System.currentTimeMillis() - startTime; |
| if (cost > ClusterDescriptor.getInstance().getConfig().getWriteOperationTimeoutMS()) { |
| errorCodePartitionGroups.add( |
| String.format( |
| "Forward change membership log %s to data group %s", log, entry.getValue())); |
| return; |
| } |
| Thread.sleep(ClusterConstant.RETRY_WAIT_TIME_MS); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| errorCodePartitionGroups.add(e.getMessage()); |
| return; |
| } |
| retryTime++; |
| } |
| } finally { |
| counter.countDown(); |
| } |
| } |
| |
| /** split a plan into several sub-plans, each belongs to only one data group. */ |
| private Map<PhysicalPlan, PartitionGroup> splitPlan(PhysicalPlan plan) |
| throws UnsupportedPlanException, CheckConsistencyException { |
| Map<PhysicalPlan, PartitionGroup> planGroupMap = null; |
| try { |
| planGroupMap = router.splitAndRoutePlan(plan); |
| } catch (StorageGroupNotSetException e) { |
| // synchronize with the leader to see if this node has unpulled storage groups |
| metaGroupMember.syncLeaderWithConsistencyCheck(true); |
| try { |
| planGroupMap = router.splitAndRoutePlan(plan); |
| } catch (MetadataException | UnknownLogTypeException ex) { |
| // ignore |
| } |
| } catch (MetadataException | UnknownLogTypeException e) { |
| logger.error("Cannot route plan {}", plan, e); |
| } |
| logger.debug("route plan {} with partitionGroup {}", plan, planGroupMap); |
| return planGroupMap; |
| } |
| |
| /** |
| * Forward plans to the DataGroupMember of one node in the corresponding group. Only when all |
| * nodes time out, will a TIME_OUT be returned. |
| * |
| * @param planGroupMap sub-plan -> belong data group pairs |
| */ |
| private TSStatus forwardPlan(Map<PhysicalPlan, PartitionGroup> planGroupMap, PhysicalPlan plan) { |
| // the error codes from the groups that cannot execute the plan |
| TSStatus status; |
| // need to create substatus for multiPlan |
| |
| // InsertTabletPlan, InsertMultiTabletPlan, InsertRowsPlan and CreateMultiTimeSeriesPlan |
| // contains many rows, |
| // each will correspond to a TSStatus as its execution result, |
| // as the plan is split and the sub-plans may have interleaving ranges, |
| // we must assure that each TSStatus is placed to the right position |
| // e.g., an InsertTabletPlan contains 3 rows, row1 and row3 belong to NodeA and row2 |
| // belongs to NodeB, when NodeA returns a success while NodeB returns a failure, the |
| // failure and success should be placed into proper positions in TSStatus.subStatus |
| if (plan instanceof InsertTabletPlan |
| || plan instanceof InsertMultiTabletPlan |
| || plan instanceof CreateMultiTimeSeriesPlan |
| || plan instanceof InsertRowsPlan) { |
| status = forwardMultiSubPlan(planGroupMap, plan); |
| } else if (planGroupMap.size() == 1) { |
| status = forwardToSingleGroup(planGroupMap.entrySet().iterator().next()); |
| } else { |
| status = forwardToMultipleGroup(planGroupMap); |
| } |
| boolean hasCreated = false; |
| try { |
| if (plan instanceof InsertPlan |
| && status.getCode() == TSStatusCode.TIMESERIES_NOT_EXIST.getStatusCode() |
| && ClusterDescriptor.getInstance().getConfig().isEnableAutoCreateSchema()) { |
| hasCreated = createTimeseriesForFailedInsertion(((InsertPlan) plan)); |
| } |
| } catch (MetadataException | CheckConsistencyException e) { |
| logger.error("{}: Cannot auto-create timeseries for {}", name, plan, e); |
| return StatusUtils.getStatus(StatusUtils.EXECUTE_STATEMENT_ERROR, e.getMessage()); |
| } |
| if (hasCreated) { |
| status = forwardPlan(planGroupMap, plan); |
| } |
| if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode() |
| && status.isSetRedirectNode()) { |
| status.setCode(TSStatusCode.NEED_REDIRECTION.getStatusCode()); |
| } |
| logger.debug("{}: executed {} with answer {}", name, plan, status); |
| return status; |
| } |
| |
| private boolean createTimeseriesForFailedInsertion(InsertPlan plan) |
| throws CheckConsistencyException, IllegalPathException { |
| // try to create timeseries |
| if (plan.getFailedMeasurements() != null) { |
| plan.getPlanFromFailed(); |
| } |
| return ((CMManager) IoTDB.metaManager).createTimeseries(plan); |
| } |
| |
| private TSStatus forwardToSingleGroup(Map.Entry<PhysicalPlan, PartitionGroup> entry) { |
| TSStatus result; |
| if (entry.getValue().contains(thisNode)) { |
| // the query should be handled by a group the local node is in, handle it with in the group |
| long startTime = |
| Timer.Statistic.META_GROUP_MEMBER_EXECUTE_NON_QUERY_IN_LOCAL_GROUP |
| .getOperationStartTime(); |
| logger.debug( |
| "Execute {} in a local group of {}", entry.getKey(), entry.getValue().getHeader()); |
| result = |
| metaGroupMember |
| .getLocalDataMember(entry.getValue().getHeader()) |
| .executeNonQueryPlan(entry.getKey()); |
| Timer.Statistic.META_GROUP_MEMBER_EXECUTE_NON_QUERY_IN_LOCAL_GROUP |
| .calOperationCostTimeFromStart(startTime); |
| } else { |
| // forward the query to the group that should handle it |
| long startTime = |
| Timer.Statistic.META_GROUP_MEMBER_EXECUTE_NON_QUERY_IN_REMOTE_GROUP |
| .getOperationStartTime(); |
| logger.debug( |
| "Forward {} to a remote group of {}", entry.getKey(), entry.getValue().getHeader()); |
| result = forwardPlan(entry.getKey(), entry.getValue()); |
| Timer.Statistic.META_GROUP_MEMBER_EXECUTE_NON_QUERY_IN_REMOTE_GROUP |
| .calOperationCostTimeFromStart(startTime); |
| } |
| return result; |
| } |
| |
| /** |
| * forward each sub-plan to its corresponding data group, if some groups goes wrong, the error |
| * messages from each group will be compacted into one string. |
| * |
| * @param planGroupMap sub-plan -> data group pairs |
| */ |
| private TSStatus forwardToMultipleGroup(Map<PhysicalPlan, PartitionGroup> planGroupMap) { |
| List<String> errorCodePartitionGroups = new ArrayList<>(); |
| TSStatus tmpStatus; |
| boolean allRedirect = true; |
| EndPoint endPoint = null; |
| for (Map.Entry<PhysicalPlan, PartitionGroup> entry : planGroupMap.entrySet()) { |
| tmpStatus = forwardToSingleGroup(entry); |
| if (tmpStatus.isSetRedirectNode()) { |
| endPoint = tmpStatus.getRedirectNode(); |
| } else { |
| allRedirect = false; |
| } |
| if (tmpStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { |
| // execution failed, record the error message |
| errorCodePartitionGroups.add( |
| String.format( |
| "[%s@%s:%s]", |
| tmpStatus.getCode(), entry.getValue().getHeader(), tmpStatus.getMessage())); |
| } |
| } |
| TSStatus status; |
| if (errorCodePartitionGroups.isEmpty()) { |
| if (allRedirect) { |
| status = StatusUtils.getStatus(TSStatusCode.NEED_REDIRECTION, endPoint); |
| } else { |
| status = StatusUtils.OK; |
| } |
| } else { |
| status = |
| StatusUtils.getStatus( |
| StatusUtils.EXECUTE_STATEMENT_ERROR, MSG_MULTIPLE_ERROR + errorCodePartitionGroups); |
| } |
| return status; |
| } |
| |
| /** |
| * Forward each sub-plan to its belonging data group, and combine responses from the groups. |
| * |
| * @param planGroupMap sub-plan -> data group pairs |
| */ |
| @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning |
| private TSStatus forwardMultiSubPlan( |
| Map<PhysicalPlan, PartitionGroup> planGroupMap, PhysicalPlan parentPlan) { |
| List<String> errorCodePartitionGroups = new ArrayList<>(); |
| TSStatus tmpStatus; |
| TSStatus[] subStatus = null; |
| boolean noFailure = true; |
| boolean isBatchFailure = false; |
| EndPoint endPoint = null; |
| int totalRowNum = parentPlan.getPaths().size(); |
| // send sub-plans to each belonging data group and collect results |
| for (Map.Entry<PhysicalPlan, PartitionGroup> entry : planGroupMap.entrySet()) { |
| tmpStatus = forwardToSingleGroup(entry); |
| logger.debug("{}: from {},{},{}", name, entry.getKey(), entry.getValue(), tmpStatus); |
| noFailure = (tmpStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) && noFailure; |
| isBatchFailure = |
| (tmpStatus.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) || isBatchFailure; |
| if (tmpStatus.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) { |
| if (parentPlan instanceof InsertTabletPlan) { |
| totalRowNum = ((InsertTabletPlan) parentPlan).getRowCount(); |
| } else if (parentPlan instanceof InsertMultiTabletPlan) { |
| // the subStatus is the two-dimensional array, |
| // The first dimension is the number of InsertTabletPlans, |
| // and the second dimension is the number of rows per InsertTabletPlan |
| totalRowNum = ((InsertMultiTabletPlan) parentPlan).getTabletsSize(); |
| } else if (parentPlan instanceof CreateMultiTimeSeriesPlan) { |
| totalRowNum = parentPlan.getPaths().size(); |
| } else if (parentPlan instanceof InsertRowsPlan) { |
| totalRowNum = ((InsertRowsPlan) parentPlan).getRowCount(); |
| } |
| |
| if (subStatus == null) { |
| subStatus = new TSStatus[totalRowNum]; |
| Arrays.fill(subStatus, RpcUtils.SUCCESS_STATUS); |
| } |
| // set the status from one group to the proper positions of the overall status |
| if (parentPlan instanceof InsertMultiTabletPlan) { |
| InsertMultiTabletPlan tmpMultiTabletPlan = ((InsertMultiTabletPlan) entry.getKey()); |
| for (int i = 0; i < tmpMultiTabletPlan.getInsertTabletPlanList().size(); i++) { |
| InsertTabletPlan tmpInsertTabletPlan = tmpMultiTabletPlan.getInsertTabletPlan(i); |
| int parentIndex = tmpMultiTabletPlan.getParentIndex(i); |
| int parentPlanRowCount = ((InsertMultiTabletPlan) parentPlan).getRowCount(parentIndex); |
| if (subStatus[parentIndex].subStatus == null) { |
| TSStatus[] tmpSubTsStatus = new TSStatus[parentPlanRowCount]; |
| Arrays.fill(tmpSubTsStatus, RpcUtils.SUCCESS_STATUS); |
| subStatus[parentIndex].subStatus = Arrays.asList(tmpSubTsStatus); |
| } |
| TSStatus[] reorderTsStatus = |
| subStatus[parentIndex].subStatus.toArray(new TSStatus[] {}); |
| |
| PartitionUtils.reordering( |
| tmpInsertTabletPlan, |
| reorderTsStatus, |
| tmpStatus.subStatus.toArray(new TSStatus[] {})); |
| subStatus[parentIndex].subStatus = Arrays.asList(reorderTsStatus); |
| } |
| } else if (parentPlan instanceof InsertTabletPlan) { |
| PartitionUtils.reordering( |
| (InsertTabletPlan) entry.getKey(), |
| subStatus, |
| tmpStatus.subStatus.toArray(new TSStatus[] {})); |
| } else if (parentPlan instanceof CreateMultiTimeSeriesPlan) { |
| CreateMultiTimeSeriesPlan subPlan = (CreateMultiTimeSeriesPlan) entry.getKey(); |
| for (int i = 0; i < subPlan.getIndexes().size(); i++) { |
| subStatus[subPlan.getIndexes().get(i)] = tmpStatus.subStatus.get(i); |
| } |
| } else if (parentPlan instanceof InsertRowsPlan) { |
| InsertRowsPlan subPlan = (InsertRowsPlan) entry.getKey(); |
| for (int i = 0; i < subPlan.getInsertRowPlanIndexList().size(); i++) { |
| subStatus[subPlan.getInsertRowPlanIndexList().get(i)] = tmpStatus.subStatus.get(i); |
| } |
| } |
| } |
| |
| if (tmpStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { |
| // execution failed, record the error message |
| errorCodePartitionGroups.add( |
| String.format( |
| "[%s@%s:%s:%s]", |
| tmpStatus.getCode(), |
| entry.getValue().getHeader(), |
| tmpStatus.getMessage(), |
| tmpStatus.subStatus)); |
| } |
| |
| if (tmpStatus.isSetRedirectNode()) { |
| boolean isLastInsertTabletPlan = |
| parentPlan instanceof InsertTabletPlan |
| && ((InsertTabletPlan) entry.getKey()).getMaxTime() |
| == ((InsertTabletPlan) parentPlan).getMaxTime(); |
| |
| boolean isLastInsertMultiTabletPlan = |
| parentPlan instanceof InsertMultiTabletPlan |
| && ((InsertMultiTabletPlan) entry.getKey()).getMaxTime() |
| == ((InsertMultiTabletPlan) parentPlan).getMaxTime(); |
| |
| if (isLastInsertTabletPlan || isLastInsertMultiTabletPlan) { |
| endPoint = tmpStatus.getRedirectNode(); |
| } |
| } |
| } |
| return concludeFinalStatus( |
| parentPlan, |
| totalRowNum, |
| noFailure, |
| endPoint, |
| isBatchFailure, |
| subStatus, |
| errorCodePartitionGroups); |
| } |
| |
| private TSStatus concludeFinalStatus( |
| PhysicalPlan parentPlan, |
| int totalRowNum, |
| boolean noFailure, |
| EndPoint endPoint, |
| boolean isBatchFailure, |
| TSStatus[] subStatus, |
| List<String> errorCodePartitionGroups) { |
| if (parentPlan instanceof InsertMultiTabletPlan |
| && !((InsertMultiTabletPlan) parentPlan).getResults().isEmpty()) { |
| if (subStatus == null) { |
| subStatus = new TSStatus[totalRowNum]; |
| Arrays.fill(subStatus, RpcUtils.SUCCESS_STATUS); |
| } |
| noFailure = false; |
| isBatchFailure = true; |
| for (Map.Entry<Integer, TSStatus> integerTSStatusEntry : |
| ((InsertMultiTabletPlan) parentPlan).getResults().entrySet()) { |
| subStatus[integerTSStatusEntry.getKey()] = integerTSStatusEntry.getValue(); |
| } |
| } |
| |
| if (parentPlan instanceof CreateMultiTimeSeriesPlan |
| && !((CreateMultiTimeSeriesPlan) parentPlan).getResults().isEmpty()) { |
| if (subStatus == null) { |
| subStatus = new TSStatus[totalRowNum]; |
| Arrays.fill(subStatus, RpcUtils.SUCCESS_STATUS); |
| } |
| noFailure = false; |
| isBatchFailure = true; |
| for (Map.Entry<Integer, TSStatus> integerTSStatusEntry : |
| ((CreateMultiTimeSeriesPlan) parentPlan).getResults().entrySet()) { |
| subStatus[integerTSStatusEntry.getKey()] = integerTSStatusEntry.getValue(); |
| } |
| } |
| |
| if (parentPlan instanceof InsertRowsPlan |
| && !((InsertRowsPlan) parentPlan).getResults().isEmpty()) { |
| if (subStatus == null) { |
| subStatus = new TSStatus[totalRowNum]; |
| Arrays.fill(subStatus, RpcUtils.SUCCESS_STATUS); |
| } |
| noFailure = false; |
| isBatchFailure = true; |
| for (Map.Entry<Integer, TSStatus> integerTSStatusEntry : |
| ((InsertRowsPlan) parentPlan).getResults().entrySet()) { |
| subStatus[integerTSStatusEntry.getKey()] = integerTSStatusEntry.getValue(); |
| } |
| } |
| |
| TSStatus status; |
| if (noFailure) { |
| status = StatusUtils.OK; |
| if (endPoint != null) { |
| status = StatusUtils.getStatus(status, endPoint); |
| } |
| } else if (isBatchFailure) { |
| status = RpcUtils.getStatus(Arrays.asList(subStatus)); |
| } else { |
| status = |
| StatusUtils.getStatus( |
| StatusUtils.EXECUTE_STATEMENT_ERROR, MSG_MULTIPLE_ERROR + errorCodePartitionGroups); |
| } |
| return status; |
| } |
| |
| /** |
| * Forward a plan to the DataGroupMember of one node in the group. Only when all nodes time out, |
| * will a TIME_OUT be returned. |
| */ |
| private TSStatus forwardPlan(PhysicalPlan plan, PartitionGroup group) { |
| for (Node node : group) { |
| TSStatus status; |
| try { |
| // only data plans are partitioned, so it must be processed by its data server instead of |
| // meta server |
| if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { |
| status = forwardDataPlanAsync(plan, node, group.getHeader()); |
| } else { |
| status = forwardDataPlanSync(plan, node, group.getHeader()); |
| } |
| } catch (IOException e) { |
| status = StatusUtils.getStatus(StatusUtils.EXECUTE_STATEMENT_ERROR, e.getMessage()); |
| } |
| if (!StatusUtils.TIME_OUT.equals(status)) { |
| if (!status.isSetRedirectNode()) { |
| status.setRedirectNode(new EndPoint(node.getClientIp(), node.getClientPort())); |
| } |
| return status; |
| } else { |
| logger.warn("Forward {} to {} timed out", plan, node); |
| } |
| } |
| logger.warn("Forward {} to {} timed out", plan, group); |
| return StatusUtils.TIME_OUT; |
| } |
| |
| /** |
| * Forward a non-query plan to the data port of "receiver" |
| * |
| * @param plan a non-query plan |
| * @param header to determine which DataGroupMember of "receiver" will process the request. |
| * @return a TSStatus indicating if the forwarding is successful. |
| */ |
| private TSStatus forwardDataPlanAsync(PhysicalPlan plan, Node receiver, RaftNode header) |
| throws IOException { |
| RaftService.AsyncClient client = |
| metaGroupMember |
| .getClientProvider() |
| .getAsyncDataClient(receiver, RaftServer.getWriteOperationTimeoutMS()); |
| return this.metaGroupMember.forwardPlanAsync(plan, receiver, header, client); |
| } |
| |
| private TSStatus forwardDataPlanSync(PhysicalPlan plan, Node receiver, RaftNode header) |
| throws IOException { |
| RaftService.Client client; |
| try { |
| client = |
| metaGroupMember |
| .getClientProvider() |
| .getSyncDataClient(receiver, RaftServer.getWriteOperationTimeoutMS()); |
| } catch (TException e) { |
| throw new IOException(e); |
| } |
| return this.metaGroupMember.forwardPlanSync(plan, receiver, header, client); |
| } |
| |
| /** |
| * Get a thrift client that will connect to "node" using the data port. |
| * |
| * @param node the node to be connected |
| * @param timeout timeout threshold of connection |
| */ |
| public AsyncDataClient getAsyncDataClient(Node node, int timeout) throws IOException { |
| return metaGroupMember.getClientProvider().getAsyncDataClient(node, timeout); |
| } |
| |
| public Node getThisNode() { |
| return thisNode; |
| } |
| |
| /** |
| * Get a thrift client that will connect to "node" using the data port. |
| * |
| * @param node the node to be connected |
| * @param timeout timeout threshold of connection |
| */ |
| public SyncDataClient getSyncDataClient(Node node, int timeout) throws TException { |
| return metaGroupMember.getClientProvider().getSyncDataClient(node, timeout); |
| } |
| } |
