apache · Caideyipi · Sep 15, 2026
diff --git a/...iotdb/subscription/it/consensus/local/tablemodel/ConsensusSubscriptionTableITSupport.java b/...iotdb/subscription/it/consensus/local/tablemodel/ConsensusSubscriptionTableITSupport.java
@@ -125,6 +125,16 @@ static void createConsensusTopic(
       final String tablePattern,
       final String columnFilter)
       throws Exception {
+    createConsensusTopic(topicName, databasePattern, tablePattern, columnFilter, null);
+  }
+
+  static void createConsensusTopic(
+      final String topicName,
+      final String databasePattern,
+      final String tablePattern,
+      final String columnFilter,
+      final String tagFilter)
+      throws Exception {
     final String host = EnvFactory.getEnv().getIP();
     final int port = Integer.parseInt(EnvFactory.getEnv().getPort());
 
@@ -141,6 +151,9 @@ static void createConsensusTopic(
       if (columnFilter != null) {
         config.put(TopicConstant.COLUMN_FILTER_KEY, columnFilter);
       }
+      if (tagFilter != null) {
+        config.put(TopicConstant.TAG_FILTER_KEY, tagFilter);
+      }
       session.createTopic(topicName, config);
     }
   }
@@ -160,6 +173,21 @@ static void alterConsensusTopicColumnFilter(final String topicName, final String
     }
   }
 
+  static void alterConsensusTopicTagFilter(final String topicName, final String tagFilter)
+      throws Exception {
+    final String host = EnvFactory.getEnv().getIP();
+    final int port = Integer.parseInt(EnvFactory.getEnv().getPort());
+
+    try (final ISubscriptionTableSession session =
+        new SubscriptionTableSessionBuilder().host(host).port(port).build()) {
+      session.open();
+
+      final Properties config = new Properties();
+      config.put(TopicConstant.TAG_FILTER_KEY, tagFilter);
+      session.alterTopic(topicName, config);
+    }
+  }
+
   static SubscriptionTablePullConsumer createConsumer(
       final String consumerId, final String consumerGroupId) throws Exception {
     final SubscriptionTablePullConsumer consumer =

diff --git a/...b/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionFilterTableIT.java b/...b/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionFilterTableIT.java
@@ -282,6 +282,80 @@ public void testColumnFilteringWithNoMatchedColumnsReturnsNothing() throws Excep
     }
   }
 
