From 16873f7a8bbed14c82ce2518380df0e797249ccc Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 13 Aug 2026 17:14:23 +0800 Subject: [PATCH 1/2] [Subscription] Isolate tree and table consensus topics --- .../ConsensusLogToTabletConverter.java | 42 ++++++++++++-- .../ConsensusSubscriptionSetupHandler.java | 46 ++++++++++----- .../ConsensusLogToTabletConverterTest.java | 56 +++++++++++++++++++ ...ConsensusSubscriptionSetupHandlerTest.java | 44 +++++++++++++++ 4 files changed, 169 insertions(+), 19 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java index 0b27de77b123..3b7d96156326 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java @@ -78,6 +78,14 @@ public class ConsensusLogToTabletConverter { */ private final String databaseName; + private boolean isTreeModelConverter() { + return treePattern != null && tablePattern == null; + } + + private boolean isTableModelConverter() { + return tablePattern != null; + } + public ConsensusLogToTabletConverter( final TreePattern treePattern, final TablePattern tablePattern, @@ -239,6 +247,10 @@ public List convert(final InsertNode insertNode) { // ======================== Tree Model Conversion ======================== private List convertInsertRowNode(final InsertRowNode node) { + if (isTableModelConverter()) { + return Collections.emptyList(); + } + final IDeviceID deviceId = node.getDeviceID(); // Device-level path filtering @@ -288,7 +300,12 @@ private List convertInsertRowNode(final InsertRowNode node) { private List convertInsertTabletNode(final InsertTabletNode node) { if (node instanceof RelationalInsertTabletNode) { - return convertRelationalInsertTabletNode((RelationalInsertTabletNode) node); + return !isTreeModelConverter() + ? convertRelationalInsertTabletNode((RelationalInsertTabletNode) node) + : Collections.emptyList(); + } + if (isTableModelConverter()) { + return Collections.emptyList(); } final IDeviceID deviceId = node.getDeviceID(); @@ -355,7 +372,9 @@ private List convertInsertRowsNode(final InsertRowsNode node) { if (rowNode instanceof RelationalInsertRowNode) { tablets.addAll(convertTreeInsertRowNodes(pendingTreeRows)); pendingTreeRows.clear(); - tablets.addAll(convertRelationalInsertRowNode((RelationalInsertRowNode) rowNode)); + if (!isTreeModelConverter()) { + tablets.addAll(convertRelationalInsertRowNode((RelationalInsertRowNode) rowNode)); + } } else { pendingTreeRows.add(rowNode); } @@ -369,7 +388,7 @@ private List convertInsertRowsOfOneDeviceNode(final InsertRowsOfOneDevic } private List convertTreeInsertRowNodes(final List rowNodes) { - if (rowNodes.isEmpty()) { + if (isTableModelConverter() || rowNodes.isEmpty()) { return Collections.emptyList(); } @@ -457,7 +476,10 @@ private List convertInsertMultiTabletsNode(final InsertMultiTabletsNode // so merged relational tablets arrive as InsertMultiTabletsNode (tree) with // RelationalInsertTabletNode children. Dispatch correctly by checking the actual child type. if (tabletNode instanceof RelationalInsertTabletNode) { - tablets.addAll(convertRelationalInsertTabletNode((RelationalInsertTabletNode) tabletNode)); + if (!isTreeModelConverter()) { + tablets.addAll( + convertRelationalInsertTabletNode((RelationalInsertTabletNode) tabletNode)); + } } else { tablets.addAll(convertInsertTabletNode(tabletNode)); } @@ -468,6 +490,10 @@ private List convertInsertMultiTabletsNode(final InsertMultiTabletsNode // ======================== Table Model Conversion ======================== private List convertRelationalInsertRowNode(final RelationalInsertRowNode node) { + if (isTreeModelConverter()) { + return Collections.emptyList(); + } + final String tableName = node.getTableName(); // Table-level pattern filtering @@ -499,6 +525,10 @@ private List convertRelationalInsertRowNode(final RelationalInsertRowNod } private List convertRelationalInsertTabletNode(final RelationalInsertTabletNode node) { + if (isTreeModelConverter()) { + return Collections.emptyList(); + } + final String tableName = node.getTableName(); // Table-level pattern filtering @@ -571,6 +601,10 @@ private List convertRelationalInsertTabletNode(final RelationalInsertTab } private List convertRelationalInsertRowsNode(final RelationalInsertRowsNode node) { + if (isTreeModelConverter()) { + return Collections.emptyList(); + } + final List tablets = new ArrayList<>(); for (final InsertRowNode rowNode : node.getInsertRowNodeList()) { tablets.addAll(convertRelationalInsertRowNode((RelationalInsertRowNode) rowNode)); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java index 3fea7020e72b..18a4369053a5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java @@ -184,8 +184,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(dataRegion, topicConfig, isTableModel)) { continue; } @@ -464,10 +463,10 @@ interface ConsensusTopicSetup { *

This method discovers local DataRegion consensus groups that match the topic filter and * binds one consensus subscription queue to each matching region. * - *

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. + *

Only regions using the same data model as the consumer group are candidates. 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, @@ -528,15 +527,18 @@ private static void setupConsensusQueueForTopic( final String dbRaw = dataRegion.getDatabaseName(); final String dbTableModel = dbRaw.startsWith("root.") ? dbRaw.substring(5) : 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(dataRegion, topicConfig, isTableModel)) { + if (dataRegion.isTableModel() == isTableModel + && topicConfig.isTableTopic() == isTableModel) { + 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 +656,20 @@ static boolean matchesTopicDatabase( || buildTablePattern(topicConfig).matchesDatabase(actualDatabaseName)); } + static boolean matchesTopicDataRegion( + final DataRegion dataRegion, final TopicConfig topicConfig, final boolean isTableModel) { + if (dataRegion == null + || topicConfig == null + || dataRegion.isTableModel() != isTableModel + || topicConfig.isTableTopic() != isTableModel) { + return false; + } + final String databaseName = dataRegion.getDatabaseName(); + 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, diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java index 1b8ad9683653..b176f2765045 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java @@ -25,11 +25,13 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher; @@ -252,6 +254,45 @@ public void testConvertRelationalInsertNodeReturnsEmptyWhenNoColumnsMatch() { Assert.assertTrue(converter.convert(StatementTestUtils.genInsertTabletNode(2, 0)).isEmpty()); } + @Test + public void testTreeConverterRejectsRelationalInsertNodes() { + final ConsensusLogToTabletConverter converter = createTreeConverter(); + final RelationalInsertRowNode relationalRow = StatementTestUtils.genInsertRowNode(0); + final RelationalInsertTabletNode relationalTablet = + StatementTestUtils.genInsertTabletNode(2, 0); + final InsertRowsNode rowsNode = new InsertRowsNode(new PlanNodeId("rows")); + rowsNode.addOneInsertRowNode(relationalRow, 0); + final RelationalInsertRowsNode relationalRowsNode = + new RelationalInsertRowsNode(new PlanNodeId("relationalRows")); + relationalRowsNode.addOneInsertRowNode(relationalRow, 0); + final InsertMultiTabletsNode multiTabletsNode = + new InsertMultiTabletsNode(new PlanNodeId("multiTablets")); + multiTabletsNode.addInsertTabletNode(relationalTablet, 0); + + Assert.assertTrue(converter.convert(relationalRow).isEmpty()); + Assert.assertTrue(converter.convert(relationalTablet).isEmpty()); + Assert.assertTrue(converter.convert(rowsNode).isEmpty()); + Assert.assertTrue(converter.convert(relationalRowsNode).isEmpty()); + Assert.assertTrue(converter.convert(multiTabletsNode).isEmpty()); + } + + @Test + public void testTableConverterRejectsTreeInsertNodes() throws IllegalPathException { + final ConsensusLogToTabletConverter converter = createConverter("m1"); + final InsertRowNode treeRow = createTreeRow("root.sg.d1", 1L, new Object[] {1, 1.0}); + final InsertTabletNode treeTablet = createTreeTablet("root.sg.d1"); + final InsertRowsNode rowsNode = new InsertRowsNode(new PlanNodeId("rows")); + rowsNode.addOneInsertRowNode(treeRow, 0); + final InsertMultiTabletsNode multiTabletsNode = + new InsertMultiTabletsNode(new PlanNodeId("multiTablets")); + multiTabletsNode.addInsertTabletNode(treeTablet, 0); + + Assert.assertTrue(converter.convert(treeRow).isEmpty()); + Assert.assertTrue(converter.convert(treeTablet).isEmpty()); + Assert.assertTrue(converter.convert(rowsNode).isEmpty()); + Assert.assertTrue(converter.convert(multiTabletsNode).isEmpty()); + } + @Test public void testConvertInsertRowsOfOneDeviceNodeGroupsRowsWithSameSchema() throws IllegalPathException { @@ -354,6 +395,21 @@ private static InsertRowNode createTreeRow( false); } + private static InsertTabletNode createTreeTablet(final String devicePath) + throws IllegalPathException { + return new InsertTabletNode( + new PlanNodeId("tablet"), + new PartialPath(devicePath), + false, + new String[] {"s1"}, + new TSDataType[] {TSDataType.INT32}, + new MeasurementSchema[] {new MeasurementSchema("s1", TSDataType.INT32)}, + new long[] {1L}, + null, + new Object[] {new int[] {1}}, + 1); + } + private static String toUtf8(final Binary value) { return new String(value.getValues(), StandardCharsets.UTF_8); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java index 97ee9df9d3c5..e1d73dae152c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.storageengine.dataregion.DataRegion; import org.apache.iotdb.rpc.subscription.config.TopicConfig; import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.rpc.subscription.exception.SubscriptionException; @@ -44,6 +45,8 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; public class ConsensusSubscriptionSetupHandlerTest { @@ -156,6 +159,47 @@ public void testAuditDatabaseNeverMatchesTopic() { assertTrue(ConsensusSubscriptionSetupHandler.matchesTopicDatabase(treeTopicConfig, "user_db")); } + @Test + public void testTopicDataRegionModelIsolation() { + final TopicConfig treeTopicConfig = new TopicConfig(Collections.emptyMap()); + final Map 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); + + final DataRegion treeDataRegion = mock(DataRegion.class); + when(treeDataRegion.isTableModel()).thenReturn(false); + when(treeDataRegion.getDatabaseName()).thenReturn("root.tree_db"); + + final DataRegion tableDataRegion = mock(DataRegion.class); + when(tableDataRegion.isTableModel()).thenReturn(true); + when(tableDataRegion.getDatabaseName()).thenReturn("table_db"); + + final DataRegion otherTableDataRegion = mock(DataRegion.class); + when(otherTableDataRegion.isTableModel()).thenReturn(true); + when(otherTableDataRegion.getDatabaseName()).thenReturn("other_table_db"); + + assertTrue( + ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( + treeDataRegion, treeTopicConfig, false)); + assertFalse( + ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( + tableDataRegion, treeTopicConfig, false)); + assertFalse( + ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( + treeDataRegion, tableTopicConfig, true)); + assertTrue( + ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( + tableDataRegion, tableTopicConfig, true)); + assertFalse( + ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( + otherTableDataRegion, tableTopicConfig, true)); + assertFalse( + ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( + tableDataRegion, tableTopicConfig, false)); + } + private static void failOnSecondTopic( final String topicName, final Set attemptedTopicNames) { attemptedTopicNames.add(topicName); From 8ab350edad7635f5c8d5feaf932dacc58d4e19ea Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:34:44 +0800 Subject: [PATCH 2/2] [Subscription] Bind consensus queues by DataRegion model --- .../ConsensusLogToTabletConverter.java | 42 ++------------ .../ConsensusSubscriptionSetupHandler.java | 25 +++++---- .../ConsensusLogToTabletConverterTest.java | 56 ------------------- ...ConsensusSubscriptionSetupHandlerTest.java | 33 ++++------- 4 files changed, 29 insertions(+), 127 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java index 3b7d96156326..0b27de77b123 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java @@ -78,14 +78,6 @@ public class ConsensusLogToTabletConverter { */ private final String databaseName; - private boolean isTreeModelConverter() { - return treePattern != null && tablePattern == null; - } - - private boolean isTableModelConverter() { - return tablePattern != null; - } - public ConsensusLogToTabletConverter( final TreePattern treePattern, final TablePattern tablePattern, @@ -247,10 +239,6 @@ public List convert(final InsertNode insertNode) { // ======================== Tree Model Conversion ======================== private List convertInsertRowNode(final InsertRowNode node) { - if (isTableModelConverter()) { - return Collections.emptyList(); - } - final IDeviceID deviceId = node.getDeviceID(); // Device-level path filtering @@ -300,12 +288,7 @@ private List convertInsertRowNode(final InsertRowNode node) { private List convertInsertTabletNode(final InsertTabletNode node) { if (node instanceof RelationalInsertTabletNode) { - return !isTreeModelConverter() - ? convertRelationalInsertTabletNode((RelationalInsertTabletNode) node) - : Collections.emptyList(); - } - if (isTableModelConverter()) { - return Collections.emptyList(); + return convertRelationalInsertTabletNode((RelationalInsertTabletNode) node); } final IDeviceID deviceId = node.getDeviceID(); @@ -372,9 +355,7 @@ private List convertInsertRowsNode(final InsertRowsNode node) { if (rowNode instanceof RelationalInsertRowNode) { tablets.addAll(convertTreeInsertRowNodes(pendingTreeRows)); pendingTreeRows.clear(); - if (!isTreeModelConverter()) { - tablets.addAll(convertRelationalInsertRowNode((RelationalInsertRowNode) rowNode)); - } + tablets.addAll(convertRelationalInsertRowNode((RelationalInsertRowNode) rowNode)); } else { pendingTreeRows.add(rowNode); } @@ -388,7 +369,7 @@ private List convertInsertRowsOfOneDeviceNode(final InsertRowsOfOneDevic } private List convertTreeInsertRowNodes(final List rowNodes) { - if (isTableModelConverter() || rowNodes.isEmpty()) { + if (rowNodes.isEmpty()) { return Collections.emptyList(); } @@ -476,10 +457,7 @@ private List convertInsertMultiTabletsNode(final InsertMultiTabletsNode // so merged relational tablets arrive as InsertMultiTabletsNode (tree) with // RelationalInsertTabletNode children. Dispatch correctly by checking the actual child type. if (tabletNode instanceof RelationalInsertTabletNode) { - if (!isTreeModelConverter()) { - tablets.addAll( - convertRelationalInsertTabletNode((RelationalInsertTabletNode) tabletNode)); - } + tablets.addAll(convertRelationalInsertTabletNode((RelationalInsertTabletNode) tabletNode)); } else { tablets.addAll(convertInsertTabletNode(tabletNode)); } @@ -490,10 +468,6 @@ private List convertInsertMultiTabletsNode(final InsertMultiTabletsNode // ======================== Table Model Conversion ======================== private List convertRelationalInsertRowNode(final RelationalInsertRowNode node) { - if (isTreeModelConverter()) { - return Collections.emptyList(); - } - final String tableName = node.getTableName(); // Table-level pattern filtering @@ -525,10 +499,6 @@ private List convertRelationalInsertRowNode(final RelationalInsertRowNod } private List convertRelationalInsertTabletNode(final RelationalInsertTabletNode node) { - if (isTreeModelConverter()) { - return Collections.emptyList(); - } - final String tableName = node.getTableName(); // Table-level pattern filtering @@ -601,10 +571,6 @@ private List convertRelationalInsertTabletNode(final RelationalInsertTab } private List convertRelationalInsertRowsNode(final RelationalInsertRowsNode node) { - if (isTreeModelConverter()) { - return Collections.emptyList(); - } - final List tablets = new ArrayList<>(); for (final InsertRowNode rowNode : node.getInsertRowNodeList()) { tablets.addAll(convertRelationalInsertRowNode((RelationalInsertRowNode) rowNode)); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java index 18a4369053a5..01e9b3d8232d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java +++ b/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,7 +185,7 @@ private static void onNewRegionCreated( final String dbRaw = dataRegion.getDatabaseName(); final String dbTableModel = dbRaw.startsWith("root.") ? dbRaw.substring(5) : dbRaw; - if (!matchesTopicDataRegion(dataRegion, topicConfig, isTableModel)) { + if (!matchesTopicDataRegion(dbRaw, topicConfig, isTableModel)) { continue; } @@ -463,10 +464,11 @@ interface ConsensusTopicSetup { *

This method discovers local DataRegion consensus groups that match the topic filter and * binds one consensus subscription queue to each matching region. * - *

Only regions using the same data model as the consumer group are candidates. 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. + *

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, @@ -526,10 +528,10 @@ 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 (!matchesTopicDataRegion(dataRegion, topicConfig, isTableModel)) { - if (dataRegion.isTableModel() == isTableModel - && topicConfig.isTableTopic() == isTableModel) { + if (!matchesTopicDataRegion(dbRaw, topicConfig, isTableModel)) { + if (isTableModel && dataRegionIsTableModel && topicConfig.isTableTopic()) { LOGGER.info( DataNodePipeMessages .PIPE_LOG_SKIPPING_REGION_DATABASE_FOR_TABLE_TOPIC_DATABASE_KEY_2DA27A84, @@ -657,14 +659,13 @@ static boolean matchesTopicDatabase( } static boolean matchesTopicDataRegion( - final DataRegion dataRegion, final TopicConfig topicConfig, final boolean isTableModel) { - if (dataRegion == null + final String databaseName, final TopicConfig topicConfig, final boolean isTableModel) { + if (databaseName == null || topicConfig == null - || dataRegion.isTableModel() != isTableModel + || isTableModelDatabase(databaseName) != isTableModel || topicConfig.isTableTopic() != isTableModel) { return false; } - final String databaseName = dataRegion.getDatabaseName(); final String actualDatabaseName = databaseName.startsWith("root.") ? databaseName.substring(5) : databaseName; return matchesTopicDatabase(topicConfig, actualDatabaseName); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java index b176f2765045..1b8ad9683653 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java @@ -25,13 +25,11 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher; @@ -254,45 +252,6 @@ public void testConvertRelationalInsertNodeReturnsEmptyWhenNoColumnsMatch() { Assert.assertTrue(converter.convert(StatementTestUtils.genInsertTabletNode(2, 0)).isEmpty()); } - @Test - public void testTreeConverterRejectsRelationalInsertNodes() { - final ConsensusLogToTabletConverter converter = createTreeConverter(); - final RelationalInsertRowNode relationalRow = StatementTestUtils.genInsertRowNode(0); - final RelationalInsertTabletNode relationalTablet = - StatementTestUtils.genInsertTabletNode(2, 0); - final InsertRowsNode rowsNode = new InsertRowsNode(new PlanNodeId("rows")); - rowsNode.addOneInsertRowNode(relationalRow, 0); - final RelationalInsertRowsNode relationalRowsNode = - new RelationalInsertRowsNode(new PlanNodeId("relationalRows")); - relationalRowsNode.addOneInsertRowNode(relationalRow, 0); - final InsertMultiTabletsNode multiTabletsNode = - new InsertMultiTabletsNode(new PlanNodeId("multiTablets")); - multiTabletsNode.addInsertTabletNode(relationalTablet, 0); - - Assert.assertTrue(converter.convert(relationalRow).isEmpty()); - Assert.assertTrue(converter.convert(relationalTablet).isEmpty()); - Assert.assertTrue(converter.convert(rowsNode).isEmpty()); - Assert.assertTrue(converter.convert(relationalRowsNode).isEmpty()); - Assert.assertTrue(converter.convert(multiTabletsNode).isEmpty()); - } - - @Test - public void testTableConverterRejectsTreeInsertNodes() throws IllegalPathException { - final ConsensusLogToTabletConverter converter = createConverter("m1"); - final InsertRowNode treeRow = createTreeRow("root.sg.d1", 1L, new Object[] {1, 1.0}); - final InsertTabletNode treeTablet = createTreeTablet("root.sg.d1"); - final InsertRowsNode rowsNode = new InsertRowsNode(new PlanNodeId("rows")); - rowsNode.addOneInsertRowNode(treeRow, 0); - final InsertMultiTabletsNode multiTabletsNode = - new InsertMultiTabletsNode(new PlanNodeId("multiTablets")); - multiTabletsNode.addInsertTabletNode(treeTablet, 0); - - Assert.assertTrue(converter.convert(treeRow).isEmpty()); - Assert.assertTrue(converter.convert(treeTablet).isEmpty()); - Assert.assertTrue(converter.convert(rowsNode).isEmpty()); - Assert.assertTrue(converter.convert(multiTabletsNode).isEmpty()); - } - @Test public void testConvertInsertRowsOfOneDeviceNodeGroupsRowsWithSameSchema() throws IllegalPathException { @@ -395,21 +354,6 @@ private static InsertRowNode createTreeRow( false); } - private static InsertTabletNode createTreeTablet(final String devicePath) - throws IllegalPathException { - return new InsertTabletNode( - new PlanNodeId("tablet"), - new PartialPath(devicePath), - false, - new String[] {"s1"}, - new TSDataType[] {TSDataType.INT32}, - new MeasurementSchema[] {new MeasurementSchema("s1", TSDataType.INT32)}, - new long[] {1L}, - null, - new Object[] {new int[] {1}}, - 1); - } - private static String toUtf8(final Binary value) { return new String(value.getValues(), StandardCharsets.UTF_8); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java index e1d73dae152c..2bec37f65d23 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java @@ -22,7 +22,6 @@ import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; import org.apache.iotdb.db.conf.IoTDBDescriptor; -import org.apache.iotdb.db.storageengine.dataregion.DataRegion; import org.apache.iotdb.rpc.subscription.config.TopicConfig; import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.rpc.subscription.exception.SubscriptionException; @@ -45,8 +44,6 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; public class ConsensusSubscriptionSetupHandlerTest { @@ -168,36 +165,30 @@ public void testTopicDataRegionModelIsolation() { tableTopicAttributes.put(TopicConstant.DATABASE_KEY, "table_db"); final TopicConfig tableTopicConfig = new TopicConfig(tableTopicAttributes); - final DataRegion treeDataRegion = mock(DataRegion.class); - when(treeDataRegion.isTableModel()).thenReturn(false); - when(treeDataRegion.getDatabaseName()).thenReturn("root.tree_db"); - - final DataRegion tableDataRegion = mock(DataRegion.class); - when(tableDataRegion.isTableModel()).thenReturn(true); - when(tableDataRegion.getDatabaseName()).thenReturn("table_db"); - - final DataRegion otherTableDataRegion = mock(DataRegion.class); - when(otherTableDataRegion.isTableModel()).thenReturn(true); - when(otherTableDataRegion.getDatabaseName()).thenReturn("other_table_db"); - assertTrue( ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( - treeDataRegion, treeTopicConfig, false)); + "root.tree_db", treeTopicConfig, false)); assertFalse( ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( - tableDataRegion, treeTopicConfig, false)); + "table_db", treeTopicConfig, false)); assertFalse( ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( - treeDataRegion, tableTopicConfig, true)); + "table_db", treeTopicConfig, true)); + assertFalse( + ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( + "root.table_db", tableTopicConfig, true)); assertTrue( ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( - tableDataRegion, tableTopicConfig, true)); + "table_db", tableTopicConfig, true)); + assertFalse( + ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( + "other_table_db", tableTopicConfig, true)); assertFalse( ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( - otherTableDataRegion, tableTopicConfig, true)); + "table_db", tableTopicConfig, false)); assertFalse( ConsensusSubscriptionSetupHandler.matchesTopicDataRegion( - tableDataRegion, tableTopicConfig, false)); + "root.table_db", tableTopicConfig, false)); } private static void failOnSecondTopic(