Pipe: Fixed the audit db filter on config receiver && Added the judgments for table model audit DB && Optimized the logger for receiver status (#17219)

Caideyipi · web-flow · commit 4e7f3e589616 · 2026-02-25T21:44:26.000+08:00
* fix

* table-audit

* defend

* except

* fix
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeLeaderChangeHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeLeaderChangeHandler.java
@@ -22,6 +22,7 @@
 import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
 import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
 import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.schema.table.Audit;
 import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.manager.ConfigManager;
 import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusGroupStatistics;
@@ -78,7 +79,9 @@ public void onConsensusGroupStatisticsChanged(ConsensusGroupStatisticsChangeEven
                   || (!databaseName.equals(SchemaConstant.SYSTEM_DATABASE)
                       && !databaseName.startsWith(SchemaConstant.SYSTEM_DATABASE + ".")
                       && !databaseName.equals(SchemaConstant.AUDIT_DATABASE)
-                      && !databaseName.startsWith(SchemaConstant.AUDIT_DATABASE + "."))) {
+                      && !databaseName.startsWith(SchemaConstant.AUDIT_DATABASE + ".")
+                      && !databaseName.equals(Audit.TABLE_MODEL_AUDIT_DATABASE)
+                      && !databaseName.startsWith(Audit.TABLE_MODEL_AUDIT_DATABASE + "."))) {
                 // null or -1 means empty origin leader
                 final int oldLeaderNodeId = (pair.left == null ? -1 : pair.left.getLeaderId());
                 final int newLeaderNodeId = (pair.right == null ? -1 : pair.right.getLeaderId());
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/ConfigRegionListeningFilter.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/ConfigRegionListeningFilter.java
@@ -24,10 +24,13 @@
 import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
 import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.schema.table.Audit;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
 import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan;
 import org.apache.iotdb.confignode.consensus.request.write.database.DeleteDatabasePlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeCreateTableOrViewPlan;
+import org.apache.iotdb.confignode.consensus.request.write.table.AbstractTablePlan;
 import org.apache.iotdb.confignode.consensus.request.write.template.CommitSetSchemaTemplatePlan;
 import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
 
@@ -252,34 +255,61 @@ public class ConfigRegionListeningFilter {
   static boolean shouldPlanBeListened(final ConfigPhysicalPlan plan) {
     final ConfigPhysicalPlanType type = plan.getType();
 
-    // Do not transfer roll back set template plan
-    if (type.equals(ConfigPhysicalPlanType.CommitSetSchemaTemplate)
-        && ((CommitSetSchemaTemplatePlan) plan).isRollback()) {
-      return false;
+    switch (type) {
+      // Do not transfer roll back set template plan
+      case CommitSetSchemaTemplate:
+        return !((CommitSetSchemaTemplatePlan) plan).isRollback();
+      // system / audit DB
+      case DeleteDatabase:
+        return !((DeleteDatabasePlan) plan).getName().equals(SchemaConstant.AUDIT_DATABASE)
+            && !((DeleteDatabasePlan) plan).getName().equals(SchemaConstant.SYSTEM_DATABASE)
+            && !((DeleteDatabasePlan) plan).getName().equals(Audit.TABLE_MODEL_AUDIT_DATABASE);
+      case CreateDatabase:
+      case AlterDatabase:
+        return !(((DatabaseSchemaPlan) plan)
+                .getSchema()
+                .getName()
+                .equals(SchemaConstant.SYSTEM_DATABASE)
+            && !((DatabaseSchemaPlan) plan)
+                .getSchema()
+                .getName()
+                .equals(SchemaConstant.AUDIT_DATABASE)
+            && !((DatabaseSchemaPlan) plan)
+                .getSchema()
+                .getName()
+                .equals(Audit.TABLE_MODEL_AUDIT_DATABASE));
+      // Table under audit db
+      case PipeCreateTableOrView:
+        return !((PipeCreateTableOrViewPlan) plan)
+            .getDatabase()
+            .equals(Audit.TABLE_MODEL_AUDIT_DATABASE);
+      case CommitCreateTable:
+      case AddTableColumn:
+      case AddViewColumn:
+      case SetTableProperties:
+      case SetViewProperties:
+      case SetTableComment:
+      case SetViewComment:
+      case SetTableColumnComment:
+      case RenameTable:
+      case RenameView:
+      case RenameTableColumn:
+      case RenameViewColumn:
+      case AlterColumnDataType:
+      case CommitDeleteTable:
+      case CommitDeleteView:
+      case CommitDeleteColumn:
+      case CommitDeleteViewColumn:
+      case PipeDeleteDevices:
+        return !((AbstractTablePlan) plan).getDatabase().equals(Audit.TABLE_MODEL_AUDIT_DATABASE);
+      // PipeEnriched & UnsetTemplate are not listened directly,
+      // but their inner plan or converted plan are listened.
+      case PipeEnriched:
+      case UnsetTemplate:
+        return true;
+      default:
+        return OPTION_PLAN_MAP.values().stream().anyMatch(types -> types.contains(type));
     }
-
-    // system / audit DB
-    if (type.equals(ConfigPhysicalPlanType.DeleteDatabase)
-            && (((DeleteDatabasePlan) plan).getName().equals(SchemaConstant.AUDIT_DATABASE)
-                || ((DeleteDatabasePlan) plan).getName().equals(SchemaConstant.SYSTEM_DATABASE))
-        || (type.equals(ConfigPhysicalPlanType.CreateDatabase)
-                || type.equals(ConfigPhysicalPlanType.AlterDatabase))
-            && (((DatabaseSchemaPlan) plan)
-                    .getSchema()
-                    .getName()
-                    .equals(SchemaConstant.SYSTEM_DATABASE)
-                || ((DatabaseSchemaPlan) plan)
-                    .getSchema()
-                    .getName()
-                    .equals(SchemaConstant.AUDIT_DATABASE))) {
-      return false;
-    }
-
-    // PipeEnriched & UnsetTemplate are not listened directly,
-    // but their inner plan or converted plan are listened.
-    return type.equals(ConfigPhysicalPlanType.PipeEnriched)
-        || type.equals(ConfigPhysicalPlanType.UnsetTemplate)
-        || OPTION_PLAN_MAP.values().stream().anyMatch(types -> types.contains(type));
   }
 
   public static Set<ConfigPhysicalPlanType> parseListeningPlanTypeSet(
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/IoTDBConfigRegionSource.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/IoTDBConfigRegionSource.java
@@ -344,7 +344,8 @@ public static boolean isTypeListened(
       final IoTDBTreePatternOperations treePattern,
       final TablePattern tablePattern) {
     final Boolean isTableDatabasePlan = isTableDatabasePlan(plan);
-    return listenedTypeSet.contains(plan.getType())
+    return ConfigRegionListeningFilter.shouldPlanBeListened(plan)
+        && listenedTypeSet.contains(plan.getType())
         && (Objects.isNull(isTableDatabasePlan)
             || Boolean.TRUE.equals(isTableDatabasePlan)
                 && tablePattern.isTableModelDataAllowedToBeCaptured()
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
@@ -27,6 +27,7 @@
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.path.PathPatternTree;
 import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.schema.table.Audit;
 import org.apache.iotdb.commons.schema.table.NonCommittableTsTable;
 import org.apache.iotdb.commons.schema.table.TableNodeStatus;
 import org.apache.iotdb.commons.schema.table.TreeViewSchema;
@@ -180,7 +181,8 @@ public TSStatus setDatabase(
       clusterSchemaInfo.isDatabaseNameValid(
           schema.getName(), schema.isSetIsTableModel() && schema.isIsTableModel());
       if (!schema.getName().equals(SchemaConstant.SYSTEM_DATABASE)
-          && !schema.getName().equals(SchemaConstant.AUDIT_DATABASE)) {
+          && !schema.getName().equals(SchemaConstant.AUDIT_DATABASE)
+          && !schema.getName().equals(Audit.TABLE_MODEL_AUDIT_DATABASE)) {
         clusterSchemaInfo.checkDatabaseLimit();
       }
       // Cache DatabaseSchema
@@ -488,7 +490,8 @@ public synchronized void adjustMaxRegionGroupNum() {
     for (final TDatabaseSchema databaseSchema : databaseSchemaMap.values()) {
       if (!isDatabaseExist(databaseSchema.getName())
           || databaseSchema.getName().equals(SchemaConstant.SYSTEM_DATABASE)
-          || databaseSchema.getName().equals(SchemaConstant.AUDIT_DATABASE)) {
+          || databaseSchema.getName().equals(SchemaConstant.AUDIT_DATABASE)
+          || databaseSchema.getName().equals(Audit.TABLE_MODEL_AUDIT_DATABASE)) {
         // filter the pre deleted database and the system database
         databaseNum--;
       }
@@ -498,7 +501,8 @@ public synchronized void adjustMaxRegionGroupNum() {
         new AdjustMaxRegionGroupNumPlan();
     for (final TDatabaseSchema databaseSchema : databaseSchemaMap.values()) {
       if (databaseSchema.getName().equals(SchemaConstant.SYSTEM_DATABASE)
-          || databaseSchema.getName().equals(SchemaConstant.AUDIT_DATABASE)) {
+          || databaseSchema.getName().equals(SchemaConstant.AUDIT_DATABASE)
+          || databaseSchema.getName().equals(Audit.TABLE_MODEL_AUDIT_DATABASE)) {
         // filter the system database
         continue;
       }
@@ -828,7 +832,9 @@ public static TSStatus enrichDatabaseSchemaWithDefaultProperties(
     TSStatus errorResp = null;
     final boolean isSystemDatabase =
         databaseSchema.getName().equals(SchemaConstant.SYSTEM_DATABASE);
-    final boolean isAuditDatabase = databaseSchema.getName().equals(SchemaConstant.AUDIT_DATABASE);
+    final boolean isAuditDatabase =
+        databaseSchema.getName().equals(SchemaConstant.AUDIT_DATABASE)
+            || databaseSchema.getName().equals(Audit.TABLE_MODEL_AUDIT_DATABASE);
 
     if (databaseSchema.getTTL() < 0) {
       errorResp =
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/CNPhysicalPlanGenerator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/CNPhysicalPlanGenerator.java
@@ -26,6 +26,7 @@
 import org.apache.iotdb.commons.schema.SchemaConstant;
 import org.apache.iotdb.commons.schema.node.role.IDatabaseMNode;
 import org.apache.iotdb.commons.schema.node.utils.IMNodeFactory;
+import org.apache.iotdb.commons.schema.table.Audit;
 import org.apache.iotdb.commons.schema.table.TsTable;
 import org.apache.iotdb.commons.schema.template.Template;
 import org.apache.iotdb.commons.utils.AuthUtils;
@@ -481,8 +482,10 @@ private void generateDatabasePhysicalPlan() {
             }
             stack.push(new Pair<>(databaseMNode, true));
             name = databaseMNode.getName();
-            for (final TsTable table : tableSet) {
-              planDeque.add(new PipeCreateTableOrViewPlan(name, table));
+            if (!name.equals(Audit.TABLE_MODEL_AUDIT_DATABASE)) {
+              for (final TsTable table : tableSet) {
+                planDeque.add(new PipeCreateTableOrViewPlan(name, table));
+              }
             }
             tableSet.clear();
             break;
@@ -552,7 +555,8 @@ private IConfigMNode deserializeDatabaseMNode(final InputStream inputStream) thr
 
     final TDatabaseSchema schema = databaseMNode.getAsMNode().getDatabaseSchema();
     if (!schema.getName().equals(SchemaConstant.AUDIT_DATABASE)
-        && !schema.getName().equals(SchemaConstant.SYSTEM_DATABASE)) {
+        && !schema.getName().equals(SchemaConstant.SYSTEM_DATABASE)
+        && !schema.getName().equals(Audit.TABLE_MODEL_AUDIT_DATABASE)) {
       final DatabaseSchemaPlan createDBPlan =
           new DatabaseSchemaPlan(ConfigPhysicalPlanType.CreateDatabase, schema);
       planDeque.add(createDBPlan);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
@@ -30,6 +30,7 @@
 import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
 import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
 import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.schema.table.Audit;
 import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.task.AlterPipePlanV2;
@@ -191,6 +192,8 @@ public void executeFromCalculateInfoForTask(final ConfigNodeProcedureEnv env) {
                     && !databaseName.startsWith(SchemaConstant.SYSTEM_DATABASE + ".")
                     && !databaseName.equals(SchemaConstant.AUDIT_DATABASE)
                     && !databaseName.startsWith(SchemaConstant.AUDIT_DATABASE + ".")
+                    && !databaseName.equals(Audit.TABLE_MODEL_AUDIT_DATABASE)
+                    && !databaseName.startsWith(Audit.TABLE_MODEL_AUDIT_DATABASE + ".")
                     && !Objects.isNull(currentPipeTaskMeta)
                     && !(PipeTaskAgent.isHistoryOnlyPipe(
                             currentPipeStaticMeta.getSourceParameters())
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
@@ -34,6 +34,7 @@
 import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
 import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
 import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.schema.table.Audit;
 import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2;
@@ -313,7 +314,9 @@ public void executeFromCalculateInfoForTask(final ConfigNodeProcedureEnv env) {
                     && !databaseName.equals(SchemaConstant.SYSTEM_DATABASE)
                     && !databaseName.startsWith(SchemaConstant.SYSTEM_DATABASE + ".")
                     && !databaseName.equals(SchemaConstant.AUDIT_DATABASE)
-                    && !databaseName.startsWith(SchemaConstant.AUDIT_DATABASE + ".")) {
+                    && !databaseName.startsWith(SchemaConstant.AUDIT_DATABASE + ".")
+                    && !databaseName.equals(Audit.TABLE_MODEL_AUDIT_DATABASE)
+                    && !databaseName.startsWith(Audit.TABLE_MODEL_AUDIT_DATABASE + ".")) {
                   // Pipe only collect user's data, filter out metric database here.
                   consensusGroupIdToTaskMetaMap.put(
                       regionGroupId.getId(),
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
@@ -42,6 +42,7 @@
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
 import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.schema.table.Audit;
 import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
 import org.apache.iotdb.commons.utils.PathUtils;
 import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
@@ -300,6 +301,8 @@ private void createDatabaseAndUpdateCache(
             databaseNamesNeedCreated.add(SchemaConstant.SYSTEM_DATABASE);
           } else if (PathUtils.isStartWith(deviceID, SchemaConstant.AUDIT_DATABASE)) {
             databaseNamesNeedCreated.add(SchemaConstant.AUDIT_DATABASE);
+          } else if (PathUtils.isStartWith(deviceID, Audit.TABLE_MODEL_AUDIT_DATABASE)) {
+            databaseNamesNeedCreated.add(Audit.TABLE_MODEL_AUDIT_DATABASE);
           } else {
             final PartialPath databaseNameNeedCreated =
                 MetaUtils.getDatabasePathByLevel(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
@@ -27,6 +27,7 @@
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.path.PathPatternTree;
 import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.schema.table.Audit;
 import org.apache.iotdb.commons.utils.PathUtils;
 import org.apache.iotdb.commons.utils.TimePartitionUtils;
 import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
@@ -655,12 +656,13 @@ public List<PlanNode> visitSchemaQueryMerge(
           .getSchemaPartitionInfo()
           .getSchemaPartitionMap()
           .forEach(
-              (storageGroup, deviceGroup) -> {
-                if (storageGroup.equals(SchemaConstant.SYSTEM_DATABASE)) {
+              (database, deviceGroup) -> {
+                if (database.equals(SchemaConstant.SYSTEM_DATABASE)) {
                   deviceGroup.forEach(
                       (deviceGroupId, schemaRegionReplicaSet) ->
                           regionsOfSystemDatabase.add(schemaRegionReplicaSet));
-                } else if (storageGroup.equals(SchemaConstant.AUDIT_DATABASE)) {
+                } else if (database.equals(SchemaConstant.AUDIT_DATABASE)
+                    || database.equals(Audit.TABLE_MODEL_AUDIT_DATABASE)) {
                   deviceGroup.forEach(
                       (deviceGroupId, schemaRegionReplicaSet) ->
                           regionsOfAuditDatabase.add(schemaRegionReplicaSet));
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -31,6 +31,7 @@
 import org.apache.iotdb.commons.path.IFullPath;
 import org.apache.iotdb.commons.path.MeasurementPath;
 import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.schema.table.Audit;
 import org.apache.iotdb.commons.schema.table.InformationSchema;
 import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil;
 import org.apache.iotdb.commons.schema.table.TsTable;
@@ -3799,6 +3800,9 @@ public static Optional<String> getNonSystemDatabaseName(String databaseName) {
     if (databaseName.startsWith(SchemaConstant.AUDIT_DATABASE)) {
       return Optional.empty();
     }
+    if (databaseName.startsWith(Audit.TABLE_MODEL_AUDIT_DATABASE)) {
+      return Optional.empty();
+    }
     int lastIndex = databaseName.lastIndexOf("-");
     if (lastIndex == -1) {
       lastIndex = databaseName.length();
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/PipeReceiverStatusHandler.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/PipeReceiverStatusHandler.java
@@ -135,8 +135,9 @@ public void handle(
         {
           PipeLogger.log(
               LOGGER::info,
-              "Temporary unavailable exception: will retry forever. status: %s",
-              status);
+              "Temporary unavailable exception: will retry forever. status: %s, message: %s",
+              status,
+              exceptionMessage);
           throw new PipeRuntimeSinkNonReportTimeConfigurableException(
               exceptionMessage, Long.MAX_VALUE);
         }
@@ -243,9 +244,10 @@ private synchronized void handleOtherExceptions(
     if (retryMaxMillisWhenOtherExceptionsOccur == Long.MAX_VALUE) {
       PipeLogger.log(
           LOGGER::warn,
-          "%s: will retry forever. status: %s",
+          "%s: will retry forever. status: %s, message: %s",
           getNoPermission(noPermission),
-          status);
+          status,
+          exceptionMessage);
     } else {
       LOGGER.warn(
           "{}: will retry for at least {} seconds. status: {}",
