/*
 * 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.mpp.plan.analyze;

import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.partition.DataPartition;
import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
import org.apache.iotdb.commons.partition.SchemaNodeManagementPartition;
import org.apache.iotdb.commons.partition.SchemaPartition;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.exception.metadata.template.TemplateImcompatibeException;
import org.apache.iotdb.db.exception.sql.MeasurementNotExistException;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.exception.sql.StatementAnalyzeException;
import org.apache.iotdb.db.metadata.path.MeasurementPath;
import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.common.header.ColumnHeader;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
import org.apache.iotdb.db.mpp.common.schematree.DeviceSchemaInfo;
import org.apache.iotdb.db.mpp.common.schematree.ISchemaTree;
import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
import org.apache.iotdb.db.mpp.plan.expression.Expression;
import org.apache.iotdb.db.mpp.plan.expression.ExpressionType;
import org.apache.iotdb.db.mpp.plan.expression.leaf.TimeSeriesOperand;
import org.apache.iotdb.db.mpp.plan.expression.multi.FunctionExpression;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.FillDescriptor;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByTimeParameter;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.OrderByParameter;
import org.apache.iotdb.db.mpp.plan.statement.Statement;
import org.apache.iotdb.db.mpp.plan.statement.StatementNode;
import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
import org.apache.iotdb.db.mpp.plan.statement.component.FillComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.GroupByTimeComponent;
import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
import org.apache.iotdb.db.mpp.plan.statement.component.ResultColumn;
import org.apache.iotdb.db.mpp.plan.statement.component.SortItem;
import org.apache.iotdb.db.mpp.plan.statement.component.SortKey;
import org.apache.iotdb.db.mpp.plan.statement.component.WhereCondition;
import org.apache.iotdb.db.mpp.plan.statement.crud.DeleteDataStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertMultiTabletsStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertRowStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertRowsOfOneDeviceStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertRowsStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertTabletStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.QueryStatement;
import org.apache.iotdb.db.mpp.plan.statement.internal.InternalCreateTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.internal.SchemaFetchStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.AlterTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountDevicesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountLevelTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountNodesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateAlignedTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateMultiTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowChildNodesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowChildPathsStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowClusterStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowDevicesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowTTLStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ActivateTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.CreateSchemaTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.SetSchemaTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ShowNodesInSchemaTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ShowPathSetTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ShowPathsUsingTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ShowSchemaTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.ExplainStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.ShowVersionStatement;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.filter.GroupByFilter;
import org.apache.iotdb.tsfile.read.filter.GroupByMonthFilter;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.filter.factory.FilterFactory;
import org.apache.iotdb.tsfile.utils.Pair;

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

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TimeZone;
import java.util.stream.Collectors;

import static com.google.common.base.Preconditions.checkState;
import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD;
import static org.apache.iotdb.commons.conf.IoTDBConstant.ONE_LEVEL_PATH_WILDCARD;
import static org.apache.iotdb.db.metadata.MetadataConstant.ALL_RESULT_NODES;

/** This visitor is used to analyze each type of Statement and returns the {@link Analysis}. */
public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> {

  private static final Logger logger = LoggerFactory.getLogger(Analyzer.class);

  private final IPartitionFetcher partitionFetcher;
  private final ISchemaFetcher schemaFetcher;
  private final TypeProvider typeProvider;
  private final MPPQueryContext context;

  public AnalyzeVisitor(
      IPartitionFetcher partitionFetcher,
      ISchemaFetcher schemaFetcher,
      TypeProvider typeProvider,
      MPPQueryContext context) {
    this.context = context;
    this.partitionFetcher = partitionFetcher;
    this.schemaFetcher = schemaFetcher;
    this.typeProvider = typeProvider;
  }

  private String getLogHeader() {
    return String.format("Query[%s]:", context.getQueryId());
  }

  @Override
  public Analysis visitNode(StatementNode node, MPPQueryContext context) {
    throw new UnsupportedOperationException(
        "Unsupported statement type: " + node.getClass().getName());
  }

  @Override
  public Analysis visitExplain(ExplainStatement explainStatement, MPPQueryContext context) {
    Analysis analysis = visitQuery(explainStatement.getQueryStatement(), context);
    analysis.setStatement(explainStatement);
    analysis.setFinishQueryAfterAnalyze(true);
    return analysis;
  }

  @Override
  public Analysis visitQuery(QueryStatement queryStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    try {
      // check for semantic errors
      queryStatement.semanticCheck();

      // concat path and construct path pattern tree
      PathPatternTree patternTree = new PathPatternTree();
      queryStatement =
          (QueryStatement) new ConcatPathRewriter().rewrite(queryStatement, patternTree);
      analysis.setStatement(queryStatement);

      // request schema fetch API
      logger.info("{} fetch query schema...", getLogHeader());
      ISchemaTree schemaTree = schemaFetcher.fetchSchema(patternTree);
      logger.info("{} fetch schema done", getLogHeader());
      // If there is no leaf node in the schema tree, the query should be completed immediately
      if (schemaTree.isEmpty()) {
        analysis.setFinishQueryAfterAnalyze(true);
        return analysis;
      }

      // extract global time filter from query filter and determine if there is a value filter
      Pair<Filter, Boolean> resultPair = analyzeGlobalTimeFilter(queryStatement);
      Filter globalTimeFilter = resultPair.left;
      boolean hasValueFilter = resultPair.right;
      analysis.setGlobalTimeFilter(globalTimeFilter);
      analysis.setHasValueFilter(hasValueFilter);

      if (queryStatement.isLastQuery()) {
        if (hasValueFilter) {
          throw new SemanticException("Only time filters are supported in LAST query");
        }
        analysis.setMergeOrderParameter(analyzeOrderBy(queryStatement));
        return analyzeLast(analysis, schemaTree.getAllMeasurement(), schemaTree);
      }

      // Example 1: select s1, s1 + s2 as t, udf(udf(s1)) from root.sg.d1
      //   outputExpressions: [<root.sg.d1.s1,null>, <root.sg.d1.s1 + root.sg.d1.s2,t>,
      //                       <udf(udf(root.sg.d1.s1)),null>]
      //   transformExpressions: [root.sg.d1.s1, root.sg.d1.s1 + root.sg.d1.s2,
      //                       udf(udf(root.sg.d1.s1))]
      //   sourceExpressions: {root.sg.d1 -> [root.sg.d1.s1, root.sg.d1.s2]}
      //
      // Example 2: select s1, s2, s3 as t from root.sg.* align by device
      //   outputExpressions: [<s1,null>, <s2,null>, <s1,t>]
      //   transformExpressions: [root.sg.d1.s1, root.sg.d1.s2, root.sg.d1.s3,
      //                       root.sg.d2.s1, root.sg.d2.s2]
      //   sourceExpressions: {root.sg.d1 -> [root.sg.d1.s1, root.sg.d1.s2, root.sg.d1.s2],
      //                       root.sg.d2 -> [root.sg.d2.s1, root.sg.d2.s2]}
      //
      // Example 3: select sum(s1) + 1 as t, count(s2) from root.sg.d1
      //   outputExpressions: [<sum(root.sg.d1.s1) + 1,t>, <count(root.sg.d1.s2),t>]
      //   transformExpressions: [sum(root.sg.d1.s1) + 1, count(root.sg.d1.s2)]
      //   aggregationExpressions: {root.sg.d1 -> [sum(root.sg.d1.s1), count(root.sg.d1.s2)]}
      //   sourceExpressions: {root.sg.d1 -> [sum(root.sg.d1.s1), count(root.sg.d1.s2)]}
      //
      // Example 4: select sum(s1) + 1 as t, count(s2) from root.sg.d1 where s1 > 1
      //   outputExpressions: [<sum(root.sg.d1.s1) + 1,t>, <count(root.sg.d1.s2),t>]
      //   transformExpressions: [sum(root.sg.d1.s1) + 1, count(root.sg.d1.s2)]
      //   aggregationExpressions: {root.sg.d1 -> [sum(root.sg.d1.s1), count(root.sg.d1.s2)]}
      //   sourceExpressions: {root.sg.d1 -> [root.sg.d1.s1, root.sg.d1.s2]}
      List<Pair<Expression, String>> outputExpressions;
      if (queryStatement.isAlignByDevice()) {
        Map<String, Set<Expression>> deviceToTransformExpressions = new HashMap<>();
        Map<String, Set<Expression>> deviceToTransformExpressionsInHaving = new HashMap<>();

        // all selected device
        Set<PartialPath> deviceList = analyzeFrom(queryStatement, schemaTree);

        Map<String, Set<String>> deviceToMeasurementsMap = new HashMap<>();
        outputExpressions =
            analyzeSelect(
                queryStatement,
                schemaTree,
                deviceList,
                deviceToTransformExpressions,
                deviceToMeasurementsMap);

        if (queryStatement.hasHaving()) {
          List<PartialPath> measurementNotExistDevices = new ArrayList<>();
          for (PartialPath device : deviceList) {
            try {
              deviceToTransformExpressionsInHaving.put(
                  device.toString(),
                  ExpressionAnalyzer.removeWildcardInFilterByDevice(
                          queryStatement.getHavingCondition().getPredicate(),
                          device,
                          schemaTree,
                          typeProvider,
                          false)
                      .stream()
                      .collect(Collectors.toSet()));
            } catch (SemanticException e) {
              if (e instanceof MeasurementNotExistException) {
                logger.warn(e.getMessage());
                measurementNotExistDevices.add(device);
                continue;
              }
              throw e;
            }
          }
          for (PartialPath measurementNotExistDevice : measurementNotExistDevices) {
            deviceList.remove(measurementNotExistDevice);
            deviceToTransformExpressions.remove(measurementNotExistDevice.getFullPath());
          }
        }

        Map<String, List<Integer>> deviceToMeasurementIndexesMap = new HashMap<>();
        List<String> allMeasurements =
            outputExpressions.stream()
                .map(Pair::getLeft)
                .map(Expression::getExpressionString)
                .distinct()
                .collect(Collectors.toList());
        for (String deviceName : deviceToMeasurementsMap.keySet()) {
          List<String> measurementsUnderDevice =
              new ArrayList<>(deviceToMeasurementsMap.get(deviceName));
          List<Integer> indexes = new ArrayList<>();
          for (String measurement : measurementsUnderDevice) {
            indexes.add(
                allMeasurements.indexOf(measurement) + 1); // add 1 to skip the device column
          }
          deviceToMeasurementIndexesMap.put(deviceName, indexes);
        }
        analysis.setDeviceToMeasurementIndexesMap(deviceToMeasurementIndexesMap);

        Map<String, Set<Expression>> deviceToSourceExpressions = new HashMap<>();
        boolean isValueFilterAggregation = queryStatement.isAggregationQuery() && hasValueFilter;

        Map<String, Boolean> deviceToIsRawDataSource = new HashMap<>();

        Map<String, Set<Expression>> deviceToAggregationExpressions = new HashMap<>();
        Map<String, Set<Expression>> deviceToAggregationTransformExpressions = new HashMap<>();
        Map<String, Expression> deviceToHavingExpression =
            new HashMap<>(); // store filter of every device
        for (String deviceName : deviceToTransformExpressions.keySet()) {
          Set<Expression> transformExpressions = deviceToTransformExpressions.get(deviceName);
          Set<Expression> transformExpressionsInHaving =
              deviceToTransformExpressionsInHaving.get(deviceName);
          Set<Expression> aggregationExpressions = new LinkedHashSet<>();
          Set<Expression> aggregationTransformExpressions = new LinkedHashSet<>();

          boolean isHasRawDataInputAggregation = false;
          if (queryStatement.isAggregationQuery()) {
            // true if nested expressions and UDFs exist in aggregation function
            // i.e. select sum(s1 + 1) from root.sg.d1 align by device
            isHasRawDataInputAggregation =
                analyzeAggregation(
                    transformExpressions, aggregationExpressions, aggregationTransformExpressions);

            if (queryStatement.hasHaving() && transformExpressionsInHaving != null) {
              List<Expression> aggregationExpressionsInHaving = new ArrayList<>();
              isHasRawDataInputAggregation |=
                  analyzeAggregationInHaving(
                      transformExpressionsInHaving,
                      aggregationExpressionsInHaving,
                      aggregationExpressions,
                      aggregationTransformExpressions);

              Expression havingExpression;
              havingExpression =
                  analyzeHavingSplitByDevice(
                      transformExpressionsInHaving); // construct Filter from Having

              havingExpression.inferTypes(typeProvider);
              deviceToHavingExpression.put(deviceName, havingExpression);
            }

            deviceToAggregationExpressions.put(deviceName, aggregationExpressions);
            deviceToAggregationTransformExpressions.put(
                deviceName, aggregationTransformExpressions);
          }

          boolean isRawDataSource =
              !queryStatement.isAggregationQuery()
                  || isValueFilterAggregation
                  || isHasRawDataInputAggregation;

          for (Expression expression : transformExpressions) {
            updateSource(
                expression,
                deviceToSourceExpressions.computeIfAbsent(deviceName, key -> new LinkedHashSet<>()),
                isRawDataSource);
          }

          if (queryStatement.hasHaving() && transformExpressionsInHaving != null) {
            for (Expression expression : transformExpressionsInHaving) {
              updateSource(
                  expression,
                  deviceToSourceExpressions.computeIfAbsent(
                      deviceName, key -> new LinkedHashSet<>()),
                  isRawDataSource);
            }
          }
          deviceToIsRawDataSource.put(deviceName, isRawDataSource);
        }
        analysis.setDeviceToAggregationExpressions(deviceToAggregationExpressions);
        analysis.setDeviceToAggregationTransformExpressions(
            deviceToAggregationTransformExpressions);
        analysis.setDeviceToIsRawDataSource(deviceToIsRawDataSource);
        analysis.setDeviceToHavingExpression(deviceToHavingExpression);

        if (queryStatement.getWhereCondition() != null) {
          Map<String, Expression> deviceToQueryFilter = new HashMap<>();
          Iterator<PartialPath> deviceIterator = deviceList.iterator();
          while (deviceIterator.hasNext()) {
            PartialPath devicePath = deviceIterator.next();
            Expression queryFilter = null;
            try {
              queryFilter = analyzeWhereSplitByDevice(queryStatement, devicePath, schemaTree);
            } catch (SemanticException e) {
              if (e instanceof MeasurementNotExistException) {
                logger.warn(e.getMessage());
                deviceIterator.remove();
                deviceToSourceExpressions.remove(devicePath.getFullPath());
                continue;
              }
              throw e;
            }
            deviceToQueryFilter.put(devicePath.getFullPath(), queryFilter);
            queryFilter.inferTypes(typeProvider);
            updateSource(
                queryFilter,
                deviceToSourceExpressions.computeIfAbsent(
                    devicePath.getFullPath(), key -> new LinkedHashSet<>()),
                true);
          }
          analysis.setDeviceToQueryFilter(deviceToQueryFilter);
        }
        analysis.setDeviceToSourceExpressions(deviceToSourceExpressions);
        analysis.setDeviceToTransformExpressions(deviceToTransformExpressions);
      } else {
        outputExpressions = analyzeSelect(queryStatement, schemaTree);
        Set<Expression> transformExpressions =
            outputExpressions.stream()
                .map(Pair::getLeft)
                .collect(Collectors.toCollection(LinkedHashSet::new));

        // get removeWildcard Expressions in Having
        // used to analyzeAggregation in Having expression and updateSource
        Set<Expression> transformExpressionsInHaving =
            queryStatement.hasHaving()
                ? ExpressionAnalyzer.removeWildcardInFilter(
                        queryStatement.getHavingCondition().getPredicate(),
                        queryStatement.getFromComponent().getPrefixPaths(),
                        schemaTree,
                        typeProvider,
                        false)
                    .stream()
                    .collect(Collectors.toSet())
                : null;

        if (queryStatement.isGroupByLevel()) {
          // map from grouped expression to set of input expressions
          Map<Expression, Expression> rawPathToGroupedPathMap = new HashMap<>();
          Map<Expression, Set<Expression>> groupByLevelExpressions =
              analyzeGroupByLevel(
                  queryStatement, outputExpressions, transformExpressions, rawPathToGroupedPathMap);
          analysis.setGroupByLevelExpressions(groupByLevelExpressions);
          analysis.setRawPathToGroupedPathMap(rawPathToGroupedPathMap);
        }

        // true if nested expressions and UDFs exist in aggregation function
        // i.e. select sum(s1 + 1) from root.sg.d1
        boolean isHasRawDataInputAggregation = false;
        if (queryStatement.isAggregationQuery()) {
          Set<Expression> aggregationExpressions = new HashSet<>();
          Set<Expression> aggregationTransformExpressions = new HashSet<>();
          List<Expression> aggregationExpressionsInHaving =
              new ArrayList<>(); // as input of GroupByLevelController
          isHasRawDataInputAggregation =
              analyzeAggregation(
                  transformExpressions, aggregationExpressions, aggregationTransformExpressions);
          if (queryStatement.hasHaving()) {
            isHasRawDataInputAggregation |=
                analyzeAggregationInHaving(
                    transformExpressionsInHaving,
                    aggregationExpressionsInHaving,
                    aggregationExpressions,
                    aggregationTransformExpressions);

            Expression havingExpression = // construct Filter from Having
                analyzeHaving(
                    queryStatement,
                    analysis.getGroupByLevelExpressions(),
                    transformExpressionsInHaving,
                    aggregationExpressionsInHaving);
            havingExpression.inferTypes(typeProvider);
            analysis.setHavingExpression(havingExpression);
          }
          analysis.setAggregationExpressions(aggregationExpressions);
          analysis.setAggregationTransformExpressions(aggregationTransformExpressions);
        }

        // generate sourceExpression according to transformExpressions
        Set<Expression> sourceExpressions = new HashSet<>();
        boolean isValueFilterAggregation = queryStatement.isAggregationQuery() && hasValueFilter;
        boolean isRawDataSource =
            !queryStatement.isAggregationQuery()
                || isValueFilterAggregation
                || isHasRawDataInputAggregation;
        for (Expression expression : transformExpressions) {
          updateSource(expression, sourceExpressions, isRawDataSource);
        }

        if (queryStatement.hasHaving()) {
          for (Expression expression : transformExpressionsInHaving) {
            updateSource(expression, sourceExpressions, isRawDataSource);
          }
        }

        if (queryStatement.getWhereCondition() != null) {
          Expression queryFilter = analyzeWhere(queryStatement, schemaTree);

          // update sourceExpression according to queryFilter
          queryFilter.inferTypes(typeProvider);
          updateSource(queryFilter, sourceExpressions, isRawDataSource);
          analysis.setQueryFilter(queryFilter);
        }
        analysis.setRawDataSource(isRawDataSource);
        analysis.setSourceExpressions(sourceExpressions);
        analysis.setTransformExpressions(transformExpressions);
      }

      if (queryStatement.isGroupByTime()) {
        GroupByTimeComponent groupByTimeComponent = queryStatement.getGroupByTimeComponent();
        if ((groupByTimeComponent.isIntervalByMonth()
                || groupByTimeComponent.isSlidingStepByMonth())
            && queryStatement.getResultTimeOrder() == Ordering.DESC) {
          throw new SemanticException("Group by month doesn't support order by time desc now.");
        }
        analysis.setGroupByTimeParameter(new GroupByTimeParameter(groupByTimeComponent));
      }

      if (queryStatement.getFillComponent() != null) {
        FillComponent fillComponent = queryStatement.getFillComponent();
        List<Expression> fillColumnList =
            outputExpressions.stream().map(Pair::getLeft).distinct().collect(Collectors.toList());
        analysis.setFillDescriptor(
            new FillDescriptor(fillComponent.getFillPolicy(), fillComponent.getFillValue()));
      }

      // generate result set header according to output expressions
      DatasetHeader datasetHeader = analyzeOutput(queryStatement, outputExpressions);
      analysis.setRespDatasetHeader(datasetHeader);
      analysis.setTypeProvider(typeProvider);

      // fetch partition information
      Set<String> deviceSet = new HashSet<>();
      if (queryStatement.isAlignByDevice()) {
        deviceSet = analysis.getDeviceToSourceExpressions().keySet();
      } else {
        for (Expression expression : analysis.getSourceExpressions()) {
          deviceSet.add(ExpressionAnalyzer.getDeviceNameInSourceExpression(expression));
        }
      }
      DataPartition dataPartition = fetchDataPartitionByDevices(deviceSet, schemaTree);
      analysis.setDataPartitionInfo(dataPartition);
    } catch (StatementAnalyzeException e) {
      logger.error("Meet error when analyzing the query statement: ", e);
      throw new StatementAnalyzeException(
          "Meet error when analyzing the query statement: " + e.getMessage());
    }
    return analysis;
  }

  private List<Pair<Expression, String>> analyzeSelect(
      QueryStatement queryStatement, ISchemaTree schemaTree) {
    List<Pair<Expression, String>> outputExpressions = new ArrayList<>();
    boolean isGroupByLevel = queryStatement.isGroupByLevel();
    ColumnPaginationController paginationController =
        new ColumnPaginationController(
            queryStatement.getSeriesLimit(),
            queryStatement.getSeriesOffset(),
            queryStatement.isLastQuery() || isGroupByLevel);

    for (ResultColumn resultColumn : queryStatement.getSelectComponent().getResultColumns()) {
      boolean hasAlias = resultColumn.hasAlias();
      List<Expression> resultExpressions =
          ExpressionAnalyzer.removeWildcardInExpression(resultColumn.getExpression(), schemaTree);
      if (hasAlias && !queryStatement.isGroupByLevel() && resultExpressions.size() > 1) {
        throw new SemanticException(
            String.format(
                "alias '%s' can only be matched with one time series", resultColumn.getAlias()));
      }
      for (Expression expression : resultExpressions) {
        if (paginationController.hasCurOffset()) {
          paginationController.consumeOffset();
          continue;
        }
        if (paginationController.hasCurLimit()) {
          if (isGroupByLevel) {
            ExpressionAnalyzer.updateTypeProvider(expression, typeProvider);
            expression.inferTypes(typeProvider);
            outputExpressions.add(new Pair<>(expression, resultColumn.getAlias()));
            if (resultColumn.getExpression() instanceof FunctionExpression) {
              queryStatement
                  .getGroupByLevelComponent()
                  .updateIsCountStar((FunctionExpression) resultColumn.getExpression());
            }
          } else {
            Expression expressionWithoutAlias =
                ExpressionAnalyzer.removeAliasFromExpression(expression);
            String alias =
                !Objects.equals(expressionWithoutAlias, expression)
                    ? expression.getExpressionString()
                    : null;
            alias = hasAlias ? resultColumn.getAlias() : alias;

            ExpressionAnalyzer.updateTypeProvider(expressionWithoutAlias, typeProvider);
            expressionWithoutAlias.inferTypes(typeProvider);
            outputExpressions.add(new Pair<>(expressionWithoutAlias, alias));
          }
          paginationController.consumeLimit();
        } else {
          break;
        }
      }
    }
    return outputExpressions;
  }

  private Set<PartialPath> analyzeFrom(QueryStatement queryStatement, ISchemaTree schemaTree) {
    // device path patterns in FROM clause
    List<PartialPath> devicePatternList = queryStatement.getFromComponent().getPrefixPaths();

    Set<PartialPath> deviceList = new LinkedHashSet<>();
    for (PartialPath devicePattern : devicePatternList) {
      // get all matched devices
      deviceList.addAll(
          schemaTree.getMatchedDevices(devicePattern).stream()
              .map(DeviceSchemaInfo::getDevicePath)
              .collect(Collectors.toList()));
    }
    return deviceList;
  }

  private List<Pair<Expression, String>> analyzeSelect(
      QueryStatement queryStatement,
      ISchemaTree schemaTree,
      Set<PartialPath> deviceList,
      Map<String, Set<Expression>> deviceToTransformExpressions,
      Map<String, Set<String>> deviceToMeasurementsMap) {
    List<Pair<Expression, String>> outputExpressions = new ArrayList<>();
    ColumnPaginationController paginationController =
        new ColumnPaginationController(
            queryStatement.getSeriesLimit(), queryStatement.getSeriesOffset(), false);

    for (ResultColumn resultColumn : queryStatement.getSelectComponent().getResultColumns()) {
      Expression selectExpression = resultColumn.getExpression();
      boolean hasAlias = resultColumn.hasAlias();

      // measurement expression after removing wildcard
      // use LinkedHashMap for order-preserving
      Map<Expression, Map<String, Expression>> measurementToDeviceTransformExpressions =
          new LinkedHashMap<>();
      for (PartialPath device : deviceList) {
        List<Expression> transformExpressions =
            ExpressionAnalyzer.concatDeviceAndRemoveWildcard(
                selectExpression, device, schemaTree, typeProvider);
        for (Expression transformExpression : transformExpressions) {
          measurementToDeviceTransformExpressions
              .computeIfAbsent(
                  ExpressionAnalyzer.getMeasurementExpression(transformExpression),
                  key -> new LinkedHashMap<>())
              .put(
                  device.getFullPath(),
                  ExpressionAnalyzer.removeAliasFromExpression(transformExpression));
        }
      }

      if (hasAlias && measurementToDeviceTransformExpressions.keySet().size() > 1) {
        throw new SemanticException(
            String.format(
                "alias '%s' can only be matched with one time series", resultColumn.getAlias()));
      }

      for (Expression measurementExpression : measurementToDeviceTransformExpressions.keySet()) {
        if (paginationController.hasCurOffset()) {
          paginationController.consumeOffset();
          continue;
        }
        if (paginationController.hasCurLimit()) {
          Map<String, Expression> deviceToTransformExpressionOfOneMeasurement =
              measurementToDeviceTransformExpressions.get(measurementExpression);
          deviceToTransformExpressionOfOneMeasurement
              .values()
              .forEach(expression -> expression.inferTypes(typeProvider));
          // check whether the datatype of paths which has the same measurement name are
          // consistent
          // if not, throw a SemanticException
          checkDataTypeConsistencyInAlignByDevice(
              new ArrayList<>(deviceToTransformExpressionOfOneMeasurement.values()));

          // add outputExpressions
          Expression measurementExpressionWithoutAlias =
              ExpressionAnalyzer.removeAliasFromExpression(measurementExpression);
          String alias =
              !Objects.equals(measurementExpressionWithoutAlias, measurementExpression)
                  ? measurementExpression.getExpressionString()
                  : null;
          alias = hasAlias ? resultColumn.getAlias() : alias;
          ExpressionAnalyzer.updateTypeProvider(measurementExpressionWithoutAlias, typeProvider);
          measurementExpressionWithoutAlias.inferTypes(typeProvider);
          outputExpressions.add(new Pair<>(measurementExpressionWithoutAlias, alias));

          // add deviceToTransformExpressions
          for (String deviceName : deviceToTransformExpressionOfOneMeasurement.keySet()) {
            Expression transformExpression =
                deviceToTransformExpressionOfOneMeasurement.get(deviceName);
            ExpressionAnalyzer.updateTypeProvider(transformExpression, typeProvider);
            transformExpression.inferTypes(typeProvider);
            deviceToTransformExpressions
                .computeIfAbsent(deviceName, key -> new LinkedHashSet<>())
                .add(ExpressionAnalyzer.removeAliasFromExpression(transformExpression));
            deviceToMeasurementsMap
                .computeIfAbsent(deviceName, key -> new LinkedHashSet<>())
                .add(measurementExpressionWithoutAlias.getExpressionString());
          }
          paginationController.consumeLimit();
        } else {
          break;
        }
      }
    }

    return outputExpressions;
  }

  private Pair<Filter, Boolean> analyzeGlobalTimeFilter(QueryStatement queryStatement) {
    Filter globalTimeFilter = null;
    boolean hasValueFilter = false;
    if (queryStatement.getWhereCondition() != null) {
      Expression predicate = queryStatement.getWhereCondition().getPredicate();
      WhereCondition whereCondition = queryStatement.getWhereCondition();
      Pair<Filter, Boolean> resultPair =
          ExpressionAnalyzer.transformToGlobalTimeFilter(predicate, true, true);
      predicate = ExpressionAnalyzer.evaluatePredicate(predicate);

      // set where condition to null if predicate is true
      if (predicate.getExpressionType().equals(ExpressionType.CONSTANT)
          && Boolean.parseBoolean(predicate.getExpressionString())) {
        queryStatement.setWhereCondition(null);
      } else {
        whereCondition.setPredicate(predicate);
      }
      globalTimeFilter = resultPair.left;
      hasValueFilter = resultPair.right;
    }
    if (queryStatement.isGroupByTime()) {
      GroupByTimeComponent groupByTimeComponent = queryStatement.getGroupByTimeComponent();
      Filter groupByFilter = initGroupByFilter(groupByTimeComponent);
      if (globalTimeFilter == null) {
        globalTimeFilter = groupByFilter;
      } else {
        // TODO: optimize the filter
        globalTimeFilter = FilterFactory.and(globalTimeFilter, groupByFilter);
      }
    }
    return new Pair<>(globalTimeFilter, hasValueFilter);
  }

  private void updateSource(
      Expression selectExpr, Set<Expression> sourceExpressions, boolean isRawDataSource) {
    sourceExpressions.addAll(
        ExpressionAnalyzer.searchSourceExpressions(selectExpr, isRawDataSource));
  }

  private boolean analyzeAggregation(
      Set<Expression> transformExpressions,
      Set<Expression> aggregationExpressions,
      Set<Expression> aggregationTransformExpressions) {
    // true if nested expressions and UDFs exist in aggregation function
    // i.e. select sum(s1 + 1) from root.sg.d1 align by device
    boolean isHasRawDataInputAggregation = false;
    for (Expression expression : transformExpressions) {
      for (Expression aggregationExpression :
          ExpressionAnalyzer.searchAggregationExpressions(expression)) {
        aggregationExpressions.add(aggregationExpression);
        aggregationTransformExpressions.addAll(aggregationExpression.getExpressions());
      }
    }
    for (Expression aggregationTransformExpression : aggregationTransformExpressions) {
      if (ExpressionAnalyzer.checkIsNeedTransform(aggregationTransformExpression)) {
        isHasRawDataInputAggregation = true;
        break;
      }
    }
    return isHasRawDataInputAggregation;
  }

  private boolean analyzeAggregationInHaving(
      Set<Expression> expressions,
      List<Expression> aggregationExpressionsInHaving,
      Set<Expression> aggregationExpressions,
      Set<Expression> aggregationTransformExpressions) {
    // true if nested expressions and UDFs exist in aggregation function
    // i.e. select sum(s1 + 1) from root.sg.d1 align by device
    boolean isHasRawDataInputAggregation = false;
    for (Expression expression : expressions) {
      for (Expression aggregationExpression :
          ExpressionAnalyzer.searchAggregationExpressions(expression)) {
        aggregationExpressionsInHaving.add(aggregationExpression);
        aggregationExpressions.add(aggregationExpression);
        aggregationTransformExpressions.addAll(aggregationExpression.getExpressions());
      }
    }

    for (Expression aggregationTransformExpression : aggregationTransformExpressions) {
      if (ExpressionAnalyzer.checkIsNeedTransform(aggregationTransformExpression)) {
        isHasRawDataInputAggregation = true;
        break;
      }
    }
    return isHasRawDataInputAggregation;
  }

  private Expression analyzeWhere(QueryStatement queryStatement, ISchemaTree schemaTree) {
    List<Expression> rewrittenPredicates =
        ExpressionAnalyzer.removeWildcardInFilter(
            queryStatement.getWhereCondition().getPredicate(),
            queryStatement.getFromComponent().getPrefixPaths(),
            schemaTree,
            typeProvider,
            true);
    return ExpressionUtils.constructQueryFilter(
        rewrittenPredicates.stream().distinct().collect(Collectors.toList()));
  }

  private Expression analyzeWhereSplitByDevice(
      QueryStatement queryStatement, PartialPath devicePath, ISchemaTree schemaTree) {
    List<Expression> rewrittenPredicates =
        ExpressionAnalyzer.removeWildcardInFilterByDevice(
            queryStatement.getWhereCondition().getPredicate(),
            devicePath,
            schemaTree,
            typeProvider,
            true);
    return ExpressionUtils.constructQueryFilter(
        rewrittenPredicates.stream().distinct().collect(Collectors.toList()));
  }

  private Expression analyzeHaving(
      QueryStatement queryStatement,
      Map<Expression, Set<Expression>> groupByLevelExpressions,
      Set<Expression> transformExpressionsInHaving,
      List<Expression> aggregationExpressionsInHaving) {

    if (queryStatement.isGroupByLevel()) {
      Map<Expression, Expression> RawPathToGroupedPathMapInHaving =
          analyzeGroupByLevelInHaving(
              queryStatement, aggregationExpressionsInHaving, groupByLevelExpressions);
      List<Expression> convertedPredicates = new ArrayList<>();
      for (Expression expression : transformExpressionsInHaving) {
        convertedPredicates.add(
            ExpressionAnalyzer.replaceRawPathWithGroupedPath(
                expression, RawPathToGroupedPathMapInHaving));
      }
      return ExpressionUtils.constructQueryFilter(
          convertedPredicates.stream().distinct().collect(Collectors.toList()));
    }

    return ExpressionUtils.constructQueryFilter(
        transformExpressionsInHaving.stream().distinct().collect(Collectors.toList()));
  }

  private Expression analyzeHavingSplitByDevice(Set<Expression> transformExpressionsInHaving) {
    return ExpressionUtils.constructQueryFilter(
        transformExpressionsInHaving.stream().distinct().collect(Collectors.toList()));
  }

  private Map<Expression, Expression> analyzeGroupByLevelInHaving(
      QueryStatement queryStatement,
      List<Expression> inputExpressions,
      Map<Expression, Set<Expression>> groupByLevelExpressions) {
    GroupByLevelController groupByLevelController =
        new GroupByLevelController(
            queryStatement.getGroupByLevelComponent().getLevels(), typeProvider);
    for (Expression inputExpression : inputExpressions) {
      groupByLevelController.control(false, inputExpression, null);
    }
    Map<Expression, Set<Expression>> groupedPathMap = groupByLevelController.getGroupedPathMap();
    groupByLevelExpressions.putAll(groupedPathMap);
    return groupByLevelController.getRawPathToGroupedPathMap();
  }

  private Map<Expression, Set<Expression>> analyzeGroupByLevel(
      QueryStatement queryStatement,
      List<Pair<Expression, String>> outputExpressions,
      Set<Expression> transformExpressions,
      Map<Expression, Expression> rawPathToGroupedPathMap) {
    GroupByLevelController groupByLevelController =
        new GroupByLevelController(
            queryStatement.getGroupByLevelComponent().getLevels(), typeProvider);
    for (int i = 0; i < outputExpressions.size(); i++) {
      Pair<Expression, String> measurementWithAlias = outputExpressions.get(i);
      boolean isCountStar = queryStatement.getGroupByLevelComponent().isCountStar(i);
      groupByLevelController.control(
          isCountStar, measurementWithAlias.left, measurementWithAlias.right);
    }
    Map<Expression, Set<Expression>> rawGroupByLevelExpressions =
        groupByLevelController.getGroupedPathMap();
    rawPathToGroupedPathMap.putAll(groupByLevelController.getRawPathToGroupedPathMap());

    Map<Expression, Set<Expression>> groupByLevelExpressions = new LinkedHashMap<>();
    outputExpressions.clear();
    ColumnPaginationController paginationController =
        new ColumnPaginationController(
            queryStatement.getSeriesLimit(), queryStatement.getSeriesOffset(), false);
    for (Expression groupedExpression : rawGroupByLevelExpressions.keySet()) {
      if (paginationController.hasCurOffset()) {
        paginationController.consumeOffset();
        continue;
      }
      if (paginationController.hasCurLimit()) {
        Pair<Expression, String> outputExpression =
            removeAliasFromExpression(
                groupedExpression,
                groupByLevelController.getAlias(groupedExpression.getExpressionString()));
        Expression groupedExpressionWithoutAlias = outputExpression.left;

        Set<Expression> rawExpressions = rawGroupByLevelExpressions.get(groupedExpression);
        rawExpressions.forEach(
            expression -> ExpressionAnalyzer.updateTypeProvider(expression, typeProvider));
        rawExpressions.forEach(expression -> expression.inferTypes(typeProvider));

        Set<Expression> rawExpressionsWithoutAlias =
            rawExpressions.stream()
                .map(ExpressionAnalyzer::removeAliasFromExpression)
                .collect(Collectors.toSet());
        rawExpressionsWithoutAlias.forEach(
            expression -> ExpressionAnalyzer.updateTypeProvider(expression, typeProvider));
        rawExpressionsWithoutAlias.forEach(expression -> expression.inferTypes(typeProvider));

        groupByLevelExpressions.put(groupedExpressionWithoutAlias, rawExpressionsWithoutAlias);

        TSDataType dataType =
            typeProvider.getType(
                new ArrayList<>(groupByLevelExpressions.get(groupedExpressionWithoutAlias))
                    .get(0)
                    .getExpressionString());
        typeProvider.setType(groupedExpression.getExpressionString(), dataType);
        typeProvider.setType(groupedExpressionWithoutAlias.getExpressionString(), dataType);
        outputExpressions.add(outputExpression);
        paginationController.consumeLimit();
      } else {
        break;
      }
    }

    // reset transformExpressions after applying SLIMIT/SOFFSET
    transformExpressions.clear();
    transformExpressions.addAll(
        groupByLevelExpressions.values().stream().flatMap(Set::stream).collect(Collectors.toSet()));
    return groupByLevelExpressions;
  }

  private Pair<Expression, String> removeAliasFromExpression(
      Expression rawExpression, String rawAlias) {
    Expression expressionWithoutAlias = ExpressionAnalyzer.removeAliasFromExpression(rawExpression);
    String alias =
        !Objects.equals(expressionWithoutAlias, rawExpression)
            ? rawExpression.getExpressionString()
            : null;
    alias = rawAlias == null ? alias : rawAlias;
    return new Pair<>(expressionWithoutAlias, alias);
  }

  private DatasetHeader analyzeOutput(
      QueryStatement queryStatement, List<Pair<Expression, String>> outputExpressions) {
    boolean isIgnoreTimestamp =
        queryStatement.isAggregationQuery() && !queryStatement.isGroupByTime();
    List<ColumnHeader> columnHeaders = new ArrayList<>();
    if (queryStatement.isAlignByDevice()) {
      columnHeaders.add(new ColumnHeader(HeaderConstant.COLUMN_DEVICE, TSDataType.TEXT, null));
      typeProvider.setType(HeaderConstant.COLUMN_DEVICE, TSDataType.TEXT);
    }
    columnHeaders.addAll(
        outputExpressions.stream()
            .map(
                expressionWithAlias -> {
                  String columnName = expressionWithAlias.left.getExpressionString();
                  String alias = expressionWithAlias.right;
                  return new ColumnHeader(columnName, typeProvider.getType(columnName), alias);
                })
            .collect(Collectors.toList()));
    return new DatasetHeader(columnHeaders, isIgnoreTimestamp);
  }

  private Analysis analyzeLast(
      Analysis analysis, List<MeasurementPath> allSelectedPath, ISchemaTree schemaTree) {
    Set<Expression> sourceExpressions;
    List<SortItem> sortItemList = analysis.getMergeOrderParameter().getSortItemList();
    if (sortItemList.size() > 0) {
      checkState(
          sortItemList.size() == 1 && sortItemList.get(0).getSortKey() == SortKey.TIMESERIES,
          "Last queries only support sorting by timeseries now.");
      boolean isAscending = sortItemList.get(0).getOrdering() == Ordering.ASC;
      sourceExpressions =
          allSelectedPath.stream()
              .map(TimeSeriesOperand::new)
              .sorted(
                  (o1, o2) ->
                      isAscending
                          ? o1.getExpressionString().compareTo(o2.getExpressionString())
                          : o2.getExpressionString().compareTo(o1.getExpressionString()))
              .collect(Collectors.toCollection(LinkedHashSet::new));
    } else {
      sourceExpressions =
          allSelectedPath.stream()
              .map(TimeSeriesOperand::new)
              .collect(Collectors.toCollection(LinkedHashSet::new));
    }

    sourceExpressions.forEach(
        expression -> ExpressionAnalyzer.updateTypeProvider(expression, typeProvider));
    analysis.setSourceExpressions(sourceExpressions);

    analysis.setRespDatasetHeader(HeaderConstant.LAST_QUERY_HEADER);
    typeProvider.setType(HeaderConstant.COLUMN_TIMESERIES, TSDataType.TEXT);
    typeProvider.setType(HeaderConstant.COLUMN_VALUE, TSDataType.TEXT);
    typeProvider.setType(HeaderConstant.COLUMN_TIMESERIES_DATATYPE, TSDataType.TEXT);

    Set<String> deviceSet =
        allSelectedPath.stream().map(MeasurementPath::getDevice).collect(Collectors.toSet());
    Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap = new HashMap<>();
    for (String devicePath : deviceSet) {
      DataPartitionQueryParam queryParam = new DataPartitionQueryParam();
      queryParam.setDevicePath(devicePath);
      sgNameToQueryParamsMap
          .computeIfAbsent(schemaTree.getBelongedStorageGroup(devicePath), key -> new ArrayList<>())
          .add(queryParam);
    }
    DataPartition dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
    analysis.setDataPartitionInfo(dataPartition);

    return analysis;
  }

  private DataPartition fetchDataPartitionByDevices(Set<String> deviceSet, ISchemaTree schemaTree) {
    Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap = new HashMap<>();
    for (String devicePath : deviceSet) {
      DataPartitionQueryParam queryParam = new DataPartitionQueryParam();
      queryParam.setDevicePath(devicePath);
      sgNameToQueryParamsMap
          .computeIfAbsent(schemaTree.getBelongedStorageGroup(devicePath), key -> new ArrayList<>())
          .add(queryParam);
    }
    return partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
  }

  private OrderByParameter analyzeOrderBy(QueryStatement queryStatement) {
    return new OrderByParameter(queryStatement.getSortItemList());
  }

  /**
   * Check datatype consistency in ALIGN BY DEVICE.
   *
   * <p>an inconsistent example: select s0 from root.sg1.d1, root.sg1.d2 align by device, return
   * false while root.sg1.d1.s0 is INT32 and root.sg1.d2.s0 is FLOAT.
   */
  private void checkDataTypeConsistencyInAlignByDevice(List<Expression> expressions) {
    TSDataType checkedDataType = typeProvider.getType(expressions.get(0).getExpressionString());
    for (Expression expression : expressions) {
      if (typeProvider.getType(expression.getExpressionString()) != checkedDataType) {
        throw new SemanticException(
            "ALIGN BY DEVICE: the data types of the same measurement column should be the same across devices.");
      }
    }
  }

  @Override
  public Analysis visitInsert(InsertStatement insertStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);
    insertStatement.semanticCheck();
    long[] timeArray = insertStatement.getTimes();
    PartialPath devicePath = insertStatement.getDevice();
    String[] measurementList = insertStatement.getMeasurementList();
    if (timeArray.length == 1) {
      // construct insert row statement
      InsertRowStatement insertRowStatement = new InsertRowStatement();
      insertRowStatement.setDevicePath(devicePath);
      insertRowStatement.setTime(timeArray[0]);
      insertRowStatement.setMeasurements(measurementList);
      insertRowStatement.setDataTypes(new TSDataType[measurementList.length]);
      Object[] values = new Object[measurementList.length];
      System.arraycopy(insertStatement.getValuesList().get(0), 0, values, 0, values.length);
      insertRowStatement.setValues(values);
      insertRowStatement.setNeedInferType(true);
      insertRowStatement.setAligned(insertStatement.isAligned());
      return insertRowStatement.accept(this, context);
    } else {
      // construct insert rows statement
      // construct insert statement
      InsertRowsStatement insertRowsStatement = new InsertRowsStatement();
      List<InsertRowStatement> insertRowStatementList = new ArrayList<>();
      for (int i = 0; i < timeArray.length; i++) {
        InsertRowStatement statement = new InsertRowStatement();
        statement.setDevicePath(devicePath);
        String[] measurements = new String[measurementList.length];
        System.arraycopy(measurementList, 0, measurements, 0, measurements.length);
        statement.setMeasurements(measurements);
        statement.setTime(timeArray[i]);
        statement.setDataTypes(new TSDataType[measurementList.length]);
        Object[] values = new Object[measurementList.length];
        System.arraycopy(insertStatement.getValuesList().get(i), 0, values, 0, values.length);
        statement.setValues(values);
        statement.setAligned(insertStatement.isAligned());
        statement.setNeedInferType(true);
        insertRowStatementList.add(statement);
      }
      insertRowsStatement.setInsertRowStatementList(insertRowStatementList);
      return insertRowsStatement.accept(this, context);
    }
  }

  @Override
  public Analysis visitCreateTimeseries(
      CreateTimeSeriesStatement createTimeSeriesStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);
    if (createTimeSeriesStatement.getTags() != null
        && !createTimeSeriesStatement.getTags().isEmpty()
        && createTimeSeriesStatement.getAttributes() != null
        && !createTimeSeriesStatement.getAttributes().isEmpty()) {
      for (String tagKey : createTimeSeriesStatement.getTags().keySet()) {
        if (createTimeSeriesStatement.getAttributes().containsKey(tagKey)) {
          throw new SemanticException(
              String.format("Tag and attribute shouldn't have the same property key [%s]", tagKey));
        }
      }
    }
    Analysis analysis = new Analysis();
    analysis.setStatement(createTimeSeriesStatement);

    checkIsTemplateCompatible(
        createTimeSeriesStatement.getPath(), createTimeSeriesStatement.getAlias());

    PathPatternTree patternTree = new PathPatternTree();
    patternTree.appendFullPath(createTimeSeriesStatement.getPath());
    SchemaPartition schemaPartitionInfo = partitionFetcher.getOrCreateSchemaPartition(patternTree);
    analysis.setSchemaPartitionInfo(schemaPartitionInfo);
    return analysis;
  }

  private void checkIsTemplateCompatible(PartialPath timeseriesPath, String alias) {
    Pair<Template, PartialPath> templateInfo = schemaFetcher.checkTemplateSetInfo(timeseriesPath);
    if (templateInfo != null) {
      if (templateInfo.left.hasSchema(timeseriesPath.getMeasurement())) {
        throw new RuntimeException(
            new TemplateImcompatibeException(
                timeseriesPath.getFullPath(),
                templateInfo.left.getName(),
                timeseriesPath.getMeasurement()));
      }

      if (alias != null && templateInfo.left.hasSchema(alias)) {
        throw new RuntimeException(
            new TemplateImcompatibeException(
                timeseriesPath.getDevicePath().concatNode(alias).getFullPath(),
                templateInfo.left.getName(),
                alias));
      }
    }
  }

  private void checkIsTemplateCompatible(
      PartialPath devicePath, List<String> measurements, List<String> aliasList) {
    Pair<Template, PartialPath> templateInfo = schemaFetcher.checkTemplateSetInfo(devicePath);
    if (templateInfo != null) {
      Template template = templateInfo.left;
      for (String measurement : measurements) {
        if (template.hasSchema(measurement)) {
          throw new RuntimeException(
              new TemplateImcompatibeException(
                  devicePath.concatNode(measurement).getFullPath(),
                  templateInfo.left.getName(),
                  measurement));
        }
      }

      if (aliasList == null) {
        return;
      }

      for (String alias : aliasList) {
        if (template.hasSchema(alias)) {
          throw new RuntimeException(
              new TemplateImcompatibeException(
                  devicePath.concatNode(alias).getFullPath(), templateInfo.left.getName(), alias));
        }
      }
    }
  }

  @Override
  public Analysis visitCreateAlignedTimeseries(
      CreateAlignedTimeSeriesStatement createAlignedTimeSeriesStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);
    List<String> measurements = createAlignedTimeSeriesStatement.getMeasurements();
    Set<String> measurementsSet = new HashSet<>(measurements);
    if (measurementsSet.size() < measurements.size()) {
      throw new SemanticException(
          "Measurement under an aligned device is not allowed to have the same measurement name");
    }

    Analysis analysis = new Analysis();
    analysis.setStatement(createAlignedTimeSeriesStatement);

    checkIsTemplateCompatible(
        createAlignedTimeSeriesStatement.getDevicePath(),
        createAlignedTimeSeriesStatement.getMeasurements(),
        createAlignedTimeSeriesStatement.getAliasList());

    PathPatternTree pathPatternTree = new PathPatternTree();
    for (String measurement : createAlignedTimeSeriesStatement.getMeasurements()) {
      pathPatternTree.appendFullPath(createAlignedTimeSeriesStatement.getDevicePath(), measurement);
    }

    SchemaPartition schemaPartitionInfo;
    schemaPartitionInfo = partitionFetcher.getOrCreateSchemaPartition(pathPatternTree);
    analysis.setSchemaPartitionInfo(schemaPartitionInfo);
    return analysis;
  }

  @Override
  public Analysis visitInternalCreateTimeseries(
      InternalCreateTimeSeriesStatement internalCreateTimeSeriesStatement,
      MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);

    Analysis analysis = new Analysis();
    analysis.setStatement(internalCreateTimeSeriesStatement);

    checkIsTemplateCompatible(
        internalCreateTimeSeriesStatement.getDevicePath(),
        internalCreateTimeSeriesStatement.getMeasurements(),
        null);

    PathPatternTree pathPatternTree = new PathPatternTree();
    for (String measurement : internalCreateTimeSeriesStatement.getMeasurements()) {
      pathPatternTree.appendFullPath(
          internalCreateTimeSeriesStatement.getDevicePath(), measurement);
    }

    SchemaPartition schemaPartitionInfo;
    schemaPartitionInfo = partitionFetcher.getOrCreateSchemaPartition(pathPatternTree);
    analysis.setSchemaPartitionInfo(schemaPartitionInfo);
    return analysis;
  }

  @Override
  public Analysis visitCreateMultiTimeseries(
      CreateMultiTimeSeriesStatement createMultiTimeSeriesStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);
    Analysis analysis = new Analysis();
    analysis.setStatement(createMultiTimeSeriesStatement);

    List<PartialPath> timeseriesPathList = createMultiTimeSeriesStatement.getPaths();
    List<String> aliasList = createMultiTimeSeriesStatement.getAliasList();
    for (int i = 0; i < timeseriesPathList.size(); i++) {
      checkIsTemplateCompatible(
          timeseriesPathList.get(i), aliasList == null ? null : aliasList.get(i));
    }

    PathPatternTree patternTree = new PathPatternTree();
    for (PartialPath path : createMultiTimeSeriesStatement.getPaths()) {
      patternTree.appendFullPath(path);
    }
    SchemaPartition schemaPartitionInfo = partitionFetcher.getOrCreateSchemaPartition(patternTree);
    analysis.setSchemaPartitionInfo(schemaPartitionInfo);
    return analysis;
  }

  @Override
  public Analysis visitAlterTimeseries(
      AlterTimeSeriesStatement alterTimeSeriesStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);
    Analysis analysis = new Analysis();
    analysis.setStatement(alterTimeSeriesStatement);

    if (alterTimeSeriesStatement.getAlias() != null) {
      checkIsTemplateCompatible(
          alterTimeSeriesStatement.getPath(), alterTimeSeriesStatement.getAlias());
    }

    PathPatternTree patternTree = new PathPatternTree();
    patternTree.appendFullPath(alterTimeSeriesStatement.getPath());
    SchemaPartition schemaPartitionInfo;
    schemaPartitionInfo = partitionFetcher.getSchemaPartition(patternTree);
    analysis.setSchemaPartitionInfo(schemaPartitionInfo);
    return analysis;
  }

  @Override
  public Analysis visitInsertTablet(
      InsertTabletStatement insertTabletStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);

    DataPartitionQueryParam dataPartitionQueryParam = new DataPartitionQueryParam();
    dataPartitionQueryParam.setDevicePath(insertTabletStatement.getDevicePath().getFullPath());
    dataPartitionQueryParam.setTimePartitionSlotList(insertTabletStatement.getTimePartitionSlots());

    DataPartition dataPartition =
        partitionFetcher.getOrCreateDataPartition(
            Collections.singletonList(dataPartitionQueryParam));

    Analysis analysis = new Analysis();
    analysis.setStatement(insertTabletStatement);
    analysis.setDataPartitionInfo(dataPartition);

    return analysis;
  }

  @Override
  public Analysis visitInsertRow(InsertRowStatement insertRowStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);

    DataPartitionQueryParam dataPartitionQueryParam = new DataPartitionQueryParam();
    dataPartitionQueryParam.setDevicePath(insertRowStatement.getDevicePath().getFullPath());
    dataPartitionQueryParam.setTimePartitionSlotList(insertRowStatement.getTimePartitionSlots());

    DataPartition dataPartition =
        partitionFetcher.getOrCreateDataPartition(
            Collections.singletonList(dataPartitionQueryParam));

    Analysis analysis = new Analysis();
    analysis.setStatement(insertRowStatement);
    analysis.setDataPartitionInfo(dataPartition);

    return analysis;
  }

  @Override
  public Analysis visitInsertRows(
      InsertRowsStatement insertRowsStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);

    List<DataPartitionQueryParam> dataPartitionQueryParams = new ArrayList<>();
    for (InsertRowStatement insertRowStatement : insertRowsStatement.getInsertRowStatementList()) {
      DataPartitionQueryParam dataPartitionQueryParam = new DataPartitionQueryParam();
      dataPartitionQueryParam.setDevicePath(insertRowStatement.getDevicePath().getFullPath());
      dataPartitionQueryParam.setTimePartitionSlotList(insertRowStatement.getTimePartitionSlots());
      dataPartitionQueryParams.add(dataPartitionQueryParam);
    }

    DataPartition dataPartition =
        partitionFetcher.getOrCreateDataPartition(dataPartitionQueryParams);

    Analysis analysis = new Analysis();
    analysis.setStatement(insertRowsStatement);
    analysis.setDataPartitionInfo(dataPartition);

    return analysis;
  }

  @Override
  public Analysis visitInsertMultiTablets(
      InsertMultiTabletsStatement insertMultiTabletsStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);

    List<DataPartitionQueryParam> dataPartitionQueryParams = new ArrayList<>();
    for (InsertTabletStatement insertTabletStatement :
        insertMultiTabletsStatement.getInsertTabletStatementList()) {
      DataPartitionQueryParam dataPartitionQueryParam = new DataPartitionQueryParam();
      dataPartitionQueryParam.setDevicePath(insertTabletStatement.getDevicePath().getFullPath());
      dataPartitionQueryParam.setTimePartitionSlotList(
          insertTabletStatement.getTimePartitionSlots());
      dataPartitionQueryParams.add(dataPartitionQueryParam);
    }

    DataPartition dataPartition =
        partitionFetcher.getOrCreateDataPartition(dataPartitionQueryParams);

    Analysis analysis = new Analysis();
    analysis.setStatement(insertMultiTabletsStatement);
    analysis.setDataPartitionInfo(dataPartition);

    return analysis;
  }

  @Override
  public Analysis visitInsertRowsOfOneDevice(
      InsertRowsOfOneDeviceStatement insertRowsOfOneDeviceStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);

    DataPartitionQueryParam dataPartitionQueryParam = new DataPartitionQueryParam();
    dataPartitionQueryParam.setDevicePath(
        insertRowsOfOneDeviceStatement.getDevicePath().getFullPath());
    dataPartitionQueryParam.setTimePartitionSlotList(
        insertRowsOfOneDeviceStatement.getTimePartitionSlots());

    DataPartition dataPartition =
        partitionFetcher.getOrCreateDataPartition(
            Collections.singletonList(dataPartitionQueryParam));

    Analysis analysis = new Analysis();
    analysis.setStatement(insertRowsOfOneDeviceStatement);
    analysis.setDataPartitionInfo(dataPartition);

    return analysis;
  }

  @Override
  public Analysis visitShowTimeSeries(
      ShowTimeSeriesStatement showTimeSeriesStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(showTimeSeriesStatement);

    PathPatternTree patternTree = new PathPatternTree();
    patternTree.appendPathPattern(showTimeSeriesStatement.getPathPattern());
    SchemaPartition schemaPartitionInfo = partitionFetcher.getSchemaPartition(patternTree);
    analysis.setSchemaPartitionInfo(schemaPartitionInfo);

    Map<Integer, Template> templateMap =
        schemaFetcher.checkAllRelatedTemplate(showTimeSeriesStatement.getPathPattern());
    analysis.setRelatedTemplateInfo(templateMap);

    if (showTimeSeriesStatement.isOrderByHeat()) {
      patternTree.constructTree();
      // request schema fetch API
      logger.info("{} fetch query schema...", getLogHeader());
      ISchemaTree schemaTree = schemaFetcher.fetchSchema(patternTree);
      logger.info("{} fetch schema done", getLogHeader());
      List<MeasurementPath> allSelectedPath = schemaTree.getAllMeasurement();

      Set<Expression> sourceExpressions =
          allSelectedPath.stream()
              .map(TimeSeriesOperand::new)
              .collect(Collectors.toCollection(LinkedHashSet::new));
      analysis.setSourceExpressions(sourceExpressions);

      Set<String> deviceSet =
          allSelectedPath.stream().map(MeasurementPath::getDevice).collect(Collectors.toSet());
      Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap = new HashMap<>();
      for (String devicePath : deviceSet) {
        DataPartitionQueryParam queryParam = new DataPartitionQueryParam();
        queryParam.setDevicePath(devicePath);
        sgNameToQueryParamsMap
            .computeIfAbsent(
                schemaTree.getBelongedStorageGroup(devicePath), key -> new ArrayList<>())
            .add(queryParam);
      }
      DataPartition dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
      analysis.setDataPartitionInfo(dataPartition);
    }

    analysis.setRespDatasetHeader(HeaderConstant.showTimeSeriesHeader);
    return analysis;
  }

  @Override
  public Analysis visitShowStorageGroup(
      ShowStorageGroupStatement showStorageGroupStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(showStorageGroupStatement);
    analysis.setRespDatasetHeader(HeaderConstant.showStorageGroupHeader);
    return analysis;
  }

  @Override
  public Analysis visitShowTTL(ShowTTLStatement showTTLStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(showTTLStatement);
    analysis.setRespDatasetHeader(HeaderConstant.showTTLHeader);
    return analysis;
  }

  @Override
  public Analysis visitShowDevices(
      ShowDevicesStatement showDevicesStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(showDevicesStatement);

    PathPatternTree patternTree = new PathPatternTree();
    patternTree.appendPathPattern(
        showDevicesStatement.getPathPattern().concatNode(IoTDBConstant.ONE_LEVEL_PATH_WILDCARD));
    SchemaPartition schemaPartitionInfo = partitionFetcher.getSchemaPartition(patternTree);

    analysis.setSchemaPartitionInfo(schemaPartitionInfo);
    analysis.setRespDatasetHeader(
        showDevicesStatement.hasSgCol()
            ? HeaderConstant.showDevicesWithSgHeader
            : HeaderConstant.showDevicesHeader);
    return analysis;
  }

  @Override
  public Analysis visitShowCluster(
      ShowClusterStatement showClusterStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(showClusterStatement);
    analysis.setRespDatasetHeader(HeaderConstant.showClusterHeader);
    return analysis;
  }

  @Override
  public Analysis visitCountStorageGroup(
      CountStorageGroupStatement countStorageGroupStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(countStorageGroupStatement);
    analysis.setRespDatasetHeader(HeaderConstant.countStorageGroupHeader);
    return analysis;
  }

  @Override
  public Analysis visitSchemaFetch(
      SchemaFetchStatement schemaFetchStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(schemaFetchStatement);

    SchemaPartition schemaPartition =
        partitionFetcher.getSchemaPartition(schemaFetchStatement.getPatternTree());
    analysis.setSchemaPartitionInfo(schemaPartition);

    if (schemaPartition.isEmpty()) {
      analysis.setFinishQueryAfterAnalyze(true);
    }

    return analysis;
  }

  @Override
  public Analysis visitCountDevices(
      CountDevicesStatement countDevicesStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(countDevicesStatement);

    PathPatternTree patternTree = new PathPatternTree();
    patternTree.appendPathPattern(
        countDevicesStatement.getPartialPath().concatNode(IoTDBConstant.ONE_LEVEL_PATH_WILDCARD));
    SchemaPartition schemaPartitionInfo = partitionFetcher.getSchemaPartition(patternTree);

    analysis.setSchemaPartitionInfo(schemaPartitionInfo);
    analysis.setRespDatasetHeader(HeaderConstant.countDevicesHeader);
    return analysis;
  }

  @Override
  public Analysis visitCountTimeSeries(
      CountTimeSeriesStatement countTimeSeriesStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(countTimeSeriesStatement);

    PathPatternTree patternTree = new PathPatternTree();
    patternTree.appendPathPattern(countTimeSeriesStatement.getPartialPath());
    SchemaPartition schemaPartitionInfo = partitionFetcher.getSchemaPartition(patternTree);

    analysis.setSchemaPartitionInfo(schemaPartitionInfo);
    analysis.setRespDatasetHeader(HeaderConstant.countTimeSeriesHeader);
    return analysis;
  }

  @Override
  public Analysis visitCountLevelTimeSeries(
      CountLevelTimeSeriesStatement countLevelTimeSeriesStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(countLevelTimeSeriesStatement);

    PathPatternTree patternTree = new PathPatternTree();
    patternTree.appendPathPattern(countLevelTimeSeriesStatement.getPartialPath());
    SchemaPartition schemaPartitionInfo = partitionFetcher.getSchemaPartition(patternTree);

    analysis.setSchemaPartitionInfo(schemaPartitionInfo);
    analysis.setRespDatasetHeader(HeaderConstant.countLevelTimeSeriesHeader);
    return analysis;
  }

  @Override
  public Analysis visitCountNodes(CountNodesStatement countStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(countStatement);

    PathPatternTree patternTree = new PathPatternTree();
    patternTree.appendPathPattern(countStatement.getPartialPath());
    SchemaNodeManagementPartition schemaNodeManagementPartition =
        partitionFetcher.getSchemaNodeManagementPartitionWithLevel(
            patternTree, countStatement.getLevel());

    if (schemaNodeManagementPartition == null) {
      return analysis;
    }
    if (!schemaNodeManagementPartition.getMatchedNode().isEmpty()
        && schemaNodeManagementPartition.getSchemaPartition().getSchemaPartitionMap().size() == 0) {
      analysis.setFinishQueryAfterAnalyze(true);
    }
    analysis.setMatchedNodes(schemaNodeManagementPartition.getMatchedNode());
    analysis.setSchemaPartitionInfo(schemaNodeManagementPartition.getSchemaPartition());
    analysis.setRespDatasetHeader(HeaderConstant.countNodesHeader);
    return analysis;
  }

  @Override
  public Analysis visitShowChildPaths(
      ShowChildPathsStatement showChildPathsStatement, MPPQueryContext context) {
    return visitSchemaNodeManagementPartition(
        showChildPathsStatement,
        showChildPathsStatement.getPartialPath(),
        HeaderConstant.showChildPathsHeader);
  }

  @Override
  public Analysis visitShowChildNodes(
      ShowChildNodesStatement showChildNodesStatement, MPPQueryContext context) {
    return visitSchemaNodeManagementPartition(
        showChildNodesStatement,
        showChildNodesStatement.getPartialPath(),
        HeaderConstant.showChildNodesHeader);
  }

  @Override
  public Analysis visitShowVersion(
      ShowVersionStatement showVersionStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(showVersionStatement);
    analysis.setRespDatasetHeader(HeaderConstant.showVersionHeader);
    analysis.setFinishQueryAfterAnalyze(true);
    return analysis;
  }

  private Analysis visitSchemaNodeManagementPartition(
      Statement statement, PartialPath path, DatasetHeader header) {
    Analysis analysis = new Analysis();
    analysis.setStatement(statement);

    PathPatternTree patternTree = new PathPatternTree();
    patternTree.appendPathPattern(path);
    SchemaNodeManagementPartition schemaNodeManagementPartition =
        partitionFetcher.getSchemaNodeManagementPartition(patternTree);

    if (schemaNodeManagementPartition == null) {
      return analysis;
    }
    if (!schemaNodeManagementPartition.getMatchedNode().isEmpty()
        && schemaNodeManagementPartition.getSchemaPartition().getSchemaPartitionMap().size() == 0) {
      analysis.setFinishQueryAfterAnalyze(true);
    }
    analysis.setMatchedNodes(schemaNodeManagementPartition.getMatchedNode());
    analysis.setSchemaPartitionInfo(schemaNodeManagementPartition.getSchemaPartition());
    analysis.setRespDatasetHeader(header);
    return analysis;
  }

  @Override
  public Analysis visitDeleteData(
      DeleteDataStatement deleteDataStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);
    Analysis analysis = new Analysis();
    analysis.setStatement(deleteDataStatement);

    PathPatternTree patternTree = new PathPatternTree();
    for (PartialPath pathPattern : deleteDataStatement.getPathList()) {
      patternTree.appendPathPattern(pathPattern);
    }

    ISchemaTree schemaTree = schemaFetcher.fetchSchema(patternTree);
    analysis.setSchemaTree(schemaTree);

    Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap = new HashMap<>();

    schemaTree
        .getMatchedDevices(new PartialPath(ALL_RESULT_NODES))
        .forEach(
            deviceSchemaInfo -> {
              PartialPath devicePath = deviceSchemaInfo.getDevicePath();
              DataPartitionQueryParam queryParam = new DataPartitionQueryParam();
              queryParam.setDevicePath(devicePath.getFullPath());
              sgNameToQueryParamsMap
                  .computeIfAbsent(
                      schemaTree.getBelongedStorageGroup(devicePath), key -> new ArrayList<>())
                  .add(queryParam);
            });

    DataPartition dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
    analysis.setDataPartitionInfo(dataPartition);

    if (dataPartition.isEmpty()) {
      analysis.setFinishQueryAfterAnalyze(true);
    }

    return analysis;
  }

  @Override
  public Analysis visitCreateSchemaTemplate(
      CreateSchemaTemplateStatement createTemplateStatement, MPPQueryContext context) {

    context.setQueryType(QueryType.WRITE);
    List<List<String>> measurementsList = createTemplateStatement.getMeasurements();
    for (List measurements : measurementsList) {
      Set<String> measurementsSet = new HashSet<>(measurements);
      if (measurementsSet.size() < measurements.size()) {
        throw new SemanticException(
            "Measurement under an aligned device is not allowed to have the same measurement name");
      }
    }
    Analysis analysis = new Analysis();
    analysis.setStatement(createTemplateStatement);
    return analysis;
  }

  @Override
  public Analysis visitShowNodesInSchemaTemplate(
      ShowNodesInSchemaTemplateStatement showNodesInSchemaTemplateStatement,
      MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(showNodesInSchemaTemplateStatement);
    analysis.setRespDatasetHeader(HeaderConstant.showNodesInSchemaTemplate);
    return analysis;
  }

  @Override
  public Analysis visitShowSchemaTemplate(
      ShowSchemaTemplateStatement showSchemaTemplateStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(showSchemaTemplateStatement);
    analysis.setRespDatasetHeader(HeaderConstant.showSchemaTemplate);
    return analysis;
  }

  private GroupByFilter initGroupByFilter(GroupByTimeComponent groupByTimeComponent) {
    if (groupByTimeComponent.isIntervalByMonth() || groupByTimeComponent.isSlidingStepByMonth()) {
      return new GroupByMonthFilter(
          groupByTimeComponent.getInterval(),
          groupByTimeComponent.getSlidingStep(),
          groupByTimeComponent.getStartTime(),
          groupByTimeComponent.getEndTime(),
          groupByTimeComponent.isSlidingStepByMonth(),
          groupByTimeComponent.isIntervalByMonth(),
          TimeZone.getTimeZone("+00:00"));
    } else {
      return new GroupByFilter(
          groupByTimeComponent.getInterval(),
          groupByTimeComponent.getSlidingStep(),
          groupByTimeComponent.getStartTime(),
          groupByTimeComponent.getEndTime());
    }
  }

  @Override
  public Analysis visitSetSchemaTemplate(
      SetSchemaTemplateStatement setSchemaTemplateStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);
    Analysis analysis = new Analysis();
    analysis.setStatement(setSchemaTemplateStatement);
    return analysis;
  }

  @Override
  public Analysis visitShowPathSetTemplate(
      ShowPathSetTemplateStatement showPathSetTemplateStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(showPathSetTemplateStatement);
    analysis.setRespDatasetHeader(HeaderConstant.showPathSetTemplate);
    return analysis;
  }

  @Override
  public Analysis visitActivateTemplate(
      ActivateTemplateStatement activateTemplateStatement, MPPQueryContext context) {
    context.setQueryType(QueryType.WRITE);
    Analysis analysis = new Analysis();
    analysis.setStatement(activateTemplateStatement);

    PartialPath activatePath = activateTemplateStatement.getPath();

    Pair<Template, PartialPath> templateSetInfo = schemaFetcher.checkTemplateSetInfo(activatePath);
    if (templateSetInfo == null) {
      throw new StatementAnalyzeException(
          new MetadataException(
              String.format(
                  "Path [%s] has not been set any template.", activatePath.getFullPath())));
    }
    analysis.setTemplateSetInfo(
        new Pair<>(templateSetInfo.left, Collections.singletonList(templateSetInfo.right)));

    PathPatternTree patternTree = new PathPatternTree();
    patternTree.appendPathPattern(activatePath.concatNode(ONE_LEVEL_PATH_WILDCARD));
    SchemaPartition partition = partitionFetcher.getOrCreateSchemaPartition(patternTree);

    analysis.setSchemaPartitionInfo(partition);

    return analysis;
  }

  @Override
  public Analysis visitShowPathsUsingTemplate(
      ShowPathsUsingTemplateStatement showPathsUsingTemplateStatement, MPPQueryContext context) {
    Analysis analysis = new Analysis();
    analysis.setStatement(showPathsUsingTemplateStatement);
    analysis.setRespDatasetHeader(HeaderConstant.showPathsUsingTemplate);

    Pair<Template, List<PartialPath>> templateSetInfo =
        schemaFetcher.getAllPathsSetTemplate(showPathsUsingTemplateStatement.getTemplateName());
    analysis.setTemplateSetInfo(templateSetInfo);
    if (templateSetInfo == null
        || templateSetInfo.right == null
        || templateSetInfo.right.isEmpty()) {
      analysis.setFinishQueryAfterAnalyze(true);
      return analysis;
    }

    PathPatternTree patternTree = new PathPatternTree();
    templateSetInfo.right.forEach(
        path -> {
          patternTree.appendPathPattern(path);
          patternTree.appendPathPattern(path.concatNode(MULTI_LEVEL_PATH_WILDCARD));
        });

    SchemaPartition partition = partitionFetcher.getOrCreateSchemaPartition(patternTree);
    analysis.setSchemaPartitionInfo(partition);
    if (partition.isEmpty()) {
      analysis.setFinishQueryAfterAnalyze(true);
      return analysis;
    }

    return analysis;
  }
}