/*
 * 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.confignode.manager.partition;

import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.cluster.RegionRoleType;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.confignode.client.DataNodeRequestType;
import org.apache.iotdb.confignode.client.async.AsyncDataNodeClientPool;
import org.apache.iotdb.confignode.client.async.handlers.AsyncClientHandler;
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.CountTimeSlotListPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.GetDataPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.GetNodePathsPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.GetOrCreateDataPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.GetOrCreateSchemaPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.GetSchemaPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.GetSeriesSlotListPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.GetTimeSlotListPlan;
import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionIdPlan;
import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionInfoListPlan;
import org.apache.iotdb.confignode.consensus.request.write.database.PreDeleteDatabasePlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateDataPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateSchemaPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.UpdateRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan;
import org.apache.iotdb.confignode.consensus.request.write.region.PollSpecificRegionMaintainTaskPlan;
import org.apache.iotdb.confignode.consensus.response.partition.CountTimeSlotListResp;
import org.apache.iotdb.confignode.consensus.response.partition.DataPartitionResp;
import org.apache.iotdb.confignode.consensus.response.partition.GetRegionIdResp;
import org.apache.iotdb.confignode.consensus.response.partition.GetSeriesSlotListResp;
import org.apache.iotdb.confignode.consensus.response.partition.GetTimeSlotListResp;
import org.apache.iotdb.confignode.consensus.response.partition.RegionInfoListResp;
import org.apache.iotdb.confignode.consensus.response.partition.SchemaNodeManagementResp;
import org.apache.iotdb.confignode.consensus.response.partition.SchemaPartitionResp;
import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.exception.NotEnoughDataNodeException;
import org.apache.iotdb.confignode.manager.IManager;
import org.apache.iotdb.confignode.manager.ProcedureManager;
import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
import org.apache.iotdb.confignode.manager.load.LoadManager;
import org.apache.iotdb.confignode.manager.node.NodeManager;
import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.persistence.partition.PartitionInfo;
import org.apache.iotdb.confignode.persistence.partition.maintainer.RegionCreateTask;
import org.apache.iotdb.confignode.persistence.partition.maintainer.RegionDeleteTask;
import org.apache.iotdb.confignode.persistence.partition.maintainer.RegionMaintainTask;
import org.apache.iotdb.confignode.persistence.partition.maintainer.RegionMaintainType;
import org.apache.iotdb.confignode.rpc.thrift.TCountTimeSlotListReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetRegionIdReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetSeriesSlotListReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetTimeSlotListReq;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
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.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/** The {@link PartitionManager} manages cluster PartitionTable read and write requests. */
public class PartitionManager {

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

  private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf();
  private static final RegionGroupExtensionPolicy SCHEMA_REGION_GROUP_EXTENSION_POLICY =
      CONF.getSchemaRegionGroupExtensionPolicy();
  private static final RegionGroupExtensionPolicy DATA_REGION_GROUP_EXTENSION_POLICY =
      CONF.getDataRegionGroupExtensionPolicy();

  private final IManager configManager;
  private final PartitionInfo partitionInfo;

  private SeriesPartitionExecutor executor;

  private static final String CONSENSUS_READ_ERROR =
      "Failed in the read API executing the consensus layer due to: ";

  private static final String CONSENSUS_WRITE_ERROR =
      "Failed in the write API executing the consensus layer due to: ";

  /** Region cleaner. */
  // Monitor for leadership change
  private final Object scheduleMonitor = new Object();
  // Try to delete Regions in every 10s
  private static final int REGION_MAINTAINER_WORK_INTERVAL = 10;
  private final ScheduledExecutorService regionMaintainer;
  private Future<?> currentRegionMaintainerFuture;

