[Subscription] Make column filters follow schema changes by default

integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java

@@ -76,6 +76,64 @@ public void tearDown() throws Exception {
     super.tearDown();
   }
 
+  @Test
+  public void testStreamingColumnFilterIncludesNewTagsAndFieldsOnThreeDataNodes() throws Exception {
+    final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
+        ConsensusSubscriptionTableITSupport.newIdentifiers("streaming_columns");
+    final String database = ids.getDatabase();
+    final String table = "t1";
+    SubscriptionTablePullConsumer consumer = null;
+    try {
+      ConsensusSubscriptionTableITSupport.createDatabaseAndTable(
+          database, table, "tag1 STRING TAG, s1 INT64 FIELD, s2 DOUBLE FIELD, s3 BOOLEAN FIELD");
+      ConsensusSubscriptionTableITSupport.insertRows(database, table, 0L, 1, true, true, true);
+      createOwnedConsensusTopic(ids.getTopic(), database, table, "column_name IN (\"s1\", \"s4\")");
+      consumer = createOwnerConsumer(ids.consumer("owner1"), ids.consumerGroup("owner"), 1L);
+      consumer.subscribe(ids.getTopic());
+      final Set<String> initialRows =
+          ConsensusSubscriptionTableITSupport.insertRows(
+              database, table, 100L, 3, true, true, true);
+      final ConsensusSubscriptionTableITSupport.ConsumedRecords initial =
+          ConsensusSubscriptionTableITSupport.pollAndCommitUntilContains(consumer, initialRows, 60);
+      ConsensusSubscriptionTableITSupport.assertExactRowKeys(initialRows, initial);
+      Assert.assertEquals(
+          Collections.singleton(
+              ConsensusSubscriptionTableITSupport.normalizeColumnSignature("tag1", "s1")),
+          initial.getSeenColumnSignatures());
+
+      addColumn(database, table, "tag2 STRING TAG");
+      addColumn(database, table, "s4 INT32 FIELD");
+      final Set<String> changedRows = new LinkedHashSet<>();
+      try (final ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) {
+        session.executeNonQueryStatement("use " + database);
+        for (long timestamp = 200; timestamp < 203; timestamp++) {
+          session.executeNonQueryStatement(
+              String.format(
+                  Locale.ROOT,
+                  "insert into %s(tag1, tag2, s1, s2, s3, s4, time) "
+                      + "values ('tag_%d', 'new_tag', %d, 1.0, true, %d, %d)",
+                  table,
+                  timestamp,
+                  timestamp,
+                  timestamp,
+                  timestamp));
+          changedRows.add(ConsensusSubscriptionTableITSupport.rowKey(database, table, timestamp));
+        }
+        session.executeNonQueryStatement("flush");
+      }
+      final ConsensusSubscriptionTableITSupport.ConsumedRecords changed =
+          ConsensusSubscriptionTableITSupport.pollAndCommitUntilContains(consumer, changedRows, 80);
+      Assert.assertTrue(changed.getRowKeys().containsAll(changedRows));
+      Assert.assertEquals(
+          Collections.singleton(
+              ConsensusSubscriptionTableITSupport.normalizeColumnSignature(
+                  "tag1", "tag2", "s1", "s4")),
+          changed.getSeenColumnSignatures());
+    } finally {
+      ConsensusSubscriptionTableITSupport.cleanup(consumer, ids.getTopic(), database);
+    }
+  }
+
   @Test
   public void testAlterColumnFilterRebindsAfterOwnerTransferOnThreeDataNodes() throws Exception {
     final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =

integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java

@@ -63,7 +63,6 @@ public class IoTDBSubscriptionColumnFilterIT extends AbstractSubscriptionDualIT
       "tag1 STRING TAG, s1 INT64 FIELD, s2 DOUBLE FIELD, s3 BOOLEAN FIELD";
   private static final String COLUMN_FILTER = "column_name IN (\"time\", \"tag1\", \"s2\")";
   private static final String FIELD_CATEGORY_FILTER = "category = \"FIELD\"";
-  private static final String FIELD_CATEGORY_REBIND_FILTER = "category IN (\"FIELD\")";
   private static final String MIXED_COLUMN_FILTER =
       "( CATEGORY = \"field\" AND datatype IN (\"int64\", \"double\") )"
           + " OR \"column_name\" REGEXP \"tag.*\"";
@@ -622,11 +621,11 @@ public void testLiveRecordHandlerColumnFilterSurvivesSenderRestart() throws Exce
   }
 
   @Test
