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

import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.consensus.SchemaRegionId;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.IConsensus;
import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
import org.apache.iotdb.db.consensus.SchemaRegionConsensusImpl;
import org.apache.iotdb.db.exception.metadata.MeasurementAlreadyExistException;
import org.apache.iotdb.db.exception.metadata.SchemaQuotaExceededException;
import org.apache.iotdb.db.protocol.thrift.impl.DataNodeRegionManager;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.ActivateTemplateNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.AlterTimeSeriesNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.BatchActivateTemplateNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateAlignedTimeSeriesNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateMultiTimeSeriesNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.InternalBatchActivateTemplateNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.InternalCreateMultiTimeSeriesNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.InternalCreateTimeSeriesNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.MeasurementGroup;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.view.CreateLogicalViewNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedDeleteDataNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedInsertNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedWritePlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.schemaengine.SchemaEngine;
import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager;
import org.apache.iotdb.db.schemaengine.template.Template;
import org.apache.iotdb.db.trigger.executor.TriggerFireResult;
import org.apache.iotdb.db.trigger.executor.TriggerFireVisitor;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.trigger.api.enums.TriggerEvent;
import org.apache.iotdb.tsfile.utils.Pair;

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

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class RegionWriteExecutor {

  private static final Logger LOGGER = LoggerFactory.getLogger(RegionWriteExecutor.class);

  private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS =
      PerformanceOverviewMetrics.getInstance();

  private static final String METADATA_ERROR_MSG = "Metadata error: ";

  private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();

  private final IConsensus dataRegionConsensus;

  private final IConsensus schemaRegionConsensus;

  private final DataNodeRegionManager regionManager;

  private final SchemaEngine schemaEngine;

  private final ClusterTemplateManager clusterTemplateManager;

  private final TriggerFireVisitor triggerFireVisitor;

  private final WritePlanNodeExecutionVisitor executionVisitor;

  private final PipeEnrichedWriteSchemaNodeExecutionVisitor pipeExecutionVisitor;

  public RegionWriteExecutor() {
    dataRegionConsensus = DataRegionConsensusImpl.getInstance();
    schemaRegionConsensus = SchemaRegionConsensusImpl.getInstance();
    regionManager = DataNodeRegionManager.getInstance();
    schemaEngine = SchemaEngine.getInstance();
    clusterTemplateManager = ClusterTemplateManager.getInstance();
    triggerFireVisitor = new TriggerFireVisitor();
    executionVisitor = new WritePlanNodeExecutionVisitor();
    pipeExecutionVisitor = new PipeEnrichedWriteSchemaNodeExecutionVisitor(executionVisitor);
  }

  @TestOnly
  public RegionWriteExecutor(
      IConsensus dataRegionConsensus,
      IConsensus schemaRegionConsensus,
      DataNodeRegionManager regionManager,
      SchemaEngine schemaEngine,
      ClusterTemplateManager clusterTemplateManager,
      TriggerFireVisitor triggerFireVisitor) {
    this.dataRegionConsensus = dataRegionConsensus;
    this.schemaRegionConsensus = schemaRegionConsensus;
    this.regionManager = regionManager;
    this.schemaEngine = schemaEngine;
    this.clusterTemplateManager = clusterTemplateManager;
    this.triggerFireVisitor = triggerFireVisitor;
    executionVisitor = new WritePlanNodeExecutionVisitor();
    pipeExecutionVisitor = new PipeEnrichedWriteSchemaNodeExecutionVisitor(executionVisitor);
  }

  @SuppressWarnings("squid:S1181")
  public RegionExecutionResult execute(ConsensusGroupId groupId, PlanNode planNode) {
    try {
      WritePlanNodeExecutionContext context =
          new WritePlanNodeExecutionContext(groupId, regionManager.getRegionLock(groupId));
      return planNode.accept(executionVisitor, context);
    } catch (Throwable e) {
      LOGGER.warn(e.getMessage(), e);
      RegionExecutionResult result = new RegionExecutionResult();
      result.setAccepted(false);
      result.setMessage(e.getMessage());
      result.setStatus(RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage()));
      return result;
    }
  }

  private class WritePlanNodeExecutionVisitor
      extends PlanVisitor<RegionExecutionResult, WritePlanNodeExecutionContext> {

    @Override
    public RegionExecutionResult visitPlan(PlanNode node, WritePlanNodeExecutionContext context) {
      RegionExecutionResult response = new RegionExecutionResult();

      if (CommonDescriptor.getInstance().getConfig().isReadOnly()) {
        response.setAccepted(false);
        response.setMessage("Fail to do non-query operations because system is read-only.");
        response.setStatus(
            RpcUtils.getStatus(
                TSStatusCode.SYSTEM_READ_ONLY,
                "Fail to do non-query operations because system is read-only."));
        return response;
      }

      try {
        TSStatus status = executePlanNodeInConsensusLayer(context.getRegionId(), node);
        response.setAccepted(TSStatusCode.SUCCESS_STATUS.getStatusCode() == status.getCode());
        response.setMessage(status.getMessage());
        response.setStatus(status);
      } catch (ConsensusException e) {
        LOGGER.warn("Failed in the write API executing the consensus layer due to: ", e);
        response.setAccepted(false);
        response.setMessage(e.toString());
        response.setStatus(
            RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage()));
      }
      return response;
    }

    private TSStatus executePlanNodeInConsensusLayer(ConsensusGroupId groupId, PlanNode planNode)
        throws ConsensusException {
      if (groupId instanceof DataRegionId) {
        return dataRegionConsensus.write(groupId, planNode);
      } else {
        return schemaRegionConsensus.write(groupId, planNode);
      }
    }

    @Override
    public RegionExecutionResult visitInsertRow(
        InsertRowNode node, WritePlanNodeExecutionContext context) {
      return executeDataInsert(node, context);
    }

    @Override
    public RegionExecutionResult visitInsertTablet(
        InsertTabletNode node, WritePlanNodeExecutionContext context) {
      return executeDataInsert(node, context);
    }

    @Override
    public RegionExecutionResult visitInsertRows(
        InsertRowsNode node, WritePlanNodeExecutionContext context) {
      return executeDataInsert(node, context);
    }

    @Override
    public RegionExecutionResult visitInsertMultiTablets(
        InsertMultiTabletsNode node, WritePlanNodeExecutionContext context) {
      return executeDataInsert(node, context);
    }

    @Override
    public RegionExecutionResult visitInsertRowsOfOneDevice(
        InsertRowsOfOneDeviceNode node, WritePlanNodeExecutionContext context) {
      return executeDataInsert(node, context);
    }

    @Override
    public RegionExecutionResult visitPipeEnrichedInsertNode(
        PipeEnrichedInsertNode node, WritePlanNodeExecutionContext context) {
      return executeDataInsert(node, context);
    }

    private RegionExecutionResult executeDataInsert(
        InsertNode insertNode, WritePlanNodeExecutionContext context) {
      RegionExecutionResult response = new RegionExecutionResult();
      context.getRegionWriteValidationRWLock().readLock().lock();
      try {
        TSStatus status = fireTriggerAndInsert(context.getRegionId(), insertNode);
        response.setAccepted(TSStatusCode.SUCCESS_STATUS.getStatusCode() == status.getCode());
        response.setMessage(status.message);
        if (!response.isAccepted()) {
          response.setStatus(status);
        }
        return response;
      } catch (ConsensusException e) {
        LOGGER.warn("Failed in the write API executing the consensus layer due to: ", e);
        response.setAccepted(false);
        response.setMessage(e.toString());
        response.setStatus(RpcUtils.getStatus(TSStatusCode.WRITE_PROCESS_ERROR, e.toString()));
        return response;
      } finally {
        context.getRegionWriteValidationRWLock().readLock().unlock();
      }
    }

    private TSStatus fireTriggerAndInsert(ConsensusGroupId groupId, InsertNode insertNode)
        throws ConsensusException {
      long triggerCostTime = 0;
      TSStatus status;
      long startTime = System.nanoTime();
      // fire Trigger before the insertion
      TriggerFireResult result = triggerFireVisitor.process(insertNode, TriggerEvent.BEFORE_INSERT);
      triggerCostTime += (System.nanoTime() - startTime);
      if (result.equals(TriggerFireResult.TERMINATION)) {
        status =
            RpcUtils.getStatus(
                TSStatusCode.TRIGGER_FIRE_ERROR.getStatusCode(),
                "Failed to complete the insertion because trigger error before the insertion.");
      } else {
        long startWriteTime = System.nanoTime();
        status = dataRegionConsensus.write(groupId, insertNode);
        PERFORMANCE_OVERVIEW_METRICS.recordScheduleStorageCost(System.nanoTime() - startWriteTime);

        // fire Trigger after the insertion
        startTime = System.nanoTime();
        boolean hasFailedTriggerBeforeInsertion =
            result.equals(TriggerFireResult.FAILED_NO_TERMINATION);
        result = triggerFireVisitor.process(insertNode, TriggerEvent.AFTER_INSERT);
        if (hasFailedTriggerBeforeInsertion || !result.equals(TriggerFireResult.SUCCESS)) {
          status =
              RpcUtils.getStatus(
                  TSStatusCode.TRIGGER_FIRE_ERROR.getStatusCode(),
                  "Meet trigger error before/after the insertion, the insertion itself is completed.");
        }
        triggerCostTime += (System.nanoTime() - startTime);
      }
      PERFORMANCE_OVERVIEW_METRICS.recordScheduleTriggerCost(triggerCostTime);
      return status;
    }

    @Override
    public RegionExecutionResult visitPipeEnrichedDeleteDataNode(
        PipeEnrichedDeleteDataNode node, WritePlanNodeExecutionContext context) {
      // data deletion should block data insertion, especially when executed for deleting timeseries
      context.getRegionWriteValidationRWLock().writeLock().lock();
      try {
        return super.visitPipeEnrichedDeleteDataNode(node, context);
      } finally {
        context.getRegionWriteValidationRWLock().writeLock().unlock();
      }
    }

    @Override
    public RegionExecutionResult visitDeleteData(
        DeleteDataNode node, WritePlanNodeExecutionContext context) {
      // data deletion don't need to block data insertion, but there are some creation operation
      // require write lock on data region.
      context.getRegionWriteValidationRWLock().writeLock().lock();
      try {
        return super.visitDeleteData(node, context);
      } finally {
        context.getRegionWriteValidationRWLock().writeLock().unlock();
      }
    }

    @Override
    public RegionExecutionResult visitCreateTimeSeries(
        CreateTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return executeCreateTimeSeries(node, context, false);
    }

    private RegionExecutionResult executeCreateTimeSeries(
        CreateTimeSeriesNode node,
        WritePlanNodeExecutionContext context,
        boolean receivedFromPipe) {
      ISchemaRegion schemaRegion =
          schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
      RegionExecutionResult result =
          checkQuotaBeforeCreatingTimeSeries(schemaRegion, node.getPath().getDevicePath(), 1);
      if (result != null) {
        return result;
      }
      if (CONFIG.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS)) {
        context.getRegionWriteValidationRWLock().writeLock().lock();
        try {
          Map<Integer, MetadataException> failingMeasurementMap =
              schemaRegion.checkMeasurementExistence(
                  node.getPath().getDevicePath(),
                  Collections.singletonList(node.getPath().getMeasurement()),
                  Collections.singletonList(node.getAlias()));
          if (failingMeasurementMap.isEmpty()) {
            return receivedFromPipe
                ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
                : super.visitCreateTimeSeries(node, context);
          } else {
            MetadataException metadataException = failingMeasurementMap.get(0);
            LOGGER.warn(METADATA_ERROR_MSG, metadataException);
            result = new RegionExecutionResult();
            result.setAccepted(false);
            result.setMessage(metadataException.getMessage());
            result.setStatus(
                RpcUtils.getStatus(
                    metadataException.getErrorCode(), metadataException.getMessage()));
            return result;
          }
        } finally {
          context.getRegionWriteValidationRWLock().writeLock().unlock();
        }
      } else {
        return receivedFromPipe
            ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
            : super.visitCreateTimeSeries(node, context);
      }
    }

    @Override
    public RegionExecutionResult visitCreateAlignedTimeSeries(
        CreateAlignedTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return executeCreateAlignedTimeSeries(node, context, false);
    }

    private RegionExecutionResult executeCreateAlignedTimeSeries(
        CreateAlignedTimeSeriesNode node,
        WritePlanNodeExecutionContext context,
        boolean receivedFromPipe) {
      ISchemaRegion schemaRegion =
          schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
      RegionExecutionResult result =
          checkQuotaBeforeCreatingTimeSeries(
              schemaRegion, node.getDevicePath(), node.getMeasurements().size());
      if (result != null) {
        return result;
      }
      if (CONFIG.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS)) {
        context.getRegionWriteValidationRWLock().writeLock().lock();
        try {
          Map<Integer, MetadataException> failingMeasurementMap =
              schemaRegion.checkMeasurementExistence(
                  node.getDevicePath(), node.getMeasurements(), node.getAliasList());
          if (failingMeasurementMap.isEmpty()) {
            return receivedFromPipe
                ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
                : super.visitCreateAlignedTimeSeries(node, context);
          } else {
            MetadataException metadataException = failingMeasurementMap.values().iterator().next();
            LOGGER.warn(METADATA_ERROR_MSG, metadataException);
            result = new RegionExecutionResult();
            result.setAccepted(false);
            result.setMessage(metadataException.getMessage());
            result.setStatus(
                RpcUtils.getStatus(
                    metadataException.getErrorCode(), metadataException.getMessage()));
            return result;
          }
        } finally {
          context.getRegionWriteValidationRWLock().writeLock().unlock();
        }
      } else {
        return receivedFromPipe
            ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
            : super.visitCreateAlignedTimeSeries(node, context);
      }
    }

    @Override
    public RegionExecutionResult visitCreateMultiTimeSeries(
        CreateMultiTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return executeCreateMultiTimeSeries(node, context, false);
    }

    private RegionExecutionResult executeCreateMultiTimeSeries(
        CreateMultiTimeSeriesNode node,
        WritePlanNodeExecutionContext context,
        boolean receivedFromPipe) {
      ISchemaRegion schemaRegion =
          schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
      RegionExecutionResult result;
      for (Map.Entry<PartialPath, MeasurementGroup> entry :
          node.getMeasurementGroupMap().entrySet()) {
        result =
            checkQuotaBeforeCreatingTimeSeries(
                schemaRegion, entry.getKey(), entry.getValue().getMeasurements().size());
        if (result != null) {
          return result;
        }
      }
      if (CONFIG.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS)) {
        context.getRegionWriteValidationRWLock().writeLock().lock();
        try {
          List<TSStatus> failingStatus = new ArrayList<>();
          Map<PartialPath, MeasurementGroup> measurementGroupMap = node.getMeasurementGroupMap();
          List<PartialPath> emptyDeviceList = new ArrayList<>();

          checkMeasurementExistence(
              measurementGroupMap, schemaRegion, failingStatus, emptyDeviceList);

          for (PartialPath emptyDevice : emptyDeviceList) {
            measurementGroupMap.remove(emptyDevice);
          }

          RegionExecutionResult failingResult =
              registerTimeSeries(
                  measurementGroupMap, node, context, failingStatus, receivedFromPipe);

          if (failingResult != null) {
            return failingResult;
          }

          TSStatus status = RpcUtils.getStatus(failingStatus);
          failingResult = new RegionExecutionResult();
          failingResult.setAccepted(false);
          failingResult.setMessage(status.getMessage());
          failingResult.setStatus(status);
          return failingResult;
        } finally {
          context.getRegionWriteValidationRWLock().writeLock().unlock();
        }
      } else {
        return receivedFromPipe
            ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
            : super.visitCreateMultiTimeSeries(node, context);
      }
    }

    private void checkMeasurementExistence(
        Map<PartialPath, MeasurementGroup> measurementGroupMap,
        ISchemaRegion schemaRegion,
        List<TSStatus> failingStatus,
        List<PartialPath> emptyDeviceList) {
      for (Map.Entry<PartialPath, MeasurementGroup> entry : measurementGroupMap.entrySet()) {
        Map<Integer, MetadataException> failingMeasurementMap =
            schemaRegion.checkMeasurementExistence(
                entry.getKey(),
                entry.getValue().getMeasurements(),
                entry.getValue().getAliasList());
        if (failingMeasurementMap.isEmpty()) {
          continue;
        }

        for (Map.Entry<Integer, MetadataException> failingMeasurement :
            failingMeasurementMap.entrySet()) {
          LOGGER.warn(METADATA_ERROR_MSG, failingMeasurement.getValue());
          failingStatus.add(
              RpcUtils.getStatus(
                  failingMeasurement.getValue().getErrorCode(),
                  failingMeasurement.getValue().getMessage()));
        }
        entry.getValue().removeMeasurements(failingMeasurementMap.keySet());

        if (entry.getValue().isEmpty()) {
          emptyDeviceList.add(entry.getKey());
        }
      }
    }

    private RegionExecutionResult registerTimeSeries(
        Map<PartialPath, MeasurementGroup> measurementGroupMap,
        CreateMultiTimeSeriesNode node,
        WritePlanNodeExecutionContext context,
        List<TSStatus> failingStatus,
        boolean receivedFromPipe) {
      if (!measurementGroupMap.isEmpty()) {
        // try registering the rest timeseries
        RegionExecutionResult executionResult =
            receivedFromPipe
                ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
                : super.visitCreateMultiTimeSeries(node, context);
        if (failingStatus.isEmpty()) {
          return executionResult;
        }

        TSStatus executionStatus = executionResult.getStatus();
        if (executionStatus.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
          failingStatus.addAll(executionStatus.getSubStatus());
        } else if (executionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
          failingStatus.add(executionStatus);
        }
      }
      return null;
    }

    @Override
    public RegionExecutionResult visitInternalCreateTimeSeries(
        InternalCreateTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return executeInternalCreateTimeSeries(node, context, false);
    }

    private RegionExecutionResult executeInternalCreateTimeSeries(
        InternalCreateTimeSeriesNode node,
        WritePlanNodeExecutionContext context,
        boolean receivedFromPipe) {
      ISchemaRegion schemaRegion =
          schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
      RegionExecutionResult result =
          checkQuotaBeforeCreatingTimeSeries(
              schemaRegion, node.getDevicePath(), node.getMeasurementGroup().size());
      if (result != null) {
        return result;
      }
      if (CONFIG.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS)) {
        context.getRegionWriteValidationRWLock().writeLock().lock();
        try {
          List<TSStatus> failingStatus = new ArrayList<>();
          List<TSStatus> alreadyExistingStatus = new ArrayList<>();
          MeasurementGroup measurementGroup = node.getMeasurementGroup();
          Map<Integer, MetadataException> failingMeasurementMap =
              schemaRegion.checkMeasurementExistence(
                  node.getDevicePath(),
                  measurementGroup.getMeasurements(),
                  measurementGroup.getAliasList());
          MetadataException metadataException;
          // filter failed measurement and keep the rest for execution
          for (Map.Entry<Integer, MetadataException> failingMeasurement :
              failingMeasurementMap.entrySet()) {
            metadataException = failingMeasurement.getValue();
            if (metadataException.getErrorCode()
                == TSStatusCode.TIMESERIES_ALREADY_EXIST.getStatusCode()) {
              // There's no need to internal create timeseries.
              alreadyExistingStatus.add(
                  RpcUtils.getStatus(
                      metadataException.getErrorCode(),
                      MeasurementPath.transformDataToString(
                          ((MeasurementAlreadyExistException) metadataException)
                              .getMeasurementPath())));
            } else {
              int errorCode = metadataException.getErrorCode();
              if (errorCode != TSStatusCode.PATH_ALREADY_EXIST.getStatusCode()
                  || errorCode != TSStatusCode.ALIAS_ALREADY_EXIST.getStatusCode()) {
                LOGGER.warn(METADATA_ERROR_MSG, metadataException);
              }
              failingStatus.add(
                  RpcUtils.getStatus(
                      metadataException.getErrorCode(), metadataException.getMessage()));
            }
          }
          measurementGroup.removeMeasurements(failingMeasurementMap.keySet());

          return processExecutionResultOfInternalCreateSchema(
              receivedFromPipe
                  ? super.visitPipeEnrichedWritePlanNode(
                      new PipeEnrichedWritePlanNode(node), context)
                  : super.visitInternalCreateTimeSeries(node, context),
              failingStatus,
              alreadyExistingStatus);
        } finally {
          context.getRegionWriteValidationRWLock().writeLock().unlock();
        }
      } else {
        return receivedFromPipe
            ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
            : super.visitInternalCreateTimeSeries(node, context);
      }
    }

    @Override
    public RegionExecutionResult visitInternalCreateMultiTimeSeries(
        InternalCreateMultiTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return executeInternalCreateMultiTimeSeries(node, context, false);
    }

    private RegionExecutionResult executeInternalCreateMultiTimeSeries(
        InternalCreateMultiTimeSeriesNode node,
        WritePlanNodeExecutionContext context,
        boolean receivedFromPipe) {
      ISchemaRegion schemaRegion =
          schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
      RegionExecutionResult result;
      for (Map.Entry<PartialPath, Pair<Boolean, MeasurementGroup>> deviceEntry :
          node.getDeviceMap().entrySet()) {
        result =
            checkQuotaBeforeCreatingTimeSeries(
                schemaRegion, deviceEntry.getKey(), deviceEntry.getValue().getRight().size());
        if (result != null) {
          return result;
        }
      }
      if (CONFIG.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS)) {
        context.getRegionWriteValidationRWLock().writeLock().lock();
        try {
          List<TSStatus> failingStatus = new ArrayList<>();
          List<TSStatus> alreadyExistingStatus = new ArrayList<>();

          MeasurementGroup measurementGroup;
          Map<Integer, MetadataException> failingMeasurementMap;
          MetadataException metadataException;
          for (Map.Entry<PartialPath, Pair<Boolean, MeasurementGroup>> deviceEntry :
              node.getDeviceMap().entrySet()) {
            measurementGroup = deviceEntry.getValue().right;
            failingMeasurementMap =
                schemaRegion.checkMeasurementExistence(
                    deviceEntry.getKey(),
                    measurementGroup.getMeasurements(),
                    measurementGroup.getAliasList());
            // filter failed measurement and keep the rest for execution
            for (Map.Entry<Integer, MetadataException> failingMeasurement :
                failingMeasurementMap.entrySet()) {
              metadataException = failingMeasurement.getValue();
              if (metadataException.getErrorCode()
                  == TSStatusCode.TIMESERIES_ALREADY_EXIST.getStatusCode()) {
                // There's no need to internal create timeseries.
                alreadyExistingStatus.add(
                    RpcUtils.getStatus(
                        metadataException.getErrorCode(),
                        MeasurementPath.transformDataToString(
                            ((MeasurementAlreadyExistException) metadataException)
                                .getMeasurementPath())));
              } else {
                LOGGER.warn(METADATA_ERROR_MSG, metadataException);
                failingStatus.add(
                    RpcUtils.getStatus(
                        metadataException.getErrorCode(), metadataException.getMessage()));
              }
            }
            measurementGroup.removeMeasurements(failingMeasurementMap.keySet());
          }

          return processExecutionResultOfInternalCreateSchema(
              receivedFromPipe
                  ? super.visitPipeEnrichedWritePlanNode(
                      new PipeEnrichedWritePlanNode(node), context)
                  : super.visitInternalCreateMultiTimeSeries(node, context),
              failingStatus,
              alreadyExistingStatus);
        } finally {
          context.getRegionWriteValidationRWLock().writeLock().unlock();
        }
      } else {
        return receivedFromPipe
            ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
            : super.visitInternalCreateMultiTimeSeries(node, context);
      }
    }

    /**
     * Check the quota before creating time series.
     *
     * @return null if the quota is not exceeded, otherwise return the execution result.
     */
    private RegionExecutionResult checkQuotaBeforeCreatingTimeSeries(
        ISchemaRegion schemaRegion, PartialPath path, int size) {
      try {
        schemaRegion.checkSchemaQuota(path, size);
      } catch (SchemaQuotaExceededException e) {
        RegionExecutionResult result = new RegionExecutionResult();
        result.setAccepted(false);
        result.setMessage(e.getMessage());
        result.setStatus(RpcUtils.getStatus(e.getErrorCode(), e.getMessage()));
        return result;
      }
      return null;
    }

    private RegionExecutionResult processExecutionResultOfInternalCreateSchema(
        RegionExecutionResult executionResult,
        List<TSStatus> failingStatus,
        List<TSStatus> alreadyExistingStatus) {
      TSStatus executionStatus = executionResult.getStatus();

      separateMeasurementAlreadyExistException(
          failingStatus, executionStatus, alreadyExistingStatus);

      RegionExecutionResult result = new RegionExecutionResult();
      TSStatus status;
      if (failingStatus.isEmpty() && alreadyExistingStatus.isEmpty()) {
        status = RpcUtils.SUCCESS_STATUS;
        result.setAccepted(true);
      } else if (failingStatus.isEmpty()) {
        status = RpcUtils.getStatus(alreadyExistingStatus);
        result.setAccepted(true);
      } else {
        status = RpcUtils.getStatus(failingStatus);
        result.setAccepted(false);
      }

      result.setMessage(status.getMessage());
      result.setStatus(status);
      return result;
    }

    private void separateMeasurementAlreadyExistException(
        List<TSStatus> failingStatus,
        TSStatus executionStatus,
        List<TSStatus> alreadyExistingStatus) {
      // separate the measurement_already_exist exception and other exceptions process,
      // measurement_already_exist exception is acceptable due to concurrent timeseries creation
      if (failingStatus.isEmpty()) {
        if (executionStatus.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
          if (executionStatus.getSubStatus().get(0).getCode()
              == TSStatusCode.TIMESERIES_ALREADY_EXIST.getStatusCode()) {
            // there's only measurement_already_exist exception
            alreadyExistingStatus.addAll(executionStatus.getSubStatus());
          } else {
            failingStatus.addAll(executionStatus.getSubStatus());
          }
        } else if (executionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
          failingStatus.add(executionStatus);
        }
      } else {
        if (executionStatus.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
          if (executionStatus.getSubStatus().get(0).getCode()
              != TSStatusCode.TIMESERIES_ALREADY_EXIST.getStatusCode()) {
            failingStatus.addAll(executionStatus.getSubStatus());
          }
        } else if (executionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
          failingStatus.add(executionStatus);
        }
      }
    }

    @Override
    public RegionExecutionResult visitAlterTimeSeries(
        AlterTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return executeAlterTimeSeries(node, context, false);
    }

    private RegionExecutionResult executeAlterTimeSeries(
        AlterTimeSeriesNode node, WritePlanNodeExecutionContext context, boolean receivedFromPipe) {
      ISchemaRegion schemaRegion =
          schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
      try {
        MeasurementPath measurementPath = schemaRegion.fetchMeasurementPath(node.getPath());
        if (node.isAlterView() && !measurementPath.getMeasurementSchema().isLogicalView()) {
          throw new MetadataException(
              String.format("%s is not view.", measurementPath.getFullPath()));
        }
        return receivedFromPipe
            ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
            : super.visitAlterTimeSeries(node, context);
      } catch (MetadataException e) {
        RegionExecutionResult result = new RegionExecutionResult();
        result.setAccepted(true);
        result.setMessage(e.getMessage());
        result.setStatus(RpcUtils.getStatus(e.getErrorCode(), e.getMessage()));
        return result;
      }
    }

    @Override
    public RegionExecutionResult visitActivateTemplate(
        ActivateTemplateNode node, WritePlanNodeExecutionContext context) {
      return executeActivateTemplate(node, context, false);
    }

    private RegionExecutionResult executeActivateTemplate(
        ActivateTemplateNode node,
        WritePlanNodeExecutionContext context,
        boolean receivedFromPipe) {
      // activate template operation shall be blocked by unset template check
      context.getRegionWriteValidationRWLock().readLock().lock();
      try {
        Pair<Template, PartialPath> templateSetInfo =
            clusterTemplateManager.checkTemplateSetInfo(node.getActivatePath());
        if (templateSetInfo == null) {
          // The activation has already been validated during analyzing.
          // That means the template is being unset during the activation plan transport.
          RegionExecutionResult result = new RegionExecutionResult();
          result.setAccepted(false);
          String message =
              String.format(
                  "Template is being unsetting from path %s. Please try activating later.",
                  node.getPathSetTemplate());
          result.setMessage(message);
          result.setStatus(RpcUtils.getStatus(TSStatusCode.METADATA_ERROR, message));
          return result;
        }
        ISchemaRegion schemaRegion =
            schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
        RegionExecutionResult result =
            checkQuotaBeforeCreatingTimeSeries(
                schemaRegion, node.getActivatePath(), templateSetInfo.left.getMeasurementNumber());
        if (result == null) {
          return receivedFromPipe
              ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
              : super.visitActivateTemplate(node, context);
        } else {
          return result;
        }
      } finally {
        context.getRegionWriteValidationRWLock().readLock().unlock();
      }
    }

    @Override
    public RegionExecutionResult visitBatchActivateTemplate(
        BatchActivateTemplateNode node, WritePlanNodeExecutionContext context) {
      return executeBatchActivateTemplate(node, context, false);
    }

    private RegionExecutionResult executeBatchActivateTemplate(
        BatchActivateTemplateNode node,
        WritePlanNodeExecutionContext context,
        boolean receivedFromPipe) {
      // activate template operation shall be blocked by unset template check
      context.getRegionWriteValidationRWLock().readLock().lock();
      try {
        ISchemaRegion schemaRegion =
            schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
        for (PartialPath devicePath : node.getTemplateActivationMap().keySet()) {
          Pair<Template, PartialPath> templateSetInfo =
              clusterTemplateManager.checkTemplateSetInfo(devicePath);
          if (templateSetInfo == null) {
            // The activation has already been validated during analyzing.
            // That means the template is being unset during the activation plan transport.
            RegionExecutionResult result = new RegionExecutionResult();
            result.setAccepted(false);
            String message =
                String.format(
                    "Template is being unsetting from path %s. Please try activating later.",
                    node.getPathSetTemplate(devicePath));
            result.setMessage(message);
            result.setStatus(RpcUtils.getStatus(TSStatusCode.METADATA_ERROR, message));
            return result;
          }
          RegionExecutionResult result =
              checkQuotaBeforeCreatingTimeSeries(
                  schemaRegion, devicePath, templateSetInfo.left.getMeasurementNumber());
          if (result != null) {
            return result;
          }
        }

        return receivedFromPipe
            ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
            : super.visitBatchActivateTemplate(node, context);
      } finally {
        context.getRegionWriteValidationRWLock().readLock().unlock();
      }
    }

    @Override
    public RegionExecutionResult visitInternalBatchActivateTemplate(
        InternalBatchActivateTemplateNode node, WritePlanNodeExecutionContext context) {
      return executeInternalBatchActivateTemplate(node, context, false);
    }

    private RegionExecutionResult executeInternalBatchActivateTemplate(
        InternalBatchActivateTemplateNode node,
        WritePlanNodeExecutionContext context,
        boolean receivedFromPipe) {
      // activate template operation shall be blocked by unset template check
      context.getRegionWriteValidationRWLock().readLock().lock();
      try {
        ISchemaRegion schemaRegion =
            schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
        for (Map.Entry<PartialPath, Pair<Integer, Integer>> entry :
            node.getTemplateActivationMap().entrySet()) {
          Pair<Template, PartialPath> templateSetInfo =
              clusterTemplateManager.checkTemplateSetInfo(entry.getKey());
          if (templateSetInfo == null) {
            // The activation has already been validated during analyzing.
            // That means the template is being unset during the activation plan transport.
            RegionExecutionResult result = new RegionExecutionResult();
            result.setAccepted(false);
            String message =
                String.format(
                    "Template is being unsetting from prefix path of %s. Please try activating later.",
                    new PartialPath(
                            Arrays.copyOf(entry.getKey().getNodes(), entry.getValue().right + 1))
                        .getFullPath());
            result.setMessage(message);
            result.setStatus(RpcUtils.getStatus(TSStatusCode.METADATA_ERROR, message));
            return result;
          }
          RegionExecutionResult result =
              checkQuotaBeforeCreatingTimeSeries(
                  schemaRegion, entry.getKey(), templateSetInfo.left.getMeasurementNumber());
          if (result != null) {
            return result;
          }
        }

        return receivedFromPipe
            ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
            : super.visitInternalBatchActivateTemplate(node, context);
      } finally {
        context.getRegionWriteValidationRWLock().readLock().unlock();
      }
    }

    @Override
    public RegionExecutionResult visitCreateLogicalView(
        CreateLogicalViewNode node, WritePlanNodeExecutionContext context) {
      return executeCreateLogicalView(node, context, false);
    }

    private RegionExecutionResult executeCreateLogicalView(
        CreateLogicalViewNode node,
        WritePlanNodeExecutionContext context,
        boolean receivedFromPipe) {
      ISchemaRegion schemaRegion =
          schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
      if (CONFIG.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS)) {
        context.getRegionWriteValidationRWLock().writeLock().lock();
        try {
          // step 1. make sure all target paths do NOT exist.
          List<PartialPath> targetPaths = node.getViewPathList();
          List<MetadataException> failingMetadataException = new ArrayList<>();
          for (PartialPath thisPath : targetPaths) {
            // no alias list for a view, so the third parameter is null
            Map<Integer, MetadataException> failingMeasurementMap =
                schemaRegion.checkMeasurementExistence(
                    thisPath.getDevicePath(),
                    Collections.singletonList(thisPath.getMeasurement()),
                    null);
            // merge all exceptions into one map
            for (Map.Entry<Integer, MetadataException> entry : failingMeasurementMap.entrySet()) {
              failingMetadataException.add(entry.getValue());
            }
          }
          // if there are some exceptions, handle each exception and return first of them.
          if (!failingMetadataException.isEmpty()) {
            MetadataException metadataException = failingMetadataException.get(0);
            LOGGER.warn(METADATA_ERROR_MSG, metadataException);
            RegionExecutionResult result = new RegionExecutionResult();
            result.setAccepted(false);
            result.setMessage(metadataException.getMessage());
            result.setStatus(
                RpcUtils.getStatus(
                    metadataException.getErrorCode(), metadataException.getMessage()));
            return result;
          }
          // step 2. make sure all source paths exist.
          return receivedFromPipe
              ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
              : super.visitCreateLogicalView(node, context);
        } finally {
          context.getRegionWriteValidationRWLock().writeLock().unlock();
        }
      } else {
        return receivedFromPipe
            ? super.visitPipeEnrichedWritePlanNode(new PipeEnrichedWritePlanNode(node), context)
            : super.visitCreateLogicalView(node, context);
      }
      // end of visitCreateLogicalView
    }

    @Override
    public RegionExecutionResult visitPipeEnrichedWritePlanNode(
        PipeEnrichedWritePlanNode node, WritePlanNodeExecutionContext context) {
      return node.getWritePlanNode().accept(pipeExecutionVisitor, context);
    }
  }

  private class PipeEnrichedWriteSchemaNodeExecutionVisitor
      extends PlanVisitor<RegionExecutionResult, WritePlanNodeExecutionContext> {

    WritePlanNodeExecutionVisitor visitor;

    private PipeEnrichedWriteSchemaNodeExecutionVisitor(WritePlanNodeExecutionVisitor visitor) {
      this.visitor = visitor;
    }

    @Override
    public RegionExecutionResult visitPlan(PlanNode node, WritePlanNodeExecutionContext context) {
      throw new UnsupportedOperationException(
          "PipeEnrichedWriteSchemaNodeExecutionVisitor does not support visiting general plan.");
    }

    @Override
    public RegionExecutionResult visitCreateTimeSeries(
        CreateTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return visitor.executeCreateTimeSeries(node, context, true);
    }

    @Override
    public RegionExecutionResult visitCreateAlignedTimeSeries(
        CreateAlignedTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return visitor.executeCreateAlignedTimeSeries(node, context, true);
    }

    @Override
    public RegionExecutionResult visitCreateMultiTimeSeries(
        CreateMultiTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return visitor.executeCreateMultiTimeSeries(node, context, true);
    }

    @Override
    public RegionExecutionResult visitInternalCreateTimeSeries(
        InternalCreateTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return visitor.executeInternalCreateTimeSeries(node, context, true);
    }

    @Override
    public RegionExecutionResult visitInternalCreateMultiTimeSeries(
        InternalCreateMultiTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return visitor.executeInternalCreateMultiTimeSeries(node, context, true);
    }

    @Override
    public RegionExecutionResult visitAlterTimeSeries(
        AlterTimeSeriesNode node, WritePlanNodeExecutionContext context) {
      return visitor.executeAlterTimeSeries(node, context, true);
    }

    @Override
    public RegionExecutionResult visitActivateTemplate(
        ActivateTemplateNode node, WritePlanNodeExecutionContext context) {
      return visitor.executeActivateTemplate(node, context, true);
    }

    @Override
    public RegionExecutionResult visitBatchActivateTemplate(
        BatchActivateTemplateNode node, WritePlanNodeExecutionContext context) {
      return visitor.executeBatchActivateTemplate(node, context, true);
    }

    @Override
    public RegionExecutionResult visitInternalBatchActivateTemplate(
        InternalBatchActivateTemplateNode node, WritePlanNodeExecutionContext context) {
      return visitor.executeInternalBatchActivateTemplate(node, context, true);
    }

    @Override
    public RegionExecutionResult visitCreateLogicalView(
        CreateLogicalViewNode node, WritePlanNodeExecutionContext context) {
      return visitor.executeCreateLogicalView(node, context, true);
    }
  }

  private static class WritePlanNodeExecutionContext {

    private final ConsensusGroupId regionId;

    private final ReentrantReadWriteLock regionRWLock;

    WritePlanNodeExecutionContext(ConsensusGroupId regionId, ReentrantReadWriteLock regionRWLock) {
      this.regionId = regionId;
      this.regionRWLock = regionRWLock;
    }

    public ConsensusGroupId getRegionId() {
      return regionId;
    }

    public ReentrantReadWriteLock getRegionWriteValidationRWLock() {
      return regionRWLock;
    }
  }
}
