Add proactive RegionGroup allocation and policy hot reload (#18692)

Author: d-wang-commit

integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBProactiveRegionGroupDistributionIT.java

@@ -1047,7 +1047,13 @@ public void testShowDevicesWithLimit() throws SQLException {
     try (Connection connection = EnvFactory.getEnv().getConnection();
         Statement statement = connection.createStatement()) {
       try {
-        List<String> exps = Arrays.asList("root.sg1.d0,false", "root.sg1.d1,false");
+        List<String> exps =
+            new ArrayList<>(
+                Arrays.asList(
+                    "root.sg1.d0,false",
+                    "root.sg1.d1,false",
+                    "root.sg1.d2,false",
+                    "root.sg1.d3,false"));
 
         statement.execute("INSERT INTO root.sg1.d0(timestamp, s1) VALUES (5, 5)");
         statement.execute("INSERT INTO root.sg1.d1(timestamp, s2) VALUES (5, 5)");
@@ -1057,8 +1063,8 @@ public void testShowDevicesWithLimit() throws SQLException {
         int count = 0;
         try (ResultSet resultSet = statement.executeQuery("show devices limit 2")) {
           while (resultSet.next()) {
-            Assert.assertEquals(
-                exps.get(count), resultSet.getString(1) + "," + resultSet.getString(2));
+            String device = resultSet.getString(1) + "," + resultSet.getString(2);
+            Assert.assertTrue(device, exps.remove(device));
             ++count;
           }
         }

integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBProactiveRegionGroupExtensionIT.java

@@ -1380,14 +1380,22 @@ public void maxByMinByExtremeTest() {
     tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
 
     expectedHeader = buildHeaders(30);
+    // Exclude l4 so both blob extrema are unique; tied rows may arrive in either order.
     sql =
-        "select max_by(time,blob),max_by(device,blob),max_by(level,blob),max_by(attr1,blob),max_by(attr2,blob),max_by(num,blob),max_by(bignum,blob),max_by(floatnum,blob),max_by(str,blob),max_by(bool,blob),max_by(date,blob),max_by(blob,blob),max_by(ts,blob),max_by(stringv,blob),max_by(doubleNum,blob),min_by(time,blob),min_by(device,blob),min_by(level,blob),min_by(attr1,blob),min_by(attr2,blob),min_by(num,blob),min_by(bignum,blob),min_by(floatnum,blob),min_by(str,blob),min_by(bool,blob),min_by(date,blob),min_by(blob,blob),min_by(ts,blob),min_by(stringv,blob),min_by(doubleNum,blob) from table0";
+        "select max_by(time,blob),max_by(device,blob),max_by(level,blob),max_by(attr1,blob),max_by(attr2,blob),max_by(num,blob),max_by(bignum,blob),max_by(floatnum,blob),max_by(str,blob),max_by(bool,blob),max_by(date,blob),max_by(blob,blob),max_by(ts,blob),max_by(stringv,blob),max_by(doubleNum,blob),min_by(time,blob),min_by(device,blob),min_by(level,blob),min_by(attr1,blob),min_by(attr2,blob),min_by(num,blob),min_by(bignum,blob),min_by(floatnum,blob),min_by(str,blob),min_by(bool,blob),min_by(date,blob),min_by(blob,blob),min_by(ts,blob),min_by(stringv,blob),min_by(doubleNum,blob) from table0 where level != 'l4'";
     retArray =
         new String[] {
           "1971-01-01T00:00:10.000Z,d1,l5,null,null,7,2147983648,213.112,lemon,true,null,0x108dcd63,2024-09-25T06:15:35.000Z,null,null,1970-01-01T00:00:00.020Z,d1,l2,yy,zz,2,2147483648,434.12,pineapple,true,null,0x108dcd62,2024-09-24T06:15:35.000Z,null,6666.8,",
         };
     tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
 
+    // Retain the unfiltered ties: either associated value is valid, but the extrema are fixed.
+    expectedHeader = buildHeaders(4);
+    sql =
+        "select max_by(num,blob) in (7,9),min_by(num,blob) in (2,5),max_by(blob,blob),min_by(blob,blob) from table0";
+    retArray = new String[] {"true,true,0x108dcd63,0x108dcd62,"};
+    tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+
     expectedHeader = buildHeaders(3);
     sql = "select extreme(num),extreme(bignum),extreme(floatnum) from table0";
     retArray = new String[] {"15,3147483648,4654.231,"};

integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSimpleQueryIT.java

@@ -259,7 +259,7 @@ public void testPartitionByAndOrderByWithNulls() {
           "2021-01-01T09:20:00.000Z,null,null,2,",
         };
     tableResultSetEqualTest(
-        "SELECT *, count(value) OVER (PARTITION BY device ORDER BY value) AS cnt FROM demo2 ORDER BY device",
+        "SELECT *, count(value) OVER (PARTITION BY device ORDER BY value) AS cnt FROM demo2 ORDER BY device,value,time",
         expectedHeader,
         retArray,
         DATABASE_NAME);

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiTAGsWithAttributesTableIT.java

@@ -175,15 +175,16 @@ public void test() throws Exception {
           "select current from view1 where time > 1",
           "select current from table1 where time > 1" + (aligned ? "" : " and current is not null"),
           true);
+      // FILL runs before an outer ORDER BY, so give both sources the same ordered input.
       compareQueryResults(
           session,
-          "select * from view1 fill method linear",
-          "select * from table1 fill method linear",
+          "select * from (select * from view1 order by battery, time) fill method linear",
+          "select * from (select * from table1 order by battery, time) fill method linear",
           true);
       compareQueryResults(
           session,
-          "select * from view1 fill method previous",
-          "select * from table1 fill method previous",
+          "select * from (select * from view1 order by battery, time) fill method previous",
+          "select * from (select * from table1 order by battery, time) fill method previous",
           true);
       compareQueryResults(
           session,

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowFunctionIT.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;
 

integration-test/src/test/java/org/apache/iotdb/relational/it/query/view/recent/IoTDBTableViewQueryIT.java

@@ -272,11 +272,10 @@ public void updateSpaceQuotaUsage() {
           .forEach(
               schemaRegionId -> {
                 if (deviceNum.containsKey(schemaRegionId)) {
-                  deviceCount.addAndGet(deviceCount.get() + deviceNum.get(schemaRegionId));
+                  deviceCount.addAndGet(deviceNum.get(schemaRegionId));
                 }
                 if (timeSeriesNum.containsKey(schemaRegionId)) {
-                  timeSeriesCount.addAndGet(
-                      timeSeriesCount.get() + timeSeriesNum.get(schemaRegionId));
+                  timeSeriesCount.addAndGet(timeSeriesNum.get(schemaRegionId));
                 }
               });
       quotaInfo.getSpaceQuotaUsage().get(entry.getKey()).setDeviceNum(deviceCount.get());

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.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/ClusterQuotaManager.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()