[Subscription] Isolate tree and table consensus topics

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java

@@ -59,6 +59,7 @@
 import java.util.function.Predicate;
 
 import static org.apache.iotdb.commons.schema.table.Audit.isAuditDatabase;
+import static org.apache.iotdb.commons.utils.PathUtils.isTableModelDatabase;
 
 /**
  * Handles setup and teardown of consensus-based subscription queues on DataNode.
@@ -184,8 +185,7 @@ private static void onNewRegionCreated(
           final String dbRaw = dataRegion.getDatabaseName();
           final String dbTableModel = dbRaw.startsWith("root.") ? dbRaw.substring(5) : dbRaw;
 
-          // For table topics, skip if this region's database doesn't match the topic filter.
-          if (!matchesTopicDatabase(topicConfig, dbTableModel)) {
+          if (!matchesTopicDataRegion(dbRaw, topicConfig, isTableModel)) {
             continue;
           }
 
@@ -464,10 +464,11 @@ interface ConsensusTopicSetup {
    * <p>This method discovers local DataRegion consensus groups that match the topic filter and
    * binds one consensus subscription queue to each matching region.
    *
-   * <p>For table-model topics, only regions whose database matches the topic's {@code DATABASE_KEY}
-   * filter are bound. For tree-model topics, all local data regions are candidates. Additionally,
-   * the {@link #onNewRegionCreated} callback ensures that regions created after this method runs
-   * are also automatically bound.
+   * <p>Only regions whose database names identify the same data model as the consumer group are
+   * candidates. A database name with the {@code root.} prefix identifies a tree-model region. For
+   * table-model topics, the candidate region's database must also match the topic's {@code
+   * DATABASE_KEY} filter. Additionally, the {@link #onNewRegionCreated} callback ensures that
+   * regions created after this method runs are also automatically bound.
    */
   private static void setupConsensusQueueForTopic(
       final String consumerGroupId,
@@ -527,16 +528,19 @@ private static void setupConsensusQueueForTopic(
       }
       final String dbRaw = dataRegion.getDatabaseName();
       final String dbTableModel = dbRaw.startsWith("root.") ? dbRaw.substring(5) : dbRaw;
+      final boolean dataRegionIsTableModel = isTableModelDatabase(dbRaw);
 
-      if (!matchesTopicDatabase(topicConfig, dbTableModel)) {
-        LOGGER.info(
-            DataNodePipeMessages
-                .PIPE_LOG_SKIPPING_REGION_DATABASE_FOR_TABLE_TOPIC_DATABASE_KEY_2DA27A84,
-            groupId,
-            dbTableModel,
-            topicName,
-            topicConfig.getStringOrDefault(
-                TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE));
+      if (!matchesTopicDataRegion(dbRaw, topicConfig, isTableModel)) {
+        if (isTableModel && dataRegionIsTableModel && topicConfig.isTableTopic()) {
+          LOGGER.info(
+              DataNodePipeMessages
+                  .PIPE_LOG_SKIPPING_REGION_DATABASE_FOR_TABLE_TOPIC_DATABASE_KEY_2DA27A84,
+              groupId,
+              dbTableModel,
+              topicName,
+              topicConfig.getStringOrDefault(
+                  TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE));
+        }
         continue;
       }
 
@@ -654,6 +658,19 @@ static boolean matchesTopicDatabase(
             || buildTablePattern(topicConfig).matchesDatabase(actualDatabaseName));
   }
 
+  static boolean matchesTopicDataRegion(
+      final String databaseName, final TopicConfig topicConfig, final boolean isTableModel) {
+    if (databaseName == null
+        || topicConfig == null
+        || isTableModelDatabase(databaseName) != isTableModel
+        || topicConfig.isTableTopic() != isTableModel) {
+      return false;
+    }
+    final String actualDatabaseName =
+        databaseName.startsWith("root.") ? databaseName.substring(5) : databaseName;
+    return matchesTopicDatabase(topicConfig, actualDatabaseName);
+  }
+
   private static TablePattern buildTablePattern(final TopicConfig topicConfig) {
     return new TablePattern(
         true,

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java

@@ -156,6 +156,41 @@ public void testAuditDatabaseNeverMatchesTopic() {
     assertTrue(ConsensusSubscriptionSetupHandler.matchesTopicDatabase(treeTopicConfig, "user_db"));
   }
 
+  @Test
+  public void testTopicDataRegionModelIsolation() {
+    final TopicConfig treeTopicConfig = new TopicConfig(Collections.emptyMap());
+    final Map<String, String> tableTopicAttributes = new HashMap<>();
+    tableTopicAttributes.put(
+        SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TABLE_VALUE);
+    tableTopicAttributes.put(TopicConstant.DATABASE_KEY, "table_db");
+    final TopicConfig tableTopicConfig = new TopicConfig(tableTopicAttributes);
+
+    assertTrue(
+        ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+            "root.tree_db", treeTopicConfig, false));
+    assertFalse(
+        ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+            "table_db", treeTopicConfig, false));
+    assertFalse(
+        ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+            "table_db", treeTopicConfig, true));
+    assertFalse(
+        ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+            "root.table_db", tableTopicConfig, true));
+    assertTrue(
+        ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+            "table_db", tableTopicConfig, true));
+    assertFalse(
+        ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+            "other_table_db", tableTopicConfig, true));
+    assertFalse(
+        ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+            "table_db", tableTopicConfig, false));
+    assertFalse(
+        ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+            "root.table_db", tableTopicConfig, false));
+  }
+
   private static void failOnSecondTopic(
       final String topicName, final Set<String> attemptedTopicNames) {
     attemptedTopicNames.add(topicName);