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 index c8faf2f059a4..dad19bbdfde2 100644 --- 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,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 countTimeSeriesNumBySchemaRegion(final List 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 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))); } }