public SchemaPartitionResp getOrCreateSchemaPartition(final GetOrCreateSchemaPartitionPlan req) {
    // Check if the related Databases exist
    for (final 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;
      }

      // Here we check if the related Databases exist again,
      // due to we don't have a transaction mechanism.
      for (final 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 does not exist",
                          database)),
              false,
              null);
        }
      }

      // Filter unassigned SchemaPartitionSlots
      final 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>
      final 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;
      }

      final Map<String, SchemaPartitionTable> assignedSchemaPartition;
      try {
        assignedSchemaPartition =
            getLoadManager().allocateSchemaPartition(unassignedSchemaPartitionSlotsMap);
      } catch (final 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
      final 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
      final AtomicInteger totalSlotNum = new AtomicInteger();
      req.getPartitionSlotsMap()
          .forEach((database, partitionSlots) -> totalSlotNum.addAndGet(partitionSlots.size()));

      final AtomicInteger unassignedSlotNum = new AtomicInteger();
      final Map<String, List<TSeriesPartitionSlot>> unassignedSchemaPartitionSlotsMap =
          partitionInfo.filterUnassignedSchemaPartitionSlots(req.getPartitionSlotsMap());
      StringJoiner errDatabases = new StringJoiner(", ", "[", "]");
      unassignedSchemaPartitionSlotsMap.forEach(
          (database, unassignedSchemaPartitionSlots) -> {
            if (!unassignedSchemaPartitionSlots.isEmpty()) {
              errDatabases.add(database);
              unassignedSlotNum.addAndGet(unassignedSchemaPartitionSlots.size());
            }
          });

      final String errMsg =
          String.format(
              "Lacked %d/%d SchemaPartition allocation result when get or create schema partitions for databases: %s",
              unassignedSlotNum.get(), totalSlotNum.get(), errDatabases);
      LOGGER.error(errMsg);
      resp.setStatus(
          new TSStatus(TSStatusCode.LACK_PARTITION_ALLOCATION.getStatusCode()).setMessage(errMsg));
      return resp;
    }
    return resp;
  }
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.
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java