From ef27ef63f1fa9110750792a7c8d37953e7026c37 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:53:23 +0800 Subject: [PATCH] [Subscription] Make column filters follow schema changes by default --- ...nsusSubscriptionColumnFilterClusterIT.java | 58 ++++++ .../IoTDBSubscriptionColumnFilterIT.java | 170 ++++++++++++++---- iotdb-client/subscription/README.md | 71 ++++++++ .../table/DataNodeTableCache.java | 9 + .../db/schemaengine/table/ITableCache.java | 3 + .../agent/SubscriptionBrokerAgent.java | 52 +----- .../columnfilter/ColumnFilterBinder.java | 7 + .../columnfilter/ColumnFilterMatcher.java | 146 ++++++++++++++- .../columnfilter/TabletColumnPruner.java | 17 +- .../table/DataNodeTableCacheLeaseTest.java | 1 + .../table/DataNodeTableCacheTest.java | 20 +++ .../columnfilter/ColumnFilterBinderTest.java | 92 ++++++++-- 12 files changed, 531 insertions(+), 115 deletions(-) create mode 100644 iotdb-client/subscription/README.md diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java index e90d94af7ea7f..0875265b6f234 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java +++ b/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 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 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 = diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java index bfeb7e6b7aa7b..d300e2c9bf6aa 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java +++ b/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 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 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 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 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 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, diff --git a/iotdb-client/subscription/README.md b/iotdb-client/subscription/README.md new file mode 100644 index 0000000000000..3764d75754a30 --- /dev/null +++ b/iotdb-client/subscription/README.md @@ -0,0 +1,71 @@ + + +# 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. diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java index 525bf511f4e4b..3db7af6f89a4d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java +++ b/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, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java index 13193fbff13e0..b87b6db7147b2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java @@ -52,6 +52,9 @@ void commitUpdateTable( TsTable getTableInWrite(final String database, final String tableName); + /** Returns the committed local schema without contacting ConfigNode. */ + TsTable getTableInCache(final String database, final String tableName); + TsTable getTable(final String database, final String tableName); TsTable getTable(String database, final String tableName, final boolean force); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java index aafcf903dc43e..b5bd049b9c215 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java @@ -20,7 +20,6 @@ package org.apache.iotdb.db.subscription.agent; import org.apache.iotdb.commons.consensus.ConsensusGroupId; -import org.apache.iotdb.commons.schema.table.TsTable; import org.apache.iotdb.commons.subscription.meta.consumer.SubscriptionProgressSnapshot; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.consensus.IConsensus; @@ -40,7 +39,6 @@ import org.apache.iotdb.db.subscription.broker.consensus.ConsensusRegionRuntimeState; import org.apache.iotdb.db.subscription.broker.consensus.ConsensusSubscriptionCommitManager; import org.apache.iotdb.db.subscription.broker.consensus.ConsensusSubscriptionSetupHandler; -import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterBinder; import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher; import org.apache.iotdb.db.subscription.event.SubscriptionEvent; import org.apache.iotdb.db.subscription.resource.SubscriptionDataNodeResourceManager; @@ -48,7 +46,6 @@ import org.apache.iotdb.db.subscription.task.subtask.SubscriptionSinkSubtask; import org.apache.iotdb.rpc.subscription.config.ConsumerConfig; import org.apache.iotdb.rpc.subscription.config.TopicConfig; -import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.rpc.subscription.exception.SubscriptionException; import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress; import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext; @@ -90,7 +87,6 @@ public class SubscriptionBrokerAgent { private final Map topicNameToColumnFilterMatcher = new ConcurrentHashMap<>(); - private final ColumnFilterBinder columnFilterBinder = new ColumnFilterBinder(); //////////////////////////// provided for subscription agent //////////////////////////// @@ -732,17 +728,9 @@ public void wakeUpConsensusPrefetchingQueues(final String topicName, final boole public void refreshColumnFilter(final String topicName, final TopicConfig topicConfig) { final ColumnFilterMatcher matcher; try { - final Map> bindingTables = - getColumnFilterBindingTables(topicConfig); - if (Objects.isNull(bindingTables)) { - topicNameToColumnFilterMatcher.remove(topicName); - LOGGER.info( - DataNodeMiscMessages.SUBSCRIPTION_COLUMN_FILTER_SCHEMA_NOT_AVAILABLE, topicName); - return; - } matcher = - ColumnFilterMatcher.fromBoundColumnFilter( - columnFilterBinder.bind(topicConfig, bindingTables)); + ColumnFilterMatcher.fromTopicConfig( + topicConfig, DataNodeTableCache.getInstance()::getTableInCache); } catch (final Exception e) { LOGGER.warn(DataNodeMiscMessages.SUBSCRIPTION_REFRESH_COLUMN_FILTER_FAILED, topicName, e); topicNameToColumnFilterMatcher.put(topicName, EMPTY_COLUMN_FILTER_MATCHER); @@ -794,42 +782,6 @@ private static ColumnFilterMatcher getDefaultColumnFilterMatcher(final TopicConf : ColumnFilterMatcher.matchAll(); } - private static Map> getColumnFilterBindingTables( - final TopicConfig topicConfig) { - if (Objects.isNull(topicConfig) - || !topicConfig.isTableTopic() - || topicConfig.isColumnFilterTrivial()) { - return Collections.emptyMap(); - } - - final String database = - topicConfig.getStringOrDefault( - TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE); - final String tableName = - topicConfig.getStringOrDefault(TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE); - if (isDefaultTopicPattern(database, TopicConstant.DATABASE_DEFAULT_VALUE) - || isDefaultTopicPattern(tableName, TopicConstant.TABLE_DEFAULT_VALUE) - || !isLiteralTopicPattern(database) - || !isLiteralTopicPattern(tableName)) { - return DataNodeTableCache.getInstance().getTableSnapshot(); - } - - final TsTable table = DataNodeTableCache.getInstance().getTable(database, tableName, false); - return Objects.isNull(table) - ? null - : Collections.singletonMap(database, Collections.singletonMap(tableName, table)); - } - - private static boolean isLiteralTopicPattern(final String pattern) { - final String regexMetaCharacters = ".*+?[](){}\\|^$"; - return Objects.nonNull(pattern) - && pattern.chars().noneMatch(c -> regexMetaCharacters.indexOf((char) c) >= 0); - } - - private static boolean isDefaultTopicPattern(final String pattern, final String defaultPattern) { - return Objects.isNull(pattern) || defaultPattern.equals(pattern.trim()); - } - public void dropColumnFilter(final String topicName) { topicNameToColumnFilterMatcher.remove(topicName); LOGGER.info(DataNodeMiscMessages.SUBSCRIPTION_DROP_COLUMN_FILTER, topicName); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterBinder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterBinder.java index 82f15db0fadf2..9417f6f12300d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterBinder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterBinder.java @@ -125,6 +125,13 @@ private static BindTableResult bindTable( return new BindTableResult(selectedColumnNames, timeSelected); } + /** Compiles a table's current schema without parsing the topic expression again. */ + static ColumnFilterMatcher bindTableFilter( + final Expression expression, final String database, final TsTable table) { + final BindTableResult result = bindTable(expression, database, table); + return ColumnFilterMatcher.forTableSelection(result.selectedColumnNames, result.timeSelected); + } + private static boolean evaluate( final Expression expression, final String database, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterMatcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterMatcher.java index 8755bb9dc0dd4..1c9246710b6d4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterMatcher.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterMatcher.java @@ -19,9 +19,12 @@ package org.apache.iotdb.db.subscription.columnfilter; +import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression; +import org.apache.iotdb.commons.schema.table.TsTable; import org.apache.iotdb.commons.subscription.columnfilter.ColumnMetadata; import org.apache.iotdb.rpc.subscription.config.TopicConfig; +import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.rpc.subscription.exception.SubscriptionException; import org.apache.tsfile.enums.ColumnCategory; @@ -34,6 +37,9 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.function.BiFunction; public class ColumnFilterMatcher { @@ -42,6 +48,11 @@ public class ColumnFilterMatcher { private final BoundColumnFilter boundColumnFilter; private final Set selectedColumnNames; private final Expression expression; + private final TablePattern tablePattern; + private final BiFunction tableProvider; + private final Boolean tableTimeSelected; + private final ConcurrentMap tableFilters = + new ConcurrentHashMap<>(); private ColumnFilterMatcher(final Set selectedColumnNames, final Expression expression) { this(null, selectedColumnNames, expression); @@ -51,9 +62,38 @@ private ColumnFilterMatcher( final BoundColumnFilter boundColumnFilter, final Set selectedColumnNames, final Expression expression) { + this(boundColumnFilter, selectedColumnNames, expression, null, null); + } + + private ColumnFilterMatcher( + final BoundColumnFilter boundColumnFilter, + final Set selectedColumnNames, + final Expression expression, + final TablePattern tablePattern, + final BiFunction tableProvider) { + this(boundColumnFilter, selectedColumnNames, expression, tablePattern, tableProvider, null); + } + + private ColumnFilterMatcher( + final BoundColumnFilter boundColumnFilter, + final Set selectedColumnNames, + final Expression expression, + final TablePattern tablePattern, + final BiFunction tableProvider, + final Boolean tableTimeSelected) { this.boundColumnFilter = boundColumnFilter; this.selectedColumnNames = selectedColumnNames; this.expression = expression; + this.tablePattern = tablePattern; + this.tableProvider = tableProvider; + this.tableTimeSelected = tableTimeSelected; + } + + /** A resolved table plan needs only column membership checks and its declared TIME selection. */ + static ColumnFilterMatcher forTableSelection( + final Set selectedColumnNames, final boolean timeSelected) { + return new ColumnFilterMatcher( + null, Collections.unmodifiableSet(selectedColumnNames), null, null, null, timeSelected); } public static ColumnFilterMatcher matchAll() { @@ -79,6 +119,65 @@ public static ColumnFilterMatcher fromTopicConfig(final TopicConfig topicConfig) null, new ColumnFilterParser().parseAndValidate(topicConfig.getColumnFilter())); } + /** + * Creates a streaming matcher. The expression is parsed once and table plans are reused until + * their local schema is replaced, including when a previously unknown table becomes available. + */ + public static ColumnFilterMatcher fromTopicConfig( + final TopicConfig topicConfig, final BiFunction tableProvider) + throws SubscriptionException { + if (Objects.isNull(topicConfig) + || !topicConfig.isTableTopic() + || topicConfig.isColumnFilterTrivial()) { + return matchAll(); + } + return new ColumnFilterMatcher( + null, + null, + new ColumnFilterParser().parseAndValidate(topicConfig.getColumnFilter()), + new TablePattern( + true, + topicConfig.getStringOrDefault( + TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE), + topicConfig.getStringOrDefault( + TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE)), + tableProvider); + } + + /** Resolves the schema once per Tablet, keeping expression evaluation out of the column loop. */ + public ColumnFilterMatcher forTable(final String databaseName, final String tableName) { + if (Objects.isNull(tableProvider)) { + return this; + } + final BoundColumnFilter.TableKey key = BoundColumnFilter.TableKey.of(databaseName, tableName); + final CachedTableFilter cached = tableFilters.get(key); + if (Objects.isNull(cached) + && (!tablePattern.matchesDatabase(key.getDatabaseName()) + || !tablePattern.matchesTable(key.getTableName()))) { + return ofSelectedColumnNames(Collections.emptySet()); + } + final TsTable table = tableProvider.apply(key.getDatabaseName(), key.getTableName()); + if (Objects.isNull(table)) { + tableFilters.remove(key); + return ofSelectedColumnNames(Collections.emptySet()); + } + // DataNodeTableCache replaces entire TsTable instances on schema changes. Comparing identity + // avoids copying schemas or evaluating predicates for every Tablet (and every row). + if (Objects.nonNull(cached) && cached.table == table) { + return cached.matcher; + } + return tableFilters.compute( + key, + (ignored, previous) -> + Objects.nonNull(previous) && previous.table == table + ? previous + : new CachedTableFilter( + table, + ColumnFilterBinder.bindTableFilter( + expression, key.getDatabaseName(), table))) + .matcher; + } + public static ColumnFilterMatcher ofSelectedColumnNames(final Set selectedColumnNames) { if (Objects.isNull(selectedColumnNames)) { return matchAll(); @@ -98,11 +197,13 @@ public boolean isMatchAll() { && Objects.isNull(expression); } - public boolean shouldAutoRetainTagsAtRuntime() { - return Objects.isNull(boundColumnFilter); - } - public boolean isTimeSelected() { + if (Objects.nonNull(tableTimeSelected)) { + return tableTimeSelected; + } + if (Objects.nonNull(tableProvider)) { + return tableFilters.values().stream().anyMatch(cached -> cached.matcher.isTimeSelected()); + } if (isMatchAll()) { return true; } @@ -116,6 +217,12 @@ public boolean isTimeSelected() { } public boolean isTimeSelected(final String databaseName, final String tableName) { + if (Objects.nonNull(tableTimeSelected)) { + return tableTimeSelected; + } + if (Objects.nonNull(tableProvider)) { + return forTable(databaseName, tableName).isTimeSelected(databaseName, tableName); + } if (isMatchAll()) { return true; } @@ -136,6 +243,23 @@ public boolean isTimeSelected(final String databaseName, final String tableName) } public Map> getTimeSelectedByTable(final String databaseName) { + if (Objects.nonNull(tableProvider)) { + final Map> result = new HashMap<>(); + tableFilters + .keySet() + .forEach( + key -> { + if (Objects.isNull(databaseName) + || key.getDatabaseName().equals(normalize(databaseName))) { + result + .computeIfAbsent(key.getDatabaseName(), ignored -> new HashMap<>()) + .put( + key.getTableName(), + isTimeSelected(key.getDatabaseName(), key.getTableName())); + } + }); + return result; + } if (Objects.isNull(boundColumnFilter) || boundColumnFilter.getTimeSelectedByTable().isEmpty()) { return Collections.emptyMap(); } @@ -173,6 +297,10 @@ public boolean match( final String columnName, final TSDataType dataType, final ColumnCategory category) { + if (Objects.nonNull(tableProvider)) { + return forTable(databaseName, tableName) + .match(databaseName, tableName, columnName, dataType, category); + } if (isMatchAll()) { return true; } @@ -199,4 +327,14 @@ private static String normalize(final String value) { private static String normalizeNullable(final String value) { return Objects.nonNull(value) ? value.trim() : ""; } + + private static final class CachedTableFilter { + private final TsTable table; + private final ColumnFilterMatcher matcher; + + private CachedTableFilter(final TsTable table, final ColumnFilterMatcher matcher) { + this.table = table; + this.matcher = matcher; + } + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/TabletColumnPruner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/TabletColumnPruner.java index 2f438350b64d6..22e71a141c211 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/TabletColumnPruner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/TabletColumnPruner.java @@ -45,7 +45,9 @@ public static Tablet pruneTableModelTablet( } final ColumnFilterMatcher effectiveMatcher = - Objects.nonNull(matcher) ? matcher : ColumnFilterMatcher.matchAll(); + Objects.nonNull(matcher) + ? matcher.forTable(databaseName, tablet.getTableName()) + : ColumnFilterMatcher.matchAll(); if (effectiveMatcher.isMatchAll()) { return tablet; } @@ -67,6 +69,9 @@ public static Tablet pruneTableModelTablet( } final IMeasurementSchema schema = schemas.get(i); final ColumnCategory category = categories.get(i); + // TAGs identify the device, including TAGs added after a cached plan was compiled. They are + // retained only if the table has a match; otherwise the entire Tablet is discarded below. + selectedColumns[i] = category == ColumnCategory.TAG && hasValueArray(values, i); if (effectiveMatcher.match( databaseName, tablet.getTableName(), @@ -84,16 +89,6 @@ public static Tablet pruneTableModelTablet( return null; } - if (effectiveMatcher.shouldAutoRetainTagsAtRuntime()) { - for (int i = 0; i < schemas.size(); i++) { - if (isValidColumn(schemas, values, i) - && categories.get(i) == ColumnCategory.TAG - && hasValueArray(values, i)) { - selectedColumns[i] = true; - } - } - } - final List selectedIndices = getSelectedIndices(selectedColumns); if (selectedIndices.isEmpty()) { return timeSelected ? pruneAllPhysicalColumns(tablet) : null; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheLeaseTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheLeaseTest.java index 00c5388e747f0..865ef633248b9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheLeaseTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheLeaseTest.java @@ -69,6 +69,7 @@ public void tearDown() { public void fencedLeaseFailsClosedForReadApis() { nowNanos.addAndGet((T_FENCE_MS + 1) * 1_000_000L); assertLeaseFenced(() -> tableCache.getTableSnapshot()); + assertLeaseFenced(() -> tableCache.getTableInCache("root.db", "t")); assertLeaseFenced(() -> tableCache.getTableInWrite("root.db", "t")); assertLeaseFenced(() -> tableCache.getTable("root.db", "t", false)); assertLeaseFenced(() -> tableCache.isDatabaseExist("root.db")); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java index bec226b21ab09..db9b36c7555af 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java @@ -52,6 +52,26 @@ public class DataNodeTableCacheTest { private static final String TABLE_CACHE_TEST_DATABASE = "root.table_cache_test"; private static final String TABLE_NAME = "table1"; + @Test + public void localReadUsesCommittedSchemaWithoutResolvingPendingUpdates() { + final ITableCache cache = DataNodeTableCache.getInstance(); + cache.invalid(TABLE_CACHE_TEST_DATABASE); + try { + final TsTable initial = createTable(TABLE_NAME); + cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, initial, null); + Assert.assertNull(cache.getTableInCache(TABLE_CACHE_TEST_DATABASE, TABLE_NAME)); + cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); + Assert.assertSame(initial, cache.getTableInCache(TABLE_CACHE_TEST_DATABASE, TABLE_NAME)); + final TsTable updated = new TsTable(initial); + cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, updated, null); + Assert.assertSame(initial, cache.getTableInCache(TABLE_CACHE_TEST_DATABASE, TABLE_NAME)); + cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null); + Assert.assertSame(updated, cache.getTableInCache(TABLE_CACHE_TEST_DATABASE, TABLE_NAME)); + } finally { + cache.invalid(TABLE_CACHE_TEST_DATABASE); + } + } + @Test public void interruptedFetchDoesNotLeakSemaphorePermit() throws Exception { final ITableCache cache = DataNodeTableCache.getInstance(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterBinderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterBinderTest.java index 6e876d1eb5f6c..22904af8e57aa 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterBinderTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterBinderTest.java @@ -40,26 +40,27 @@ import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; public class ColumnFilterBinderTest { @Test - public void testBindSnapshotKeepsOnlyBoundTagsAtRuntime() { - final BoundColumnFilter boundFilter = - new ColumnFilterBinder() - .bind( - createTableTopicConfig("column_name = \"temperature\""), - Collections.singletonMap( - "db", Collections.singletonMap("sensors", createTableSchema()))); - final ColumnFilterMatcher matcher = ColumnFilterMatcher.fromBoundColumnFilter(boundFilter); + public void testStreamingFilterKeepsNewTagsAtRuntime() { + final TsTable table = createTableSchema(); + final ColumnFilterMatcher matcher = + ColumnFilterMatcher.fromTopicConfig( + createTableTopicConfig("column_name = \"temperature\""), (db, name) -> table); + final ColumnFilterMatcher compiled = matcher.forTable("db", "sensors"); final Tablet prunedTablet = TabletColumnPruner.pruneTableModelTablet(createRuntimeTabletWithNewTag(), "db", matcher); Assert.assertNotNull(prunedTablet); - Assert.assertEquals(2, prunedTablet.getSchemas().size()); + Assert.assertEquals(3, prunedTablet.getSchemas().size()); Assert.assertEquals("device", prunedTablet.getSchemas().get(0).getMeasurementName()); - Assert.assertEquals("temperature", prunedTablet.getSchemas().get(1).getMeasurementName()); + Assert.assertEquals("new_tag", prunedTablet.getSchemas().get(1).getMeasurementName()); + Assert.assertEquals("temperature", prunedTablet.getSchemas().get(2).getMeasurementName()); + Assert.assertSame(compiled, matcher.forTable("db", "sensors")); } @Test @@ -76,11 +77,80 @@ public void testAttributeMatchBindsTagsButNotAttribute() { TabletColumnPruner.pruneTableModelTablet(createRuntimeTabletWithNewTag(), "db", matcher); Assert.assertNotNull(prunedTablet); - Assert.assertEquals(1, prunedTablet.getSchemas().size()); + Assert.assertEquals(2, prunedTablet.getSchemas().size()); Assert.assertEquals("device", prunedTablet.getSchemas().get(0).getMeasurementName()); + Assert.assertEquals("new_tag", prunedTablet.getSchemas().get(1).getMeasurementName()); Assert.assertEquals(ColumnCategory.TAG, prunedTablet.getColumnTypes().get(0)); } + @Test + public void testStreamingFilterRefreshesChangedSchemaAndNewTables() { + final Map tables = new HashMap<>(); + final TsTable initialTable = createTableSchema(); + tables.put("sensors", initialTable); + final ColumnFilterMatcher matcher = + ColumnFilterMatcher.fromTopicConfig( + createTableTopicConfig("category = \"FIELD\" AND datatype = \"DOUBLE\""), + (db, name) -> tables.get(name)); + final ColumnFilterMatcher initialPlan = matcher.forTable("db", "sensors"); + Assert.assertTrue(initialPlan.match("db", "sensors", "temperature")); + Assert.assertFalse(initialPlan.match("db", "sensors", "new_field")); + Assert.assertSame(initialPlan, matcher.forTable("db", "sensors")); + + final TsTable updatedTable = new TsTable(initialTable); + updatedTable.addColumnSchema(new FieldColumnSchema("new_field", TSDataType.DOUBLE)); + updatedTable.addColumnSchema(new TagColumnSchema("new_tag", TSDataType.STRING)); + tables.put("sensors", updatedTable); + final ColumnFilterMatcher updatedPlan = matcher.forTable("db", "sensors"); + Assert.assertNotSame(initialPlan, updatedPlan); + Assert.assertTrue(updatedPlan.match("db", "sensors", "new_field")); + Assert.assertTrue(updatedPlan.match("db", "sensors", "new_tag")); + Assert.assertFalse(updatedPlan.match("db", "sensors", "status")); + + Assert.assertFalse(matcher.match("db", "events", "temperature")); + tables.put("events", createTableSchema("events", "event_time")); + Assert.assertTrue(matcher.match("db", "events", "temperature")); + tables.remove("sensors"); + Assert.assertFalse(matcher.match("db", "sensors", "temperature")); + } + + @Test + public void testStreamingFilterUsesLatestDatatypeAndViewAlias() { + final AtomicReference schema = new AtomicReference<>(createTreeViewSchema()); + final ColumnFilterMatcher matcher = + ColumnFilterMatcher.fromTopicConfig( + createTableTopicConfig("column_name = \"temp_alias\" AND datatype = \"INT64\""), + (db, name) -> schema.get()); + Assert.assertFalse(matcher.match("db", "sensors_view", "temperature")); + final TsTable changed = createTreeViewSchema(); + final FieldColumnSchema field = new FieldColumnSchema("temp_alias", TSDataType.INT64); + TreeViewSchema.setOriginalName(field, "temperature"); + changed.removeColumnSchema("temp_alias"); + changed.addColumnSchema(field); + schema.set(changed); + Assert.assertTrue(matcher.match("db", "sensors_view", "temperature")); + Assert.assertFalse(matcher.match("db", "sensors_view", "temp_alias")); + Assert.assertTrue(matcher.match("db", "sensors_view", "device_alias")); + } + + @Test + public void testStreamingFilterKeepsTimeSelectionPerTable() { + final Map tables = new HashMap<>(); + tables.put("sensors", createTableSchema()); + tables.put("events", createTableSchema("events", "event_time")); + final TopicConfig config = createTableTopicConfig("column_name = \"time\""); + config.getAttribute().put(TopicConstant.DATABASE_KEY, "db"); + final ColumnFilterMatcher matcher = + ColumnFilterMatcher.fromTopicConfig(config, (db, name) -> tables.get(name)); + Assert.assertTrue(matcher.isTimeSelected("db", "sensors")); + Assert.assertFalse(matcher.isTimeSelected("db", "events")); + Assert.assertFalse(matcher.isTimeSelected("other", "sensors")); + Assert.assertTrue(matcher.isTimeSelected()); + Assert.assertEquals( + Map.of("sensors", true, "events", false), matcher.getTimeSelectedByTable("db").get("db")); + Assert.assertTrue(matcher.getTimeSelectedByTable("other").isEmpty()); + } + @Test public void testBindTimeSelectionPerTable() { final Map tables = new HashMap<>();