public Map<String, DataPartitionTable> allocateDataPartition(
      Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap)
      throws DatabaseNotExistsException, NoAvailableRegionGroupException {
    Map<String, DataPartitionTable> result = new TreeMap<>();

    for (Map.Entry<String, Map<TSeriesPartitionSlot, TTimeSlotList>> slotsMapEntry :
        unassignedDataPartitionSlotsMap.entrySet()) {
      final String database = slotsMapEntry.getKey();
      final Map<TSeriesPartitionSlot, TTimeSlotList> unassignedPartitionSlotsMap =
          slotsMapEntry.getValue();

      // Filter available DataRegionGroups and
      // sort them by the number of allocated DataPartitions
      BalanceTreeMap<TConsensusGroupId, Integer> availableDataRegionGroupCounter =
          new BalanceTreeMap<>();
      List<Pair<Long, TConsensusGroupId>> regionSlotsCounter =
          getPartitionManager()
              .getSortedRegionGroupSlotsCounter(database, TConsensusGroupType.DataRegion);
      for (Pair<Long, TConsensusGroupId> pair : regionSlotsCounter) {
        availableDataRegionGroupCounter.put(pair.getRight(), pair.getLeft().intValue());
      }

      DataPartitionTable dataPartitionTable = new DataPartitionTable();
      if (!dataPartitionPolicyTableMap.containsKey(database)) {
        throw new DatabaseNotExistsException(database);
      }
      DataPartitionPolicyTable allotTable = dataPartitionPolicyTableMap.get(database);
      allotTable.acquireLock();
      try {
        // Enumerate SeriesPartitionSlot
        for (Map.Entry<TSeriesPartitionSlot, TTimeSlotList> seriesPartitionEntry :
            unassignedPartitionSlotsMap.entrySet()) {
          SeriesPartitionTable seriesPartitionTable = new SeriesPartitionTable();

          // Enumerate TimePartitionSlot in ascending order
          TSeriesPartitionSlot seriesPartitionSlot = seriesPartitionEntry.getKey();
          List<TTimePartitionSlot> timePartitionSlots =
              seriesPartitionEntry.getValue().getTimePartitionSlots();
          timePartitionSlots.sort(Comparator.comparingLong(TTimePartitionSlot::getStartTime));

          for (TTimePartitionSlot timePartitionSlot : timePartitionSlots) {

            // 1. The historical DataPartition will try to inherit successor DataPartition first
            TConsensusGroupId successor =
                getPartitionManager()
                    .getSuccessorDataPartition(database, seriesPartitionSlot, timePartitionSlot);
            if (successor != null && availableDataRegionGroupCounter.containsKey(successor)) {
              seriesPartitionTable.putDataPartition(timePartitionSlot, successor);
              availableDataRegionGroupCounter.put(
                  successor, availableDataRegionGroupCounter.get(successor) + 1);
              continue;
            }

            // 2. Assign DataPartition base on the DataAllotTable
            TConsensusGroupId allotGroupId =
                allotTable.getRegionGroupIdOrActivateIfNecessary(seriesPartitionSlot);
            if (availableDataRegionGroupCounter.containsKey(allotGroupId)) {
              seriesPartitionTable.putDataPartition(timePartitionSlot, allotGroupId);
              availableDataRegionGroupCounter.put(
                  allotGroupId, availableDataRegionGroupCounter.get(allotGroupId) + 1);
              continue;
            }

            // 3. The allotDataRegionGroup is unavailable,
            // try to inherit predecessor DataPartition
            TConsensusGroupId predecessor =
                getPartitionManager()
                    .getPredecessorDataPartition(database, seriesPartitionSlot, timePartitionSlot);
            if (predecessor != null && availableDataRegionGroupCounter.containsKey(predecessor)) {
              seriesPartitionTable.putDataPartition(timePartitionSlot, predecessor);
              availableDataRegionGroupCounter.put(
                  predecessor, availableDataRegionGroupCounter.get(predecessor) + 1);
              continue;
            }

            // 4. Assign the DataPartition to DataRegionGroup with the least DataPartitions
            // If the above DataRegionGroups are unavailable
            TConsensusGroupId greedyGroupId = availableDataRegionGroupCounter.getKeyWithMinValue();
            seriesPartitionTable.putDataPartition(timePartitionSlot, greedyGroupId);
            availableDataRegionGroupCounter.put(
                greedyGroupId, availableDataRegionGroupCounter.get(greedyGroupId) + 1);
            LOGGER.warn(
                "[PartitionBalancer] The SeriesSlot: {} in TimeSlot: {} will be allocated to DataRegionGroup: {}, because the original target: {} is currently unavailable.",
                seriesPartitionSlot,
                timePartitionSlot,
                greedyGroupId,
                allotGroupId);
          }

          dataPartitionTable
              .getDataPartitionMap()
              .put(seriesPartitionEntry.getKey(), seriesPartitionTable);
        }
      } finally {
        allotTable.releaseLock();
      }
      result.put(database, dataPartitionTable);
    }

    return result;
  }
Allocate DataPartitions

@param unassignedDataPartitionSlotsMap DataPartitionSlots that should be assigned
@throws DatabaseNotExistsException If some specific Databases don't exist
@throws NoAvailableRegionGroupException If there are no available RegionGroups
@return Map<DatabaseName, DataPartitionTable>, the allocating result
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java