/*
 * 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.qp.executor;

import org.apache.iotdb.db.auth.AuthException;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.auth.authorizer.BasicAuthorizer;
import org.apache.iotdb.db.auth.authorizer.IAuthorizer;
import org.apache.iotdb.db.auth.entity.PathPrivilege;
import org.apache.iotdb.db.auth.entity.Role;
import org.apache.iotdb.db.auth.entity.User;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.cq.ContinuousQueryService;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.cache.ChunkCache;
import org.apache.iotdb.db.engine.cache.TimeSeriesMetadataCache;
import org.apache.iotdb.db.engine.flush.pool.FlushTaskPoolManager;
import org.apache.iotdb.db.engine.merge.manage.MergeManager;
import org.apache.iotdb.db.engine.merge.manage.MergeManager.TaskStatus;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.engine.trigger.service.TriggerRegistrationService;
import org.apache.iotdb.db.exception.BatchProcessException;
import org.apache.iotdb.db.exception.ContinuousQueryException;
import org.apache.iotdb.db.exception.QueryIdNotExsitException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.TriggerExecutionException;
import org.apache.iotdb.db.exception.TriggerManagementException;
import org.apache.iotdb.db.exception.UDFRegistrationException;
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.metadata.mnode.IMNode;
import org.apache.iotdb.db.metadata.mnode.IMeasurementMNode;
import org.apache.iotdb.db.metadata.mnode.IStorageGroupMNode;
import org.apache.iotdb.db.metadata.mnode.MeasurementMNode;
import org.apache.iotdb.db.monitor.StatMonitor;
import org.apache.iotdb.db.qp.constant.SQLConstant;
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator.AuthorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan;
import org.apache.iotdb.db.qp.physical.crud.CreateTemplatePlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePartitionPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.GroupByTimeFillPlan;
import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
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.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowsOfOneDevicePlan;
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.LastQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryIndexPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.SetSchemaTemplatePlan;
import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.CountPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateFunctionPlan;
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.CreateTriggerPlan;
import org.apache.iotdb.db.qp.physical.sys.DataAuthPlan;
import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
import org.apache.iotdb.db.qp.physical.sys.DropFunctionPlan;
import org.apache.iotdb.db.qp.physical.sys.DropTriggerPlan;
import org.apache.iotdb.db.qp.physical.sys.FlushPlan;
import org.apache.iotdb.db.qp.physical.sys.KillQueryPlan;
import org.apache.iotdb.db.qp.physical.sys.LoadConfigurationPlan;
import org.apache.iotdb.db.qp.physical.sys.MergePlan;
import org.apache.iotdb.db.qp.physical.sys.OperateFilePlan;
import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowChildNodesPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowChildPathsPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowDevicesPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowFunctionsPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowLockInfoPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowStorageGroupPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowTTLPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.StartTriggerPlan;
import org.apache.iotdb.db.qp.physical.sys.StopTriggerPlan;
import org.apache.iotdb.db.qp.physical.sys.TracingPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryTimeManager;
import org.apache.iotdb.db.query.control.QueryTimeManager.QueryInfo;
import org.apache.iotdb.db.query.control.TracingManager;
import org.apache.iotdb.db.query.dataset.AlignByDeviceDataSet;
import org.apache.iotdb.db.query.dataset.ListDataSet;
import org.apache.iotdb.db.query.dataset.ShowContinuousQueriesResult;
import org.apache.iotdb.db.query.dataset.ShowDevicesDataSet;
import org.apache.iotdb.db.query.dataset.ShowTimeseriesDataSet;
import org.apache.iotdb.db.query.dataset.SingleDataSet;
import org.apache.iotdb.db.query.executor.IQueryRouter;
import org.apache.iotdb.db.query.executor.QueryRouter;
import org.apache.iotdb.db.query.udf.service.UDFRegistrationInformation;
import org.apache.iotdb.db.query.udf.service.UDFRegistrationService;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.tools.TsFileRewriteTool;
import org.apache.iotdb.db.utils.AuthUtils;
import org.apache.iotdb.db.utils.FileLoaderUtils;
import org.apache.iotdb.db.utils.TypeInferenceUtils;
import org.apache.iotdb.db.utils.UpgradeUtils;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.ChunkGroupMetadata;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.TimeseriesMetadata;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.iotdb.tsfile.read.query.dataset.EmptyDataSet;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CANCELLED;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CHILD_NODES;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CHILD_PATHS;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_COLUMN;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CONTINUOUS_QUERY_EVERY_INTERVAL;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CONTINUOUS_QUERY_FOR_INTERVAL;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CONTINUOUS_QUERY_NAME;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CONTINUOUS_QUERY_QUERY_SQL;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CONTINUOUS_QUERY_TARGET_PATH;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_COUNT;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CREATED_TIME;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_DEVICES;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_DONE;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_FUNCTION_CLASS;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_FUNCTION_NAME;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_FUNCTION_TYPE;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_ITEM;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_LOCK_INFO;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PRIVILEGE;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PROGRESS;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_ROLE;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_STORAGE_GROUP;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TASK_NAME;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TTL;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_USER;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_VALUE;
import static org.apache.iotdb.db.conf.IoTDBConstant.FUNCTION_TYPE_BUILTIN_UDAF;
import static org.apache.iotdb.db.conf.IoTDBConstant.FUNCTION_TYPE_BUILTIN_UDTF;
import static org.apache.iotdb.db.conf.IoTDBConstant.FUNCTION_TYPE_EXTERNAL_UDAF;
import static org.apache.iotdb.db.conf.IoTDBConstant.FUNCTION_TYPE_EXTERNAL_UDTF;
import static org.apache.iotdb.db.conf.IoTDBConstant.FUNCTION_TYPE_NATIVE;
import static org.apache.iotdb.db.conf.IoTDBConstant.QUERY_ID;
import static org.apache.iotdb.db.conf.IoTDBConstant.STATEMENT;
import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;

public class PlanExecutor implements IPlanExecutor {

  private static final Logger logger = LoggerFactory.getLogger(PlanExecutor.class);
  private static final Logger AUDIT_LOGGER =
      LoggerFactory.getLogger(IoTDBConstant.AUDIT_LOGGER_NAME);
  // for data query
  protected IQueryRouter queryRouter;
  // for administration
  private IAuthorizer authorizer;

  private static final String INSERT_MEASUREMENTS_FAILED_MESSAGE = "failed to insert measurements ";

  public PlanExecutor() throws QueryProcessException {
    queryRouter = new QueryRouter();
    try {
      authorizer = BasicAuthorizer.getInstance();
    } catch (AuthException e) {
      throw new QueryProcessException(e.getMessage());
    }
  }

  @Override
  public QueryDataSet processQuery(PhysicalPlan queryPlan, QueryContext context)
      throws IOException, StorageEngineException, QueryFilterOptimizationException,
          QueryProcessException, MetadataException, InterruptedException {
    if (queryPlan instanceof QueryPlan) {
      return processDataQuery((QueryPlan) queryPlan, context);
    } else if (queryPlan instanceof AuthorPlan) {
      return processAuthorQuery((AuthorPlan) queryPlan);
    } else if (queryPlan instanceof ShowPlan) {
      return processShowQuery((ShowPlan) queryPlan, context);
    } else {
      throw new QueryProcessException(String.format("Unrecognized query plan %s", queryPlan));
    }
  }

  @Override
  public boolean processNonQuery(PhysicalPlan plan)
      throws QueryProcessException, StorageGroupNotSetException, StorageEngineException {
    switch (plan.getOperatorType()) {
      case DELETE:
        delete((DeletePlan) plan);
        return true;
      case INSERT:
        insert((InsertRowPlan) plan);
        return true;
      case BATCH_INSERT_ONE_DEVICE:
        insert((InsertRowsOfOneDevicePlan) plan);
        return true;
      case BATCH_INSERT_ROWS:
        insert((InsertRowsPlan) plan);
        return true;
      case BATCH_INSERT:
        insertTablet((InsertTabletPlan) plan);
        return true;
      case MULTI_BATCH_INSERT:
        insertTablet((InsertMultiTabletPlan) plan);
        return true;
      case CREATE_ROLE:
      case DELETE_ROLE:
      case CREATE_USER:
      case REVOKE_USER_ROLE:
      case REVOKE_ROLE_PRIVILEGE:
      case REVOKE_USER_PRIVILEGE:
      case GRANT_ROLE_PRIVILEGE:
      case GRANT_USER_PRIVILEGE:
      case GRANT_USER_ROLE:
      case MODIFY_PASSWORD:
      case DELETE_USER:
        AuthorPlan author = (AuthorPlan) plan;
        return operateAuthor(author);
      case GRANT_WATERMARK_EMBEDDING:
        return operateWatermarkEmbedding(((DataAuthPlan) plan).getUsers(), true);
      case REVOKE_WATERMARK_EMBEDDING:
        return operateWatermarkEmbedding(((DataAuthPlan) plan).getUsers(), false);
      case DELETE_TIMESERIES:
        return deleteTimeSeries((DeleteTimeSeriesPlan) plan);
      case CREATE_TIMESERIES:
        return createTimeSeries((CreateTimeSeriesPlan) plan);
      case CREATE_ALIGNED_TIMESERIES:
        return createAlignedTimeSeries((CreateAlignedTimeSeriesPlan) plan);
      case CREATE_MULTI_TIMESERIES:
        return createMultiTimeSeries((CreateMultiTimeSeriesPlan) plan);
      case ALTER_TIMESERIES:
        return alterTimeSeries((AlterTimeSeriesPlan) plan);
      case SET_STORAGE_GROUP:
        return setStorageGroup((SetStorageGroupPlan) plan);
      case DELETE_STORAGE_GROUP:
        return deleteStorageGroups((DeleteStorageGroupPlan) plan);
      case TTL:
        operateTTL((SetTTLPlan) plan);
        return true;
      case LOAD_CONFIGURATION:
        loadConfiguration((LoadConfigurationPlan) plan);
        return true;
      case LOAD_FILES:
        operateLoadFiles((OperateFilePlan) plan);
        return true;
      case REMOVE_FILE:
        operateRemoveFile((OperateFilePlan) plan);
        return true;
      case MOVE_FILE:
        operateMoveFile((OperateFilePlan) plan);
        return true;
      case FLUSH:
        operateFlush((FlushPlan) plan);
        return true;
      case MERGE:
      case FULL_MERGE:
        operateMerge((MergePlan) plan);
        return true;
      case TRACING:
        operateTracing((TracingPlan) plan);
        return true;
      case CLEAR_CACHE:
        operateClearCache();
        return true;
      case DELETE_PARTITION:
        DeletePartitionPlan p = (DeletePartitionPlan) plan;
        TimePartitionFilter filter =
            (storageGroupName, partitionId) ->
                storageGroupName.equals(
                        ((DeletePartitionPlan) plan).getStorageGroupName().getFullPath())
                    && p.getPartitionId().contains(partitionId);
        StorageEngine.getInstance()
            .removePartitions(((DeletePartitionPlan) plan).getStorageGroupName(), filter);
        return true;
      case CREATE_SCHEMA_SNAPSHOT:
        operateCreateSnapshot();
        return true;
      case CREATE_FUNCTION:
        return operateCreateFunction((CreateFunctionPlan) plan);
      case DROP_FUNCTION:
        return operateDropFunction((DropFunctionPlan) plan);
      case CREATE_TRIGGER:
        return operateCreateTrigger((CreateTriggerPlan) plan);
      case DROP_TRIGGER:
        return operateDropTrigger((DropTriggerPlan) plan);
      case START_TRIGGER:
        return operateStartTrigger((StartTriggerPlan) plan);
      case STOP_TRIGGER:
        return operateStopTrigger((StopTriggerPlan) plan);
      case CREATE_INDEX:
        throw new QueryProcessException("Create index hasn't been supported yet");
      case DROP_INDEX:
        throw new QueryProcessException("Drop index hasn't been supported yet");
      case KILL:
        try {
          operateKillQuery((KillQueryPlan) plan);
        } catch (QueryIdNotExsitException e) {
          throw new QueryProcessException(e.getMessage());
        }
        return true;
      case CREATE_TEMPLATE:
        return createSchemaTemplate((CreateTemplatePlan) plan);
      case SET_DEVICE_TEMPLATE:
        return setSchemaTemplate((SetSchemaTemplatePlan) plan);
      case CREATE_CONTINUOUS_QUERY:
        return operateCreateContinuousQuery((CreateContinuousQueryPlan) plan);
      case DROP_CONTINUOUS_QUERY:
        return operateDropContinuousQuery((DropContinuousQueryPlan) plan);
      default:
        throw new UnsupportedOperationException(
            String.format("operation %s is not supported", plan.getOperatorType()));
    }
  }

  private boolean createSchemaTemplate(CreateTemplatePlan createTemplatePlan)
      throws QueryProcessException {
    try {
      IoTDB.metaManager.createSchemaTemplate(createTemplatePlan);
    } catch (MetadataException e) {
      throw new QueryProcessException(e);
    }
    return true;
  }

  private boolean setSchemaTemplate(SetSchemaTemplatePlan setSchemaTemplatePlan)
      throws QueryProcessException {
    try {
      IoTDB.metaManager.setSchemaTemplate(setSchemaTemplatePlan);
    } catch (MetadataException e) {
      throw new QueryProcessException(e);
    }
    return true;
  }

  private boolean operateCreateFunction(CreateFunctionPlan plan) throws UDFRegistrationException {
    UDFRegistrationService.getInstance()
        .register(plan.getUdfName(), plan.getClassName(), plan.isTemporary(), true);
    return true;
  }

  private boolean operateDropFunction(DropFunctionPlan plan) throws UDFRegistrationException {
    UDFRegistrationService.getInstance().deregister(plan.getUdfName());
    return true;
  }

  private boolean operateCreateTrigger(CreateTriggerPlan plan)
      throws TriggerManagementException, TriggerExecutionException {
    TriggerRegistrationService.getInstance().register(plan);
    return true;
  }

  private boolean operateDropTrigger(DropTriggerPlan plan) throws TriggerManagementException {
    TriggerRegistrationService.getInstance().deregister(plan);
    return true;
  }

  private boolean operateStartTrigger(StartTriggerPlan plan)
      throws TriggerManagementException, TriggerExecutionException {
    TriggerRegistrationService.getInstance().activate(plan);
    return true;
  }

  private boolean operateStopTrigger(StopTriggerPlan plan) throws TriggerManagementException {
    TriggerRegistrationService.getInstance().inactivate(plan);
    return true;
  }

  private void operateMerge(MergePlan plan) throws StorageEngineException {
    if (plan.getOperatorType() == OperatorType.FULL_MERGE) {
      StorageEngine.getInstance().mergeAll(true);
    } else {
      StorageEngine.getInstance().mergeAll(false);
    }
  }

  private void operateClearCache() {
    ChunkCache.getInstance().clear();
    TimeSeriesMetadataCache.getInstance().clear();
  }

  private void operateCreateSnapshot() {
    IoTDB.metaManager.createMTreeSnapshot();
  }

  private void operateKillQuery(KillQueryPlan killQueryPlan) throws QueryIdNotExsitException {
    QueryTimeManager queryTimeManager = QueryTimeManager.getInstance();
    long killQueryId = killQueryPlan.getQueryId();
    if (killQueryId != -1) {
      if (queryTimeManager.getQueryInfoMap().get(killQueryId) != null) {
        queryTimeManager.killQuery(killQueryId);
      } else {
        throw new QueryIdNotExsitException(
            String.format(
                "Query Id %d is not exist, please check it.", killQueryPlan.getQueryId()));
      }
    } else {
      // if queryId is not specified, kill all running queries
      if (!queryTimeManager.getQueryInfoMap().isEmpty()) {
        synchronized (queryTimeManager.getQueryInfoMap()) {
          List<Long> queryIdList = new ArrayList<>(queryTimeManager.getQueryInfoMap().keySet());
          for (Long queryId : queryIdList) {
            queryTimeManager.killQuery(queryId);
          }
        }
      }
    }
  }

  /** when tracing off need Close the stream */
  private void operateTracing(TracingPlan plan) {
    IoTDBDescriptor.getInstance().getConfig().setEnablePerformanceTracing(plan.isTracingOn());
    if (!plan.isTracingOn()) {
      TracingManager.getInstance().close();
    } else {
      if (!TracingManager.getInstance().getWriterStatus()) {
        TracingManager.getInstance().openTracingWriteStream();
      }
    }
  }

  private void operateFlush(FlushPlan plan) throws StorageGroupNotSetException {
    if (plan.getPaths().isEmpty()) {
      StorageEngine.getInstance().syncCloseAllProcessor();
    } else {
      flushSpecifiedStorageGroups(plan);
    }

    if (!plan.getPaths().isEmpty()) {
      List<PartialPath> noExistSg = checkStorageGroupExist(plan.getPaths());
      if (!noExistSg.isEmpty()) {
        StringBuilder sb = new StringBuilder();
        noExistSg.forEach(storageGroup -> sb.append(storageGroup.getFullPath()).append(","));
        throw new StorageGroupNotSetException(sb.subSequence(0, sb.length() - 1).toString(), true);
      }
    }
  }

  private boolean operateCreateContinuousQuery(CreateContinuousQueryPlan plan)
      throws ContinuousQueryException {
    return ContinuousQueryService.getInstance().register(plan, true);
  }

  private boolean operateDropContinuousQuery(DropContinuousQueryPlan plan)
      throws ContinuousQueryException {
    return ContinuousQueryService.getInstance().deregister(plan);
  }

  public static void flushSpecifiedStorageGroups(FlushPlan plan)
      throws StorageGroupNotSetException {
    Map<PartialPath, List<Pair<Long, Boolean>>> storageGroupMap =
        plan.getStorageGroupPartitionIds();
    for (Entry<PartialPath, List<Pair<Long, Boolean>>> entry : storageGroupMap.entrySet()) {
      PartialPath storageGroupName = entry.getKey();
      // normal flush
      if (entry.getValue() == null) {
        if (plan.isSeq() == null) {
          StorageEngine.getInstance()
              .closeStorageGroupProcessor(storageGroupName, true, plan.isSync());
          StorageEngine.getInstance()
              .closeStorageGroupProcessor(storageGroupName, false, plan.isSync());
        } else {
          StorageEngine.getInstance()
              .closeStorageGroupProcessor(storageGroupName, plan.isSeq(), plan.isSync());
        }
      }
      // partition specified flush, for snapshot flush plan
      else {
        List<Pair<Long, Boolean>> partitionIdSequencePairs = entry.getValue();
        for (Pair<Long, Boolean> pair : partitionIdSequencePairs) {
          StorageEngine.getInstance()
              .closeStorageGroupProcessor(storageGroupName, pair.left, pair.right, true);
        }
      }
    }
  }

  protected QueryDataSet processDataQuery(QueryPlan queryPlan, QueryContext context)
      throws StorageEngineException, QueryFilterOptimizationException, QueryProcessException,
          IOException, InterruptedException {
    QueryDataSet queryDataSet;
    if (queryPlan instanceof AlignByDevicePlan) {
      queryDataSet = getAlignByDeviceDataSet((AlignByDevicePlan) queryPlan, context, queryRouter);
    } else {
      if (queryPlan.getPaths() == null || queryPlan.getPaths().isEmpty()) {
        // no time series are selected, return EmptyDataSet
        return new EmptyDataSet();
      } else if (queryPlan instanceof UDTFPlan) {
        UDTFPlan udtfPlan = (UDTFPlan) queryPlan;
        queryDataSet = queryRouter.udtfQuery(udtfPlan, context);
      } else if (queryPlan instanceof GroupByTimeFillPlan) {
        GroupByTimeFillPlan groupByFillPlan = (GroupByTimeFillPlan) queryPlan;
        queryDataSet = queryRouter.groupByFill(groupByFillPlan, context);
      } else if (queryPlan instanceof GroupByTimePlan) {
        GroupByTimePlan groupByTimePlan = (GroupByTimePlan) queryPlan;
        queryDataSet = queryRouter.groupBy(groupByTimePlan, context);
      } else if (queryPlan instanceof QueryIndexPlan) {
        throw new QueryProcessException("Query index hasn't been supported yet");
      } else if (queryPlan instanceof AggregationPlan) {
        AggregationPlan aggregationPlan = (AggregationPlan) queryPlan;
        queryDataSet = queryRouter.aggregate(aggregationPlan, context);
      } else if (queryPlan instanceof FillQueryPlan) {
        FillQueryPlan fillQueryPlan = (FillQueryPlan) queryPlan;
        queryDataSet = queryRouter.fill(fillQueryPlan, context);
      } else if (queryPlan instanceof LastQueryPlan) {
        queryDataSet = queryRouter.lastQuery((LastQueryPlan) queryPlan, context);
      } else {
        queryDataSet = queryRouter.rawDataQuery((RawDataQueryPlan) queryPlan, context);
      }
    }
    queryDataSet.setRowLimit(queryPlan.getRowLimit());
    queryDataSet.setRowOffset(queryPlan.getRowOffset());
    queryDataSet.setWithoutAllNull(queryPlan.isWithoutAllNull());
    queryDataSet.setWithoutAnyNull(queryPlan.isWithoutAnyNull());
    return queryDataSet;
  }

  protected AlignByDeviceDataSet getAlignByDeviceDataSet(
      AlignByDevicePlan plan, QueryContext context, IQueryRouter router) {
    return new AlignByDeviceDataSet(plan, context, router);
  }

  protected QueryDataSet processShowQuery(ShowPlan showPlan, QueryContext context)
      throws QueryProcessException, MetadataException {
    switch (showPlan.getShowContentType()) {
      case TTL:
        return processShowTTLQuery((ShowTTLPlan) showPlan);
      case FLUSH_TASK_INFO:
        return processShowFlushTaskInfo();
      case VERSION:
        return processShowVersion();
      case TIMESERIES:
        return processShowTimeseries((ShowTimeSeriesPlan) showPlan, context);
      case STORAGE_GROUP:
        return processShowStorageGroup((ShowStorageGroupPlan) showPlan);
      case LOCK_INFO:
        return processShowLockInfo((ShowLockInfoPlan) showPlan);
      case DEVICES:
        return processShowDevices((ShowDevicesPlan) showPlan);
      case CHILD_PATH:
        return processShowChildPaths((ShowChildPathsPlan) showPlan);
      case CHILD_NODE:
        return processShowChildNodes((ShowChildNodesPlan) showPlan);
      case COUNT_TIMESERIES:
        return processCountTimeSeries((CountPlan) showPlan);
      case COUNT_NODE_TIMESERIES:
        return processCountNodeTimeSeries((CountPlan) showPlan);
      case COUNT_DEVICES:
        return processCountDevices((CountPlan) showPlan);
      case COUNT_STORAGE_GROUP:
        return processCountStorageGroup((CountPlan) showPlan);
      case COUNT_NODES:
        return processCountNodes((CountPlan) showPlan);
      case MERGE_STATUS:
        return processShowMergeStatus();
      case QUERY_PROCESSLIST:
        return processShowQueryProcesslist();
      case FUNCTIONS:
        return processShowFunctions((ShowFunctionsPlan) showPlan);
      case TRIGGERS:
        return processShowTriggers();
      case CONTINUOUS_QUERY:
        return processShowContinuousQueries();
      default:
        throw new QueryProcessException(String.format("Unrecognized show plan %s", showPlan));
    }
  }

  private QueryDataSet processCountNodes(CountPlan countPlan) throws MetadataException {
    int num = getNodesNumInGivenLevel(countPlan.getPath(), countPlan.getLevel());
    return createSingleDataSet(COLUMN_COUNT, TSDataType.INT32, num);
  }

  private QueryDataSet processCountNodeTimeSeries(CountPlan countPlan) throws MetadataException {
    // get the nodes that need to group by first
    List<PartialPath> nodes = getNodesList(countPlan.getPath(), countPlan.getLevel());
    ListDataSet listDataSet =
        new ListDataSet(
            Arrays.asList(
                new PartialPath(COLUMN_COLUMN, false), new PartialPath(COLUMN_COUNT, false)),
            Arrays.asList(TSDataType.TEXT, TSDataType.INT32));
    for (PartialPath columnPath : nodes) {
      RowRecord record = new RowRecord(0);
      Field field = new Field(TSDataType.TEXT);
      field.setBinaryV(new Binary(columnPath.getFullPath()));
      Field field1 = new Field(TSDataType.INT32);
      // get the count of every group
      field1.setIntV(getPathsNum(columnPath));
      record.addField(field);
      record.addField(field1);
      listDataSet.putRecord(record);
    }
    return listDataSet;
  }

  private QueryDataSet processCountDevices(CountPlan countPlan) throws MetadataException {
    int num = getDevicesNum(countPlan.getPath());
    return createSingleDataSet(COLUMN_DEVICES, TSDataType.INT32, num);
  }

  private QueryDataSet processCountStorageGroup(CountPlan countPlan) throws MetadataException {
    int num = getStorageGroupNum(countPlan.getPath());
    return createSingleDataSet(COLUMN_STORAGE_GROUP, TSDataType.INT32, num);
  }

  private QueryDataSet createSingleDataSet(String columnName, TSDataType columnType, Object val) {
    SingleDataSet singleDataSet =
        new SingleDataSet(
            Collections.singletonList(new PartialPath(columnName, false)),
            Collections.singletonList(columnType));
    Field field = new Field(columnType);
    switch (columnType) {
      case TEXT:
        field.setBinaryV(((Binary) val));
        break;
      case FLOAT:
        field.setFloatV(((float) val));
        break;
      case INT32:
        field.setIntV(((int) val));
        break;
      case INT64:
        field.setLongV(((long) val));
        break;
      case DOUBLE:
        field.setDoubleV(((double) val));
        break;
      case BOOLEAN:
        field.setBoolV(((boolean) val));
        break;
      default:
        throw new UnSupportedDataTypeException("Unsupported data type" + columnType);
    }
    RowRecord record = new RowRecord(0);
    record.addField(field);
    singleDataSet.setRecord(record);
    return singleDataSet;
  }

  private int getDevicesNum(PartialPath path) throws MetadataException {
    return IoTDB.metaManager.getDevicesNum(path);
  }

  private int getStorageGroupNum(PartialPath path) throws MetadataException {
    return IoTDB.metaManager.getStorageGroupNum(path);
  }

  protected int getPathsNum(PartialPath path) throws MetadataException {
    return IoTDB.metaManager.getAllTimeseriesCount(path);
  }

  protected int getNodesNumInGivenLevel(PartialPath path, int level) throws MetadataException {
    return IoTDB.metaManager.getNodesCountInGivenLevel(path, level);
  }

  protected List<PartialPath> getPathsName(PartialPath path) throws MetadataException {
    return IoTDB.metaManager.getAllTimeseriesPath(path);
  }

  protected List<PartialPath> getNodesList(PartialPath schemaPattern, int level)
      throws MetadataException {
    return IoTDB.metaManager.getNodesList(schemaPattern, level);
  }

  private QueryDataSet processCountTimeSeries(CountPlan countPlan) throws MetadataException {
    int num = getPathsNum(countPlan.getPath());
    return createSingleDataSet(COLUMN_COUNT, TSDataType.INT32, num);
  }

  private QueryDataSet processShowDevices(ShowDevicesPlan showDevicesPlan)
      throws MetadataException {
    return new ShowDevicesDataSet(showDevicesPlan);
  }

  protected Set<PartialPath> getDevices(PartialPath path) throws MetadataException {
    return IoTDB.metaManager.getDevices(path);
  }

  private QueryDataSet processShowChildPaths(ShowChildPathsPlan showChildPathsPlan)
      throws MetadataException {
    Set<String> childPathsList = getPathNextChildren(showChildPathsPlan.getPath());
    ListDataSet listDataSet =
        new ListDataSet(
            Collections.singletonList(new PartialPath(COLUMN_CHILD_PATHS, false)),
            Collections.singletonList(TSDataType.TEXT));
    for (String s : childPathsList) {
      RowRecord record = new RowRecord(0);
      Field field = new Field(TSDataType.TEXT);
      field.setBinaryV(new Binary(s));
      record.addField(field);
      listDataSet.putRecord(record);
    }
    return listDataSet;
  }

  protected Set<String> getPathNextChildren(PartialPath path) throws MetadataException {
    return IoTDB.metaManager.getChildNodePathInNextLevel(path);
  }

  private QueryDataSet processShowChildNodes(ShowChildNodesPlan showChildNodesPlan)
      throws MetadataException {
    // getNodeNextChildren
    Set<String> childNodesList = getNodeNextChildren(showChildNodesPlan.getPath());
    ListDataSet listDataSet =
        new ListDataSet(
            Collections.singletonList(new PartialPath(COLUMN_CHILD_NODES, false)),
            Collections.singletonList(TSDataType.TEXT));
    for (String s : childNodesList) {
      RowRecord record = new RowRecord(0);
      Field field = new Field(TSDataType.TEXT);
      field.setBinaryV(new Binary(s));
      record.addField(field);
      listDataSet.putRecord(record);
    }
    return listDataSet;
  }

  protected Set<String> getNodeNextChildren(PartialPath path) throws MetadataException {
    return IoTDB.metaManager.getChildNodeInNextLevel(path);
  }

  protected List<PartialPath> getStorageGroupNames(PartialPath path) throws MetadataException {
    return IoTDB.metaManager.getStorageGroupPaths(path);
  }

  private QueryDataSet processShowStorageGroup(ShowStorageGroupPlan showStorageGroupPlan)
      throws MetadataException {
    ListDataSet listDataSet =
        new ListDataSet(
            Collections.singletonList(new PartialPath(COLUMN_STORAGE_GROUP, false)),
            Collections.singletonList(TSDataType.TEXT));
    List<PartialPath> storageGroupList = getStorageGroupNames(showStorageGroupPlan.getPath());
    addToDataSet(storageGroupList, listDataSet);
    return listDataSet;
  }

  private void addToDataSet(Collection<PartialPath> paths, ListDataSet dataSet) {
    for (PartialPath s : paths) {
      RowRecord record = new RowRecord(0);
      Field field = new Field(TSDataType.TEXT);
      field.setBinaryV(new Binary(s.getFullPath()));
      record.addField(field);
      dataSet.putRecord(record);
    }
  }

  private QueryDataSet processShowLockInfo(ShowLockInfoPlan showLockInfoPlan)
      throws MetadataException {
    ListDataSet listDataSet =
        new ListDataSet(
            Arrays.asList(
                new PartialPath(COLUMN_STORAGE_GROUP, false),
                new PartialPath(COLUMN_LOCK_INFO, false)),
            Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));
    try {
      List<PartialPath> storageGroupList = getStorageGroupNames(showLockInfoPlan.getPath());
      List<String> lockHolderList = StorageEngine.getInstance().getLockInfo(storageGroupList);
      addLockInfoToDataSet(storageGroupList, lockHolderList, listDataSet);
    } catch (StorageEngineException e) {
      throw new MetadataException(e);
    }
    return listDataSet;
  }

  private void addLockInfoToDataSet(
      List<PartialPath> paths, List<String> lockHolderList, ListDataSet dataSet) {
    for (int i = 0; i < paths.size(); i++) {
      RowRecord record = new RowRecord(0);
      Field field = new Field(TSDataType.TEXT);
      field.setBinaryV(new Binary(paths.get(i).getFullPath()));
      record.addField(field);
      field = new Field(TSDataType.TEXT);
      field.setBinaryV(new Binary(lockHolderList.get(i)));
      record.addField(field);
      dataSet.putRecord(record);
    }
  }

  private QueryDataSet processShowTimeseries(
      ShowTimeSeriesPlan showTimeSeriesPlan, QueryContext context) throws MetadataException {
    return new ShowTimeseriesDataSet(showTimeSeriesPlan, context);
  }

  protected List<IStorageGroupMNode> getAllStorageGroupNodes() {
    return IoTDB.metaManager.getAllStorageGroupNodes();
  }

  private QueryDataSet processShowTTLQuery(ShowTTLPlan showTTLPlan) {
    ListDataSet listDataSet =
        new ListDataSet(
            Arrays.asList(
                new PartialPath(COLUMN_STORAGE_GROUP, false), new PartialPath(COLUMN_TTL, false)),
            Arrays.asList(TSDataType.TEXT, TSDataType.INT64));
    List<PartialPath> selectedSgs = showTTLPlan.getStorageGroups();

    List<IStorageGroupMNode> storageGroups = getAllStorageGroupNodes();
    int timestamp = 0;
    for (IStorageGroupMNode mNode : storageGroups) {
      PartialPath sgName = mNode.getPartialPath();
      if (!selectedSgs.isEmpty() && !selectedSgs.contains(sgName)) {
        continue;
      }
      RowRecord rowRecord = new RowRecord(timestamp++);
      Field sg = new Field(TSDataType.TEXT);
      Field ttl;
      sg.setBinaryV(new Binary(sgName.getFullPath()));
      if (mNode.getDataTTL() != Long.MAX_VALUE) {
        ttl = new Field(TSDataType.INT64);
        ttl.setLongV(mNode.getDataTTL());
      } else {
        ttl = null;
      }
      rowRecord.addField(sg);
      rowRecord.addField(ttl);
      listDataSet.putRecord(rowRecord);
    }

    return listDataSet;
  }

  private QueryDataSet processShowVersion() {
    SingleDataSet singleDataSet =
        new SingleDataSet(
            Collections.singletonList(new PartialPath(IoTDBConstant.COLUMN_VERSION, false)),
            Collections.singletonList(TSDataType.TEXT));
    Field field = new Field(TSDataType.TEXT);
    field.setBinaryV(new Binary(IoTDBConstant.VERSION));
    RowRecord rowRecord = new RowRecord(0);
    rowRecord.addField(field);
    singleDataSet.setRecord(rowRecord);
    return singleDataSet;
  }

  private QueryDataSet processShowFlushTaskInfo() {
    ListDataSet listDataSet =
        new ListDataSet(
            Arrays.asList(
                new PartialPath(COLUMN_ITEM, false), new PartialPath(COLUMN_VALUE, false)),
            Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));

    int timestamp = 0;
    addRowRecordForShowQuery(
        listDataSet,
        timestamp++,
        "total number of flush tasks",
        Integer.toString(FlushTaskPoolManager.getInstance().getTotalTasks()));
    addRowRecordForShowQuery(
        listDataSet,
        timestamp++,
        "number of working flush tasks",
        Integer.toString(FlushTaskPoolManager.getInstance().getWorkingTasksNumber()));
    addRowRecordForShowQuery(
        listDataSet,
        timestamp,
        "number of waiting flush tasks",
        Integer.toString(FlushTaskPoolManager.getInstance().getWaitingTasksNumber()));
    return listDataSet;
  }

  private QueryDataSet processShowFunctions(ShowFunctionsPlan showPlan)
      throws QueryProcessException {
    ListDataSet listDataSet =
        new ListDataSet(
            Arrays.asList(
                new PartialPath(COLUMN_FUNCTION_NAME, false),
                new PartialPath(COLUMN_FUNCTION_TYPE, false),
                new PartialPath(COLUMN_FUNCTION_CLASS, false)),
            Arrays.asList(TSDataType.TEXT, TSDataType.TEXT, TSDataType.TEXT));

    appendUDFs(listDataSet, showPlan);
    appendNativeFunctions(listDataSet, showPlan);

    listDataSet.sort(
        (r1, r2) ->
            String.CASE_INSENSITIVE_ORDER.compare(
                r1.getFields().get(0).getStringValue(), r2.getFields().get(0).getStringValue()));
    return listDataSet;
  }

  @SuppressWarnings("squid:S3776")
  private void appendUDFs(ListDataSet listDataSet, ShowFunctionsPlan showPlan)
      throws QueryProcessException {
    for (UDFRegistrationInformation info :
        UDFRegistrationService.getInstance().getRegistrationInformation()) {
      if (showPlan.showTemporary() && !info.isTemporary()) {
        continue;
      }

      RowRecord rowRecord = new RowRecord(0); // ignore timestamp
      rowRecord.addField(Binary.valueOf(info.getFunctionName()), TSDataType.TEXT);
      String functionType = "";
      try {
        if (info.isBuiltin()) {
          if (info.isUDTF()) {
            functionType = FUNCTION_TYPE_BUILTIN_UDTF;
          } else if (info.isUDAF()) {
            functionType = FUNCTION_TYPE_BUILTIN_UDAF;
          }
        } else {
          if (info.isUDTF()) {
            functionType = FUNCTION_TYPE_EXTERNAL_UDTF;
          } else if (info.isUDAF()) {
            functionType = FUNCTION_TYPE_EXTERNAL_UDAF;
          }
        }
      } catch (InstantiationException
          | InvocationTargetException
          | NoSuchMethodException
          | IllegalAccessException e) {
        throw new QueryProcessException(e.toString());
      }
      rowRecord.addField(Binary.valueOf(functionType), TSDataType.TEXT);
      rowRecord.addField(Binary.valueOf(info.getClassName()), TSDataType.TEXT);
      listDataSet.putRecord(rowRecord);
    }
  }

  private QueryDataSet processShowContinuousQueries() {
    ListDataSet listDataSet =
        new ListDataSet(
            Arrays.asList(
                new PartialPath(COLUMN_CONTINUOUS_QUERY_NAME, false),
                new PartialPath(COLUMN_CONTINUOUS_QUERY_EVERY_INTERVAL, false),
                new PartialPath(COLUMN_CONTINUOUS_QUERY_FOR_INTERVAL, false),
                new PartialPath(COLUMN_CONTINUOUS_QUERY_QUERY_SQL, false),
                new PartialPath(COLUMN_CONTINUOUS_QUERY_TARGET_PATH, false)),
            Arrays.asList(
                TSDataType.TEXT,
                TSDataType.INT64,
                TSDataType.INT64,
                TSDataType.TEXT,
                TSDataType.TEXT));

    List<ShowContinuousQueriesResult> continuousQueriesList =
        ContinuousQueryService.getInstance().getShowContinuousQueriesResultList();

    for (ShowContinuousQueriesResult result : continuousQueriesList) {
      RowRecord record = new RowRecord(0);
      record.addField(Binary.valueOf(result.getContinuousQueryName()), TSDataType.TEXT);
      record.addField(result.getEveryInterval(), TSDataType.INT64);
      record.addField(result.getForInterval(), TSDataType.INT64);
      record.addField(Binary.valueOf(result.getQuerySql()), TSDataType.TEXT);
      record.addField(Binary.valueOf(result.getTargetPath().getFullPath()), TSDataType.TEXT);
      listDataSet.putRecord(record);
    }

    return listDataSet;
  }

  private void appendNativeFunctions(ListDataSet listDataSet, ShowFunctionsPlan showPlan) {
    if (showPlan.showTemporary()) {
      return;
    }

    final Binary functionType = Binary.valueOf(FUNCTION_TYPE_NATIVE);
    final Binary className = Binary.valueOf("");
    for (String functionName : SQLConstant.getNativeFunctionNames()) {
      RowRecord rowRecord = new RowRecord(0); // ignore timestamp
      rowRecord.addField(Binary.valueOf(functionName.toUpperCase()), TSDataType.TEXT);
      rowRecord.addField(functionType, TSDataType.TEXT);
      rowRecord.addField(className, TSDataType.TEXT);
      listDataSet.putRecord(rowRecord);
    }
  }

  private QueryDataSet processShowTriggers() {
    return TriggerRegistrationService.getInstance().show();
  }

  private void addRowRecordForShowQuery(
      ListDataSet listDataSet, int timestamp, String item, String value) {
    RowRecord rowRecord = new RowRecord(timestamp);
    Field itemField = new Field(TSDataType.TEXT);
    itemField.setBinaryV(new Binary(item));
    Field valueField = new Field(TSDataType.TEXT);
    valueField.setBinaryV(new Binary(value));
    rowRecord.addField(itemField);
    rowRecord.addField(valueField);
    listDataSet.putRecord(rowRecord);
  }

  @Override
  public void delete(DeletePlan deletePlan) throws QueryProcessException {
    AUDIT_LOGGER.info(
        "delete data from {} in [{},{}]",
        deletePlan.getPaths(),
        deletePlan.getDeleteStartTime(),
        deletePlan.getDeleteEndTime());
    for (PartialPath path : deletePlan.getPaths()) {
      delete(
          path,
          deletePlan.getDeleteStartTime(),
          deletePlan.getDeleteEndTime(),
          deletePlan.getIndex());
    }
  }

  private void operateLoadFiles(OperateFilePlan plan) throws QueryProcessException {
    File file = plan.getFile();
    if (!file.exists()) {
      throw new QueryProcessException(
          String.format("File path %s doesn't exists.", file.getPath()));
    }
    if (file.isDirectory()) {
      recursionFileDir(file, plan);
    } else {
      loadFile(file, plan);
    }
  }

  private void recursionFileDir(File curFile, OperateFilePlan plan) throws QueryProcessException {
    File[] files = curFile.listFiles();
    for (File file : files) {
      if (file.isDirectory()) {
        recursionFileDir(file, plan);
      } else {
        loadFile(file, plan);
      }
    }
  }

  private void loadFile(File file, OperateFilePlan plan) throws QueryProcessException {
    if (!file.getName().endsWith(TSFILE_SUFFIX)) {
      return;
    }
    TsFileResource tsFileResource = new TsFileResource(file);
    tsFileResource.setClosed(true);
    try {
      // check file
      RestorableTsFileIOWriter restorableTsFileIOWriter = new RestorableTsFileIOWriter(file);
      if (restorableTsFileIOWriter.hasCrashed()) {
        restorableTsFileIOWriter.close();
        throw new QueryProcessException(
            String.format(
                "Cannot load file %s because the file has crashed.", file.getAbsolutePath()));
      }
      Map<Path, IMeasurementSchema> schemaMap = new HashMap<>();

      List<ChunkGroupMetadata> chunkGroupMetadataList = new ArrayList<>();
      try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getAbsolutePath(), false)) {
        reader.selfCheck(schemaMap, chunkGroupMetadataList, false);
        if (plan.getVerifyMetadata()) {
          loadNewTsFileVerifyMetadata(reader);
        }
      } catch (IOException e) {
        logger.warn("can not get timeseries metadata from {}.", file.getAbsoluteFile());
        throw new QueryProcessException(e.getMessage());
      }

      FileLoaderUtils.checkTsFileResource(tsFileResource);
      if (UpgradeUtils.isNeedUpgrade(tsFileResource)) {
        throw new QueryProcessException(
            String.format(
                "Cannot load file %s because the file's version is old which needs to be upgraded.",
                file.getAbsolutePath()));
      }

      // create schemas if they doesn't exist
      if (plan.isAutoCreateSchema()) {
        createSchemaAutomatically(chunkGroupMetadataList, schemaMap, plan.getSgLevel());
      }

      List<TsFileResource> splitResources = new ArrayList();
      if (tsFileResource.isSpanMultiTimePartitions()) {
        logger.info(
            "try to split the tsFile={} du to it spans multi partitions",
            tsFileResource.getTsFile().getPath());
        TsFileRewriteTool.rewriteTsFile(tsFileResource, splitResources);
        logger.info(
            "after split, the old tsFile was split to {} new tsFiles", splitResources.size());
      }

      if (splitResources.isEmpty()) {
        splitResources.add(tsFileResource);
      }

      for (TsFileResource resource : splitResources) {
        StorageEngine.getInstance().loadNewTsFile(resource);
      }
    } catch (Exception e) {
      throw new QueryProcessException(
          String.format("Cannot load file %s because %s", file.getAbsolutePath(), e.getMessage()));
    }
  }

  private void loadNewTsFileVerifyMetadata(TsFileSequenceReader tsFileSequenceReader)
      throws MetadataException, QueryProcessException, IOException {
    Map<String, List<TimeseriesMetadata>> metadataSet =
        tsFileSequenceReader.getAllTimeseriesMetadata();
    for (Map.Entry<String, List<TimeseriesMetadata>> entry : metadataSet.entrySet()) {
      String deviceId = entry.getKey();
      PartialPath devicePath = new PartialPath(deviceId);
      if (!IoTDB.metaManager.isPathExist(devicePath)) {
        continue;
      }
      for (TimeseriesMetadata metadata : entry.getValue()) {
        PartialPath fullPath =
            new PartialPath(deviceId + TsFileConstant.PATH_SEPARATOR + metadata.getMeasurementId());
        if (IoTDB.metaManager.isPathExist(fullPath)) {
          TSDataType dataType = IoTDB.metaManager.getSeriesSchema(fullPath).getType();
          if (dataType != metadata.getTSDataType()) {
            throw new QueryProcessException(
                fullPath.getFullPath()
                    + " is "
                    + metadata.getTSDataType().name()
                    + " in the loading TsFile but is "
                    + dataType.name()
                    + " in IoTDB.");
          }
        }
      }
    }
  }

  @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
  private void createSchemaAutomatically(
      List<ChunkGroupMetadata> chunkGroupMetadataList,
      Map<Path, IMeasurementSchema> knownSchemas,
      int sgLevel)
      throws QueryProcessException, MetadataException, IOException {
    if (chunkGroupMetadataList.isEmpty()) {
      return;
    }

    Set<PartialPath> registeredSeries = new HashSet<>();
    for (ChunkGroupMetadata chunkGroupMetadata : chunkGroupMetadataList) {
      String device = chunkGroupMetadata.getDevice();
      IMNode node =
          IoTDB.metaManager.getDeviceNodeWithAutoCreate(
                  new PartialPath(device), true, true, sgLevel)
              .left;
      for (ChunkMetadata chunkMetadata : chunkGroupMetadata.getChunkMetadataList()) {
        PartialPath series =
            new PartialPath(
                chunkGroupMetadata.getDevice()
                    + TsFileConstant.PATH_SEPARATOR
                    + chunkMetadata.getMeasurementUid());
        if (!registeredSeries.contains(series)) {
          registeredSeries.add(series);
          IMeasurementSchema schema =
              knownSchemas.get(new Path(series.getDevice(), series.getMeasurement()));
          if (schema == null) {
            throw new MetadataException(
                String.format(
                    "Can not get the schema of measurement [%s]",
                    chunkMetadata.getMeasurementUid()));
          }
          if (!node.hasChild(chunkMetadata.getMeasurementUid())) {
            IoTDB.metaManager.createTimeseries(
                series,
                schema.getType(),
                schema.getEncodingType(),
                schema.getCompressor(),
                Collections.emptyMap());
          } else if (!(node.getChild(chunkMetadata.getMeasurementUid())
              instanceof MeasurementMNode)) {
            throw new QueryProcessException(
                String.format("Current Path is not leaf node. %s", series));
          }
        }
      }
    }
  }

  private void operateRemoveFile(OperateFilePlan plan) throws QueryProcessException {
    try {
      if (!StorageEngine.getInstance().deleteTsfile(plan.getFile())) {
        throw new QueryProcessException(
            String.format("File %s doesn't exist.", plan.getFile().getName()));
      }
    } catch (StorageEngineException | IllegalPathException e) {
      throw new QueryProcessException(
          String.format("Cannot remove file because %s", e.getMessage()));
    }
  }

  private void operateMoveFile(OperateFilePlan plan) throws QueryProcessException {
    if (!plan.getTargetDir().exists() || !plan.getTargetDir().isDirectory()) {
      throw new QueryProcessException(
          String.format("Target dir %s is invalid.", plan.getTargetDir().getPath()));
    }
    try {
      if (!StorageEngine.getInstance().moveTsfile(plan.getFile(), plan.getTargetDir())) {
        throw new QueryProcessException(
            String.format("File %s doesn't exist.", plan.getFile().getName()));
      }
    } catch (StorageEngineException | IllegalPathException e) {
      throw new QueryProcessException(
          String.format(
              "Cannot move file %s to target directory %s because %s",
              plan.getFile().getPath(), plan.getTargetDir().getPath(), e.getMessage()));
    }
  }

  private void operateTTL(SetTTLPlan plan) throws QueryProcessException {
    try {
      List<PartialPath> storageGroupPaths =
          IoTDB.metaManager.getStorageGroupPaths(plan.getStorageGroup());
      for (PartialPath storagePath : storageGroupPaths) {
        IoTDB.metaManager.setTTL(storagePath, plan.getDataTTL());
        StorageEngine.getInstance().setTTL(storagePath, plan.getDataTTL());
      }
    } catch (MetadataException e) {
      throw new QueryProcessException(e);
    } catch (IOException e) {
      throw new QueryProcessException(e.getMessage());
    }
  }

  @Override
  public void update(PartialPath path, long startTime, long endTime, String value) {
    throw new UnsupportedOperationException("update is not supported now");
  }

  @Override
  public void delete(PartialPath path, long startTime, long endTime, long planIndex)
      throws QueryProcessException {
    try {
      StorageEngine.getInstance().delete(path, startTime, endTime, planIndex);
    } catch (StorageEngineException e) {
      throw new QueryProcessException(e);
    }
  }

  protected IMNode getSeriesSchemas(InsertPlan insertPlan) throws MetadataException {
    try {
      return IoTDB.metaManager.getSeriesSchemasAndReadLockDevice(insertPlan);
    } catch (IOException e) {
      throw new MetadataException(e);
    }
  }

  private void checkFailedMeasurments(InsertPlan plan)
      throws PathNotExistException, StorageEngineException {
    // check if all path not exist exceptions
    List<String> failedPaths = plan.getFailedMeasurements();
    List<Exception> exceptions = plan.getFailedExceptions();
    boolean isPathNotExistException = true;
    for (Exception e : exceptions) {
      Throwable curException = e;
      while (curException.getCause() != null) {
        curException = curException.getCause();
      }
      if (!(curException instanceof PathNotExistException)) {
        isPathNotExistException = false;
        break;
      }
    }
    if (isPathNotExistException) {
      throw new PathNotExistException(failedPaths);
    } else {
      throw new StorageEngineException(
          INSERT_MEASUREMENTS_FAILED_MESSAGE
              + plan.getFailedMeasurements()
              + (!exceptions.isEmpty() ? (" caused by " + exceptions.get(0).getMessage()) : ""));
    }
  }

  @Override
  public void insert(InsertRowsOfOneDevicePlan insertRowsOfOneDevicePlan)
      throws QueryProcessException {
    if (insertRowsOfOneDevicePlan.getRowPlans().length == 0) {
      return;
    }
    try {
      for (InsertRowPlan plan : insertRowsOfOneDevicePlan.getRowPlans()) {
        plan.setMeasurementMNodes(new IMeasurementMNode[plan.getMeasurements().length]);
        // check whether types are match
        getSeriesSchemas(plan);
        // we do not need to infer data type for insertRowsOfOneDevicePlan
        if (plan.isAligned()) {
          plan.setPrefixPath(plan.getPrefixPath().getDevicePath());
        }
      }
      // ok, we can begin to write data into the engine..
      StorageEngine.getInstance().insert(insertRowsOfOneDevicePlan);

      List<String> notExistedPaths = null;
      List<String> failedMeasurements = null;

      // If there are some exceptions, we assume they caused by the same reason.
      Exception exception = null;
      for (InsertRowPlan plan : insertRowsOfOneDevicePlan.getRowPlans()) {
        if (plan.getFailedMeasurements() != null) {
          if (notExistedPaths == null) {
            notExistedPaths = new ArrayList<>();
            failedMeasurements = new ArrayList<>();
          }
          // check if all path not exist exceptions
          List<String> failedPaths = plan.getFailedMeasurements();
          List<Exception> exceptions = plan.getFailedExceptions();
          boolean isPathNotExistException = true;
          for (Exception e : exceptions) {
            exception = e;
            Throwable curException = e;
            while (curException.getCause() != null) {
              curException = curException.getCause();
            }
            if (!(curException instanceof PathNotExistException)) {
              isPathNotExistException = false;
              break;
            }
          }
          if (isPathNotExistException) {
            notExistedPaths.addAll(failedPaths);
          } else {
            failedMeasurements.addAll(plan.getFailedMeasurements());
          }
        }
      }
      if (notExistedPaths != null && !notExistedPaths.isEmpty()) {
        throw new PathNotExistException(notExistedPaths);
      } else if (notExistedPaths != null && !failedMeasurements.isEmpty()) {
        throw new StorageEngineException(
            "failed to insert points "
                + failedMeasurements
                + (exception != null ? (" caused by " + exception.getMessage()) : ""));
      }

    } catch (StorageEngineException | MetadataException e) {
      throw new QueryProcessException(e);
    }
  }

  @Override
  public void insert(InsertRowsPlan plan) throws QueryProcessException {
    for (int i = 0; i < plan.getInsertRowPlanList().size(); i++) {
      if (plan.getResults().containsKey(i) || plan.isExecuted(i)) {
        continue;
      }
      try {
        insert(plan.getInsertRowPlanList().get(i));
      } catch (QueryProcessException e) {
        plan.getResults().put(i, RpcUtils.getStatus(e.getErrorCode(), e.getMessage()));
      }
    }
    if (!plan.getResults().isEmpty()) {
      throw new BatchProcessException(plan.getFailingStatus());
    }
  }

  @Override
  public void insert(InsertRowPlan insertRowPlan) throws QueryProcessException {
    try {
      insertRowPlan.setMeasurementMNodes(
          new IMeasurementMNode[insertRowPlan.getMeasurements().length]);
      // When insert data with sql statement, the data types will be null here.
      // We need to predicted the data types first
      if (insertRowPlan.getDataTypes()[0] == null) {
        for (int i = 0; i < insertRowPlan.getDataTypes().length; i++) {
          insertRowPlan.getDataTypes()[i] =
              TypeInferenceUtils.getPredictedDataType(
                  insertRowPlan.getValues()[i], insertRowPlan.isNeedInferType());
        }
      }
      // check whether types are match
      getSeriesSchemas(insertRowPlan);
      if (insertRowPlan.isAligned()) {
        insertRowPlan.setPrefixPathForAlignTimeSeries(
            insertRowPlan.getPrefixPath().getDevicePath());
      }
      insertRowPlan.transferType();
      StorageEngine.getInstance().insert(insertRowPlan);
      if (insertRowPlan.getFailedMeasurements() != null) {
        checkFailedMeasurments(insertRowPlan);
      }
    } catch (StorageEngineException | MetadataException e) {
      if (IoTDBDescriptor.getInstance().getConfig().isEnableStatMonitor()) {
        StatMonitor.getInstance().updateFailedStatValue();
      }
      throw new QueryProcessException(e);
    } catch (Exception e) {
      // update failed statistics
      if (IoTDBDescriptor.getInstance().getConfig().isEnableStatMonitor()) {
        StatMonitor.getInstance().updateFailedStatValue();
      }
      throw e;
    }
  }

  @Override
  public void insertTablet(InsertMultiTabletPlan insertMultiTabletPlan)
      throws QueryProcessException {
    for (int i = 0; i < insertMultiTabletPlan.getInsertTabletPlanList().size(); i++) {
      if (insertMultiTabletPlan.getResults().containsKey(i)
          || insertMultiTabletPlan.isExecuted(i)) {
        continue;
      }
      insertTablet(insertMultiTabletPlan.getInsertTabletPlanList().get(i));
    }
  }

  @Override
  public void insertTablet(InsertTabletPlan insertTabletPlan) throws QueryProcessException {
    if (insertTabletPlan.getRowCount() == 0) {
      return;
    }
    try {
      insertTabletPlan.setMeasurementMNodes(
          new IMeasurementMNode[insertTabletPlan.getMeasurements().length]);
      getSeriesSchemas(insertTabletPlan);
      if (insertTabletPlan.isAligned()) {
        insertTabletPlan.setPrefixPath(insertTabletPlan.getPrefixPath().getDevicePath());
      }
      StorageEngine.getInstance().insertTablet(insertTabletPlan);
      if (insertTabletPlan.getFailedMeasurements() != null) {
        checkFailedMeasurments(insertTabletPlan);
      }
    } catch (StorageEngineException | MetadataException e) {
      if (IoTDBDescriptor.getInstance().getConfig().isEnableStatMonitor()) {
        StatMonitor.getInstance().updateFailedStatValue();
      }
      throw new QueryProcessException(e);
    } catch (Exception e) {
      // update failed statistics
      if (IoTDBDescriptor.getInstance().getConfig().isEnableStatMonitor()) {
        StatMonitor.getInstance().updateFailedStatValue();
      }
      throw e;
    }
  }

  private boolean operateAuthor(AuthorPlan author) throws QueryProcessException {
    AuthorOperator.AuthorType authorType = author.getAuthorType();
    String userName = author.getUserName();
    String roleName = author.getRoleName();
    String password = author.getPassword();
    String newPassword = author.getNewPassword();
    Set<Integer> permissions = author.getPermissions();
    PartialPath nodeName = author.getNodeName();
    try {
      switch (authorType) {
        case UPDATE_USER:
          authorizer.updateUserPassword(userName, newPassword);
          break;
        case CREATE_USER:
          authorizer.createUser(userName, password);
          break;
        case CREATE_ROLE:
          authorizer.createRole(roleName);
          break;
        case DROP_USER:
          authorizer.deleteUser(userName);
          break;
        case DROP_ROLE:
          authorizer.deleteRole(roleName);
          break;
        case GRANT_ROLE:
          for (int i : permissions) {
            authorizer.grantPrivilegeToRole(roleName, nodeName.getFullPath(), i);
          }
          break;
        case GRANT_USER:
          for (int i : permissions) {
            authorizer.grantPrivilegeToUser(userName, nodeName.getFullPath(), i);
          }
          break;
        case GRANT_ROLE_TO_USER:
          authorizer.grantRoleToUser(roleName, userName);
          break;
        case REVOKE_USER:
          for (int i : permissions) {
            authorizer.revokePrivilegeFromUser(userName, nodeName.getFullPath(), i);
          }
          break;
        case REVOKE_ROLE:
          for (int i : permissions) {
            authorizer.revokePrivilegeFromRole(roleName, nodeName.getFullPath(), i);
          }
          break;
        case REVOKE_ROLE_FROM_USER:
          authorizer.revokeRoleFromUser(roleName, userName);
          break;
        default:
          throw new QueryProcessException("Unsupported operation " + authorType);
      }
    } catch (AuthException e) {
      throw new QueryProcessException(e.getMessage(), true);
    }
    return true;
  }

  private boolean operateWatermarkEmbedding(List<String> users, boolean useWatermark)
      throws QueryProcessException {
    try {
      for (String user : users) {
        authorizer.setUserUseWaterMark(user, useWatermark);
      }
    } catch (AuthException e) {
      throw new QueryProcessException(e.getMessage());
    }
    return true;
  }

  private boolean createTimeSeries(CreateTimeSeriesPlan createTimeSeriesPlan)
      throws QueryProcessException {
    try {
      IoTDB.metaManager.createTimeseries(createTimeSeriesPlan);
    } catch (MetadataException e) {
      throw new QueryProcessException(e);
    }
    return true;
  }

  private boolean createAlignedTimeSeries(CreateAlignedTimeSeriesPlan createAlignedTimeSeriesPlan)
      throws QueryProcessException {
    try {
      IoTDB.metaManager.createAlignedTimeSeries(createAlignedTimeSeriesPlan);
    } catch (MetadataException e) {
      throw new QueryProcessException(e);
    }
    return true;
  }

  @SuppressWarnings("squid:S3776") // high Cognitive Complexity
  private boolean createMultiTimeSeries(CreateMultiTimeSeriesPlan multiPlan)
      throws BatchProcessException {
    int dataTypeIdx = 0;
    for (int i = 0; i < multiPlan.getPaths().size(); i++) {
      if (multiPlan.getResults().containsKey(i) || multiPlan.isExecuted(i)) {
        continue;
      }
      PartialPath path = multiPlan.getPaths().get(i);
      String measurement = path.getMeasurement();
      CreateTimeSeriesPlan plan =
          new CreateTimeSeriesPlan(
              multiPlan.getPaths().get(i),
              multiPlan.getDataTypes().get(i),
              multiPlan.getEncodings().get(i),
              multiPlan.getCompressors().get(i),
              multiPlan.getProps() == null ? null : multiPlan.getProps().get(i),
              multiPlan.getTags() == null ? null : multiPlan.getTags().get(i),
              multiPlan.getAttributes() == null ? null : multiPlan.getAttributes().get(i),
              multiPlan.getAlias() == null ? null : multiPlan.getAlias().get(i));
      dataTypeIdx++;
      try {
        createTimeSeries(plan);
      } catch (QueryProcessException e) {
        multiPlan.getResults().put(i, RpcUtils.getStatus(e.getErrorCode(), e.getMessage()));
      }
    }
    if (!multiPlan.getResults().isEmpty()) {
      throw new BatchProcessException(multiPlan.getFailingStatus());
    }
    return true;
  }

  protected boolean deleteTimeSeries(DeleteTimeSeriesPlan deleteTimeSeriesPlan)
      throws QueryProcessException {
    AUDIT_LOGGER.info("delete timeseries {}", deleteTimeSeriesPlan.getPaths());
    List<PartialPath> deletePathList = deleteTimeSeriesPlan.getPaths();
    for (int i = 0; i < deletePathList.size(); i++) {
      PartialPath path = deletePathList.get(i);
      try {
        StorageEngine.getInstance().deleteTimeseries(path, deleteTimeSeriesPlan.getIndex());
        String failed = IoTDB.metaManager.deleteTimeseries(path);
        if (failed != null) {
          deleteTimeSeriesPlan
              .getResults()
              .put(i, RpcUtils.getStatus(TSStatusCode.NODE_DELETE_FAILED_ERROR, failed));
        }
      } catch (StorageEngineException | MetadataException e) {
        deleteTimeSeriesPlan
            .getResults()
            .put(i, RpcUtils.getStatus(e.getErrorCode(), e.getMessage()));
      }
    }
    if (!deleteTimeSeriesPlan.getResults().isEmpty()) {
      throw new BatchProcessException(deleteTimeSeriesPlan.getFailingStatus());
    }
    return true;
  }

  private boolean alterTimeSeries(AlterTimeSeriesPlan alterTimeSeriesPlan)
      throws QueryProcessException {
    PartialPath path = alterTimeSeriesPlan.getPath();
    Map<String, String> alterMap = alterTimeSeriesPlan.getAlterMap();
    try {
      switch (alterTimeSeriesPlan.getAlterType()) {
        case RENAME:
          String beforeName = alterMap.keySet().iterator().next();
          String currentName = alterMap.get(beforeName);
          IoTDB.metaManager.renameTagOrAttributeKey(beforeName, currentName, path);
          break;
        case SET:
          IoTDB.metaManager.setTagsOrAttributesValue(alterMap, path);
          break;
        case DROP:
          IoTDB.metaManager.dropTagsOrAttributes(alterMap.keySet(), path);
          break;
        case ADD_TAGS:
          IoTDB.metaManager.addTags(alterMap, path);
          break;
        case ADD_ATTRIBUTES:
          IoTDB.metaManager.addAttributes(alterMap, path);
          break;
        case UPSERT:
          IoTDB.metaManager.upsertTagsAndAttributes(
              alterTimeSeriesPlan.getAlias(),
              alterTimeSeriesPlan.getTagsMap(),
              alterTimeSeriesPlan.getAttributesMap(),
              path);
          break;
      }
    } catch (MetadataException e) {
      throw new QueryProcessException(e);
    } catch (IOException e) {
      throw new QueryProcessException(
          String.format(
              "Something went wrong while read/write the [%s]'s tag/attribute info.",
              path.getFullPath()));
    }
    return true;
  }

  public boolean setStorageGroup(SetStorageGroupPlan setStorageGroupPlan)
      throws QueryProcessException {
    AUDIT_LOGGER.info("set storage group to {}", setStorageGroupPlan.getPaths());
    PartialPath path = setStorageGroupPlan.getPath();
    try {
      IoTDB.metaManager.setStorageGroup(path);
    } catch (MetadataException e) {
      throw new QueryProcessException(e);
    }
    return true;
  }

  protected boolean deleteStorageGroups(DeleteStorageGroupPlan deleteStorageGroupPlan)
      throws QueryProcessException {
    AUDIT_LOGGER.info("delete storage group {}", deleteStorageGroupPlan.getPaths());
    List<PartialPath> deletePathList = new ArrayList<>();
    try {
      for (PartialPath storageGroupPath : deleteStorageGroupPlan.getPaths()) {
        List<PartialPath> allRelatedStorageGroupPath =
            IoTDB.metaManager.getStorageGroupPaths(storageGroupPath);
        if (allRelatedStorageGroupPath.isEmpty()) {
          throw new PathNotExistException(storageGroupPath.getFullPath(), true);
        }
        for (PartialPath path : allRelatedStorageGroupPath) {
          StorageEngine.getInstance().deleteStorageGroup(path);
          deletePathList.add(path);
        }
      }
      IoTDB.metaManager.deleteStorageGroups(deletePathList);
    } catch (MetadataException e) {
      throw new QueryProcessException(e);
    }
    return true;
  }

  protected QueryDataSet processAuthorQuery(AuthorPlan plan) throws QueryProcessException {
    AuthorType authorType = plan.getAuthorType();
    String userName = plan.getUserName();
    String roleName = plan.getRoleName();
    PartialPath path = plan.getNodeName();

    ListDataSet dataSet;

    try {
      switch (authorType) {
        case LIST_ROLE:
          dataSet = executeListRole(plan);
          break;
        case LIST_USER:
          dataSet = executeListUser(plan);
          break;
        case LIST_ROLE_USERS:
          dataSet = executeListRoleUsers(roleName);
          break;
        case LIST_USER_ROLES:
          dataSet = executeListUserRoles(userName);
          break;
        case LIST_ROLE_PRIVILEGE:
          dataSet = executeListRolePrivileges(roleName, path);
          break;
        case LIST_USER_PRIVILEGE:
          dataSet = executeListUserPrivileges(userName, path);
          break;
        default:
          throw new QueryProcessException("Unsupported operation " + authorType);
      }
    } catch (AuthException e) {
      throw new QueryProcessException(e.getMessage());
    }
    return dataSet;
  }

  private ListDataSet executeListRole(AuthorPlan plan) throws AuthException {
    ListDataSet dataSet =
        new ListDataSet(
            Collections.singletonList(new PartialPath(COLUMN_ROLE, false)),
            Collections.singletonList(TSDataType.TEXT));

    // check if current user is granted list_role privilege
    boolean hasListRolePrivilege =
        AuthorityChecker.check(
            plan.getLoginUserName(),
            Collections.emptyList(),
            plan.getOperatorType(),
            plan.getLoginUserName());
    if (!hasListRolePrivilege) {
      return dataSet;
    }

    List<String> roleList = authorizer.listAllRoles();
    addToDataSet(roleList, dataSet);
    return dataSet;
  }

  private void addToDataSet(List<String> strResults, ListDataSet dataSet) {
    int index = 0;
    for (String role : strResults) {
      RowRecord record = new RowRecord(index++);
      Field field = new Field(TSDataType.TEXT);
      field.setBinaryV(new Binary(role));
      record.addField(field);
      dataSet.putRecord(record);
    }
  }

  private ListDataSet executeListUser(AuthorPlan plan) throws AuthException {
    ListDataSet dataSet =
        new ListDataSet(
            Collections.singletonList(new PartialPath(COLUMN_USER, false)),
            Collections.singletonList(TSDataType.TEXT));

    // check if current user is granted list_user privilege
    boolean hasListUserPrivilege =
        AuthorityChecker.check(
            plan.getLoginUserName(),
            Collections.singletonList((plan.getNodeName())),
            plan.getOperatorType(),
            plan.getLoginUserName());
    if (!hasListUserPrivilege) {
      return dataSet;
    }

    List<String> userList = authorizer.listAllUsers();
    addToDataSet(userList, dataSet);
    return dataSet;
  }

  private ListDataSet executeListRoleUsers(String roleName) throws AuthException {
    Role role = authorizer.getRole(roleName);
    if (role == null) {
      throw new AuthException("No such role : " + roleName);
    }
    ListDataSet dataSet =
        new ListDataSet(
            Collections.singletonList(new PartialPath(COLUMN_USER, false)),
            Collections.singletonList(TSDataType.TEXT));
    List<String> userList = authorizer.listAllUsers();
    int index = 0;
    for (String userN : userList) {
      User userObj = authorizer.getUser(userN);
      if (userObj != null && userObj.hasRole(roleName)) {
        RowRecord record = new RowRecord(index++);
        Field field = new Field(TSDataType.TEXT);
        field.setBinaryV(new Binary(userN));
        record.addField(field);
        dataSet.putRecord(record);
      }
    }
    return dataSet;
  }

  private ListDataSet executeListUserRoles(String userName) throws AuthException {
    User user = authorizer.getUser(userName);
    if (user != null) {
      ListDataSet dataSet =
          new ListDataSet(
              Collections.singletonList(new PartialPath(COLUMN_ROLE, false)),
              Collections.singletonList(TSDataType.TEXT));
      int index = 0;
      for (String roleN : user.getRoleList()) {
        RowRecord record = new RowRecord(index++);
        Field field = new Field(TSDataType.TEXT);
        field.setBinaryV(new Binary(roleN));
        record.addField(field);
        dataSet.putRecord(record);
      }
      return dataSet;
    } else {
      throw new AuthException("No such user : " + userName);
    }
  }

  private ListDataSet executeListRolePrivileges(String roleName, PartialPath path)
      throws AuthException {
    Role role = authorizer.getRole(roleName);
    if (role != null) {
      List<PartialPath> headerList = new ArrayList<>();
      List<TSDataType> typeList = new ArrayList<>();
      headerList.add(new PartialPath(COLUMN_PRIVILEGE, false));
      typeList.add(TSDataType.TEXT);
      ListDataSet dataSet = new ListDataSet(headerList, typeList);
      int index = 0;
      for (PathPrivilege pathPrivilege : role.getPrivilegeList()) {
        if (path == null || AuthUtils.pathBelongsTo(path.getFullPath(), pathPrivilege.getPath())) {
          RowRecord record = new RowRecord(index++);
          Field field = new Field(TSDataType.TEXT);
          field.setBinaryV(new Binary(pathPrivilege.toString()));
          record.addField(field);
          dataSet.putRecord(record);
        }
      }
      return dataSet;
    } else {
      throw new AuthException("No such role : " + roleName);
    }
  }

  private ListDataSet executeListUserPrivileges(String userName, PartialPath path)
      throws AuthException {
    User user = authorizer.getUser(userName);
    if (user == null) {
      throw new AuthException("No such user : " + userName);
    }
    List<PartialPath> headerList = new ArrayList<>();
    List<TSDataType> typeList = new ArrayList<>();
    headerList.add(new PartialPath(COLUMN_ROLE, false));
    headerList.add(new PartialPath(COLUMN_PRIVILEGE, false));
    typeList.add(TSDataType.TEXT);
    typeList.add(TSDataType.TEXT);
    ListDataSet dataSet = new ListDataSet(headerList, typeList);
    int index = 0;
    for (PathPrivilege pathPrivilege : user.getPrivilegeList()) {
      if (path == null || AuthUtils.pathBelongsTo(path.getFullPath(), pathPrivilege.getPath())) {
        RowRecord record = new RowRecord(index++);
        Field roleF = new Field(TSDataType.TEXT);
        roleF.setBinaryV(new Binary(""));
        record.addField(roleF);
        Field privilegeF = new Field(TSDataType.TEXT);
        privilegeF.setBinaryV(new Binary(pathPrivilege.toString()));
        record.addField(privilegeF);
        dataSet.putRecord(record);
      }
    }
    for (String roleN : user.getRoleList()) {
      Role role = authorizer.getRole(roleN);
      if (role == null) {
        continue;
      }
      for (PathPrivilege pathPrivilege : role.getPrivilegeList()) {
        if (path == null || AuthUtils.pathBelongsTo(path.getFullPath(), pathPrivilege.getPath())) {
          RowRecord record = new RowRecord(index++);
          Field roleF = new Field(TSDataType.TEXT);
          roleF.setBinaryV(new Binary(roleN));
          record.addField(roleF);
          Field privilegeF = new Field(TSDataType.TEXT);
          privilegeF.setBinaryV(new Binary(pathPrivilege.toString()));
          record.addField(privilegeF);
          dataSet.putRecord(record);
        }
      }
    }
    return dataSet;
  }

  protected String deleteTimeSeries(PartialPath path) throws MetadataException {
    return IoTDB.metaManager.deleteTimeseries(path);
  }

  @SuppressWarnings("unused") // for the distributed version
  protected void loadConfiguration(LoadConfigurationPlan plan) throws QueryProcessException {
    IoTDBDescriptor.getInstance().loadHotModifiedProps();
  }

  private QueryDataSet processShowMergeStatus() {
    List<PartialPath> headerList = new ArrayList<>();
    List<TSDataType> typeList = new ArrayList<>();
    headerList.add(new PartialPath(COLUMN_STORAGE_GROUP, false));
    headerList.add(new PartialPath(COLUMN_TASK_NAME, false));
    headerList.add(new PartialPath(COLUMN_CREATED_TIME, false));
    headerList.add(new PartialPath(COLUMN_PROGRESS, false));
    headerList.add(new PartialPath(COLUMN_CANCELLED, false));
    headerList.add(new PartialPath(COLUMN_DONE, false));

    typeList.add(TSDataType.TEXT);
    typeList.add(TSDataType.TEXT);
    typeList.add(TSDataType.TEXT);
    typeList.add(TSDataType.TEXT);
    typeList.add(TSDataType.BOOLEAN);
    typeList.add(TSDataType.BOOLEAN);
    ListDataSet dataSet = new ListDataSet(headerList, typeList);
    Map<String, List<TaskStatus>>[] taskStatus = MergeManager.getINSTANCE().collectTaskStatus();
    for (Map<String, List<TaskStatus>> statusMap : taskStatus) {
      for (Entry<String, List<TaskStatus>> stringListEntry : statusMap.entrySet()) {
        for (TaskStatus status : stringListEntry.getValue()) {
          dataSet.putRecord(toRowRecord(status, stringListEntry.getKey()));
        }
      }
    }
    return dataSet;
  }

  public RowRecord toRowRecord(TaskStatus status, String storageGroup) {
    RowRecord record = new RowRecord(0);
    record.addField(new Binary(storageGroup), TSDataType.TEXT);
    record.addField(new Binary(status.getTaskName()), TSDataType.TEXT);
    record.addField(new Binary(status.getCreatedTime()), TSDataType.TEXT);
    record.addField(new Binary(status.getProgress()), TSDataType.TEXT);
    record.addField(status.isCancelled(), TSDataType.BOOLEAN);
    record.addField(status.isDone(), TSDataType.BOOLEAN);
    return record;
  }

  private QueryDataSet processShowQueryProcesslist() {
    ListDataSet listDataSet =
        new ListDataSet(
            Arrays.asList(new PartialPath(QUERY_ID, false), new PartialPath(STATEMENT, false)),
            Arrays.asList(TSDataType.INT64, TSDataType.TEXT));
    QueryTimeManager queryTimeManager = QueryTimeManager.getInstance();
    for (Entry<Long, QueryInfo> queryInfo : queryTimeManager.getQueryInfoMap().entrySet()) {
      RowRecord record = new RowRecord(queryInfo.getValue().getStartTime());
      record.addField(queryInfo.getKey(), TSDataType.INT64);
      record.addField(new Binary(queryInfo.getValue().getStatement()), TSDataType.TEXT);
      listDataSet.putRecord(record);
    }
    return listDataSet;
  }

  /**
   * @param storageGroups the storage groups to check
   * @return List of PartialPath the storage groups that not exist
   */
  List<PartialPath> checkStorageGroupExist(List<PartialPath> storageGroups) {
    List<PartialPath> noExistSg = new ArrayList<>();
    if (storageGroups == null) {
      return noExistSg;
    }
    for (PartialPath storageGroup : storageGroups) {
      if (!IoTDB.metaManager.isStorageGroup(storageGroup)) {
        noExistSg.add(storageGroup);
      }
    }
    return noExistSg;
  }
}