Add proactive RegionGroup allocation and policy hot reload

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java

@@ -95,34 +95,82 @@ public class ConfigNodeConfig {
 
   private String dataPartitionAllocationStrategy = "INHERIT";
 
-  /** The policy of extension SchemaRegionGroup for each Database. */
+  /**
+   * The policy of extending SchemaRegionGroups for each Database: CUSTOM, AUTO, or PROACTIVE.
+   *
+   * <p>CUSTOM suits known workloads that need manual resource allocation, creating the configured
+   * target number of groups per Database when schema partitions are first allocated. AUTO suits
+   * most routine workloads (roughly 80% as an approximate planning guideline), gradually meeting
+   * the configured minimum and expanding with slot occupancy up to the resource-based maximum to
+   * balance parallelism and group management overhead. PROACTIVE complements AUTO for workloads
+   * such as very few devices with many measurements and a high load, creating groups earlier for
+   * more parallelism at the cost of additional group management overhead.
+   *
+   * <p>AUTO and PROACTIVE use the same per-Database maximum calculation, including resource sharing
+   * across Databases; their group counts match when both reach the same maximum. PROACTIVE targets
+   * one group per distinct schema series slot, not per measurement, and grows toward the configured
+   * minimum incrementally like AUTO. When all groups of this type are disabled, it may add one
+   * group within the maximum. Switching policies does not remove existing groups.
+   */
   private volatile RegionGroupExtensionPolicy schemaRegionGroupExtensionPolicy =
-      RegionGroupExtensionPolicy.AUTO;
+      RegionGroupExtensionPolicy.PROACTIVE;
 
   /**
    * When set schema_region_group_extension_policy=CUSTOM, this parameter is the default number of
-   * SchemaRegionGroups for each Database. When set schema_region_group_extension_policy=AUTO, this
-   * parameter is the default minimal number of SchemaRegionGroups for each Database.
+   * SchemaRegionGroups for each Database. For AUTO and PROACTIVE, this parameter is the default
+   * minimum number of SchemaRegionGroups and a lower bound for the per-Database maximum. Both
+   * policies grow toward this minimum incrementally, adding at most the number of series slots in
+   * the pending request to satisfy the minimum. PROACTIVE also grows with the number of active
+   * series slots.
    */
   private volatile int defaultSchemaRegionGroupNumPerDatabase = 1;
 
-  /** The maximum number of SchemaRegions expected to be managed by each DataNode. */
+  /**
+   * The expected number of SchemaRegions per DataNode, used to calculate the same per-Database
+   * maximum for AUTO and PROACTIVE. This is not a hard limit on node or cluster totals:
+   * per-Database minimums, rounding, and existing groups can raise the total above this resource
+   * estimate.
+   */
   private volatile int schemaRegionPerDataNode = 1;
 
-  /** The policy of extension DataRegionGroup for each Database. */
+  /**
+   * The policy of extending DataRegionGroups for each Database: CUSTOM, AUTO, or PROACTIVE.
+   *
+   * <p>CUSTOM suits known workloads that need manual resource allocation, creating the configured
+   * target number of groups per Database when data partitions are first allocated. AUTO suits most
+   * routine workloads (roughly 80% as an approximate planning guideline), gradually meeting the
+   * configured minimum and expanding with slot occupancy up to the resource-based maximum to
+   * balance parallelism and group management overhead. PROACTIVE complements AUTO for workloads
+   * such as very few devices with many measurements and a high load, creating groups earlier for
+   * more parallelism at the cost of additional group management overhead.
+   *
+   * <p>AUTO and PROACTIVE use the same per-Database maximum calculation, including resource sharing
+   * across Databases; their group counts match when both reach the same maximum. PROACTIVE targets
+   * one group per distinct data series slot, not per measurement, and grows toward the configured
+   * minimum incrementally like AUTO. New time partitions in an existing slot can also trigger this
+   * growth toward the minimum. When all groups of this type are disabled, it may add one group
+   * within the maximum. Its allocation policy table balances active slots so new time partitions
+   * can use new groups; existing time-partition assignments are retained. Switching policies does
+   * not remove existing groups.
+   */
   private volatile RegionGroupExtensionPolicy dataRegionGroupExtensionPolicy =
-      RegionGroupExtensionPolicy.AUTO;
+      RegionGroupExtensionPolicy.PROACTIVE;
 
   /**
    * When set data_region_group_extension_policy=CUSTOM, this parameter is the default number of
-   * DataRegionGroups for each Database. When set data_region_group_extension_policy=AUTO, this
-   * parameter is the default minimal number of DataRegionGroups for each Database.
+   * DataRegionGroups for each Database. For AUTO and PROACTIVE, this parameter is the default
+   * minimum number of DataRegionGroups and a lower bound for the per-Database maximum. Both
+   * policies grow toward this minimum incrementally, adding at most the number of series slots in
+   * the pending request to satisfy the minimum. PROACTIVE also grows with the number of active
+   * series slots.
    */
   private volatile int defaultDataRegionGroupNumPerDatabase = 2;
 
   /**
-   * The maximum number of DataRegions expected to be managed by each DataNode. Set to 0 means that
-   * each dataNode automatically has the number of CPU cores / 2 regions.
+   * The expected number of DataRegions per DataNode, used to calculate the same per-Database
+   * maximum for AUTO and PROACTIVE. Set to 0 to use half the CPU core count as the expected number
+   * for each DataNode. This is not a hard limit on node or cluster totals: per-Database minimums,
+   * rounding, and existing groups can raise the total above this resource estimate.
    */
   private volatile int dataRegionPerDataNode = 0;
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -125,6 +125,7 @@
 import org.apache.iotdb.confignode.manager.node.NodeMetrics;
 import org.apache.iotdb.confignode.manager.partition.PartitionManager;
 import org.apache.iotdb.confignode.manager.partition.PartitionMetrics;
+import org.apache.iotdb.confignode.manager.partition.RegionGroupExtensionPolicy;
 import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
 import org.apache.iotdb.confignode.manager.pipe.coordinator.PipeManager;
 import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
@@ -1836,6 +1837,10 @@ public TSStatus setConfiguration(TSetConfigurationReq req) {
       long previousHeartbeatIntervalInMs = CONF.getHeartbeatIntervalInMs();
       int previousSchemaRegionPerDataNode = CONF.getSchemaRegionPerDataNode();
       int previousDataRegionPerDataNode = CONF.getDataRegionPerDataNode();
+      RegionGroupExtensionPolicy previousSchemaRegionGroupExtensionPolicy =
+          CONF.getSchemaRegionGroupExtensionPolicy();
+      RegionGroupExtensionPolicy previousDataRegionGroupExtensionPolicy =
+          CONF.getDataRegionGroupExtensionPolicy();
       boolean wasTopologyProbingEnabled = CONF.isEnableTopologyProbing();
       int previousProcedureCompletedCleanInterval = CONF.getProcedureCompletedCleanInterval();
       int previousProcedureCompletedEvictTTL = CONF.getProcedureCompletedEvictTTL();
@@ -1866,8 +1871,11 @@ public TSStatus setConfiguration(TSetConfigurationReq req) {
         return tsStatus;
       }
       handleHeartbeatIntervalHotReload(previousHeartbeatIntervalInMs);
-      handleRegionPerDataNodeHotReload(
-          previousSchemaRegionPerDataNode, previousDataRegionPerDataNode);
+      handleRegionGroupConfigHotReload(
+          previousSchemaRegionPerDataNode,
+          previousDataRegionPerDataNode,
+          previousSchemaRegionGroupExtensionPolicy,
+          previousDataRegionGroupExtensionPolicy);
       handleTopologyProbingHotReload(wasTopologyProbingEnabled);
       handleProcedureCleanerHotReload(
           previousProcedureCompletedCleanInterval, previousProcedureCompletedEvictTTL);
@@ -1914,16 +1922,31 @@ private void handleHeartbeatIntervalHotReload(long previousHeartbeatIntervalInMs
     getRetryFailedTasksThread().reloadHeartbeatInterval();
   }
 
-  private void handleRegionPerDataNodeHotReload(
-      int previousSchemaRegionPerDataNode, int previousDataRegionPerDataNode) {
+  private void handleRegionGroupConfigHotReload(
+      int previousSchemaRegionPerDataNode,
+      int previousDataRegionPerDataNode,
+      RegionGroupExtensionPolicy previousSchemaRegionGroupExtensionPolicy,
+      RegionGroupExtensionPolicy previousDataRegionGroupExtensionPolicy) {
     if (previousSchemaRegionPerDataNode == CONF.getSchemaRegionPerDataNode()
-        && previousDataRegionPerDataNode == CONF.getDataRegionPerDataNode()) {
+        && previousDataRegionPerDataNode == CONF.getDataRegionPerDataNode()
+        && previousSchemaRegionGroupExtensionPolicy == CONF.getSchemaRegionGroupExtensionPolicy()
+        && previousDataRegionGroupExtensionPolicy == CONF.getDataRegionGroupExtensionPolicy()) {
       return;
     }
     if (!getConsensusManager().isLeader()) {
       return;
     }
+    // Leaving CUSTOM must replace its stored cap with the resource-derived cap, even when
+    // per-node quotas are unchanged. This preserves existing groups and skips types still CUSTOM.
     getClusterSchemaManager().adjustMaxRegionGroupNum();
+    if (previousDataRegionGroupExtensionPolicy != RegionGroupExtensionPolicy.PROACTIVE
+        && CONF.getDataRegionGroupExtensionPolicy() == RegionGroupExtensionPolicy.PROACTIVE) {
+      // Existing groups may already meet the new target, so no creation procedure would rebalance
+      // their slot assignments. Refresh the policy table for future time partitions on this leader.
+      getClusterSchemaManager()
+          .getDatabaseNames(null)
+          .forEach(database -> getLoadManager().reBalanceDataPartitionPolicy(database));
+    }
   }
 
   private void handleTopologyProbingHotReload(boolean wasEnabled) {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/partition/DataPartitionPolicyTable.java

@@ -25,6 +25,7 @@
 import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
 import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.i18n.ManagerMessages;
+import org.apache.iotdb.confignode.manager.partition.RegionGroupExtensionPolicy;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -34,6 +35,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.locks.ReentrantLock;
 
 public class DataPartitionPolicyTable {
@@ -111,7 +113,7 @@ public void reBalanceDataPartitionPolicy(List<TConsensusGroupId> dataRegionGroup
       }
       Collections.shuffle(seriesPartitionSlots);
 
-      int mu = SERIES_SLOT_NUM / dataRegionGroups.size();
+      int mu = getRetainedSlotLimit(dataAllotMap, dataRegionGroups.size());
       for (TSeriesPartitionSlot seriesPartitionSlot : seriesPartitionSlots) {
         if (!dataAllotMap.containsKey(seriesPartitionSlot)) {
           // Skip unallocated SeriesPartitionSlot
@@ -141,7 +143,7 @@ public void setDataAllotMap(Map<TSeriesPartitionSlot, TConsensusGroupId> dataAll
     }
     dataAllotTableLock.lock();
     try {
-      int mu = SERIES_SLOT_NUM / seriesPartitionSlotCounter.size();
+      int mu = getRetainedSlotLimit(dataAllotMap, seriesPartitionSlotCounter.size());
       dataAllotMap.forEach(
           (seriesPartitionSlot, regionGroupId) -> {
             if (regionGroupId != null && seriesPartitionSlotCounter.get(regionGroupId) < mu) {
@@ -158,6 +160,18 @@ public void setDataAllotMap(Map<TSeriesPartitionSlot, TConsensusGroupId> dataAll
     }
   }
 
+  private int getRetainedSlotLimit(
+      Map<TSeriesPartitionSlot, TConsensusGroupId> assignments, int regionGroupCount) {
+    if (CONF.getDataRegionGroupExtensionPolicy() == RegionGroupExtensionPolicy.PROACTIVE) {
+      // Empty DataPartition entries have no last group and must not inflate the recovery limit.
+      long activeSlotCount = assignments.values().stream().filter(Objects::nonNull).count();
+      // New groups are created before the pending slots are activated. Keep at least one slot
+      // per existing group so that incremental growth does not discard balanced assignments.
+      return Math.max(1, (int) (activeSlotCount / regionGroupCount));
+    }
+    return SERIES_SLOT_NUM / regionGroupCount;
+  }
+
   public void logDataAllotTable(String database) {
     seriesPartitionSlotCounter
         .keySet()