Fixed the schema quota that counts audit DB (#17548)

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java

@@ -27,6 +27,7 @@
 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.SchemaConstant;
 import org.apache.iotdb.commons.utils.FileUtils;
 import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.consensus.ConsensusFactory;
@@ -44,6 +45,9 @@
 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.schemaregion.read.req.SchemaRegionReadPlanFactory;
+import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.ITimeSeriesSchemaInfo;
+import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.reader.ISchemaReader;
 import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager;
 import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatReq;
 import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatResp;
@@ -54,6 +58,7 @@
 import java.io.File;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -74,6 +79,9 @@ public class SchemaEngine {
 
   private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
 
+  private static final PartialPath AUDIT_LOG_PATH_PATTERN =
+      new PartialPath(new String[] {"root", "__system", "audit", "**"});
+
   private final SchemaRegionLoader schemaRegionLoader;
 
   @SuppressWarnings("java:S3077")
@@ -397,11 +405,52 @@ public Map<Integer, Long> countTimeSeriesNumBySchemaRegion(final List<Integer> s
         .forEach(
             entry ->
                 timeSeriesNum.put(
-                    entry.getKey().getId(),
-                    entry.getValue().getSchemaRegionStatistics().getSeriesNumber(false)));
+                    entry.getKey().getId(), getTimeSeriesNumberForQuota(entry.getValue())));
     return timeSeriesNum;
   }
 
+  /**
+   * Returns the series count used by schema quota. Audit series are stored below {@code
+   * root.__system.audit} in dev/1.3, while other internal series may share the same system database
+   * and must remain counted.
+   */
+  private long getTimeSeriesNumberForQuota(final ISchemaRegion schemaRegion) {
+    final long totalSeriesNumber = schemaRegion.getSchemaRegionStatistics().getSeriesNumber(false);
+    if (!SchemaConstant.SYSTEM_DATABASE.equals(schemaRegion.getDatabaseFullPath())) {
+      return totalSeriesNumber;
+    }
+
+    long auditSeriesNumber = 0;
+    try {
+      try (ISchemaReader<ITimeSeriesSchemaInfo> reader =
+          schemaRegion.getTimeSeriesReader(
+              SchemaRegionReadPlanFactory.getShowTimeSeriesPlan(
+                  AUDIT_LOG_PATH_PATTERN,
+                  Collections.emptyMap(),
+                  -1,
+                  0,
+                  false,
+                  null,
+                  false,
+                  SchemaConstant.ALL_MATCH_SCOPE))) {
+        while (reader.hasNext()) {
+          if (!reader.next().isLogicalView()) {
+            auditSeriesNumber++;
+          }
+        }
+        if (!reader.isSuccess()) {
+          logger.warn("Failed to count audit time series for schema quota", reader.getFailure());
+          return totalSeriesNumber;
+        }
+      }
+    } catch (Exception e) {
+      // Keep the heartbeat available if the audit-only scan fails; the next heartbeat retries it.
+      logger.warn("Failed to exclude audit time series from schema quota count", e);
+      return totalSeriesNumber;
+    }
+    return Math.max(0, totalSeriesNumber - auditSeriesNumber);
+  }
+
   /**
    * Update total count in schema quota manager and generate local count map response. If limit is
    * not -1 and deviceNumMap/timeSeriesNumMap is null, fill deviceNumMap/timeSeriesNumMap of the
@@ -448,9 +497,7 @@ public void updateAndFillSchemaCountMap(TDataNodeHeartbeatReq req, TDataNodeHear
                   tmp.put(
                       consensusGroupId.getId(),
                       Optional.ofNullable(schemaRegionMap.get(consensusGroupId))
-                          .map(
-                              schemaRegion ->
-                                  schemaRegion.getSchemaRegionStatistics().getSeriesNumber(false))
+                          .map(this::getTimeSeriesNumberForQuota)
                           .orElse(0L)));
     }
   }