This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 49ba3be90cb [Subscription] Make column filters follow schema changes
by default (#18925)
49ba3be90cb is described below
commit 49ba3be90cba1d16214606a10b45b134fd130cc0
Author: Caideyipi <[email protected]>
AuthorDate: Tue Oct 20 15:17:46 2026 +0800
[Subscription] Make column filters follow schema changes by default (#18925)
---
...ConsensusSubscriptionColumnFilterClusterIT.java | 58 +++++++
.../IoTDBSubscriptionColumnFilterIT.java | 170 ++++++++++++++++-----
iotdb-client/subscription/README.md | 71 +++++++++
.../db/schemaengine/table/DataNodeTableCache.java | 9 ++
.../iotdb/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 +
.../schemaengine/table/DataNodeTableCacheTest.java | 20 +++
.../columnfilter/ColumnFilterBinderTest.java | 92 +++++++++--
12 files changed, 531 insertions(+), 115 deletions(-)
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 e90d94af7ea..0875265b6f2 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 class IoTDBConsensusSubscriptionColumnFilterClusterIT
extends AbstractSub
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 =
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 bfeb7e6b7aa..d300e2c9bf6 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 class IoTDBSubscriptionColumnFilterIT extends
AbstractSubscriptionDualIT
}
@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 class IoTDBSubscriptionColumnFilterIT extends
AbstractSubscriptionDualIT
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 class IoTDBSubscriptionColumnFilterIT extends
AbstractSubscriptionDualIT
}
@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 class IoTDBSubscriptionColumnFilterIT extends
AbstractSubscriptionDualIT
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 class IoTDBSubscriptionColumnFilterIT extends
AbstractSubscriptionDualIT
}
@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 class IoTDBSubscriptionColumnFilterIT extends
AbstractSubscriptionDualIT
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 @@ public class IoTDBSubscriptionColumnFilterIT extends
AbstractSubscriptionDualIT
}
}
+ 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 00000000000..3764d75754a
--- /dev/null
+++ b/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.
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 525bf511f4e..3db7af6f89a 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 @@ public class DataNodeTableCache implements ITableCache {
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 13193fbff13..b87b6db7147 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 @@ public interface ITableCache {
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 aafcf903dc4..b5bd049b9c2 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.ConsensusPrefetchingQue
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.execution.ConsensusSubscriptionPref
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<String, ColumnFilterMatcher>
topicNameToColumnFilterMatcher =
new ConcurrentHashMap<>();
- private final ColumnFilterBinder columnFilterBinder = new
ColumnFilterBinder();
//////////////////////////// provided for subscription agent
////////////////////////////
@@ -732,17 +728,9 @@ public class SubscriptionBrokerAgent {
public void refreshColumnFilter(final String topicName, final TopicConfig
topicConfig) {
final ColumnFilterMatcher matcher;
try {
- final Map<String, Map<String, TsTable>> 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 @@ public class SubscriptionBrokerAgent {
: ColumnFilterMatcher.matchAll();
}
- private static Map<String, Map<String, TsTable>>
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 82f15db0fad..9417f6f1230 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 @@ public class ColumnFilterBinder {
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 8755bb9dc0d..1c9246710b6 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.Locale;
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<String> selectedColumnNames;
private final Expression expression;
+ private final TablePattern tablePattern;
+ private final BiFunction<String, String, TsTable> tableProvider;
+ private final Boolean tableTimeSelected;
+ private final ConcurrentMap<BoundColumnFilter.TableKey, CachedTableFilter>
tableFilters =
+ new ConcurrentHashMap<>();
private ColumnFilterMatcher(final Set<String> selectedColumnNames, final
Expression expression) {
this(null, selectedColumnNames, expression);
@@ -51,9 +62,38 @@ public class ColumnFilterMatcher {
final BoundColumnFilter boundColumnFilter,
final Set<String> selectedColumnNames,
final Expression expression) {
+ this(boundColumnFilter, selectedColumnNames, expression, null, null);
+ }
+
+ private ColumnFilterMatcher(
+ final BoundColumnFilter boundColumnFilter,
+ final Set<String> selectedColumnNames,
+ final Expression expression,
+ final TablePattern tablePattern,
+ final BiFunction<String, String, TsTable> tableProvider) {
+ this(boundColumnFilter, selectedColumnNames, expression, tablePattern,
tableProvider, null);
+ }
+
+ private ColumnFilterMatcher(
+ final BoundColumnFilter boundColumnFilter,
+ final Set<String> selectedColumnNames,
+ final Expression expression,
+ final TablePattern tablePattern,
+ final BiFunction<String, String, TsTable> 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<String> selectedColumnNames, final boolean timeSelected) {
+ return new ColumnFilterMatcher(
+ null, Collections.unmodifiableSet(selectedColumnNames), null, null,
null, timeSelected);
}
public static ColumnFilterMatcher matchAll() {
@@ -79,6 +119,65 @@ public class ColumnFilterMatcher {
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<String, String, TsTable>
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<String>
selectedColumnNames) {
if (Objects.isNull(selectedColumnNames)) {
return matchAll();
@@ -98,11 +197,13 @@ public class ColumnFilterMatcher {
&& 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 class ColumnFilterMatcher {
}
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 class ColumnFilterMatcher {
}
public Map<String, Map<String, Boolean>> getTimeSelectedByTable(final String
databaseName) {
+ if (Objects.nonNull(tableProvider)) {
+ final Map<String, Map<String, Boolean>> 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 class ColumnFilterMatcher {
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 @@ public class ColumnFilterMatcher {
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 2f438350b64..22e71a141c2 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 class TabletColumnPruner {
}
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 class TabletColumnPruner {
}
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 class TabletColumnPruner {
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<Integer> 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 00c5388e747..865ef633248 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 class DataNodeTableCacheLeaseTest {
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 bec226b21ab..db9b36c7555 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 6e876d1eb5f..22904af8e57 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.Arrays;
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 class ColumnFilterBinderTest {
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<String, TsTable> 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<TsTable> 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<String, TsTable> 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<String, TsTable> tables = new HashMap<>();