+  @Test
+  public void testTagFilteringAndAlterInConsensusQueue() throws Exception {
+    final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
+        ConsensusSubscriptionTableITSupport.newIdentifiers("table_filter_tags");
+    final String database = ids.getDatabase();
+    final String table = "t1";
+    final String schema = "tag1 STRING TAG, s1 INT64 FIELD, s2 DOUBLE FIELD";
+    final String expectedColumnSignature =
+        ConsensusSubscriptionTableITSupport.normalizeColumnSignature("tag1", "s2");
+    SubscriptionTablePullConsumer consumer = null;
+
+    try {
+      ConsensusSubscriptionTableITSupport.createDatabaseAndTable(database, table, schema);
+      try (final ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) {
+        session.executeNonQueryStatement("use " + database);
+        session.executeNonQueryStatement(
+            "insert into " + table + "(tag1, s1, s2, time) values ('bootstrap', 0, 0.0, 0)");
+        session.executeNonQueryStatement("flush");
+      }
+
+      ConsensusSubscriptionTableITSupport.createConsensusTopic(
+          ids.getTopic(), database, table, "column_name = \"s2\"", "tag1 = \"keep\"");
+
+      consumer =
+          ConsensusSubscriptionTableITSupport.createConsumer(
+              ids.getConsumerId(), ids.getConsumerGroupId());
+      consumer.subscribe(ids.getTopic());
+
+      final Set<String> expectedBeforeAlter = new LinkedHashSet<>();
+      try (final ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) {
+        session.executeNonQueryStatement("use " + database);
+        for (int i = 1; i <= 5; i++) {
+          final long timestamp = 100L + i;
+          final String tag = i % 2 == 1 ? "keep" : "drop";
+          session.executeNonQueryStatement(
+              String.format(
+                  Locale.ROOT,
+                  "insert into %s(tag1, s1, s2, time) values ('%s', %d, %.1f, %d)",
+                  table,
+                  tag,
+                  i * 10L,
+                  i + 0.5d,
+                  timestamp));
+          if ("keep".equals(tag)) {
+            expectedBeforeAlter.add(
+                ConsensusSubscriptionTableITSupport.rowKey(database, table, timestamp));
+          }
+        }
+        session.executeNonQueryStatement("flush");
+      }
+
+      final ConsensusSubscriptionTableITSupport.ConsumedRecords consumedBeforeAlter =
+          ConsensusSubscriptionTableITSupport.pollAndCommitUntilContains(
+              consumer, expectedBeforeAlter, 50);
+      ConsensusSubscriptionTableITSupport.assertExactRowKeys(
+          expectedBeforeAlter, consumedBeforeAlter);
+      Assert.assertEquals(
+          Collections.singleton(expectedColumnSignature),
+          consumedBeforeAlter.getSeenColumnSignatures());
+
+      ConsensusSubscriptionTableITSupport.alterConsensusTopicTagFilter(
+          ids.getTopic(), "tag1 LIKE \"t1_tag_%\"");
+      final ConsensusSubscriptionTableITSupport.ConsumedRecords consumedAfterAlter =
+          ConsensusSubscriptionTableITSupport.insertRowsAndPollUntilColumnSignature(
+              consumer, database, table, 200L, 3, true, false, expectedColumnSignature, 60);
+      Assert.assertTrue(consumedAfterAlter.getRowCount() > 0);
+      Assert.assertTrue(
+          consumedAfterAlter.getSeenColumnSignatures().contains(expectedColumnSignature));
+      ConsensusSubscriptionTableITSupport.assertNoMoreMessages(consumer, 3, Duration.ofMillis(500));
+    } finally {
+      ConsensusSubscriptionTableITSupport.cleanup(consumer, ids.getTopic(), database);
+    }
+  }
+
   @Test
   public void testColumnFilteringComplexOperatorsInConsensusQueue() throws Exception {
     final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =

diff --git a/...ava/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java b/...ava/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java
@@ -452,6 +452,73 @@ public void testLiveTsFileColumnFilter() throws Exception {
     }
   }
 
+  @Test
+  public void testLiveRecordHandlerTagFilterBeforeColumnFilter() throws Exception {
+    final String database = databaseName("tag_record");
+    final String topicName = topicName("tag_record");
+    final String consumerId = consumerName("tag_record");
+    final String consumerGroupId = consumerGroupName("tag_record");
+
+    try {
+      createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+      createTopicWithFilters(
+          topicName,
+          database,
+          TABLE_NAME,
+          TopicConstant.MODE_INITIAL_VALUE,
+          TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+          COLUMN_FILTER,
+          "tag1 IN (\"tag_1\", \"tag_3\")");
+
+      try (final ISubscriptionTablePullConsumer consumer =
+          createConsumer(consumerId, consumerGroupId)) {
+        consumer.subscribe(topicName);
+        insertRows(senderEnv, database, TABLE_NAME, 1, 5);
+
+        final Set<Long> expectedTimestamps = new LinkedHashSet<>(Arrays.asList(1L, 3L));
+        final ConsumedRecordStats stats =
+            pollRecordMessagesForTimestamps(consumer, expectedTimestamps, true);
+
+        Assert.assertEquals(expectedTimestamps, stats.timestamps);
+        Assert.assertEquals(EXPECTED_COLUMNS, stats.columnNames);
+      }
+    } finally {
+      cleanup(topicName, database);
+    }
+  }
+
+  @Test
+  public void testSnapshotTsFileTagFilter() throws Exception {
+    final String database = databaseName("tag_snapshot_tsfile");
+    final String topicName = topicName("tag_snapshot_tsfile");
+    final String consumerId = consumerName("tag_snapshot_tsfile");
+    final String consumerGroupId = consumerGroupName("tag_snapshot_tsfile");
+
+    try {
+      createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+      createDatabaseAndTable(receiverEnv, database, TABLE_NAME, TABLE_SCHEMA);
+      insertRows(senderEnv, database, TABLE_NAME, 10, 14);
+      createTopicWithFilters(
+          topicName,
+          database,
+          TABLE_NAME,
+          TopicConstant.MODE_SNAPSHOT_VALUE,
+          TopicConstant.FORMAT_TS_FILE_VALUE,
+          TopicConstant.COLUMN_FILTER_DEFAULT_VALUE,
+          "tag1 IN (\"tag_10\", \"tag_12\")");
+
+      try (final ISubscriptionTablePullConsumer consumer =
+          createConsumer(consumerId, consumerGroupId)) {
+        consumer.subscribe(topicName);
+        pollTsFileMessagesAndLoad(consumer, database, 2);
+      }
+
+      assertLoadedTimestamps(database, new LinkedHashSet<>(Arrays.asList(10L, 12L)));
+    } finally {
+      cleanup(topicName, database);
+    }
+  }
+
   @Test
   public void testSnapshotTsFileColumnFilter() throws Exception {
     final String database = databaseName("snapshot_tsfile");
@@ -1036,6 +1103,34 @@ private void createTopicWithoutColumnFilter(
         topicName, database, tableName, TopicConstant.MODE_INITIAL_VALUE, format, "", false);
   }
 