-  public void testLiveRecordHandlerRestartResnapshotsNewMatchingTable() throws Exception {
-    final String database = databaseName("restart_new");
-    final String topicName = topicName("restart_new");
-    final String consumerId = consumerName("restart_new");
-    final String consumerGroupId = consumerGroupName("restart_new");
+  public void testLiveRecordHandlerIncludesNewMatchingTableWithoutRestart() throws Exception {
+    final String database = databaseName("new_table");
+    final String topicName = topicName("new_table");
+    final String consumerId = consumerName("new_table");
+    final String consumerGroupId = consumerGroupName("new_table");
     final String initialTableName = TABLE_NAME + "_restart_old";
     final String newTableName = TABLE_NAME + "_restart_new";
 
@@ -638,13 +637,13 @@ public void testLiveRecordHandlerRestartResnapshotsNewMatchingTable() throws Exc
           TABLE_NAME + "_restart_.*",
           TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
           FIELD_CATEGORY_FILTER);
-      createTable(senderEnv, database, newTableName, TABLE_SCHEMA);
-
-      TestUtils.restartCluster(senderEnv);
-
       try (final ISubscriptionTablePullConsumer consumer =
           createConsumer(consumerId, consumerGroupId)) {
         consumer.subscribe(topicName);
+        insertRows(senderEnv, database, initialTableName, 230, 233);
+        pollRecordMessagesForTimestamps(
+            consumer, new LinkedHashSet<>(Arrays.asList(230L, 231L, 232L)), false);
+        createTable(senderEnv, database, newTableName, TABLE_SCHEMA);
         insertRows(senderEnv, database, newTableName, 240, 243);
 
         final Set<Long> expectedTimestamps = new LinkedHashSet<>(Arrays.asList(240L, 241L, 242L));
@@ -662,11 +661,11 @@ public void testLiveRecordHandlerRestartResnapshotsNewMatchingTable() throws Exc
   }
 
   @Test
-  public void testLiveRecordHandlerStrictSnapshotRequiresAlterAfterAddColumn() throws Exception {
-    final String database = databaseName("strict");
-    final String topicName = topicName("strict");
-    final String consumerId = consumerName("strict");
-    final String consumerGroupId = consumerGroupName("strict");
+  public void testLiveRecordHandlerIncludesNewFieldsWithoutAlteringTopic() throws Exception {
+    final String database = databaseName("streaming");
+    final String topicName = topicName("streaming");
+    final String consumerId = consumerName("streaming");
+    final String consumerGroupId = consumerGroupName("streaming");
 
     try {
       createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
@@ -680,32 +679,94 @@ public void testLiveRecordHandlerStrictSnapshotRequiresAlterAfterAddColumn() thr
       try (final ISubscriptionTablePullConsumer consumer =
           createConsumer(consumerId, consumerGroupId)) {
         consumer.subscribe(topicName);
+        insertRows(senderEnv, database, TABLE_NAME, 70, 74);
+        pollRecordMessagesForTimestamps(
+            consumer, new LinkedHashSet<>(Arrays.asList(70L, 71L, 72L, 73L)), false);
         addColumn(senderEnv, database, TABLE_NAME, "s4 INT32 FIELD");
         insertRowsWithS4(senderEnv, database, TABLE_NAME, 80, 84);
 
-        final Set<Long> beforeAlterTimestamps =
-            new LinkedHashSet<>(Arrays.asList(80L, 81L, 82L, 83L));
-        final ConsumedRecordStats beforeAlter =
-            pollRecordMessagesForTimestamps(consumer, beforeAlterTimestamps, false);
-        Assert.assertTrue(beforeAlter.columnNames.contains("tag1"));
-        Assert.assertTrue(beforeAlter.columnNames.contains("s1"));
-        Assert.assertTrue(beforeAlter.columnNames.contains("s2"));
-        Assert.assertTrue(beforeAlter.columnNames.contains("s3"));
-        Assert.assertFalse(beforeAlter.columnNames.contains("s4"));
+        final Set<Long> expectedTimestamps = new LinkedHashSet<>(Arrays.asList(80L, 81L, 82L, 83L));
+        final ConsumedRecordStats stats =
+            pollRecordMessagesForTimestamps(consumer, expectedTimestamps, false);
+        Assert.assertEquals(
+            new LinkedHashSet<>(Arrays.asList("tag1", "s1", "s2", "s3", "s4")), stats.columnNames);
+        Assert.assertEquals(expectedTimestamps, stats.timestamps);
+      }
+    } finally {
+      cleanup(topicName, database);
+    }
+  }
 
-        alterTopicColumnFilter(topicName, FIELD_CATEGORY_REBIND_FILTER);
-        insertRowsWithS4(senderEnv, database, TABLE_NAME, 90, 94);
+  @Test
+  public void testLiveRecordHandlerIncludesNewTagsForSelectedFields() throws Exception {
+    final String database = databaseName("new_tag");
+    final String topicName = topicName("new_tag");
+    try {
+      createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+      createTopic(
+          topicName,
+          database,
+          TABLE_NAME,
+          TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+          "column_name = \"s2\"");
+      try (final ISubscriptionTablePullConsumer consumer =
+          createConsumer(consumerName("new_tag"), consumerGroupName("new_tag"))) {
+        consumer.subscribe(topicName);
+        insertRows(senderEnv, database, TABLE_NAME, 300, 303);
+        final ConsumedRecordStats initial =
+            pollRecordMessagesForTimestamps(
+                consumer, new LinkedHashSet<>(Arrays.asList(300L, 301L, 302L)), false);
+        Assert.assertEquals(new LinkedHashSet<>(Arrays.asList("tag1", "s2")), initial.columnNames);
 
-        final Set<Long> afterAlterTimestamps =
-            new LinkedHashSet<>(Arrays.asList(90L, 91L, 92L, 93L));
-        final ConsumedRecordStats afterAlter =
-            pollRecordMessagesForTimestamps(consumer, afterAlterTimestamps, false);
-        Assert.assertTrue(afterAlter.columnNames.contains("tag1"));
-        Assert.assertTrue(afterAlter.columnNames.contains("s1"));
-        Assert.assertTrue(afterAlter.columnNames.contains("s2"));
-        Assert.assertTrue(afterAlter.columnNames.contains("s3"));
-        Assert.assertTrue(afterAlter.columnNames.contains("s4"));
-        Assert.assertEquals(afterAlterTimestamps, afterAlter.timestamps);
+        addColumn(senderEnv, database, TABLE_NAME, "tag2 STRING TAG");
+        insertRowsWithNewTag(senderEnv, database, TABLE_NAME, 310, 313);
+        final ConsumedRecordStats changed =
+            pollRecordMessagesForTimestamps(
+                consumer, new LinkedHashSet<>(Arrays.asList(310L, 311L, 312L)), false);
+        Assert.assertEquals(
+            new LinkedHashSet<>(Arrays.asList("tag1", "tag2", "s2")), changed.columnNames);
+      }
+    } finally {
+      cleanup(topicName, database);
+    }
+  }
+
+  @Test
+  public void testLiveTsFileIncludesNewTagsWithoutAlteringTopic() throws Exception {
+    final String database = databaseName("tsfile_new_tag");
+    final String topicName = topicName("tsfile_new_tag");
+    try {
+      createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+      createDatabaseAndTable(receiverEnv, database, TABLE_NAME, TABLE_SCHEMA);
+      createTopic(
+          topicName,
+          database,
+          TABLE_NAME,
+          TopicConstant.FORMAT_TS_FILE_VALUE,
+          "column_name IN (\"time\", \"s2\")");
+      try (final ISubscriptionTablePullConsumer consumer =
+          createConsumer(consumerName("tsfile_new_tag"), consumerGroupName("tsfile_new_tag"))) {
+        consumer.subscribe(topicName);
+        insertRows(senderEnv, database, TABLE_NAME, 320, 323);
+        pollTsFileMessagesAndLoad(consumer, database, 3);
+        addColumn(senderEnv, database, TABLE_NAME, "tag2 STRING TAG");
+        addColumn(receiverEnv, database, TABLE_NAME, "tag2 STRING TAG");
+        insertRowsWithNewTag(senderEnv, database, TABLE_NAME, 330, 333);
+        pollTsFileMessagesAndLoad(consumer, database, 6);
+        try (final Connection connection = receiverEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+            final Statement statement = connection.createStatement()) {
+          statement.execute("use " + database);
+          try (final java.sql.ResultSet rows =
+              statement.executeQuery("select tag2, s2 from " + TABLE_NAME + " where time >= 330")) {
+            int count = 0;
+            while (rows.next()) {
+              Assert.assertEquals("new_tag", rows.getString("tag2"));
+              Assert.assertTrue(rows.getDouble("s2") >= 330.5);
+              count++;
+            }
+            Assert.assertEquals(3, count);
+          }
+        }
       }
     } finally {
       cleanup(topicName, database);
@@ -795,7 +856,7 @@ public void testLiveRecordHandlerColumnFilterMatchesCustomTimeColumn() throws Ex
   }
 
   @Test
-  public void testLiveRecordHandlerColumnFilterUsesAlteredDataTypeAtBindTime() throws Exception {
+  public void testLiveRecordHandlerColumnFilterUsesLatestDataType() throws Exception {
     final String database = databaseName("alter_type");
     final String topicName = topicName("alter_type");
     final String consumerId = consumerName("alter_type");
@@ -805,17 +866,22 @@ public void testLiveRecordHandlerColumnFilterUsesAlteredDataTypeAtBindTime() thr
 
     try {
       createDatabaseAndTable(senderEnv, database, TABLE_NAME, initialSchema);
-      alterColumnType(senderEnv, database, TABLE_NAME, "s1", "INT64");
       createTopic(
           topicName,
           database,
           TABLE_NAME,
           TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
-          "datatype = \"INT64\"");
+          "category = \"TAG\" OR datatype = \"INT64\"");
 
       try (final ISubscriptionTablePullConsumer consumer =
           createConsumer(consumerId, consumerGroupId)) {
         consumer.subscribe(topicName);
+        insertRows(senderEnv, database, TABLE_NAME, 140, 143);
+        final ConsumedRecordStats initial =
+            pollRecordMessagesForTimestamps(
+                consumer, new LinkedHashSet<>(Arrays.asList(140L, 141L, 142L)), false);
+        Assert.assertEquals(new LinkedHashSet<>(Arrays.asList("tag1")), initial.columnNames);
+        alterColumnType(senderEnv, database, TABLE_NAME, "s1", "INT64");
         insertRows(senderEnv, database, TABLE_NAME, 150, 153);
 
         final Set<Long> expectedTimestamps = new LinkedHashSet<>(Arrays.asList(150L, 151L, 152L));
@@ -1148,6 +1214,32 @@ private static void insertRowsWithS4(
     }
   }
 
+  private static void insertRowsWithNewTag(
+      final BaseEnv env,
+      final String database,
+      final String tableName,
+      final int startInclusive,
+      final int endExclusive)
+      throws Exception {
+    try (final Connection connection = env.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        final Statement statement = connection.createStatement()) {
+      statement.execute("use " + database);
+      for (int i = startInclusive; i < endExclusive; i++) {
+        statement.execute(
+            String.format(
+                Locale.ROOT,
+                "insert into %s(tag1, tag2, s1, s2, s3, time) "
+                    + "values ('tag_%d', 'new_tag', %d, %.1f, true, %d)",
+                tableName,
+                i,
+                i * 10L,
+                i + 0.5d,
+                i));
+      }
+      statement.execute("flush");
+    }
+  }
+
   private static void insertRowsWithAttribute(
       final BaseEnv env,
       final String database,

iotdb-client/subscription/README.md

@@ -0,0 +1,71 @@
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+        http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+
+# Subscription column filters
+
+Table topics support a `column-filter` expression over column metadata:
+`database`, `table_name`, `column_name`, `datatype`, and `category`.
+String literals use double quotes, and comparisons are case insensitive.
+The default expression is `true`, which includes every column.
+
+## Streaming schema semantics
+
+Column filters follow schema changes while the subscription runs. When a
+table has a matching column, its TAG columns are retained automatically,
+including TAGs added after topic creation or subscription. For example:
+
+```sql
+CREATE TOPIC selected_measurements WITH (
+  'database' = 'factory',
+  'table' = 'sensors',
+  'column-filter' = 'column_name IN ("temperature", "pressure")'
+);
+```
+
+This selects `temperature` and `pressure` together with all TAGs. Adding a
+`location` TAG later includes it in subsequent data without altering the
+topic or restarting the consumer. FIELDs outside the predicate are excluded.
+A predicate such as `category = "FIELD"` also picks up new FIELDs, while
+`column_name LIKE "sensor%"` picks up new matching column names.
+New tables matching the topic's database and table patterns are included
+automatically. Datatype predicates use the latest committed table schema.
+Tree view predicates use view column names and return the corresponding
+source FIELDs. ATTRIBUTE columns are excluded from filtered payloads even
+when they match; matching them still retains the table's TAGs. Tables with
+no matching column produce no rows. TIME selection remains per table and
+uses its declared time column name.
+
+The predicate is applied when data is prepared for delivery, for both
+historical and incremental data and for record and TsFile payloads. Data
+already prepared for delivery keeps its existing projection. There is no
+fixed schema snapshot option for column filtering. The topic's
+`mode = snapshot` still controls the historical data cutoff; it does not
+freeze the column schema.
+
+## Evaluation cost
+
+Each topic parses its expression once. Before pruning a Tablet, the
+DataNode reads the committed schema from its local table cache. It
+reuses the compiled column selection while that schema instance is
+unchanged and recompiles only the affected table after a schema update.
+This adds one local cache lookup per Tablet, without remote schema
+requests or expression evaluation per row. If a table's schema is not
+available locally, its data is skipped until the schema becomes available.

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java

@@ -826,6 +826,15 @@ private String compareTable(final TsTable oldTable, final TsTable newTable) {
     return modified ? builder.toString() : DataNodeSchemaMessages.COMPARE_TABLE_NOT_MODIFIED;
   }
 
+  /** Returns the current local schema without fetching or resolving updates from ConfigNode. */
+  @Override
+  public TsTable getTableInCache(final String database, final String tableName) {
+    return getTableInCache(
+        PathUtils.unQualifyDatabaseName(database),
+        tableName,
+        LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
+  }
+
   private TsTable getTableInCache(
       final String database,
       final String tableName,