Implement schema quota for activation in table model

apache · Nov 18, 2024 · 5fd2c9c · 5fd2c9c
1 parent f9f2385
commit 5fd2c9c
Show file tree

Hide file tree

Showing 5 changed files with 58 additions and 9 deletions.
diff --git a/...ignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java b/...ignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
@@ -1139,15 +1139,16 @@ public Pair<Long, Long> getSchemaQuotaRemain() {
     }
   }
 
-  public void updateTimeSeriesUsage(Map<Integer, Long> seriesUsage) {
+  public void updateTimeSeriesUsage(final Map<Integer, Long> seriesUsage) {
     schemaQuotaStatistics.updateTimeSeriesUsage(seriesUsage);
   }
 
-  public void updateDeviceUsage(Map<Integer, Long> deviceUsage) {
+  public void updateDeviceUsage(final Map<Integer, Long> deviceUsage) {
     schemaQuotaStatistics.updateDeviceUsage(deviceUsage);
   }
 
-  public void updateSchemaQuotaConfiguration(long seriesThreshold, long deviceThreshold) {
+  public void updateSchemaQuotaConfiguration(
+      final long seriesThreshold, final long deviceThreshold) {
     schemaQuotaStatistics.setDeviceThreshold(deviceThreshold);
     schemaQuotaStatistics.setSeriesThreshold(seriesThreshold);
   }

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
@@ -27,7 +27,9 @@
 import org.apache.iotdb.commons.exception.IllegalPathException;
 import org.apache.iotdb.commons.exception.MetadataException;
 import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.schema.table.TsTable;
 import org.apache.iotdb.commons.utils.FileUtils;
+import org.apache.iotdb.commons.utils.PathUtils;
 import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -43,6 +45,7 @@
 import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegionParams;
 import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionLoader;
 import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionParams;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
 import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager;
 import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatReq;
 import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatResp;
@@ -56,6 +59,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
@@ -359,8 +363,8 @@ public int getSchemaRegionNumber() {
     return schemaRegionMap == null ? 0 : schemaRegionMap.size();
   }
 
-  public Map<Integer, Long> countDeviceNumBySchemaRegion(List<Integer> schemaIds) {
-    Map<Integer, Long> deviceNum = new HashMap<>();
+  public Map<Integer, Long> countDeviceNumBySchemaRegion(final List<Integer> schemaIds) {
+    final Map<Integer, Long> deviceNum = new HashMap<>();
 
     schemaRegionMap.entrySet().stream()
         .filter(
@@ -375,8 +379,8 @@ public Map<Integer, Long> countDeviceNumBySchemaRegion(List<Integer> schemaIds)
     return deviceNum;
   }
 
-  public Map<Integer, Long> countTimeSeriesNumBySchemaRegion(List<Integer> schemaIds) {
-    Map<Integer, Long> timeSeriesNum = new HashMap<>();
+  public Map<Integer, Long> countTimeSeriesNumBySchemaRegion(final List<Integer> schemaIds) {
+    final Map<Integer, Long> timeSeriesNum = new HashMap<>();
     schemaRegionMap.entrySet().stream()
         .filter(
             entry ->
@@ -386,7 +390,26 @@ public Map<Integer, Long> countTimeSeriesNumBySchemaRegion(List<Integer> schemaI
             entry ->
                 timeSeriesNum.put(
                     entry.getKey().getId(),
-                    entry.getValue().getSchemaRegionStatistics().getSeriesNumber(false)));
+                    entry.getValue().getSchemaRegionStatistics().getSeriesNumber(false)
+                        + entry
+                            .getValue()
+                            .getSchemaRegionStatistics()
+                            .getTable2DevicesNumMap()
+                            .entrySet()
+                            .stream()
+                            .map(
+                                tableEntry -> {
+                                  final TsTable table =
+                                      DataNodeTableCache.getInstance()
+                                          .getTable(
+                                              PathUtils.unQualifyDatabaseName(
+                                                  entry.getValue().getDatabaseFullPath()),
+                                              tableEntry.getKey());
+                                  return Objects.nonNull(table)
+                                      ? table.getMeasurementNum() * tableEntry.getValue()
+                                      : 0;
+                                })
+                            .reduce(0L, Long::sum)));
     return timeSeriesNum;
   }
 

diff --git a/...tanode/src/main/java/org/apache/iotdb/db/schemaengine/rescon/ISchemaRegionStatistics.java b/...tanode/src/main/java/org/apache/iotdb/db/schemaengine/rescon/ISchemaRegionStatistics.java
@@ -19,6 +19,8 @@
 
 package org.apache.iotdb.db.schemaengine.rescon;
 
+import java.util.Map;
+
 public interface ISchemaRegionStatistics {
 
   boolean isAllowToCreateNewSeries();
@@ -27,10 +29,12 @@ public interface ISchemaRegionStatistics {
 
   int getSchemaRegionId();
 
-  long getSeriesNumber(boolean includeView);
+  long getSeriesNumber(final boolean includeView);
 
   long getDevicesNumber();
 
+  Map<String, Long> getTable2DevicesNumMap();
+
   long getTableDevicesNumber(final String table);
 
   int getTemplateActivatedNumber();

diff --git a/...node/src/main/java/org/apache/iotdb/db/schemaengine/rescon/MemSchemaRegionStatistics.java b/...node/src/main/java/org/apache/iotdb/db/schemaengine/rescon/MemSchemaRegionStatistics.java
@@ -22,6 +22,7 @@
 import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager;
 import org.apache.iotdb.db.schemaengine.template.Template;
 
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -97,6 +98,11 @@ public long getDevicesNumber() {
     return devicesNumber.get();
   }
 
+  @Override
+  public Map<String, Long> getTable2DevicesNumMap() {
+    return tableDeviceNumber;
+  }
+
   @Override
   public long getTableDevicesNumber(final String table) {
     final Long deviceNumber = tableDeviceNumber.get(table);

diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTable.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTable.java
@@ -77,6 +77,7 @@ public class TsTable {
   private Map<String, String> props = null;
 
   private transient int idNums = 0;
+  private transient int measurementNum = 0;
 
   public TsTable(final String tableName) {
     this.tableName = tableName;
@@ -112,6 +113,8 @@ public void addColumnSchema(final TsTableColumnSchema columnSchema) {
       if (columnSchema.getColumnCategory().equals(TsTableColumnCategory.ID)) {
         idNums++;
         idColumnIndexMap.put(columnSchema.getColumnName(), idNums - 1);
+      } else if (columnSchema.getColumnCategory().equals(TsTableColumnCategory.MEASUREMENT)) {
+        measurementNum++;
       }
     } finally {
       readWriteLock.writeLock().unlock();
@@ -140,6 +143,9 @@ public void removeColumnSchema(final String columnName) {
         throw new SchemaExecutionException("Cannot remove an id column: " + columnName);
       } else if (columnSchema != null) {
         columnSchemaMap.remove(columnName);
+        if (columnSchema.getColumnCategory().equals(TsTableColumnCategory.MEASUREMENT)) {
+          measurementNum--;
+        }
       }
     } finally {
       readWriteLock.writeLock().unlock();
@@ -164,6 +170,15 @@ public int getIdNums() {
     }
   }
 
+  public int getMeasurementNum() {
+    readWriteLock.readLock().lock();
+    try {
+      return measurementNum;
+    } finally {
+      readWriteLock.readLock().unlock();
+    }
+  }
+
   public List<TsTableColumnSchema> getColumnList() {
     readWriteLock.readLock().lock();
     try {