+  private void createTopicWithFilters(
+      final String topicName,
+      final String database,
+      final String tableName,
+      final String mode,
+      final String format,
+      final String columnFilter,
+      final String tagFilter)
+      throws Exception {
+    try (final ISubscriptionTableSession session =
+        new SubscriptionTableSessionBuilder()
+            .host(senderEnv.getIP())
+            .port(Integer.parseInt(senderEnv.getPort()))
+            .build()) {
+      session.open();
+      session.dropTopicIfExists(topicName);
+
+      final Properties config = new Properties();
+      config.put(TopicConstant.MODE_KEY, mode);
+      config.put(TopicConstant.FORMAT_KEY, format);
+      config.put(TopicConstant.DATABASE_KEY, database);
+      config.put(TopicConstant.TABLE_KEY, tableName);
+      config.put(TopicConstant.COLUMN_FILTER_KEY, columnFilter);
+      config.put(TopicConstant.TAG_FILTER_KEY, tagFilter);
+      session.createTopic(topicName, config);
+    }
+  }
+
   private void createTopic(
       final String topicName,
       final String database,
@@ -1455,6 +1550,22 @@ private void assertLoadedTsFileRowsWithAllColumns(final String database, final i
     }
   }
 
+  private void assertLoadedTimestamps(final String database, final Set<Long> expectedTimestamps)
+      throws Exception {
+    final Set<Long> actualTimestamps = new LinkedHashSet<>();
+    try (final Connection connection = receiverEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        final Statement statement = connection.createStatement()) {
+      statement.execute("use " + database);
+      try (final java.sql.ResultSet resultSet =
+          statement.executeQuery("select time from " + TABLE_NAME + " order by time")) {
+        while (resultSet.next()) {
+          actualTimestamps.add(resultSet.getLong("time"));
+        }
+      }
+    }
+    Assert.assertEquals(expectedTimestamps, actualTimestamps);
+  }
+
   private void cleanup(final String topicName, final String database) {
     try (final ISubscriptionTableSession session =
         new SubscriptionTableSessionBuilder()

diff --git a/...ient/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java b/...ient/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java
@@ -235,6 +235,23 @@ public boolean isColumnFilterTrivial() {
     return TopicConstant.COLUMN_FILTER_DEFAULT_VALUE.equalsIgnoreCase(getColumnFilter().trim());
   }
 
+  public Map<String, String> getAttributesWithSourceTagFilter() {
+    return Collections.singletonMap(TopicConstant.TAG_FILTER_KEY, getTagFilter());
+  }
+
+  public String getTagFilter() {
+    return getStringIgnoreCase(
+        TopicConstant.TAG_FILTER_KEY, TopicConstant.TAG_FILTER_DEFAULT_VALUE);
+  }
+
+  public boolean hasTagFilter() {
+    return containsKeyIgnoreCase(TopicConstant.TAG_FILTER_KEY);
+  }
+
+  public boolean isTagFilterTrivial() {
+    return TopicConstant.TAG_FILTER_DEFAULT_VALUE.equalsIgnoreCase(getTagFilter().trim());
+  }
+
   public Map<String, String> getAttributesWithSourceTimeRange() {
     final Map<String, String> attributesWithTimeRange = new HashMap<>();
 

diff --git a/...nt/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java b/...nt/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java
@@ -31,11 +31,13 @@ public class TopicConstant {
   public static final String DATABASE_KEY = "database";
   public static final String TABLE_KEY = "table";
   public static final String COLUMN_FILTER_KEY = "column-filter";
+  public static final String TAG_FILTER_KEY = "tag-filter";
   public static final String RETENTION_BYTES_KEY = "retention.bytes";
   public static final String RETENTION_MS_KEY = "retention.ms";
   public static final String DATABASE_DEFAULT_VALUE = ".*";
   public static final String TABLE_DEFAULT_VALUE = ".*";
   public static final String COLUMN_FILTER_DEFAULT_VALUE = "true";
+  public static final String TAG_FILTER_DEFAULT_VALUE = "true";
 
   public static final String START_TIME_KEY = "start-time";
   public static final String END_TIME_KEY = "end-time";

diff --git a/...ava/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java b/...ava/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
@@ -1180,7 +1180,8 @@ private Optional<SubscriptionMessage> pollTablets(
   }
 
   private Optional<SubscriptionMessage> pollTabletsInternal(
-      final SubscriptionPollResponse initialResponse, final PollTimer timer) {
+      final SubscriptionPollResponse initialResponse, final PollTimer timer)
+      throws SubscriptionException {
     final Map<String, List<Tablet>> tablets =
         ((TabletsPayload) initialResponse.getPayload()).getTabletsWithDBInfo();
     final SubscriptionCommitContext commitContext = initialResponse.getCommitContext();
@@ -1200,6 +1201,10 @@ private Optional<SubscriptionMessage> pollTabletsInternal(
           LOGGER.warn(errorMessage);
           throw new SubscriptionRuntimeNonCriticalException(errorMessage);
         }
+        if (tabletsSize == 0) {
+          ack(Collections.singletonList(new SubscriptionMessage(commitContext, tablets)));
+          return Optional.empty();
+        }
         return Optional.of(
             new SubscriptionMessage(commitContext, tablets, timeSelected, timeSelectedByTable));
       }

diff --git a/.../subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java b/.../subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java
@@ -110,6 +110,35 @@ public void testColumnFilterTrivialWithMixedCaseKeyAndValue() {
     Assert.assertTrue(new TopicConfig(attributes).isColumnFilterTrivial());
   }
 
+  @Test
+  public void testTagFilterKeyIsCaseInsensitive() {
+    final TopicConfig topicConfig =
+        new TopicConfig(Collections.singletonMap("Tag-Filter", "region = \"north\""));
+
+    Assert.assertTrue(topicConfig.hasTagFilter());
+    Assert.assertEquals("region = \"north\"", topicConfig.getTagFilter());
+    Assert.assertEquals(
+        "region = \"north\"",
+        topicConfig.getAttributesWithSourceTagFilter().get(TopicConstant.TAG_FILTER_KEY));
+  }
+
+  @Test
+  public void testTagFilterDefaultsToTrivialWhenAbsent() {
+    final TopicConfig topicConfig = new TopicConfig(new HashMap<>());
+
+    Assert.assertFalse(topicConfig.hasTagFilter());
+    Assert.assertTrue(topicConfig.isTagFilterTrivial());
+    Assert.assertEquals(TopicConstant.TAG_FILTER_DEFAULT_VALUE, topicConfig.getTagFilter());
+  }
+
+  @Test
+  public void testTagFilterTrivialWithMixedCaseKeyAndValue() {
+    final Map<String, String> attributes = new HashMap<>();
+    attributes.put("TAG-FILTER", " TRUE ");
+
+    Assert.assertTrue(new TopicConfig(attributes).isTagFilterTrivial());
+  }
+
   private static TopicConfig topicConfigWithMode(final String mode) {
     return new TopicConfig(Collections.singletonMap(TopicConstant.MODE_KEY, mode));
   }