  public PartitionManager(IManager configManager, PartitionInfo partitionInfo) {
    this.configManager = configManager;
    this.partitionInfo = partitionInfo;
    this.regionMaintainer =
        IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
            ThreadName.CONFIG_NODE_REGION_MAINTAINER.getName());
    setSeriesPartitionExecutor();
  }

  /** Construct SeriesPartitionExecutor by iotdb-confignode{@literal .}properties. */
  private void setSeriesPartitionExecutor() {
    this.executor =
        SeriesPartitionExecutor.getSeriesPartitionExecutor(
            CONF.getSeriesPartitionExecutorClass(), CONF.getSeriesSlotNum());
  }

  // ======================================================
  // Consensus read/write interfaces
  // ======================================================

  /**
   * Thread-safely get SchemaPartition.
   *
   * @param req SchemaPartitionPlan with partitionSlotsMap
   * @return SchemaPartitionDataSet that contains only existing SchemaPartition
   */
  public SchemaPartitionResp getSchemaPartition(GetSchemaPartitionPlan req) {
    try {
      return (SchemaPartitionResp) getConsensusManager().read(req);
    } catch (ConsensusException e) {
      LOGGER.warn(CONSENSUS_READ_ERROR, e);
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      return new SchemaPartitionResp(res, false, Collections.emptyMap());
    }
  }

  /**
   * Thread-safely get DataPartition
   *
   * @param req DataPartitionPlan with Map<StorageGroupName, Map<SeriesPartitionSlot,
   *     TTimeSlotList>>
   * @return DataPartitionDataSet that contains only existing DataPartition
   */
  public DataPartitionResp getDataPartition(GetDataPartitionPlan req) {
    try {
      return (DataPartitionResp) getConsensusManager().read(req);
    } catch (ConsensusException e) {
      LOGGER.warn(CONSENSUS_READ_ERROR, e);
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      return new DataPartitionResp(res, false, Collections.emptyMap());
    }
  }

  /**
   * Get SchemaPartition and create a new one if it does not exist.
   *
   * @param req SchemaPartitionPlan with partitionSlotsMap
   * @return SchemaPartitionResp with DataPartition and TSStatus. SUCCESS_STATUS if all process
   *     finish. NOT_ENOUGH_DATA_NODE if the DataNodes is not enough to create new Regions.
   *     STORAGE_GROUP_NOT_EXIST if some StorageGroup don't exist.
   */
  public SchemaPartitionResp getOrCreateSchemaPartition(GetOrCreateSchemaPartitionPlan req) {
    // Check if the related Databases exist
    for (String database : req.getPartitionSlotsMap().keySet()) {
      if (!isDatabaseExist(database)) {
        return new SchemaPartitionResp(
            new TSStatus(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode())
                .setMessage(
                    String.format(
                        "Create SchemaPartition failed because the database: %s is not exists",
                        database)),
            false,
            null);
      }
    }

    // After all the SchemaPartitions are allocated,
    // all the read requests about SchemaPartitionTable are parallel.
    SchemaPartitionResp resp = getSchemaPartition(req);
    if (resp.isAllPartitionsExist()) {
      return resp;
    }

    // We serialize the creation process of SchemaPartitions to
    // ensure that each SchemaPartition is created by a unique CreateSchemaPartitionReq.
    // Because the number of SchemaPartitions per database is limited
    // by the number of SeriesPartitionSlots,
    // the number of serialized CreateSchemaPartitionReqs is acceptable.
    synchronized (this) {
      // Here we should check again if the SchemaPartition
      // has been created by other threads to improve concurrent performance
      resp = getSchemaPartition(req);
      if (resp.isAllPartitionsExist()) {
        return resp;
      }

      // Filter unassigned SchemaPartitionSlots
      Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap =
          partitionInfo.filterUnassignedSchemaPartitionSlots(req.getPartitionSlotsMap());

      // Here we ensure that each StorageGroup has at least one SchemaRegion.
      // And if some StorageGroups own too many slots, extend SchemaRegion for them.

      // Map<StorageGroup, unassigned SeriesPartitionSlot count>
      Map<String, Integer> unassignedSchemaPartitionSlotsCountMap = new ConcurrentHashMap<>();
      unassignedSchemaPartitionSlotsMap.forEach(
          (storageGroup, unassignedSchemaPartitionSlots) ->
              unassignedSchemaPartitionSlotsCountMap.put(
                  storageGroup, unassignedSchemaPartitionSlots.size()));
      TSStatus status =
          extendRegionGroupIfNecessary(
              unassignedSchemaPartitionSlotsCountMap, TConsensusGroupType.SchemaRegion);
      if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
        // Return an error code if Region extension failed
        resp.setStatus(status);
        return resp;
      }

      Map<String, SchemaPartitionTable> assignedSchemaPartition;
      try {
        assignedSchemaPartition =
            getLoadManager().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
      } catch (NoAvailableRegionGroupException e) {
        status = getConsensusManager().confirmLeader();
        if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
          // The allocation might fail due to leadership change
          resp.setStatus(status);
          return resp;
        }

        LOGGER.error("Create SchemaPartition failed because: ", e);
        resp.setStatus(
            new TSStatus(TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode())
                .setMessage(e.getMessage()));
        return resp;
      }

      // Cache allocating result only if the current ConfigNode still holds its leadership
      CreateSchemaPartitionPlan createPlan = new CreateSchemaPartitionPlan();
      createPlan.setAssignedSchemaPartition(assignedSchemaPartition);

      status = consensusWritePartitionResult(createPlan);
      if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
        // The allocation might fail due to consensus error
        resp.setStatus(status);
        return resp;
      }
    }

    resp = getSchemaPartition(req);
    if (!resp.isAllPartitionsExist()) {
      // Count the fail rate
      AtomicInteger totalSlotNum = new AtomicInteger();
      req.getPartitionSlotsMap()
          .forEach((database, partitionSlots) -> totalSlotNum.addAndGet(partitionSlots.size()));

      AtomicInteger unassignedSlotNum = new AtomicInteger();
      Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap =
          partitionInfo.filterUnassignedSchemaPartitionSlots(req.getPartitionSlotsMap());
      unassignedSchemaPartitionSlotsMap.forEach(
          (database, unassignedSchemaPartitionSlots) ->
              unassignedSlotNum.addAndGet(unassignedSchemaPartitionSlots.size()));

      String errMsg =
          String.format(
              "Lacked %d/%d SchemaPartition allocation result in the response of getOrCreateSchemaPartition method",
              unassignedSlotNum.get(), totalSlotNum.get());
      LOGGER.error(errMsg);
      resp.setStatus(
          new TSStatus(TSStatusCode.LACK_PARTITION_ALLOCATION.getStatusCode()).setMessage(errMsg));
      return resp;
    }
    return resp;
  }

  /**
   * Get DataPartition and create a new one if it does not exist.
   *
   * @param req DataPartitionPlan with Map<StorageGroupName, Map<SeriesPartitionSlot,
   *     List<TimePartitionSlot>>>
   * @return DataPartitionResp with DataPartition and TSStatus. SUCCESS_STATUS if all process
   *     finish. NOT_ENOUGH_DATA_NODE if the DataNodes is not enough to create new Regions.
   *     STORAGE_GROUP_NOT_EXIST if some StorageGroup don't exist.
   */
  public DataPartitionResp getOrCreateDataPartition(GetOrCreateDataPartitionPlan req) {
    // Check if the related Databases exist
    for (String database : req.getPartitionSlotsMap().keySet()) {
      if (!isDatabaseExist(database)) {
        return new DataPartitionResp(
            new TSStatus(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode())
                .setMessage(
                    String.format(
                        "Create DataPartition failed because the database: %s is not exists",
                        database)),
            false,
            null);
      }
    }

    // After all the DataPartitions are allocated,
    // all the read requests about DataPartitionTable are parallel.
    DataPartitionResp resp = getDataPartition(req);
    if (resp.isAllPartitionsExist()) {
      return resp;
    }

    // We serialize the creation process of DataPartitions to
    // ensure that each DataPartition is created by a unique CreateDataPartitionReq.
    // Because the number of DataPartitions per database is limited
    // by the number of SeriesPartitionSlots,
    // the number of serialized CreateDataPartitionReqs is acceptable.
    synchronized (this) {
      // Here we should check again if the DataPartition
      // has been created by other threads to improve concurrent performance
      resp = getDataPartition(req);
      if (resp.isAllPartitionsExist()) {
        return resp;
      }

      // Filter unassigned DataPartitionSlots
      Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap =
          partitionInfo.filterUnassignedDataPartitionSlots(req.getPartitionSlotsMap());

      // Here we ensure that each StorageGroup has at least one DataRegion.
      // And if some StorageGroups own too many slots, extend DataRegion for them.

      // Map<StorageGroup, unassigned SeriesPartitionSlot count>
      Map<String, Integer> unassignedDataPartitionSlotsCountMap = new ConcurrentHashMap<>();
      unassignedDataPartitionSlotsMap.forEach(
          (storageGroup, unassignedDataPartitionSlots) ->
              unassignedDataPartitionSlotsCountMap.put(
                  storageGroup, unassignedDataPartitionSlots.size()));
      TSStatus status =
          extendRegionGroupIfNecessary(
              unassignedDataPartitionSlotsCountMap, TConsensusGroupType.DataRegion);
      if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
        // Return an error code if Region extension failed
        resp.setStatus(status);
        return resp;
      }

      Map<String, DataPartitionTable> assignedDataPartition;
      try {
        assignedDataPartition =
            getLoadManager().allocateDataPartition(unassignedDataPartitionSlotsMap);
      } catch (NoAvailableRegionGroupException e) {
        status = getConsensusManager().confirmLeader();
        if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
          // The allocation might fail due to leadership change
          resp.setStatus(status);
          return resp;
        }

        LOGGER.error("Create DataPartition failed because: ", e);
        resp.setStatus(
            new TSStatus(TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode())
                .setMessage(e.getMessage()));
        return resp;
      }

      // Cache allocating result only if the current ConfigNode still holds its leadership
      CreateDataPartitionPlan createPlan = new CreateDataPartitionPlan();
      createPlan.setAssignedDataPartition(assignedDataPartition);

      status = consensusWritePartitionResult(createPlan);
      if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
        // The allocation might fail due to consensus error
        resp.setStatus(status);
        return resp;
      }
    }

    resp = getDataPartition(req);
    if (!resp.isAllPartitionsExist()) {
      // Count the fail rate
      AtomicInteger totalSlotNum = new AtomicInteger();
      req.getPartitionSlotsMap()
          .forEach(
              (database, partitionSlots) ->
                  partitionSlots.forEach(
                      (seriesPartitionSlot, timeSlotList) ->
                          totalSlotNum.addAndGet(timeSlotList.getTimePartitionSlots().size())));

      AtomicInteger unassignedSlotNum = new AtomicInteger();
      Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap =
          partitionInfo.filterUnassignedDataPartitionSlots(req.getPartitionSlotsMap());
      unassignedDataPartitionSlotsMap.forEach(
          (database, unassignedDataPartitionSlots) ->
              unassignedDataPartitionSlots.forEach(
                  (seriesPartitionSlot, timeSlotList) ->
                      unassignedSlotNum.addAndGet(timeSlotList.getTimePartitionSlots().size())));

      String errMsg =
          String.format(
              "Lacked %d/%d DataPartition allocation result in the response of getOrCreateDataPartition method",
              unassignedSlotNum.get(), totalSlotNum.get());
      LOGGER.error(errMsg);
      resp.setStatus(
          new TSStatus(TSStatusCode.LACK_PARTITION_ALLOCATION.getStatusCode()).setMessage(errMsg));
      return resp;
    }
    return resp;
  }

  private TSStatus consensusWritePartitionResult(ConfigPhysicalPlan plan) {
    TSStatus status = getConsensusManager().confirmLeader();
    if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
      // Here we check the leadership second time
      // since the RegionGroup creating process might take some time
      return status;
    }
    try {
      return getConsensusManager().write(plan);
    } catch (ConsensusException e) {
      // The allocation might fail due to consensus error
      LOGGER.error("Write DataPartition allocation result failed because: {}", status);
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      return res;
    }
  }

  // ======================================================
  // Leader scheduling interfaces
  // ======================================================

  /**
   * Allocate more RegionGroup to the specified StorageGroups if necessary.
   *
   * @param unassignedPartitionSlotsCountMap Map<StorageGroup, unassigned Partition count>
   * @param consensusGroupType SchemaRegion or DataRegion
   * @return SUCCESS_STATUS when RegionGroup extension successful; NOT_ENOUGH_DATA_NODE when there
   *     are not enough DataNodes; STORAGE_GROUP_NOT_EXIST when some StorageGroups don't exist
   */
  private TSStatus extendRegionGroupIfNecessary(
      Map<String, Integer> unassignedPartitionSlotsCountMap,
      TConsensusGroupType consensusGroupType) {

    TSStatus result = new TSStatus();

    try {
      if (TConsensusGroupType.SchemaRegion.equals(consensusGroupType)) {
        switch (SCHEMA_REGION_GROUP_EXTENSION_POLICY) {
          case CUSTOM:
            return customExtendRegionGroupIfNecessary(
                unassignedPartitionSlotsCountMap, consensusGroupType);
          case AUTO:
          default:
            return autoExtendRegionGroupIfNecessary(
                unassignedPartitionSlotsCountMap, consensusGroupType);
        }
      } else {
        switch (DATA_REGION_GROUP_EXTENSION_POLICY) {
          case CUSTOM:
            return customExtendRegionGroupIfNecessary(
                unassignedPartitionSlotsCountMap, consensusGroupType);
          case AUTO:
          default:
            return autoExtendRegionGroupIfNecessary(
                unassignedPartitionSlotsCountMap, consensusGroupType);
        }
      }
    } catch (NotEnoughDataNodeException e) {
      LOGGER.error(e.getMessage());
      result.setCode(TSStatusCode.NO_ENOUGH_DATANODE.getStatusCode());
      result.setMessage(e.getMessage());
    } catch (DatabaseNotExistsException e) {
      LOGGER.error(e.getMessage());
      result.setCode(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode());
      result.setMessage(e.getMessage());
    }

    return result;
  }

  private TSStatus customExtendRegionGroupIfNecessary(
      Map<String, Integer> unassignedPartitionSlotsCountMap, TConsensusGroupType consensusGroupType)
      throws DatabaseNotExistsException, NotEnoughDataNodeException {

    // Map<Database, Region allotment>
    Map<String, Integer> allotmentMap = new ConcurrentHashMap<>();

    for (Map.Entry<String, Integer> entry : unassignedPartitionSlotsCountMap.entrySet()) {
      final String database = entry.getKey();
      int minRegionGroupNum =
          getClusterSchemaManager().getMinRegionGroupNum(database, consensusGroupType);
      int allocatedRegionGroupCount =
          partitionInfo.getRegionGroupCount(database, consensusGroupType);

      // Extend RegionGroups until allocatedRegionGroupCount == minRegionGroupNum
      if (allocatedRegionGroupCount < minRegionGroupNum) {
        allotmentMap.put(database, minRegionGroupNum - allocatedRegionGroupCount);
      }
    }

    return generateAndAllocateRegionGroups(allotmentMap, consensusGroupType);
  }

  private TSStatus autoExtendRegionGroupIfNecessary(
      Map<String, Integer> unassignedPartitionSlotsCountMap, TConsensusGroupType consensusGroupType)
      throws NotEnoughDataNodeException, DatabaseNotExistsException {

    // Map<Database, Region allotment>
    Map<String, Integer> allotmentMap = new ConcurrentHashMap<>();

    for (Map.Entry<String, Integer> entry : unassignedPartitionSlotsCountMap.entrySet()) {
      final String database = entry.getKey();
      final int unassignedPartitionSlotsCount = entry.getValue();

      float allocatedRegionGroupCount =
          partitionInfo.getRegionGroupCount(database, consensusGroupType);
      // The slotCount equals to the sum of assigned slot count and unassigned slot count
      float slotCount =
          (float) partitionInfo.getAssignedSeriesPartitionSlotsCount(database)
              + unassignedPartitionSlotsCount;
      float maxRegionGroupNum =
          getClusterSchemaManager().getMaxRegionGroupNum(database, consensusGroupType);
      float maxSlotCount = CONF.getSeriesSlotNum();

      /* RegionGroup extension is required in the following cases */
      // 1. The number of current RegionGroup of the Database is less than the minimum number
      int minRegionGroupNum =
          getClusterSchemaManager().getMinRegionGroupNum(database, consensusGroupType);
      if (allocatedRegionGroupCount < minRegionGroupNum
          // Ensure the number of RegionGroups is enough
          // for current SeriesPartitionSlots after extension
          // Otherwise, more RegionGroups should be extended through case 2.
          && slotCount <= (maxSlotCount / maxRegionGroupNum) * minRegionGroupNum) {

        // Let the sum of unassignedPartitionSlotsCount and allocatedRegionGroupCount
        // no less than the minRegionGroupNum
        int delta =
            (int)
                Math.min(
                    unassignedPartitionSlotsCount, minRegionGroupNum - allocatedRegionGroupCount);
        allotmentMap.put(database, delta);

      } else if (allocatedRegionGroupCount < maxRegionGroupNum
          && slotCount / allocatedRegionGroupCount > maxSlotCount / maxRegionGroupNum) {
        // 2. The average number of partitions held by each Region will be greater than the
        // expected average number after the partition allocation is completed.

        // The delta is equal to the smallest integer solution that satisfies the inequality:
        // slotCount / (allocatedRegionGroupCount + delta) < maxSlotCount / maxRegionGroupNum
        int delta =
            Math.min(
                (int) (maxRegionGroupNum - allocatedRegionGroupCount),
                Math.max(
                    1,
                    (int)
                        Math.ceil(
                            slotCount * maxRegionGroupNum / maxSlotCount
                                - allocatedRegionGroupCount)));
        allotmentMap.put(database, delta);

      } else if (allocatedRegionGroupCount
              == filterRegionGroupThroughStatus(database, RegionGroupStatus.Disabled).size()
          && allocatedRegionGroupCount < maxRegionGroupNum) {
        // 3. All RegionGroups in the specified Database are disabled currently
        allotmentMap.put(database, 1);
      }
    }

    return generateAndAllocateRegionGroups(allotmentMap, consensusGroupType);
  }

  private TSStatus generateAndAllocateRegionGroups(
      Map<String, Integer> allotmentMap, TConsensusGroupType consensusGroupType)
      throws NotEnoughDataNodeException, DatabaseNotExistsException {
    if (!allotmentMap.isEmpty()) {
      CreateRegionGroupsPlan createRegionGroupsPlan =
          getLoadManager().allocateRegionGroups(allotmentMap, consensusGroupType);
      LOGGER.info("[CreateRegionGroups] Starting to create the following RegionGroups:");
      createRegionGroupsPlan.planLog(LOGGER);
      return getProcedureManager().createRegionGroups(consensusGroupType, createRegionGroupsPlan);
    } else {
      return RpcUtils.SUCCESS_STATUS;
    }
  }

  /**
   * Only leader use this interface. Checks whether the specified DataPartition has a successor and
   * returns if it does.
   *
   * @param database DatabaseName
   * @param seriesPartitionSlot Corresponding SeriesPartitionSlot
   * @param timePartitionSlot Corresponding TimePartitionSlot
   * @return The specific DataPartition's successor if exists, null otherwise
   */
  public TConsensusGroupId getSuccessorDataPartition(
      String database,
      TSeriesPartitionSlot seriesPartitionSlot,
      TTimePartitionSlot timePartitionSlot) {
    return partitionInfo.getSuccessorDataPartition(
        database, seriesPartitionSlot, timePartitionSlot);
  }

  /**
   * Only leader use this interface. Checks whether the specified DataPartition has a predecessor
   * and returns if it does.
   *
   * @param database DatabaseName
   * @param seriesPartitionSlot Corresponding SeriesPartitionSlot
   * @param timePartitionSlot Corresponding TimePartitionSlot
   * @return The specific DataPartition's predecessor if exists, null otherwise
   */
  public TConsensusGroupId getPredecessorDataPartition(
      String database,
      TSeriesPartitionSlot seriesPartitionSlot,
      TTimePartitionSlot timePartitionSlot) {
    return partitionInfo.getPredecessorDataPartition(
        database, seriesPartitionSlot, timePartitionSlot);
  }

  /**
   * Get the DataNodes who contain the specified Database's Schema or Data.
   *
   * @param database The specific Database's name
   * @param type SchemaRegion or DataRegion
   * @return Set {@literal <}TDataNodeLocation{@literal >}, the related DataNodes
   */
  public Set<TDataNodeLocation> getDatabaseRelatedDataNodes(
      String database, TConsensusGroupType type) {
    return partitionInfo.getDatabaseRelatedDataNodes(database, type);
  }

  /**
   * Only leader use this interface.
   *
   * @param type The specified TConsensusGroupType
   * @return Deep copy of all Regions' RegionReplicaSet and organized to Map
   */
  public Map<TConsensusGroupId, TRegionReplicaSet> getAllReplicaSetsMap(TConsensusGroupType type) {
    return partitionInfo.getAllReplicaSets(type).stream()
        .collect(
            Collectors.toMap(TRegionReplicaSet::getRegionId, regionReplicaSet -> regionReplicaSet));
  }

  /**
   * Only leader use this interface.
   *
   * @return Deep copy of all Regions' RegionReplicaSet
   */
  public List<TRegionReplicaSet> getAllReplicaSets() {
    return partitionInfo.getAllReplicaSets();
  }

  /**
   * Only leader use this interface.
   *
   * @param type The specified TConsensusGroupType
   * @return Deep copy of all Regions' RegionReplicaSet with the specified TConsensusGroupType
   */
  public List<TRegionReplicaSet> getAllReplicaSets(TConsensusGroupType type) {
    return partitionInfo.getAllReplicaSets(type);
  }

  /**
   * Only leader use this interface.
   *
   * @param database The specified Database
   * @return All Regions' RegionReplicaSet of the specified Database
   */
  public List<TRegionReplicaSet> getAllReplicaSets(String database) {
    return partitionInfo.getAllReplicaSets(database);
  }

  /**
   * Get all RegionGroups currently owned by the specified Database.
   *
   * @param dataNodeId The specified dataNodeId
   * @return Deep copy of all RegionGroups' RegionReplicaSet with the specified dataNodeId
   */
  public List<TRegionReplicaSet> getAllReplicaSets(int dataNodeId) {
    return partitionInfo.getAllReplicaSets(dataNodeId);
  }

  /**
   * Only leader use this interface.
   *
   * @param database The specified Database
   * @param regionGroupIds The specified RegionGroupIds
   * @return All Regions' RegionReplicaSet of the specified Database
   */
  public List<TRegionReplicaSet> getReplicaSets(
      String database, List<TConsensusGroupId> regionGroupIds) {
    return partitionInfo.getReplicaSets(database, regionGroupIds);
  }

  /**
   * Only leader use this interface.
   *
   * <p>Get the number of Regions currently owned by the specified DataNode
   *
   * @param dataNodeId The specified DataNode
   * @param type SchemaRegion or DataRegion
   * @return The number of Regions currently owned by the specified DataNode
   */
  public int getRegionCount(int dataNodeId, TConsensusGroupType type) {
    return partitionInfo.getRegionCount(dataNodeId, type);
  }

  /**
   * Only leader use this interface.
   *
   * <p>Count the scatter width of the specified DataNode
   *
   * @param dataNodeId The specified DataNode
   * @param type SchemaRegion or DataRegion
   * @return The schema/data scatter width of the specified DataNode. The scatter width refers to
   *     the number of other DataNodes in the cluster which have at least one identical schema/data
   *     replica as the specified DataNode.
   */
  public int countDataNodeScatterWidth(int dataNodeId, TConsensusGroupType type) {
    int clusterNodeCount = getNodeManager().getRegisteredNodeCount();
    return partitionInfo.countDataNodeScatterWidth(dataNodeId, type, clusterNodeCount);
  }

  /**
   * Only leader use this interface.
   *
   * <p>Get the number of RegionGroups currently owned by the specified Database
   *
   * @param database DatabaseName
   * @param type SchemaRegion or DataRegion
   * @return Number of Regions currently owned by the specified Database
   * @throws DatabaseNotExistsException When the specified Database doesn't exist
   */
  public int getRegionGroupCount(String database, TConsensusGroupType type)
      throws DatabaseNotExistsException {
    return partitionInfo.getRegionGroupCount(database, type);
  }

  /**
   * Only leader use this interface.
   *
   * <p>Get all the RegionGroups currently owned by the specified Database
   *
   * @param database DatabaseName
   * @param type SchemaRegion or DataRegion
   * @return List of TConsensusGroupId
   * @throws DatabaseNotExistsException When the specified Database doesn't exist
   */
  public List<TConsensusGroupId> getAllRegionGroupIds(String database, TConsensusGroupType type)
      throws DatabaseNotExistsException {
    return partitionInfo.getAllRegionGroupIds(database, type);
  }

  /**
   * Check if the specified Database exists.
   *
   * @param database The specified Database
   * @return True if the DatabaseSchema is exists and the Database is not pre-deleted
   */
  public boolean isDatabaseExist(String database) {
    return partitionInfo.isDatabaseExisted(database);
  }

  /**
   * Filter the un-exist Databases.
   *
   * @param databases the Databases to check
   * @return List of PartialPath the Databases that not exist
   */
  public List<PartialPath> filterUnExistDatabases(List<PartialPath> databases) {
    List<PartialPath> unExistDatabases = new ArrayList<>();
    if (databases == null) {
      return unExistDatabases;
    }
    for (PartialPath database : databases) {
      if (!isDatabaseExist(database.getFullPath())) {
        unExistDatabases.add(database);
      }
    }
    return unExistDatabases;
  }

  /**
   * Only leader use this interface.
   *
   * <p>Get the assigned SeriesPartitionSlots count in the specified Database
   *
   * @param database The specified Database
   * @return The assigned SeriesPartitionSlots count
   */
  public int getAssignedSeriesPartitionSlotsCount(String database) {
    return partitionInfo.getAssignedSeriesPartitionSlotsCount(database);
  }

  /**
   * Only leader use this interface.
   *
   * <p>Get the assigned TimePartitionSlots count in the specified Database
   *
   * @param database The specified Database
   * @return The assigned TimePartitionSlots count
   */
  public long getAssignedTimePartitionSlotsCount(String database) {
    return partitionInfo.getAssignedTimePartitionSlotsCount(database);
  }

  /**
   * Only leader use this interface.
   *
   * @param database DatabaseName
   * @param type SchemaRegion or DataRegion
   * @return The specific StorageGroup's Regions that sorted by the number of allocated slots
   * @throws NoAvailableRegionGroupException When all RegionGroups within the specified StorageGroup
   *     are unavailable currently
   */
  public List<Pair<Long, TConsensusGroupId>> getSortedRegionGroupSlotsCounter(
      String database, TConsensusGroupType type) throws NoAvailableRegionGroupException {
    // Collect static data
    List<Pair<Long, TConsensusGroupId>> regionGroupSlotsCounter =
        partitionInfo.getRegionGroupSlotsCounter(database, type);

    // Filter RegionGroups that have Disabled status
    List<Pair<Long, TConsensusGroupId>> result = new ArrayList<>();
    for (Pair<Long, TConsensusGroupId> slotsCounter : regionGroupSlotsCounter) {
      RegionGroupStatus status = getLoadManager().getRegionGroupStatus(slotsCounter.getRight());
      if (!RegionGroupStatus.Disabled.equals(status)) {
        result.add(slotsCounter);
      }
    }

    if (result.isEmpty()) {
      throw new NoAvailableRegionGroupException(type);
    }

    Map<TConsensusGroupId, RegionGroupStatus> regionGroupStatusMap =
        getLoadManager()
            .getRegionGroupStatus(result.stream().map(Pair::getRight).collect(Collectors.toList()));
    result.sort(
        (o1, o2) -> {
          // Use the number of partitions as the first priority
          if (o1.getLeft() < o2.getLeft()) {
            return -1;
          } else if (o1.getLeft() > o2.getLeft()) {
            return 1;
          } else {
            // Use RegionGroup status as second priority, Running > Available > Discouraged
            return regionGroupStatusMap
                .get(o1.getRight())
                .compare(regionGroupStatusMap.get(o2.getRight()));
          }
        });
    return result;
  }

  /**
   * Only leader use this interface.
   *
   * @return Integer set of all schemaengine region id
   */
  public Set<Integer> getAllSchemaPartition() {
    return partitionInfo.getAllSchemaPartition();
  }

  /**
   * Only leader use this interface.
   *
   * @return the next RegionGroupId
   */
  public int generateNextRegionGroupId() {
    return partitionInfo.generateNextRegionGroupId();
  }

  /**
   * GetNodePathsPartition.
   *
   * @param physicalPlan GetNodesPathsPartitionReq
   * @return SchemaNodeManagementPartitionDataSet that contains only existing matched
   *     SchemaPartition and matched child paths aboveMTree
   */
  public SchemaNodeManagementResp getNodePathsPartition(GetNodePathsPartitionPlan physicalPlan) {
    try {
      return (SchemaNodeManagementResp) getConsensusManager().read(physicalPlan);
    } catch (ConsensusException e) {
      LOGGER.warn(CONSENSUS_READ_ERROR, e);
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      SchemaNodeManagementResp resp = new SchemaNodeManagementResp();
      resp.setStatus(res);
      return resp;
    }
  }

  public void preDeleteDatabase(
      String database, PreDeleteDatabasePlan.PreDeleteType preDeleteType) {
    final PreDeleteDatabasePlan preDeleteDatabasePlan =
        new PreDeleteDatabasePlan(database, preDeleteType);
    try {
      getConsensusManager().write(preDeleteDatabasePlan);
    } catch (ConsensusException e) {
      LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
    }
  }

  public boolean isDatabasePreDeleted(String database) {
    return partitionInfo.isDatabasePreDeleted(database);
  }

  /**
   * Get TSeriesPartitionSlot.
   *
   * @param devicePath Full path ending with device name
   * @return SeriesPartitionSlot
   */
  public TSeriesPartitionSlot getSeriesPartitionSlot(String devicePath) {
    return executor.getSeriesPartitionSlot(devicePath);
  }

  public RegionInfoListResp getRegionInfoList(GetRegionInfoListPlan req) {
    try {
      // Get static result
      RegionInfoListResp regionInfoListResp = (RegionInfoListResp) getConsensusManager().read(req);
      // Get cached result
      Map<TConsensusGroupId, Integer> allLeadership = getLoadManager().getRegionLeaderMap();
      regionInfoListResp
          .getRegionInfoList()
          .forEach(
              regionInfo -> {
                regionInfo.setStatus(
                    getLoadManager()
                        .getRegionStatus(
                            regionInfo.getConsensusGroupId(), regionInfo.getDataNodeId())
                        .getStatus());

                String regionType =
                    regionInfo.getDataNodeId()
                            == allLeadership.getOrDefault(regionInfo.getConsensusGroupId(), -1)
                        ? RegionRoleType.Leader.toString()
                        : RegionRoleType.Follower.toString();
                regionInfo.setRoleType(regionType);
              });

      return regionInfoListResp;

    } catch (ConsensusException e) {
      LOGGER.warn(CONSENSUS_READ_ERROR, e);
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      RegionInfoListResp resp = new RegionInfoListResp();
      resp.setStatus(res);
      return resp;
    }
  }

  /**
   * Check if the specified RegionGroup exists.
   *
   * @param regionGroupId The specified RegionGroup
   */
  public boolean isRegionGroupExists(TConsensusGroupId regionGroupId) {
    return partitionInfo.isRegionGroupExisted(regionGroupId);
  }

  /**
   * Update region location.
   *
   * @param req UpdateRegionLocationReq
   * @return TSStatus
   */
  public TSStatus updateRegionLocation(UpdateRegionLocationPlan req) {
    try {
      return getConsensusManager().write(req);
    } catch (ConsensusException e) {
      LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      return res;
    }
  }

  public GetRegionIdResp getRegionId(TGetRegionIdReq req) {
    GetRegionIdPlan plan = new GetRegionIdPlan(req.getType());
    if (req.isSetDatabase()) {
      plan.setDatabase(req.getDatabase());
    } else {
      plan.setDatabase(getClusterSchemaManager().getDatabaseNameByDevice(req.getDevice()));
      plan.setSeriesSlotId(executor.getSeriesPartitionSlot(req.getDevice()));
    }
    if (Objects.equals(plan.getDatabase(), "")) {
      // Return empty result if Database not specified
      return new GetRegionIdResp(RpcUtils.SUCCESS_STATUS, new ArrayList<>());
    }
    if (req.isSetStartTimeSlot()) {
      plan.setStartTimeSlotId(req.getStartTimeSlot());
    } else {
      plan.setStartTimeSlotId(new TTimePartitionSlot(0));
    }
    if (req.isSetEndTimeSlot()) {
      plan.setEndTimeSlotId(req.getEndTimeSlot());
    } else {
      plan.setEndTimeSlotId(new TTimePartitionSlot(Long.MAX_VALUE));
    }

    try {
      return (GetRegionIdResp) getConsensusManager().read(plan);
    } catch (ConsensusException e) {
      LOGGER.warn(CONSENSUS_READ_ERROR, e);
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      return new GetRegionIdResp(res, Collections.emptyList());
    }
  }

  public GetTimeSlotListResp getTimeSlotList(TGetTimeSlotListReq req) {
    long startTime = req.isSetStartTime() ? req.getStartTime() : Long.MIN_VALUE;
    long endTime = req.isSetEndTime() ? req.getEndTime() : Long.MAX_VALUE;
    GetTimeSlotListPlan plan = new GetTimeSlotListPlan(startTime, endTime);
    if (req.isSetDatabase()) {
      plan.setDatabase(req.getDatabase());
    } else if (req.isSetDevice()) {
      plan.setDatabase(getClusterSchemaManager().getDatabaseNameByDevice(req.getDevice()));
      plan.setSeriesSlotId(executor.getSeriesPartitionSlot(req.getDevice()));
      if (Objects.equals(plan.getDatabase(), "")) {
        // Return empty result if Database not specified
        return new GetTimeSlotListResp(RpcUtils.SUCCESS_STATUS, new ArrayList<>());
      }
    } else {
      plan.setRegionId(
          new TConsensusGroupId(TConsensusGroupType.DataRegion, (int) req.getRegionId()));
    }
    try {
      return (GetTimeSlotListResp) getConsensusManager().read(plan);
    } catch (ConsensusException e) {
      LOGGER.warn(CONSENSUS_READ_ERROR, e);
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      return new GetTimeSlotListResp(res, Collections.emptyList());
    }
  }

  public CountTimeSlotListResp countTimeSlotList(TCountTimeSlotListReq req) {
    long startTime = req.isSetStartTime() ? req.getStartTime() : Long.MIN_VALUE;
    long endTime = req.isSetEndTime() ? req.getEndTime() : Long.MAX_VALUE;
    CountTimeSlotListPlan plan = new CountTimeSlotListPlan(startTime, endTime);
    if (req.isSetDatabase()) {
      plan.setDatabase(req.getDatabase());
    } else if (req.isSetDevice()) {
      plan.setDatabase(getClusterSchemaManager().getDatabaseNameByDevice(req.getDevice()));
      plan.setSeriesSlotId(executor.getSeriesPartitionSlot(req.getDevice()));
      if (Objects.equals(plan.getDatabase(), "")) {
        // Return empty result if Database not specified
        return new CountTimeSlotListResp(RpcUtils.SUCCESS_STATUS, 0);
      }
    } else {
      plan.setRegionId(
          new TConsensusGroupId(TConsensusGroupType.DataRegion, (int) req.getRegionId()));
    }
    try {
      return (CountTimeSlotListResp) getConsensusManager().read(plan);
    } catch (ConsensusException e) {
      LOGGER.warn(CONSENSUS_READ_ERROR, e);
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      return new CountTimeSlotListResp(res, 0);
    }
  }

  public GetSeriesSlotListResp getSeriesSlotList(TGetSeriesSlotListReq req) {
    GetSeriesSlotListPlan plan = new GetSeriesSlotListPlan(req.getDatabase(), req.getType());
    try {
      return (GetSeriesSlotListResp) getConsensusManager().read(plan);
    } catch (ConsensusException e) {
      LOGGER.warn(CONSENSUS_READ_ERROR, e);
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      return new GetSeriesSlotListResp(res, Collections.emptyList());
    }
  }

  /**
   * Get database for region.
   *
   * @param regionId regionId
   * @return database name
   */
  public String getRegionStorageGroup(TConsensusGroupId regionId) {
    return partitionInfo.getRegionStorageGroup(regionId);
  }

  /**
   * Called by {@link PartitionManager#regionMaintainer}.
   *
   * <p>Periodically maintain the RegionReplicas to be created or deleted
   */
  public void maintainRegionReplicas() {
    // The consensusManager of configManager may not be fully initialized at this time
    Optional.ofNullable(getConsensusManager())
        .ifPresent(
            consensusManager -> {
              if (getConsensusManager().isLeader()) {
                List<RegionMaintainTask> regionMaintainTaskList =
                    partitionInfo.getRegionMaintainEntryList();

                if (regionMaintainTaskList.isEmpty()) {
                  return;
                }

                // Group tasks by region id
                Map<TConsensusGroupId, Queue<RegionMaintainTask>> regionMaintainTaskMap =
                    new HashMap<>();
                for (RegionMaintainTask regionMaintainTask : regionMaintainTaskList) {
                  regionMaintainTaskMap
                      .computeIfAbsent(regionMaintainTask.getRegionId(), k -> new LinkedList<>())
                      .add(regionMaintainTask);
                }

                while (!regionMaintainTaskMap.isEmpty()) {
                  // Select same type task from each region group
                  List<RegionMaintainTask> selectedRegionMaintainTask = new ArrayList<>();
                  RegionMaintainType currentType = null;
                  for (Map.Entry<TConsensusGroupId, Queue<RegionMaintainTask>> entry :
                      regionMaintainTaskMap.entrySet()) {
                    RegionMaintainTask regionMaintainTask = entry.getValue().peek();
                    if (regionMaintainTask == null) {
                      continue;
                    }

                    if (currentType == null) {
                      currentType = regionMaintainTask.getType();
                      selectedRegionMaintainTask.add(entry.getValue().peek());
                    } else {
                      if (!currentType.equals(regionMaintainTask.getType())) {
                        continue;
                      }

                      if (currentType.equals(RegionMaintainType.DELETE)
                          || entry
                              .getKey()
                              .getType()
                              .equals(selectedRegionMaintainTask.get(0).getRegionId().getType())) {
                        // Delete or same create task
                        selectedRegionMaintainTask.add(entry.getValue().peek());
                      }
                    }
                  }

                  if (selectedRegionMaintainTask.isEmpty()) {
                    break;
                  }

                  Set<TConsensusGroupId> successfulTask = new HashSet<>();
                  switch (currentType) {
                    case CREATE:
                      // create region
                      switch (selectedRegionMaintainTask.get(0).getRegionId().getType()) {
                        case SchemaRegion:
                          // create SchemaRegion
                          AsyncClientHandler<TCreateSchemaRegionReq, TSStatus>
                              createSchemaRegionHandler =
                                  new AsyncClientHandler<>(
                                      DataNodeRequestType.CREATE_SCHEMA_REGION);
                          for (RegionMaintainTask regionMaintainTask : selectedRegionMaintainTask) {
                            RegionCreateTask schemaRegionCreateTask =
                                (RegionCreateTask) regionMaintainTask;
                            LOGGER.info(
                                "Start to create Region: {} on DataNode: {}",
                                schemaRegionCreateTask.getRegionReplicaSet().getRegionId(),
                                schemaRegionCreateTask.getTargetDataNode());
                            createSchemaRegionHandler.putRequest(
                                schemaRegionCreateTask.getRegionId().getId(),
                                new TCreateSchemaRegionReq(
                                    schemaRegionCreateTask.getRegionReplicaSet(),
                                    schemaRegionCreateTask.getStorageGroup()));
                            createSchemaRegionHandler.putDataNodeLocation(
                                schemaRegionCreateTask.getRegionId().getId(),
                                schemaRegionCreateTask.getTargetDataNode());
                          }

                          AsyncDataNodeClientPool.getInstance()
                              .sendAsyncRequestToDataNodeWithRetry(createSchemaRegionHandler);

                          for (Map.Entry<Integer, TSStatus> entry :
                              createSchemaRegionHandler.getResponseMap().entrySet()) {
                            if (entry.getValue().getCode()
                                == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                              successfulTask.add(
                                  new TConsensusGroupId(
                                      TConsensusGroupType.SchemaRegion, entry.getKey()));
                            }
                          }
                          break;
                        case DataRegion:
                          // Create DataRegion
                          AsyncClientHandler<TCreateDataRegionReq, TSStatus>
                              createDataRegionHandler =
                                  new AsyncClientHandler<>(DataNodeRequestType.CREATE_DATA_REGION);
                          for (RegionMaintainTask regionMaintainTask : selectedRegionMaintainTask) {
                            RegionCreateTask dataRegionCreateTask =
                                (RegionCreateTask) regionMaintainTask;
                            LOGGER.info(
                                "Start to create Region: {} on DataNode: {}",
                                dataRegionCreateTask.getRegionReplicaSet().getRegionId(),
                                dataRegionCreateTask.getTargetDataNode());
                            createDataRegionHandler.putRequest(
                                dataRegionCreateTask.getRegionId().getId(),
                                new TCreateDataRegionReq(
                                        dataRegionCreateTask.getRegionReplicaSet(),
                                        dataRegionCreateTask.getStorageGroup())
                                    .setTtl(dataRegionCreateTask.getTTL()));
                            createDataRegionHandler.putDataNodeLocation(
                                dataRegionCreateTask.getRegionId().getId(),
                                dataRegionCreateTask.getTargetDataNode());
                          }

                          AsyncDataNodeClientPool.getInstance()
                              .sendAsyncRequestToDataNodeWithRetry(createDataRegionHandler);

                          for (Map.Entry<Integer, TSStatus> entry :
                              createDataRegionHandler.getResponseMap().entrySet()) {
                            if (entry.getValue().getCode()
                                == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                              successfulTask.add(
                                  new TConsensusGroupId(
                                      TConsensusGroupType.DataRegion, entry.getKey()));
                            }
                          }
                          break;
                      }
                      break;
                    case DELETE:
                      // delete region
                      AsyncClientHandler<TConsensusGroupId, TSStatus> deleteRegionHandler =
                          new AsyncClientHandler<>(DataNodeRequestType.DELETE_REGION);
                      Map<Integer, TConsensusGroupId> regionIdMap = new HashMap<>();
                      for (RegionMaintainTask regionMaintainTask : selectedRegionMaintainTask) {
                        RegionDeleteTask regionDeleteTask = (RegionDeleteTask) regionMaintainTask;
                        LOGGER.info(
                            "Start to delete Region: {} on DataNode: {}",
                            regionDeleteTask.getRegionId(),
                            regionDeleteTask.getTargetDataNode());
                        deleteRegionHandler.putRequest(
                            regionDeleteTask.getRegionId().getId(), regionDeleteTask.getRegionId());
                        deleteRegionHandler.putDataNodeLocation(
                            regionDeleteTask.getRegionId().getId(),
                            regionDeleteTask.getTargetDataNode());
                        regionIdMap.put(
                            regionDeleteTask.getRegionId().getId(), regionDeleteTask.getRegionId());
                      }

                      long startTime = System.currentTimeMillis();
                      AsyncDataNodeClientPool.getInstance()
                          .sendAsyncRequestToDataNodeWithRetry(deleteRegionHandler);

                      LOGGER.info(
                          "Deleting regions costs {}ms", (System.currentTimeMillis() - startTime));

                      for (Map.Entry<Integer, TSStatus> entry :
                          deleteRegionHandler.getResponseMap().entrySet()) {
                        if (entry.getValue().getCode()
                            == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                          successfulTask.add(regionIdMap.get(entry.getKey()));
                        }
                      }
                      break;
                  }

                  if (successfulTask.isEmpty()) {
                    break;
                  }

                  for (TConsensusGroupId regionId : successfulTask) {
                    regionMaintainTaskMap.compute(
                        regionId,
                        (k, v) -> {
                          if (v == null) {
                            throw new IllegalStateException();
                          }
                          v.poll();
                          if (v.isEmpty()) {
                            return null;
                          } else {
                            return v;
                          }
                        });
                  }

                  // Poll the head entry if success
                  try {
                    getConsensusManager()
                        .write(new PollSpecificRegionMaintainTaskPlan(successfulTask));
                  } catch (ConsensusException e) {
                    LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
                  }

                  if (successfulTask.size() < selectedRegionMaintainTask.size()) {
                    // Here we just break and wait until next schedule task
                    // due to all the RegionMaintainEntry should be executed by
                    // the order of they were offered
                    break;
                  }
                }
              }
            });
  }

  public void startRegionCleaner() {
    synchronized (scheduleMonitor) {
      if (currentRegionMaintainerFuture == null) {
        /* Start the RegionCleaner service */
        currentRegionMaintainerFuture =
            ScheduledExecutorUtil.safelyScheduleAtFixedRate(
                regionMaintainer,
                this::maintainRegionReplicas,
                0,
                REGION_MAINTAINER_WORK_INTERVAL,
                TimeUnit.SECONDS);
        LOGGER.info("RegionCleaner is started successfully.");
      }
    }
  }

  public void stopRegionCleaner() {
    synchronized (scheduleMonitor) {
      if (currentRegionMaintainerFuture != null) {
        /* Stop the RegionCleaner service */
        currentRegionMaintainerFuture.cancel(false);
        currentRegionMaintainerFuture = null;
        LOGGER.info("RegionCleaner is stopped successfully.");
      }
    }
  }

  /**
   * Filter the RegionGroups in the specified Database through the RegionGroupStatus.
   *
   * @param database The specified Database
   * @param status The specified RegionGroupStatus
   * @return Filtered RegionGroups with the specified RegionGroupStatus
   */
  public List<TRegionReplicaSet> filterRegionGroupThroughStatus(
      String database, RegionGroupStatus... status) {
    List<TConsensusGroupId> matchedRegionGroups =
        getLoadManager().filterRegionGroupThroughStatus(status);
    return getReplicaSets(database, matchedRegionGroups);
  }

  public void getSchemaRegionIds(
      List<String> databases, Map<String, List<Integer>> schemaRegionIds) {
    partitionInfo.getSchemaRegionIds(databases, schemaRegionIds);
  }

  public void getDataRegionIds(List<String> databases, Map<String, List<Integer>> dataRegionIds) {
    partitionInfo.getDataRegionIds(databases, dataRegionIds);
  }

  /**
   * Get the {@link TConsensusGroupType} of the given integer regionId.
   *
   * @param regionId The specified integer regionId
   * @return {@link Optional#of(Object tConsensusGroupType)} of the given integer regionId, or
   *     {@link Optional#empty()} if the integer regionId does not match any of the regionGroups.
   */
  public Optional<TConsensusGroupType> getRegionType(int regionId) {
    return partitionInfo.getRegionType(regionId);
  }

  /**
   * Get the last DataAllotTable of the specified Database.
   *
   * @param database The specified Database
   * @return The last DataAllotTable
   */
  public Map<TSeriesPartitionSlot, TConsensusGroupId> getLastDataAllotTable(String database) {
    return partitionInfo.getLastDataAllotTable(database);
  }

  public ScheduledExecutorService getRegionMaintainer() {
    return regionMaintainer;
  }

  private ConsensusManager getConsensusManager() {
    return configManager.getConsensusManager();
  }

  private ClusterSchemaManager getClusterSchemaManager() {
    return configManager.getClusterSchemaManager();
  }

  private LoadManager getLoadManager() {
    return configManager.getLoadManager();
  }

  private ProcedureManager getProcedureManager() {
    return configManager.getProcedureManager();
  }

  private NodeManager getNodeManager() {
    return configManager.getNodeManager();
  }
}
