[Feature] Support TAG value filtering for table subscriptions (#18639)
* Subscription: support table-model tag filtering
* Subscription: expose tag filter diagnostics in SHOW TOPICS
* Address review feedback for table subscription TAG filters
* Preserve quoted TAG filters in topic SQL formatting
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/ConsensusSubscriptionTableITSupport.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/ConsensusSubscriptionTableITSupport.java
index 29a859d..6f4e828 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/ConsensusSubscriptionTableITSupport.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/ConsensusSubscriptionTableITSupport.java
@@ -125,6 +125,16 @@
final String tablePattern,
final String columnFilter)
throws Exception {
+ createConsensusTopic(topicName, databasePattern, tablePattern, columnFilter, null);
+ }
+
+ static void createConsensusTopic(
+ final String topicName,
+ final String databasePattern,
+ final String tablePattern,
+ final String columnFilter,
+ final String tagFilter)
+ throws Exception {
final String host = EnvFactory.getEnv().getIP();
final int port = Integer.parseInt(EnvFactory.getEnv().getPort());
@@ -141,6 +151,9 @@
if (columnFilter != null) {
config.put(TopicConstant.COLUMN_FILTER_KEY, columnFilter);
}
+ if (tagFilter != null) {
+ config.put(TopicConstant.TAG_FILTER_KEY, tagFilter);
+ }
session.createTopic(topicName, config);
}
}
@@ -160,6 +173,21 @@
}
}
+ static void alterConsensusTopicTagFilter(final String topicName, final String tagFilter)
+ throws Exception {
+ final String host = EnvFactory.getEnv().getIP();
+ final int port = Integer.parseInt(EnvFactory.getEnv().getPort());
+
+ try (final ISubscriptionTableSession session =
+ new SubscriptionTableSessionBuilder().host(host).port(port).build()) {
+ session.open();
+
+ final Properties config = new Properties();
+ config.put(TopicConstant.TAG_FILTER_KEY, tagFilter);
+ session.alterTopic(topicName, config);
+ }
+ }
+
static SubscriptionTablePullConsumer createConsumer(
final String consumerId, final String consumerGroupId) throws Exception {
final SubscriptionTablePullConsumer consumer =
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionFilterTableIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionFilterTableIT.java
index 96193a0..f311d00 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionFilterTableIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionFilterTableIT.java
@@ -283,6 +283,80 @@
}
@Test
+ public void testTagFilteringAndAlterInConsensusQueue() throws Exception {
+ final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
+ ConsensusSubscriptionTableITSupport.newIdentifiers("table_filter_tags");
+ final String database = ids.getDatabase();
+ final String table = "t1";
+ final String schema = "tag1 STRING TAG, s1 INT64 FIELD, s2 DOUBLE FIELD";
+ final String expectedColumnSignature =
+ ConsensusSubscriptionTableITSupport.normalizeColumnSignature("tag1", "s2");
+ SubscriptionTablePullConsumer consumer = null;
+
+ try {
+ ConsensusSubscriptionTableITSupport.createDatabaseAndTable(database, table, schema);
+ try (final ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) {
+ session.executeNonQueryStatement("use " + database);
+ session.executeNonQueryStatement(
+ "insert into " + table + "(tag1, s1, s2, time) values ('bootstrap', 0, 0.0, 0)");
+ session.executeNonQueryStatement("flush");
+ }
+
+ ConsensusSubscriptionTableITSupport.createConsensusTopic(
+ ids.getTopic(), database, table, "column_name = \"s2\"", "tag1 = \"keep\"");
+
+ consumer =
+ ConsensusSubscriptionTableITSupport.createConsumer(
+ ids.getConsumerId(), ids.getConsumerGroupId());
+ consumer.subscribe(ids.getTopic());
+
+ final Set<String> expectedBeforeAlter = new LinkedHashSet<>();
+ try (final ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) {
+ session.executeNonQueryStatement("use " + database);
+ for (int i = 1; i <= 5; i++) {
+ final long timestamp = 100L + i;
+ final String tag = i % 2 == 1 ? "keep" : "drop";
+ session.executeNonQueryStatement(
+ String.format(
+ Locale.ROOT,
+ "insert into %s(tag1, s1, s2, time) values ('%s', %d, %.1f, %d)",
+ table,
+ tag,
+ i * 10L,
+ i + 0.5d,
+ timestamp));
+ if ("keep".equals(tag)) {
+ expectedBeforeAlter.add(
+ ConsensusSubscriptionTableITSupport.rowKey(database, table, timestamp));
+ }
+ }
+ session.executeNonQueryStatement("flush");
+ }
+
+ final ConsensusSubscriptionTableITSupport.ConsumedRecords consumedBeforeAlter =
+ ConsensusSubscriptionTableITSupport.pollAndCommitUntilContains(
+ consumer, expectedBeforeAlter, 50);
+ ConsensusSubscriptionTableITSupport.assertExactRowKeys(
+ expectedBeforeAlter, consumedBeforeAlter);
+ Assert.assertEquals(
+ Collections.singleton(expectedColumnSignature),
+ consumedBeforeAlter.getSeenColumnSignatures());
+
+ ConsensusSubscriptionTableITSupport.alterConsensusTopicTagFilter(
+ ids.getTopic(), "tag1 LIKE \"t1_tag_%\"");
+ final ConsensusSubscriptionTableITSupport.ConsumedRecords consumedAfterAlter =
+ ConsensusSubscriptionTableITSupport.insertRowsAndPollUntilColumnSignature(
+ consumer, database, table, 200L, 3, true, false, expectedColumnSignature, 60);
+ Assert.assertTrue(consumedAfterAlter.getRowCount() > 0);
+ Assert.assertTrue(
+ consumedAfterAlter.getSeenColumnSignatures().contains(expectedColumnSignature));
+ ConsensusSubscriptionTableITSupport.assertNoMoreMessages(consumer, 3, Duration.ofMillis(500));
+ } finally {
+ ConsensusSubscriptionTableITSupport.cleanup(consumer, ids.getTopic(), database);
+ }
+ }
+
+ @Test
public void testColumnFilteringComplexOperatorsInConsensusQueue() throws Exception {
final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
ConsensusSubscriptionTableITSupport.newIdentifiers("table_filter_complex_operators");
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 d300e2c..2e4fc5f 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
@@ -452,6 +452,73 @@
}
@Test
+ public void testLiveRecordHandlerTagFilterBeforeColumnFilter() throws Exception {
+ final String database = databaseName("tag_record");
+ final String topicName = topicName("tag_record");
+ final String consumerId = consumerName("tag_record");
+ final String consumerGroupId = consumerGroupName("tag_record");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createTopicWithFilters(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.MODE_INITIAL_VALUE,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ COLUMN_FILTER,
+ "tag1 IN (\"tag_1\", \"tag_3\")");
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ insertRows(senderEnv, database, TABLE_NAME, 1, 5);
+
+ final Set<Long> expectedTimestamps = new LinkedHashSet<>(Arrays.asList(1L, 3L));
+ final ConsumedRecordStats stats =
+ pollRecordMessagesForTimestamps(consumer, expectedTimestamps, true);
+
+ Assert.assertEquals(expectedTimestamps, stats.timestamps);
+ Assert.assertEquals(EXPECTED_COLUMNS, stats.columnNames);
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void testSnapshotTsFileTagFilter() throws Exception {
+ final String database = databaseName("tag_snapshot_tsfile");
+ final String topicName = topicName("tag_snapshot_tsfile");
+ final String consumerId = consumerName("tag_snapshot_tsfile");
+ final String consumerGroupId = consumerGroupName("tag_snapshot_tsfile");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createDatabaseAndTable(receiverEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ insertRows(senderEnv, database, TABLE_NAME, 10, 14);
+ createTopicWithFilters(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.MODE_SNAPSHOT_VALUE,
+ TopicConstant.FORMAT_TS_FILE_VALUE,
+ TopicConstant.COLUMN_FILTER_DEFAULT_VALUE,
+ "tag1 IN (\"tag_10\", \"tag_12\")");
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ pollTsFileMessagesAndLoad(consumer, database, 2);
+ }
+
+ assertLoadedTimestamps(database, new LinkedHashSet<>(Arrays.asList(10L, 12L)));
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
public void testSnapshotTsFileColumnFilter() throws Exception {
final String database = databaseName("snapshot_tsfile");
final String topicName = topicName("snapshot_tsfile");
@@ -1102,6 +1169,34 @@
topicName, database, tableName, TopicConstant.MODE_INITIAL_VALUE, format, "", false);
}
+ private void createTopicWithFilters(
+ final String topicName,
+ final String database,
+ final String tableName,
+ final String mode,
+ final String format,
+ final String columnFilter,
+ final String tagFilter)
+ throws Exception {
+ try (final ISubscriptionTableSession session =
+ new SubscriptionTableSessionBuilder()
+ .host(senderEnv.getIP())
+ .port(Integer.parseInt(senderEnv.getPort()))
+ .build()) {
+ session.open();
+ session.dropTopicIfExists(topicName);
+
+ final Properties config = new Properties();
+ config.put(TopicConstant.MODE_KEY, mode);
+ config.put(TopicConstant.FORMAT_KEY, format);
+ config.put(TopicConstant.DATABASE_KEY, database);
+ config.put(TopicConstant.TABLE_KEY, tableName);
+ config.put(TopicConstant.COLUMN_FILTER_KEY, columnFilter);
+ config.put(TopicConstant.TAG_FILTER_KEY, tagFilter);
+ session.createTopic(topicName, config);
+ }
+ }
+
private void createTopic(
final String topicName,
final String database,
@@ -1547,6 +1642,22 @@
}
}
+ private void assertLoadedTimestamps(final String database, final Set<Long> expectedTimestamps)
+ throws Exception {
+ final Set<Long> actualTimestamps = new LinkedHashSet<>();
+ try (final Connection connection = receiverEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("use " + database);
+ try (final java.sql.ResultSet resultSet =
+ statement.executeQuery("select time from " + TABLE_NAME + " order by time")) {
+ while (resultSet.next()) {
+ actualTimestamps.add(resultSet.getLong("time"));
+ }
+ }
+ }
+ Assert.assertEquals(expectedTimestamps, actualTimestamps);
+ }
+
private void cleanup(final String topicName, final String database) {
try (final ISubscriptionTableSession session =
new SubscriptionTableSessionBuilder()
diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java
index bc4a545..3980f09 100644
--- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java
+++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java
@@ -28,16 +28,23 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collections;
+import java.util.EnumMap;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
public class TopicConfig extends PipeParameters {
+ private static final AtomicLong NEXT_FILTER_VERSION = new AtomicLong();
+
+ // DataNode-local revision. It is not part of the serialized topic attributes.
+ private long filterVersion = NEXT_FILTER_VERSION.incrementAndGet();
+
private static final Set<String> MODE_VALUE_SET;
private static final Set<String> ORDER_MODE_VALUE_SET;
@@ -63,6 +70,37 @@
super(attributes);
}
+ public TopicConfig(final TopicConfig config) {
+ super(new HashMap<>(config.getAttribute()));
+ filterVersion = config.filterVersion;
+ }
+
+ public long getFilterVersion() {
+ return filterVersion;
+ }
+
+ /** Called before publishing replacement topic metadata on a DataNode. */
+ public void inheritFilterVersion(final TopicConfig previousConfig) {
+ if (previousConfig != null) {
+ filterVersion =
+ getFilteringAttributes().equals(previousConfig.getFilteringAttributes())
+ ? previousConfig.filterVersion
+ : NEXT_FILTER_VERSION.incrementAndGet();
+ }
+ }
+
+ private Map<TopicProperty, String> getFilteringAttributes() {
+ final Map<TopicProperty, String> result = new EnumMap<>(TopicProperty.class);
+ attributes.forEach(
+ (key, value) -> {
+ final TopicProperty property = TopicProperty.fromKey(key);
+ if (property != null && property.isFilteringProperty()) {
+ result.put(property, value);
+ }
+ });
+ return result;
+ }
+
// TODO: hide from the client
// refer to org.apache.iotdb.commons.pipe.config.constant.SystemConstant
private static final String SQL_DIALECT_KEY = "__system.sql-dialect";
@@ -254,6 +292,23 @@
return TopicConstant.COLUMN_FILTER_DEFAULT_VALUE.equalsIgnoreCase(getColumnFilter().trim());
}
+ public Map<String, String> getAttributesWithSourceTagFilter() {
+ return Collections.singletonMap(TopicConstant.TAG_FILTER_KEY, getTagFilter());
+ }
+
+ public String getTagFilter() {
+ return getStringIgnoreCase(
+ TopicConstant.TAG_FILTER_KEY, TopicConstant.TAG_FILTER_DEFAULT_VALUE);
+ }
+
+ public boolean hasTagFilter() {
+ return containsKeyIgnoreCase(TopicConstant.TAG_FILTER_KEY);
+ }
+
+ public boolean isTagFilterTrivial() {
+ return TopicConstant.TAG_FILTER_DEFAULT_VALUE.equalsIgnoreCase(getTagFilter().trim());
+ }
+
public Map<String, String> getAttributesWithSourceTimeRange() {
final Map<String, String> attributesWithTimeRange = new HashMap<>();
diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java
index 8a8af71..f6f3b59 100644
--- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java
+++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java
@@ -31,6 +31,7 @@
public static final String DATABASE_KEY = "database";
public static final String TABLE_KEY = "table";
public static final String COLUMN_FILTER_KEY = "column-filter";
+ public static final String TAG_FILTER_KEY = "tag-filter";
public static final String RETENTION_BYTES_KEY = "retention.bytes";
public static final String RETENTION_MS_KEY = "retention.ms";
public static final String RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY =
@@ -39,6 +40,7 @@
public static final String DATABASE_DEFAULT_VALUE = ".*";
public static final String TABLE_DEFAULT_VALUE = ".*";
public static final String COLUMN_FILTER_DEFAULT_VALUE = "true";
+ public static final String TAG_FILTER_DEFAULT_VALUE = "true";
public static final String START_TIME_KEY = "start-time";
public static final String END_TIME_KEY = "end-time";
diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicProperty.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicProperty.java
new file mode 100644
index 0000000..c5b5915
--- /dev/null
+++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicProperty.java
@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.rpc.subscription.config;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+
+/** Built-in topic keys and the properties that affect a batch's filtering configuration. */
+public enum TopicProperty {
+ PATH(TopicConstant.PATH_KEY, true),
+ PATTERN(TopicConstant.PATTERN_KEY, true),
+ DATABASE(TopicConstant.DATABASE_KEY, true),
+ TABLE(TopicConstant.TABLE_KEY, true),
+ COLUMN_FILTER(TopicConstant.COLUMN_FILTER_KEY, true),
+ TAG_FILTER(TopicConstant.TAG_FILTER_KEY, true),
+ START_TIME(TopicConstant.START_TIME_KEY, true),
+ END_TIME(TopicConstant.END_TIME_KEY, true),
+ LOOSE_RANGE(TopicConstant.LOOSE_RANGE_KEY, true),
+ STRICT(TopicConstant.STRICT_KEY, true),
+ SQL_DIALECT("__system.sql-dialect", true),
+ SOURCE_CAPTURE_TREE("source.capture.tree", true),
+ EXTRACTOR_CAPTURE_TREE("extractor.capture.tree", true),
+ RETENTION_BYTES(TopicConstant.RETENTION_BYTES_KEY, false),
+ RETENTION_MS(TopicConstant.RETENTION_MS_KEY, false),
+ RETAIN_PROGRESS_AFTER_UNSUBSCRIBE(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, false),
+ MODE(TopicConstant.MODE_KEY, false),
+ ORDER_MODE(TopicConstant.ORDER_MODE_KEY, false),
+ FORMAT(TopicConstant.FORMAT_KEY, false),
+ RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS(
+ TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, false),
+ OWNER_ID(TopicConstant.OWNER_ID_KEY, false),
+ OWNER_EPOCH(TopicConstant.OWNER_EPOCH_KEY, false),
+ MAX_OWNER_EPOCH(TopicConstant.MAX_OWNER_EPOCH_KEY, false),
+ OWNER_LEASE_DURATION_MS(TopicConstant.OWNER_LEASE_DURATION_MS_KEY, false);
+
+ private static final Map<String, TopicProperty> PROPERTIES_BY_KEY;
+
+ static {
+ final Map<String, TopicProperty> properties = new HashMap<>();
+ for (final TopicProperty property : values()) {
+ properties.put(property.key, property);
+ }
+ PROPERTIES_BY_KEY = Collections.unmodifiableMap(properties);
+ }
+
+ private final String key;
+ private final boolean filteringProperty;
+
+ TopicProperty(final String key, final boolean filteringProperty) {
+ this.key = key;
+ this.filteringProperty = filteringProperty;
+ }
+
+ public String getKey() {
+ return key;
+ }
+
+ public boolean isFilteringProperty() {
+ return filteringProperty;
+ }
+
+ public static TopicProperty fromKey(final String key) {
+ return key == null ? null : PROPERTIES_BY_KEY.get(key.trim().toLowerCase(Locale.ROOT));
+ }
+
+ /** Canonicalize built-in keys while preserving extension keys and all values verbatim. */
+ public static String canonicalizeKey(final String key) {
+ final TopicProperty property = fromKey(key);
+ return property == null ? key : property.getKey();
+ }
+}
diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/AbstractSubscriptionSession.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/AbstractSubscriptionSession.java
index c09ebdf..e4b1e1c 100644
--- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/AbstractSubscriptionSession.java
+++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/AbstractSubscriptionSession.java
@@ -241,7 +241,7 @@
while (dataSet.hasNext()) {
final RowRecord record = dataSet.next();
final List<Field> fields = record.getFields();
- if (fields.size() != 2) {
+ if (fields.size() < 2) {
throw new SubscriptionException(
String.format(
SubscriptionMessages
diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
index 113971a..1cb614c 100644
--- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
+++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
@@ -1279,7 +1279,8 @@
}
private Optional<SubscriptionMessage> pollTabletsInternal(
- final SubscriptionPollResponse initialResponse, final PollTimer timer) {
+ final SubscriptionPollResponse initialResponse, final PollTimer timer)
+ throws SubscriptionException {
final Map<String, List<Tablet>> tablets =
((TabletsPayload) initialResponse.getPayload()).getTabletsWithDBInfo();
final SubscriptionCommitContext commitContext = initialResponse.getCommitContext();
@@ -1299,6 +1300,10 @@
LOGGER.warn(errorMessage);
throw new SubscriptionRuntimeNonCriticalException(errorMessage);
}
+ if (tabletsSize == 0) {
+ ack(Collections.singletonList(new SubscriptionMessage(commitContext, tablets)));
+ return Optional.empty();
+ }
return Optional.of(
new SubscriptionMessage(commitContext, tablets, timeSelected, timeSelectedByTable));
}
diff --git a/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java b/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java
index 0235feb..d3524b6 100644
--- a/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java
+++ b/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java
@@ -29,6 +29,59 @@
public class TopicConfigTest {
@Test
+ public void testFilterVersionIgnoresOwnerRetentionAndUnknownProperties() {
+ final TopicConfig original =
+ new TopicConfig(Collections.singletonMap("Tag-Filter", "region = \"north\""));
+ final Map<String, String> attributes = new HashMap<>(original.getAttribute());
+ attributes.put(TopicConstant.OWNER_ID_KEY, "owner-1");
+ attributes.put(TopicConstant.OWNER_EPOCH_KEY, "1");
+ attributes.put(TopicConstant.RETENTION_MS_KEY, "60000");
+ attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "1000");
+ attributes.put("future.non-filtering-property", "value");
+ final TopicConfig updated = new TopicConfig(attributes);
+ updated.inheritFilterVersion(original);
+
+ Assert.assertEquals(original.getFilterVersion(), updated.getFilterVersion());
+ }
+
+ @Test
+ public void testFilterVersionChangesForPredicateBindingAndTreeCapture() {
+ final TopicConfig original =
+ new TopicConfig(
+ Map.of(
+ TopicConstant.TAG_FILTER_KEY, "region = \"north\"",
+ TopicConstant.DATABASE_KEY, "db",
+ TopicConstant.TABLE_KEY, "weather"));
+ for (final Map.Entry<String, String> change :
+ Map.of(
+ TopicConstant.TAG_FILTER_KEY,
+ "region = \"North\"",
+ TopicConstant.DATABASE_KEY,
+ "db2",
+ TopicConstant.TABLE_KEY,
+ "weather2",
+ "source.capture.tree",
+ "true")
+ .entrySet()) {
+ final TopicConfig updated = new TopicConfig(original);
+ updated.getAttribute().put(change.getKey(), change.getValue());
+ updated.inheritFilterVersion(original);
+ Assert.assertNotEquals(original.getFilterVersion(), updated.getFilterVersion());
+ }
+ }
+
+ @Test
+ public void testSnapshotCopyPreservesFilterVersionAndOwnsAttributes() {
+ final TopicConfig original =
+ new TopicConfig(Collections.singletonMap(TopicConstant.TAG_FILTER_KEY, "false"));
+ final TopicConfig snapshot = new TopicConfig(original);
+ Assert.assertEquals(original.getFilterVersion(), snapshot.getFilterVersion());
+
+ snapshot.getAttribute().put(TopicConstant.TAG_FILTER_KEY, "true");
+ Assert.assertEquals("false", original.getTagFilter());
+ }
+
+ @Test
public void testModeDefaultsToInitial() {
final TopicConfig topicConfig = new TopicConfig();
@@ -133,6 +186,35 @@
}
@Test
+ public void testTagFilterKeyIsCaseInsensitive() {
+ final TopicConfig topicConfig =
+ new TopicConfig(Collections.singletonMap("Tag-Filter", "region = \"north\""));
+
+ Assert.assertTrue(topicConfig.hasTagFilter());
+ Assert.assertEquals("region = \"north\"", topicConfig.getTagFilter());
+ Assert.assertEquals(
+ "region = \"north\"",
+ topicConfig.getAttributesWithSourceTagFilter().get(TopicConstant.TAG_FILTER_KEY));
+ }
+
+ @Test
+ public void testTagFilterDefaultsToTrivialWhenAbsent() {
+ final TopicConfig topicConfig = new TopicConfig(new HashMap<>());
+
+ Assert.assertFalse(topicConfig.hasTagFilter());
+ Assert.assertTrue(topicConfig.isTagFilterTrivial());
+ Assert.assertEquals(TopicConstant.TAG_FILTER_DEFAULT_VALUE, topicConfig.getTagFilter());
+ }
+
+ @Test
+ public void testTagFilterTrivialWithMixedCaseKeyAndValue() {
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("TAG-FILTER", " TRUE ");
+
+ Assert.assertTrue(new TopicConfig(attributes).isTagFilterTrivial());
+ }
+
+ @Test
public void testRetainProgressAfterUnsubscribeDefaultsToFalse() {
Assert.assertFalse(new TopicConfig().isProgressRetainedAfterUnsubscribe());
}
diff --git a/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerMultiProviderPollTest.java b/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerMultiProviderPollTest.java
index 34ef7a8..5faeff9 100644
--- a/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerMultiProviderPollTest.java
+++ b/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerMultiProviderPollTest.java
@@ -39,6 +39,7 @@
import org.junit.Assert;
import org.junit.Test;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -93,14 +94,41 @@
}
}
+ @Test
+ public void testFullyFilteredTabletBatchIsAcknowledgedAndPollReturnsNoMessages()
+ throws SubscriptionException {
+ final TestPullConsumer consumer = new TestPullConsumer(1, 0, true);
+ try {
+ consumer.open();
+ consumer.subscribeTopic();
+
+ final List<SubscriptionMessage> messages = consumer.pollForTest(1_000L);
+
+ Assert.assertTrue(messages.isEmpty());
+ Assert.assertEquals(1, consumer.getAcknowledgedCommitContexts().size());
+ Assert.assertEquals(
+ new SubscriptionCommitContext(1, 0, TOPIC, CONSUMER_GROUP_ID, 0L),
+ consumer.getAcknowledgedCommitContexts().get(0));
+ } finally {
+ consumer.close();
+ }
+ }
+
private static class TestPullConsumer extends AbstractSubscriptionPullConsumer {
private final Map<Integer, Integer> pollCounts = new HashMap<>();
+ private final List<SubscriptionCommitContext> acknowledgedCommitContexts = new ArrayList<>();
private final int dataProviderId;
private final int emptyPollsBeforeData;
+ private final boolean emptyTabletBatch;
private int backoffCount;
private TestPullConsumer(final int dataProviderId, final int emptyPollsBeforeData) {
+ this(dataProviderId, emptyPollsBeforeData, false);
+ }
+
+ private TestPullConsumer(
+ final int dataProviderId, final int emptyPollsBeforeData, final boolean emptyTabletBatch) {
super(
new AbstractSubscriptionPullConsumerBuilder()
.host(HOST)
@@ -112,6 +140,7 @@
.autoCommit(false));
this.dataProviderId = dataProviderId;
this.emptyPollsBeforeData = emptyPollsBeforeData;
+ this.emptyTabletBatch = emptyTabletBatch;
}
@Override
@@ -141,7 +170,8 @@
connectionTimeoutInMs,
pollCounts,
dataProviderId,
- emptyPollsBeforeData);
+ emptyPollsBeforeData,
+ emptyTabletBatch);
}
private void subscribeTopic() {
@@ -161,6 +191,17 @@
return backoffCount;
}
+ private List<SubscriptionCommitContext> getAcknowledgedCommitContexts() {
+ return acknowledgedCommitContexts;
+ }
+
+ @Override
+ protected void ack(final Iterable<SubscriptionMessage> messages) throws SubscriptionException {
+ for (final SubscriptionMessage message : messages) {
+ acknowledgedCommitContexts.add(message.getCommitContext());
+ }
+ }
+
@Override
void sleepAfterEmptyPollRound() {
backoffCount++;
@@ -173,6 +214,7 @@
private final Map<Integer, Integer> pollCounts;
private final int dataProviderId;
private final int emptyPollsBeforeData;
+ private final boolean emptyTabletBatch;
private TestSubscriptionProvider(
final TEndPoint endPoint,
@@ -188,7 +230,8 @@
final int connectionTimeoutInMs,
final Map<Integer, Integer> pollCounts,
final int dataProviderId,
- final int emptyPollsBeforeData) {
+ final int emptyPollsBeforeData,
+ final boolean emptyTabletBatch) {
super(
endPoint,
username,
@@ -205,6 +248,7 @@
this.pollCounts = pollCounts;
this.dataProviderId = dataProviderId;
this.emptyPollsBeforeData = emptyPollsBeforeData;
+ this.emptyTabletBatch = emptyTabletBatch;
}
@Override
@@ -265,6 +309,18 @@
}
final SubscriptionCommitContext commitContext =
new SubscriptionCommitContext(dataNodeId, 0, TOPIC, CONSUMER_GROUP_ID, 0L);
+ if (emptyTabletBatch) {
+ final int pollCount = pollCounts.get(dataNodeId);
+ if (pollCount != emptyPollsBeforeData + 1) {
+ return Collections.emptyList();
+ }
+ return new ArrayList<>(
+ Collections.singletonList(
+ new SubscriptionPollResponse(
+ SubscriptionPollResponseType.TABLETS.getType(),
+ new TabletsPayload(Collections.emptyMap(), 1),
+ commitContext)));
+ }
final List<IMeasurementSchema> schemas =
Collections.singletonList(new MeasurementSchema("s1", TSDataType.INT64));
final Tablet tablet = new Tablet("root.sg.d1", schemas, 1);
@@ -277,5 +333,19 @@
new TabletsPayload(Collections.singletonList(tablet), -1),
commitContext));
}
+
+ @Override
+ List<SubscriptionPollResponse> pollTablets(
+ final SubscriptionCommitContext commitContext, final int offset, final long timeoutMs)
+ throws SubscriptionException {
+ if (emptyTabletBatch && dataNodeId == dataProviderId && offset == 1) {
+ return Collections.singletonList(
+ new SubscriptionPollResponse(
+ SubscriptionPollResponseType.TABLETS.getType(),
+ new TabletsPayload(Collections.emptyMap(), 0),
+ commitContext));
+ }
+ return Collections.emptyList();
+ }
}
}
diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
index 4aea615..cfde56c 100644
--- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
+++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
@@ -719,6 +719,15 @@
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_AND_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_D86CEA8E =
"Failed to create or alter topic, %s and %s are only supported for incremental topics";
public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_DUPLICATE_ARG_ATTRIBUTES_ARE_NOT_ALLOWED_27315578 =
+ "Failed to create or alter topic, duplicate %s attributes are not allowed";
+ public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_IS_ONLY_SUPPORTED_FOR_TABLE_TOPICS_A5126607 =
+ "Failed to create or alter topic, %s is only supported for table topics";
+ public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_SHOULD_NOT_BE_EMPTY_767B1148 =
+ "Failed to create or alter topic, %s should not be empty";
+ public static final String
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ATTRIBUTES_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_55CB0BF2 =
"Failed to create or alter topic, attributes %s are only supported for incremental topics";
public static final String
diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
index 35436ee..ddf8be6 100644
--- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
+++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
@@ -761,6 +761,15 @@
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_AND_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_D86CEA8E =
"创建或修改 topic 失败，%s 和 %s 仅支持 incremental 模式的 topic";
public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_DUPLICATE_ARG_ATTRIBUTES_ARE_NOT_ALLOWED_27315578 =
+ "创建或修改 Topic 失败，不允许重复的 %s 属性";
+ public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_IS_ONLY_SUPPORTED_FOR_TABLE_TOPICS_A5126607 =
+ "创建或修改 Topic 失败，%s 仅支持表模型 Topic";
+ public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_SHOULD_NOT_BE_EMPTY_767B1148 =
+ "创建或修改 Topic 失败，%s 不应为空";
+ public static final String
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ATTRIBUTES_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_55CB0BF2 =
"创建或修改 topic 失败，属性 %s 仅支持 incremental 模式的 topic";
public static final String
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
index c0b6d10..79a735d 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
@@ -54,6 +54,7 @@
import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.common.DataSet;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterParser;
import org.apache.iotdb.mpp.rpc.thrift.TTopicOwnerLeaseEntry;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -92,6 +93,7 @@
public class SubscriptionInfo implements SnapshotProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionInfo.class);
+ private static final TagFilterParser TAG_FILTER_PARSER = new TagFilterParser();
private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf();
@@ -106,6 +108,7 @@
TopicConstant.DATABASE_KEY,
TopicConstant.TABLE_KEY,
TopicConstant.COLUMN_FILTER_KEY,
+ TopicConstant.TAG_FILTER_KEY,
TopicConstant.RETENTION_BYTES_KEY,
TopicConstant.RETENTION_MS_KEY,
TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
@@ -129,6 +132,7 @@
TopicConstant.DATABASE_KEY,
TopicConstant.TABLE_KEY,
TopicConstant.COLUMN_FILTER_KEY,
+ TopicConstant.TAG_FILTER_KEY,
TopicConstant.RETENTION_BYTES_KEY,
TopicConstant.RETENTION_MS_KEY,
TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
@@ -329,6 +333,11 @@
}
}
+ public void validateUpdatedTopicAttributes(final Map<String, String> updatedAttributes)
+ throws SubscriptionException {
+ validateDuplicateTopicAttributes(new TopicConfig(safeTopicAttributes(updatedAttributes)));
+ }
+
private void checkBeforeAlteringTopicInternal(TopicMeta topicMeta) throws SubscriptionException {
validateTopicConfig(topicMeta.getConfig());
@@ -400,6 +409,7 @@
}
validateColumnFilter(topicConfig);
+ validateTagFilter(topicConfig);
validateIncrementalTopicRetentionConfig(topicConfig);
validateRecycleUncommittedEventInterval(topicConfig);
@@ -553,7 +563,8 @@
if (!seenKeys.add(normalizedKey)) {
final String exceptionMessage =
String.format(
- "Failed to create or alter topic, duplicate %s attributes are not allowed",
+ ConfigNodeMessages
+ .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_DUPLICATE_ARG_ATTRIBUTES_ARE_NOT_ALLOWED_27315578,
normalizedKey);
LOGGER.warn(exceptionMessage);
throw new SubscriptionException(exceptionMessage);
@@ -562,24 +573,50 @@
}
private void validateColumnFilter(final TopicConfig topicConfig) throws SubscriptionException {
- if (!topicConfig.hasColumnFilter()) {
+ validateTableOnlyFilter(
+ topicConfig,
+ TopicConstant.COLUMN_FILTER_KEY,
+ topicConfig.hasColumnFilter(),
+ topicConfig.getColumnFilter());
+ }
+
+ private void validateTagFilter(final TopicConfig topicConfig) throws SubscriptionException {
+ validateTableOnlyFilter(
+ topicConfig,
+ TopicConstant.TAG_FILTER_KEY,
+ topicConfig.hasTagFilter(),
+ topicConfig.getTagFilter());
+ if (topicConfig.hasTagFilter()) {
+ TAG_FILTER_PARSER.parseAndValidate(topicConfig.getTagFilter());
+ }
+ }
+
+ private void validateTableOnlyFilter(
+ final TopicConfig topicConfig,
+ final String filterKey,
+ final boolean hasFilter,
+ final String filter)
+ throws SubscriptionException {
+ if (!hasFilter) {
return;
}
if (!topicConfig.isTableTopic()) {
final String exceptionMessage =
String.format(
- "Failed to create or alter topic, %s is only supported for table topics",
- TopicConstant.COLUMN_FILTER_KEY);
+ ConfigNodeMessages
+ .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_IS_ONLY_SUPPORTED_FOR_TABLE_TOPICS_A5126607,
+ filterKey);
LOGGER.warn(exceptionMessage);
throw new SubscriptionException(exceptionMessage);
}
- if (topicConfig.getColumnFilter().trim().isEmpty()) {
+ if (filter.trim().isEmpty()) {
final String exceptionMessage =
String.format(
- "Failed to create or alter topic, %s should not be empty",
- TopicConstant.COLUMN_FILTER_KEY);
+ ConfigNodeMessages
+ .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_SHOULD_NOT_BE_EMPTY_767B1148,
+ filterKey);
LOGGER.warn(exceptionMessage);
throw new SubscriptionException(exceptionMessage);
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
index 1f6cc40..fb7ff2f 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
@@ -130,6 +130,7 @@
.deepCopyTopicMeta(
updatedTopicMeta.getTopicName(), updatedTopicMeta.visibleUnderTableModel());
if (Objects.nonNull(updatedTopicAttributes) && Objects.nonNull(existedTopicMeta)) {
+ subscriptionInfo.get().validateUpdatedTopicAttributes(updatedTopicAttributes);
updatedTopicMeta = existedTopicMeta.deepCopyWithUpdatedAttributes(updatedTopicAttributes);
}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java
index f9e7e7a..fd97beb 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java
@@ -87,6 +87,75 @@
}
@Test
+ public void testValidateTagFilterOnCreate() throws Exception {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes = newIncrementalTableTopicAttributes();
+ attributes.put(TopicConstant.TAG_FILTER_KEY, "region IN (\"north\", \"south\")");
+
+ Assert.assertTrue(
+ subscriptionInfo.validateBeforeCreatingTopic(
+ new TCreateTopicReq("table_topic").setTopicAttributes(attributes)));
+ }
+
+ @Test
+ public void testRejectInvalidTagFilterFromSessionApiOnCreate() {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes = newIncrementalTableTopicAttributes();
+ attributes.put(TopicConstant.TAG_FILTER_KEY, "region = north");
+
+ assertCreateRejected(subscriptionInfo, attributes, "Invalid tag-filter");
+ }
+
+ @Test
+ public void testRejectInvalidTagFilterFromSessionApiOnAlter() throws Exception {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> originalAttributes = newIncrementalTableTopicAttributes();
+ originalAttributes.put(TopicConstant.TAG_FILTER_KEY, "region = \"north\"");
+ subscriptionInfo.createTopic(
+ new CreateTopicPlan(new TopicMeta("table_topic", 1L, originalAttributes)));
+
+ final Map<String, String> updatedAttributes = newIncrementalTableTopicAttributes();
+ updatedAttributes.put(TopicConstant.TAG_FILTER_KEY, "region IN ()");
+ try {
+ subscriptionInfo.validateBeforeAlteringTopic(
+ new TopicMeta("table_topic", 2L, updatedAttributes));
+ Assert.fail("Expected invalid tag-filter alteration to fail");
+ } catch (final SubscriptionException e) {
+ Assert.assertTrue(e.getMessage().contains("Invalid tag-filter"));
+ }
+ }
+
+ @Test
+ public void testRejectTagFilterOnTreeTopic() {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(TopicConstant.TAG_FILTER_KEY, "region = \"north\"");
+
+ assertCreateRejected(subscriptionInfo, attributes, "only supported for table topics");
+ }
+
+ @Test
+ public void testTagFilterKeyIsCaseInsensitiveOnCreate() throws Exception {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes = newInitialTableTopicAttributes();
+ attributes.put("Tag-Filter", "region = \"north\"");
+
+ Assert.assertTrue(
+ subscriptionInfo.validateBeforeCreatingTopic(
+ new TCreateTopicReq("table_topic").setTopicAttributes(attributes)));
+ }
+
+ @Test
+ public void testRejectDuplicateTagFilterKeys() {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes = newInitialTableTopicAttributes();
+ attributes.put(TopicConstant.TAG_FILTER_KEY, "region = \"north\"");
+ attributes.put("Tag-Filter", "region = \"south\"");
+
+ assertCreateRejected(subscriptionInfo, attributes, "duplicate tag-filter");
+ }
+
+ @Test
public void testRejectTopicThatOnlySelectsAuditDatabase() {
final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
@@ -251,6 +320,15 @@
}
@Test
+ public void testRejectEmptyTagFilter() {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes = newIncrementalTableTopicAttributes();
+ attributes.put(TopicConstant.TAG_FILTER_KEY, " ");
+
+ assertCreateRejected(subscriptionInfo, attributes, "tag-filter should not be empty");
+ }
+
+ @Test
public void testAcceptAlteringColumnFilter() throws Exception {
final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
final Map<String, String> originalAttributes = newIncrementalTableTopicAttributes();
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedureTest.java
index 3f553cb..56f16b6 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedureTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedureTest.java
@@ -25,6 +25,8 @@
import org.apache.iotdb.confignode.persistence.subscription.SubscriptionInfo;
import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
import org.apache.tsfile.utils.PublicBAOS;
import org.junit.Test;
@@ -36,6 +38,8 @@
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
public class AlterTopicProcedureTest {
@@ -105,4 +109,34 @@
assertEquals("processor1", procedure.getUpdatedTopicMeta().getConfig().getString("processor"));
assertEquals("source1", procedure.getUpdatedTopicMeta().getConfig().getString("source"));
}
+
+ @Test
+ public void testRejectCaseInsensitiveDuplicateAttributesBeforeMerge() throws Exception {
+ final String topicName = "test_table_topic";
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> initialAttributes = new HashMap<>();
+ initialAttributes.put("__system.sql-dialect", "table");
+ assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo
+ .createTopic(new CreateTopicPlan(new TopicMeta(topicName, 1, initialAttributes)))
+ .getCode());
+
+ final Map<String, String> requestAttributes = new HashMap<>();
+ requestAttributes.put(TopicConstant.TAG_FILTER_KEY, "region = \"north\"");
+ requestAttributes.put("TAG-FILTER", "region = \"south\"");
+ final AlterTopicProcedure procedure =
+ new AlterTopicProcedure(
+ subscriptionInfo.deepCopyTopicMetaWithUpdatedAttributes(
+ topicName, requestAttributes, true),
+ requestAttributes,
+ new AtomicReference<>(subscriptionInfo));
+
+ try {
+ procedure.executeFromValidate(null);
+ fail("Expected duplicate tag-filter attributes to be rejected");
+ } catch (final SubscriptionException e) {
+ assertTrue(e.getMessage(), e.getMessage().contains("duplicate tag-filter"));
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
index 43cd511..b4271e3 100644
--- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
@@ -123,6 +123,84 @@
public static final String COLUMN_FILTER_LIKE_PATTERN = "LIKE pattern";
public static final String COLUMN_FILTER_LIKE_ESCAPE = "LIKE escape";
public static final String COLUMN_FILTER_REGEXP_PATTERN = "REGEXP pattern";
+ public static final String EXCEPTION_INVALID_TAG_FILTER_ARG_E4B1C1C6 =
+ "Invalid tag-filter: %s";
+ public static final String EXCEPTION_TAG_FILTER_SHOULD_NOT_BE_EMPTY_507CA5B0 =
+ "tag-filter should not be empty";
+ public static final String
+ EXCEPTION_ONLY_AND_COMPARISONS_ARE_SUPPORTED_IN_TAG_FILTER_19946958 =
+ "only =, !=, and <> comparisons are supported in tag-filter";
+ public static final String EXCEPTION_LEFT_OPERAND_MUST_BE_A_TAG_COLUMN_F9D4548B =
+ "left operand must be a TAG column";
+ public static final String EXCEPTION_TAG_FILTER_EVALUATION_FAILED_ARG_1B239B2F =
+ "Tag-filter evaluation failed: %s";
+ public static final String EXCEPTION_TAG_FILTER_SCHEMA_BINDING_FAILED_ARG_7A5D2D47 =
+ "Tag-filter schema binding failed: %s";
+ public static final String EXCEPTION_TAG_FILTER_EXCEEDS_MAXIMUM_UTF_8_LENGTH_OF_ARG_BYTES_9C8400F2 =
+ "tag-filter exceeds maximum UTF-8 length of %s bytes";
+ public static final String EXCEPTION_TAG_FILTER_AST_DEPTH_EXCEEDS_MAXIMUM_OF_ARG_64C92731 =
+ "tag-filter AST depth exceeds maximum of %s";
+ public static final String EXCEPTION_TAG_FILTER_AST_NODE_COUNT_EXCEEDS_MAXIMUM_OF_ARG_40CCE694 =
+ "tag-filter AST node count exceeds maximum of %s";
+ public static final String EXCEPTION_TAG_FILTER_IN_LIST_EXCEEDS_MAXIMUM_OF_ARG_VALUES_07E7FC6C =
+ "tag-filter IN list exceeds maximum of %s values";
+ public static final String EXCEPTION_TAG_FILTER_REGEXP_PATTERN_EXCEEDS_MAXIMUM_LENGTH_OF_ARG_CHARACTERS_67590971 =
+ "tag-filter REGEXP pattern exceeds maximum length of %s characters";
+ public static final String EXCEPTION_TABLE_SCHEMA_IS_NOT_AVAILABLE_FOR_TAG_FILTER_993AB728 =
+ "table schema is not available for tag-filter";
+ public static final String
+ EXCEPTION_TABLE_SCHEMA_ARG_ARG_IS_ABSENT_FROM_THIS_DATANODE_S_TABLE_CACHE_THE_TABLE_MAY_NOT_EXIST_OR_ITS_SCHEMA_HAS_NOT_BEEN_SYNCHRONIZED_YET_23975BDF =
+ "Table schema %s.%s is absent from this DataNode's table cache; the table may not exist or its schema has not been synchronized yet";
+ public static final String
+ EXCEPTION_TABLE_SCHEMA_KEPT_CHANGING_WHILE_BINDING_TAG_FILTER_FOR_DATABASE_PATTERN_ARG_AND_TABLE_PATTERN_ARG_RETRY_AFTER_SCHEMA_SYNCHRONIZATION_77A005E9 =
+ "Table schema kept changing while binding tag-filter for database pattern %s and table pattern %s; retry after schema synchronization";
+ public static final String EXCEPTION_REFERENCED_TAG_COLUMN_IS_MISSING_ARG_F5300BEA =
+ "referenced TAG column is missing: %s";
+ public static final String EXCEPTION_REFERENCED_COLUMN_IS_NOT_A_TAG_COLUMN_ARG_34D60881 =
+ "referenced column is not a TAG column: %s";
+ public static final String EXCEPTION_UNCOMPILED_LIKE_PREDICATE_01AEE439 =
+ "uncompiled LIKE predicate";
+ public static final String EXCEPTION_UNCOMPILED_REGEXP_PREDICATE_2B0DD646 =
+ "uncompiled REGEXP predicate";
+ public static final String EXCEPTION_MATCHER_IS_UNAVAILABLE_1A659D47 = "matcher is unavailable";
+ public static final String EXCEPTION_TABLET_SCHEMA_IS_MISSING_164075B0 = "tablet schema is missing";
+ public static final String EXCEPTION_TABLET_VALUE_COLUMNS_ARE_INCOMPLETE_DCD09F3C =
+ "tablet value columns are incomplete";
+ public static final String EXCEPTION_FAILED_TO_EVALUATE_A_TABLET_ROW_7E4E94CE =
+ "failed to evaluate a tablet row";
+ public static final String EXCEPTION_TABLE_BINDING_IS_NOT_AVAILABLE_882C5F2F =
+ "table binding is not available";
+ public static final String EXCEPTION_REFERENCED_COLUMN_IS_NOT_A_TAG_COLUMN_33FA34BC =
+ "referenced column is not a TAG column";
+ public static final String EXCEPTION_REFERENCED_COLUMN_CATEGORY_CHANGED_564594D8 =
+ "referenced column category changed";
+ public static final String EXCEPTION_REFERENCED_TAG_COLUMN_IS_MISSING_55E07377 =
+ "referenced TAG column is missing";
+ public static final String EXCEPTION_TABLET_MEASUREMENT_SCHEMA_IS_INCOMPLETE_6A813472 =
+ "tablet measurement schema is incomplete";
+ public static final String EXCEPTION_FAILED_TO_COMPACT_FILTERED_TABLET_5D4AD8AA =
+ "failed to compact filtered tablet";
+ public static final String EXCEPTION_TABLET_COLUMN_CATEGORIES_ARE_MISSING_2C660532 =
+ "tablet column categories are missing";
+ public static final String EXCEPTION_TABLET_COLUMN_CATEGORY_IS_MISSING_A812B500 =
+ "tablet column category is missing";
+ public static final String EXCEPTION_TABLET_TIMESTAMPS_ARE_INCOMPLETE_24F8CE6F =
+ "tablet timestamps are incomplete";
+ public static final String EXCEPTION_TABLET_BITMAPS_ARE_INCOMPLETE_7BBC8035 =
+ "tablet bitmaps are incomplete";
+ public static final String EXCEPTION_TABLET_VALUE_COLUMN_IS_INCOMPLETE_845721FE =
+ "tablet value column is incomplete";
+ public static final String EXCEPTION_TREE_VIEW_PROJECTOR_IS_UNAVAILABLE_FOR_FILTERED_SUBSCRIPTION_DATA_B5F396A5 =
+ "Tree View projector is unavailable for filtered subscription data";
+ public static final String
+ EXCEPTION_TOPIC_CONFIGURATION_IS_NOT_AVAILABLE_FOR_TAG_FILTER_1E023E3A =
+ "topic configuration is not available for tag-filter";
+ public static final String
+ EXCEPTION_TOPIC_CONFIGURATION_CHANGED_WHILE_CAPTURING_TAG_FILTER_SNAPSHOT_984CEB1B =
+ "topic configuration changed while capturing tag-filter snapshot";
+ public static final String
+ LOG_SUBSCRIPTION_TABLE_SCHEMA_IS_NOT_AVAILABLE_FOR_TAG_FILTER_TOPIC_ARG_E864C98F =
+ "Subscription: table schema is not available for tag-filter topic [{}]";
public static final String CREATE_NEW_REGION_ERROR_FMT = "create new region %s error, exception:%s";
public static final String CREATE_NEW_REGION_SUCCEED_FMT = "create new region %s succeed";
@@ -793,6 +871,14 @@
"Subscription: failed to lazily refresh column-filter matcher for topic [{}]";
public static final String SUBSCRIPTION_DROP_COLUMN_FILTER =
"Subscription: dropped column-filter matcher for topic [{}]";
+ public static final String
+ LOG_SUBSCRIPTION_FAILED_TO_REFRESH_TAG_FILTER_MATCHER_FOR_TOPIC_ARG_USE_EMPTY_MATCHER_TO_FAIL_CLOSED_3CB76D70 =
+ "Subscription: failed to refresh tag-filter matcher for topic [{}], use empty matcher to fail closed";
+ public static final String
+ LOG_SUBSCRIPTION_REFRESHED_TAG_FILTER_MATCHER_FOR_TOPIC_ARG_71598849 =
+ "Subscription: refreshed tag-filter matcher for topic [{}]";
+ public static final String LOG_SUBSCRIPTION_DROPPED_TAG_FILTER_MATCHER_FOR_TOPIC_ARG_9EA52458 =
+ "Subscription: dropped tag-filter matcher for topic [{}]";
public static final String SUBSCRIPTION_UNSUPPORTED_CONSENSUS_PROGRESS_FILE_VERSION_FMT =
"Unsupported consensus subscription progress file version %s";
@@ -833,6 +919,8 @@
"Exception occurred when sealing events from batch {}";
public static final String EXCEPTION_CONSTRUCT_NEW_BATCH =
"Exception occurred when construct new batch";
+ public static final String EXCEPTION_FAILED_TO_SEAL_SUBSCRIPTION_EVENT_BATCH_1FB7E92C =
+ "Failed to seal subscription event batch";
// ---------------------------------------------------------------------------
// subscription – SubscriptionPrefetchingQueue
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index 7dd677c..d3282b7 100644
--- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -4054,6 +4054,12 @@
"No materializer is available to enforce the DeviceEntry memory limit";
public static final String EXCEPTION_NO_MORE_DEVICEENTRY_RECORDS_ARE_AVAILABLE_8D51C199 =
"No more DeviceEntry records are available";
+ public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_DUPLICATE_ARG_ATTRIBUTES_ARE_NOT_ALLOWED_27315578 =
+ "Failed to create or alter topic, duplicate %s attributes are not allowed";
+ public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_IS_ONLY_SUPPORTED_FOR_TABLE_TOPICS_A5126607 =
+ "Failed to create or alter topic, %s is only supported for table topics";
public static final String EXCEPTION_ONLY_INMEMORYDEVICEENTRYDATASET_SUPPORTS_GET_INLINE_DEVICE_ENTRIES_07A52CAB =
"Only InMemoryDeviceEntryDataSet supports get inline device entries";
public static final String EXCEPTION_LOAD_CONSENSUS_INVALID_PIECE_REF_F3498507 =
diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
index 743d1aa..0139e97 100644
--- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
@@ -123,6 +123,86 @@
public static final String COLUMN_FILTER_LIKE_PATTERN = "LIKE 模式";
public static final String COLUMN_FILTER_LIKE_ESCAPE = "LIKE 转义符";
public static final String COLUMN_FILTER_REGEXP_PATTERN = "REGEXP 模式";
+ public static final String EXCEPTION_INVALID_TAG_FILTER_ARG_E4B1C1C6 =
+ "\u65e0\u6548\u7684 tag-filter\uff1a%s";
+ public static final String EXCEPTION_TAG_FILTER_SHOULD_NOT_BE_EMPTY_507CA5B0 =
+ "tag-filter \u4e0d\u5e94\u4e3a\u7a7a";
+ public static final String
+ EXCEPTION_ONLY_AND_COMPARISONS_ARE_SUPPORTED_IN_TAG_FILTER_19946958 =
+ "tag-filter \u4ec5\u652f\u6301 =\u3001!= \u548c <> \u6bd4\u8f83";
+ public static final String EXCEPTION_LEFT_OPERAND_MUST_BE_A_TAG_COLUMN_F9D4548B =
+ "\u5de6\u64cd\u4f5c\u6570\u5fc5\u987b\u662f TAG \u5217";
+ public static final String EXCEPTION_TAG_FILTER_EVALUATION_FAILED_ARG_1B239B2F =
+ "tag-filter \u6c42\u503c\u5931\u8d25\uff1a%s";
+ public static final String EXCEPTION_TAG_FILTER_SCHEMA_BINDING_FAILED_ARG_7A5D2D47 =
+ "tag-filter Schema \u7ed1\u5b9a\u5931\u8d25\uff1a%s";
+ public static final String EXCEPTION_TAG_FILTER_EXCEEDS_MAXIMUM_UTF_8_LENGTH_OF_ARG_BYTES_9C8400F2 =
+ "tag-filter \u8d85\u8fc7 UTF-8 \u6700\u5927\u957f\u5ea6 %s \u5b57\u8282";
+ public static final String EXCEPTION_TAG_FILTER_AST_DEPTH_EXCEEDS_MAXIMUM_OF_ARG_64C92731 =
+ "tag-filter AST \u6df1\u5ea6\u8d85\u8fc7\u6700\u5927\u503c %s";
+ public static final String EXCEPTION_TAG_FILTER_AST_NODE_COUNT_EXCEEDS_MAXIMUM_OF_ARG_40CCE694 =
+ "tag-filter AST \u8282\u70b9\u6570\u8d85\u8fc7\u6700\u5927\u503c %s";
+ public static final String EXCEPTION_TAG_FILTER_IN_LIST_EXCEEDS_MAXIMUM_OF_ARG_VALUES_07E7FC6C =
+ "tag-filter IN \u5217\u8868\u8d85\u8fc7\u6700\u5927\u503c %s";
+ public static final String EXCEPTION_TAG_FILTER_REGEXP_PATTERN_EXCEEDS_MAXIMUM_LENGTH_OF_ARG_CHARACTERS_67590971 =
+ "tag-filter REGEXP \u6a21\u5f0f\u8d85\u8fc7\u6700\u5927\u957f\u5ea6 %s \u5b57\u7b26";
+ public static final String
+ EXCEPTION_TABLE_SCHEMA_ARG_ARG_IS_ABSENT_FROM_THIS_DATANODE_S_TABLE_CACHE_THE_TABLE_MAY_NOT_EXIST_OR_ITS_SCHEMA_HAS_NOT_BEEN_SYNCHRONIZED_YET_23975BDF =
+ "此 DataNode 的表缓存中没有表结构 %s.%s；该表可能不存在，或其结构尚未同步";
+ public static final String
+ EXCEPTION_TABLE_SCHEMA_KEPT_CHANGING_WHILE_BINDING_TAG_FILTER_FOR_DATABASE_PATTERN_ARG_AND_TABLE_PATTERN_ARG_RETRY_AFTER_SCHEMA_SYNCHRONIZATION_77A005E9 =
+ "为数据库模式 %s 和表模式 %s 绑定 tag-filter 时表结构持续变化；请在结构同步后重试";
+ public static final String EXCEPTION_TABLE_SCHEMA_IS_NOT_AVAILABLE_FOR_TAG_FILTER_993AB728 =
+ "tag-filter \u6240\u9700\u8868\u7ed3\u6784\u4e0d\u53ef\u7528";
+ public static final String EXCEPTION_REFERENCED_TAG_COLUMN_IS_MISSING_ARG_F5300BEA =
+ "\u5f15\u7528\u7684 TAG \u5217\u4e0d\u5b58\u5728\uff1a%s";
+ public static final String EXCEPTION_REFERENCED_COLUMN_IS_NOT_A_TAG_COLUMN_ARG_34D60881 =
+ "\u5f15\u7528\u7684\u5217\u4e0d\u662f TAG \u5217\uff1a%s";
+ public static final String EXCEPTION_UNCOMPILED_LIKE_PREDICATE_01AEE439 =
+ "LIKE \u8c13\u8bcd\u672a\u7f16\u8bd1";
+ public static final String EXCEPTION_UNCOMPILED_REGEXP_PREDICATE_2B0DD646 =
+ "REGEXP \u8c13\u8bcd\u672a\u7f16\u8bd1";
+ public static final String EXCEPTION_MATCHER_IS_UNAVAILABLE_1A659D47 =
+ "tag-filter \u5339\u914d\u5668\u4e0d\u53ef\u7528";
+ public static final String EXCEPTION_TABLET_SCHEMA_IS_MISSING_164075B0 =
+ "Tablet Schema \u7f3a\u5931";
+ public static final String EXCEPTION_TABLET_VALUE_COLUMNS_ARE_INCOMPLETE_DCD09F3C =
+ "Tablet \u503c\u5217\u4e0d\u5b8c\u6574";
+ public static final String EXCEPTION_FAILED_TO_EVALUATE_A_TABLET_ROW_7E4E94CE =
+ "Tablet \u884c\u6c42\u503c\u5931\u8d25";
+ public static final String EXCEPTION_TABLE_BINDING_IS_NOT_AVAILABLE_882C5F2F =
+ "\u8868\u7ed1\u5b9a\u4e0d\u53ef\u7528";
+ public static final String EXCEPTION_REFERENCED_COLUMN_IS_NOT_A_TAG_COLUMN_33FA34BC =
+ "\u5f15\u7528\u7684\u5217\u4e0d\u662f TAG \u5217";
+ public static final String EXCEPTION_REFERENCED_COLUMN_CATEGORY_CHANGED_564594D8 =
+ "\u5f15\u7528\u5217\u7684\u7c7b\u522b\u5df2\u53d8\u66f4";
+ public static final String EXCEPTION_REFERENCED_TAG_COLUMN_IS_MISSING_55E07377 =
+ "\u5f15\u7528\u7684 TAG \u5217\u7f3a\u5931";
+ public static final String EXCEPTION_TABLET_MEASUREMENT_SCHEMA_IS_INCOMPLETE_6A813472 =
+ "Tablet \u6d4b\u91cf\u7ed3\u6784\u4e0d\u5b8c\u6574";
+ public static final String EXCEPTION_FAILED_TO_COMPACT_FILTERED_TABLET_5D4AD8AA =
+ "\u538b\u7f29\u8fc7\u6ee4\u540e\u7684 Tablet \u5931\u8d25";
+ public static final String EXCEPTION_TABLET_COLUMN_CATEGORIES_ARE_MISSING_2C660532 =
+ "Tablet \u5217\u7c7b\u522b\u7f3a\u5931";
+ public static final String EXCEPTION_TABLET_COLUMN_CATEGORY_IS_MISSING_A812B500 =
+ "Tablet \u5217\u7c7b\u522b\u7f3a\u5931";
+ public static final String EXCEPTION_TABLET_TIMESTAMPS_ARE_INCOMPLETE_24F8CE6F =
+ "Tablet \u65f6\u95f4\u6233\u4e0d\u5b8c\u6574";
+ public static final String EXCEPTION_TABLET_BITMAPS_ARE_INCOMPLETE_7BBC8035 =
+ "Tablet bitmap \u4e0d\u5b8c\u6574";
+ public static final String EXCEPTION_TABLET_VALUE_COLUMN_IS_INCOMPLETE_845721FE =
+ "Tablet \u503c\u5217\u4e0d\u5b8c\u6574";
+ public static final String EXCEPTION_TREE_VIEW_PROJECTOR_IS_UNAVAILABLE_FOR_FILTERED_SUBSCRIPTION_DATA_B5F396A5 =
+ "\u8fc7\u6ee4\u8ba2\u9605\u6570\u636e\u7684 Tree View \u6295\u5f71\u5668\u4e0d\u53ef\u7528";
+ public static final String
+ EXCEPTION_TOPIC_CONFIGURATION_IS_NOT_AVAILABLE_FOR_TAG_FILTER_1E023E3A =
+ "\u65e0\u6cd5\u83b7\u53d6 tag-filter \u7684 Topic \u914d\u7f6e";
+ public static final String
+ EXCEPTION_TOPIC_CONFIGURATION_CHANGED_WHILE_CAPTURING_TAG_FILTER_SNAPSHOT_984CEB1B =
+ "\u6355\u83b7 tag-filter \u5feb\u7167\u65f6 Topic \u914d\u7f6e\u53d1\u751f\u53d8\u5316";
+ public static final String
+ LOG_SUBSCRIPTION_TABLE_SCHEMA_IS_NOT_AVAILABLE_FOR_TAG_FILTER_TOPIC_ARG_E864C98F =
+ "\u8ba2\u9605：tag-filter Topic \u7684\u8868 Schema \u4e0d\u53ef\u7528 [{}]";
public static final String CREATE_NEW_REGION_ERROR_FMT = "创建新 region %s 错误，异常：%s";
public static final String CREATE_NEW_REGION_SUCCEED_FMT = "创建新 region %s 成功";
@@ -793,6 +873,14 @@
"\u8ba2\u9605\uff1a\u61d2\u52a0\u8f7d\u5237\u65b0\u4e3b\u9898 [{}] \u7684 column-filter \u5339\u914d\u5668\u5931\u8d25";
public static final String SUBSCRIPTION_DROP_COLUMN_FILTER =
"\u8ba2\u9605\uff1a\u5220\u9664\u4e3b\u9898 [{}] \u7684 column-filter \u5339\u914d\u5668";
+ public static final String
+ LOG_SUBSCRIPTION_FAILED_TO_REFRESH_TAG_FILTER_MATCHER_FOR_TOPIC_ARG_USE_EMPTY_MATCHER_TO_FAIL_CLOSED_3CB76D70 =
+ "\u8ba2\u9605\uff1a\u5237\u65b0\u4e3b\u9898 [{}] \u7684 tag-filter \u5339\u914d\u5668\u5931\u8d25\uff0c\u4f7f\u7528\u7a7a\u5339\u914d\u5668\u4ee5\u5b89\u5168\u5931\u8d25";
+ public static final String
+ LOG_SUBSCRIPTION_REFRESHED_TAG_FILTER_MATCHER_FOR_TOPIC_ARG_71598849 =
+ "\u8ba2\u9605\uff1a\u5df2\u5237\u65b0\u4e3b\u9898 [{}] \u7684 tag-filter \u5339\u914d\u5668";
+ public static final String LOG_SUBSCRIPTION_DROPPED_TAG_FILTER_MATCHER_FOR_TOPIC_ARG_9EA52458 =
+ "\u8ba2\u9605\uff1a\u5220\u9664\u4e3b\u9898 [{}] \u7684 tag-filter \u5339\u914d\u5668";
public static final String SUBSCRIPTION_UNSUPPORTED_CONSENSUS_PROGRESS_FILE_VERSION_FMT =
"不支持的共识订阅进度文件版本 %s";
@@ -833,6 +921,8 @@
"从批次 {} 封存事件时发生异常";
public static final String EXCEPTION_CONSTRUCT_NEW_BATCH =
"构造新批次时发生异常";
+ public static final String EXCEPTION_FAILED_TO_SEAL_SUBSCRIPTION_EVENT_BATCH_1FB7E92C =
+ "\u5c01\u5b58\u8ba2\u9605\u4e8b\u4ef6\u6279\u6b21\u5931\u8d25";
// ---------------------------------------------------------------------------
// subscription – SubscriptionPrefetchingQueue
diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index 203c755..6a50a25 100644
--- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -4810,6 +4810,12 @@
"没有可用于执行 DeviceEntry 内存限制的 materializer";
public static final String EXCEPTION_NO_MORE_DEVICEENTRY_RECORDS_ARE_AVAILABLE_8D51C199 =
"没有更多可用的 DeviceEntry 记录";
+ public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_DUPLICATE_ARG_ATTRIBUTES_ARE_NOT_ALLOWED_27315578 =
+ "\u521b\u5efa\u6216\u4fee\u6539 Topic \u5931\u8d25\uff0c\u4e0d\u5141\u8bb8\u91cd\u590d\u7684 %s \u5c5e\u6027";
+ public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_IS_ONLY_SUPPORTED_FOR_TABLE_TOPICS_A5126607 =
+ "\u521b\u5efa\u6216\u4fee\u6539 Topic \u5931\u8d25\uff0c%s \u4ec5\u652f\u6301\u8868\u6a21\u578b Topic";
public static final String EXCEPTION_ONLY_INMEMORYDEVICEENTRYDATASET_SUPPORTS_GET_INLINE_DEVICE_ENTRIES_07A52CAB =
"只有 InMemoryDeviceEntryDataSet 支持获取内存中的设备条目";
public static final String EXCEPTION_LOAD_CONSENSUS_INVALID_PIECE_REF_F3498507 =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/PipeInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/PipeInsertionEvent.java
index bab43ce..8f45510 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/PipeInsertionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/PipeInsertionEvent.java
@@ -169,6 +169,9 @@
}
public String getTableModelDatabaseName() {
+ if (!isTableModelEvent()) {
+ return null;
+ }
return tableModelDatabaseName == null
? tableModelDatabaseName = PathUtils.unQualifyDatabaseName(sourceDatabaseNameFromDataRegion)
: tableModelDatabaseName;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java
index 4793f0c..5ca1862 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java
@@ -46,6 +46,7 @@
import java.util.Objects;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiFunction;
+import java.util.stream.Collectors;
import static org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent.isTabletEmpty;
@@ -58,26 +59,33 @@
private final PipeTsFileBuilder treeModeTsFileBuilder;
private final PipeTsFileBuilder tableModeTsFileBuilder;
- private final BiFunction<String, Tablet, Tablet> tableModelTabletPruner;
+ private final TabletTransformer tabletTransformer;
private final Map<Pair<String, Long>, Double> pipeName2WeightMap = new HashMap<>();
public PipeTabletEventTsFileBatch(final int maxDelayInMs, final long requestMaxBatchSizeInBytes) {
- this(maxDelayInMs, requestMaxBatchSizeInBytes, null, null);
+ this(maxDelayInMs, requestMaxBatchSizeInBytes, null, null, null);
}
public PipeTabletEventTsFileBatch(
final int maxDelayInMs,
final long requestMaxBatchSizeInBytes,
final TriLongConsumer recordMetric) {
- this(maxDelayInMs, requestMaxBatchSizeInBytes, recordMetric, null);
+ this(maxDelayInMs, requestMaxBatchSizeInBytes, recordMetric, null, null);
}
public PipeTabletEventTsFileBatch(
final int maxDelayInMs,
final long requestMaxBatchSizeInBytes,
final BiFunction<String, Tablet, Tablet> tableModelTabletPruner) {
- this(maxDelayInMs, requestMaxBatchSizeInBytes, null, tableModelTabletPruner);
+ this(maxDelayInMs, requestMaxBatchSizeInBytes, null, tableModelTabletPruner, null);
+ }
+
+ public PipeTabletEventTsFileBatch(
+ final int maxDelayInMs,
+ final long requestMaxBatchSizeInBytes,
+ final TabletTransformer tabletTransformer) {
+ this(maxDelayInMs, requestMaxBatchSizeInBytes, null, null, tabletTransformer);
}
public PipeTabletEventTsFileBatch(
@@ -85,12 +93,31 @@
final long requestMaxBatchSizeInBytes,
final TriLongConsumer recordMetric,
final BiFunction<String, Tablet, Tablet> tableModelTabletPruner) {
+ this(maxDelayInMs, requestMaxBatchSizeInBytes, recordMetric, tableModelTabletPruner, null);
+ }
+
+ private PipeTabletEventTsFileBatch(
+ final int maxDelayInMs,
+ final long requestMaxBatchSizeInBytes,
+ final TriLongConsumer recordMetric,
+ final BiFunction<String, Tablet, Tablet> tableModelTabletPruner,
+ final TabletTransformer tabletTransformer) {
super(maxDelayInMs, requestMaxBatchSizeInBytes, recordMetric);
final AtomicLong tsFileIdGenerator = new AtomicLong(0);
treeModeTsFileBuilder = new PipeTreeModelTsFileBuilderV2(currentBatchId, tsFileIdGenerator);
tableModeTsFileBuilder = new PipeTableModelTsFileBuilderV2(currentBatchId, tsFileIdGenerator);
- this.tableModelTabletPruner = tableModelTabletPruner;
+ this.tabletTransformer =
+ Objects.nonNull(tabletTransformer)
+ ? tabletTransformer
+ : (databaseName, tablet, isTableModel, isAligned) ->
+ new TabletTransformResult(
+ Objects.nonNull(tableModelTabletPruner) && isTableModel
+ ? tableModelTabletPruner.apply(databaseName, tablet)
+ : tablet,
+ databaseName,
+ isTableModel,
+ isAligned);
}
@Override
@@ -100,25 +127,22 @@
(PipeInsertNodeTabletInsertionEvent) event;
final boolean isTableModel = insertNodeTabletInsertionEvent.isTableModelEvent();
final List<Tablet> tablets = insertNodeTabletInsertionEvent.convertToTablets();
- final List<Tablet> retainedTablets = new ArrayList<>(tablets.size());
- final List<Boolean> retainedAlignedFlags = new ArrayList<>(tablets.size());
+ final List<TabletTransformResult> retainedTablets = new ArrayList<>(tablets.size());
for (int i = 0; i < tablets.size(); ++i) {
- Tablet tablet = tablets.get(i);
+ final Tablet tablet = tablets.get(i);
if (isTabletEmpty(tablet)) {
continue;
}
- if (isTableModel) {
- tablet =
- pruneTableModelTablet(
- tablet, insertNodeTabletInsertionEvent.getTableModelDatabaseName());
- if (isTabletEmpty(tablet)) {
- continue;
- }
+ final TabletTransformResult transformedTablet =
+ transformTablet(
+ insertNodeTabletInsertionEvent.getTableModelDatabaseName(),
+ tablet,
+ isTableModel,
+ isTableModel || insertNodeTabletInsertionEvent.isAligned(i));
+ if (Objects.isNull(transformedTablet) || isTabletEmpty(transformedTablet.getTablet())) {
+ continue;
}
- retainedTablets.add(tablet);
- if (!isTableModel) {
- retainedAlignedFlags.add(insertNodeTabletInsertionEvent.isAligned(i));
- }
+ retainedTablets.add(transformedTablet);
}
// Pruning can remove all rows/columns from a tablet. Account only for data that is
@@ -127,50 +151,58 @@
if (retainedTablets.isEmpty()) {
return false;
}
- increaseTotalBufferSizeAndUpdateMemoryBlock(calculateTabletsSizeInBytes(retainedTablets));
- for (int i = 0; i < retainedTablets.size(); ++i) {
- final Tablet tablet = retainedTablets.get(i);
- if (isTableModel) {
+ increaseTotalBufferSizeAndUpdateMemoryBlock(
+ calculateTabletsSizeInBytes(
+ retainedTablets.stream()
+ .map(TabletTransformResult::getTablet)
+ .collect(Collectors.toList())));
+ for (final TabletTransformResult transformedTablet : retainedTablets) {
+ if (transformedTablet.isTableModel()) {
bufferTableModelTablet(
insertNodeTabletInsertionEvent.getPipeName(),
insertNodeTabletInsertionEvent.getCreationTime(),
- tablet,
- insertNodeTabletInsertionEvent.getTableModelDatabaseName());
+ transformedTablet.getTablet(),
+ transformedTablet.getDatabaseName());
} else {
bufferTreeModelTablet(
insertNodeTabletInsertionEvent.getPipeName(),
insertNodeTabletInsertionEvent.getCreationTime(),
- tablet,
- retainedAlignedFlags.get(i));
+ transformedTablet.getTablet(),
+ transformedTablet.isAligned());
}
}
return true;
} else if (event instanceof PipeRawTabletInsertionEvent) {
final PipeRawTabletInsertionEvent rawTabletInsertionEvent =
(PipeRawTabletInsertionEvent) event;
- Tablet tablet = rawTabletInsertionEvent.convertToTablet();
+ final Tablet tablet = rawTabletInsertionEvent.convertToTablet();
if (isTabletEmpty(tablet)) {
return false;
}
- if (rawTabletInsertionEvent.isTableModelEvent()) {
- tablet = pruneTableModelTablet(tablet, rawTabletInsertionEvent.getTableModelDatabaseName());
- if (isTabletEmpty(tablet)) {
- return false;
- }
+ final boolean isTableModel = rawTabletInsertionEvent.isTableModelEvent();
+ final TabletTransformResult transformedTablet =
+ transformTablet(
+ rawTabletInsertionEvent.getTableModelDatabaseName(),
+ tablet,
+ isTableModel,
+ isTableModel || rawTabletInsertionEvent.isAligned());
+ if (Objects.isNull(transformedTablet) || isTabletEmpty(transformedTablet.getTablet())) {
+ return false;
}
- increaseTotalBufferSizeAndUpdateMemoryBlock(calculateTabletSizeInBytes(tablet));
- if (rawTabletInsertionEvent.isTableModelEvent()) {
+ increaseTotalBufferSizeAndUpdateMemoryBlock(
+ calculateTabletSizeInBytes(transformedTablet.getTablet()));
+ if (transformedTablet.isTableModel()) {
bufferTableModelTablet(
rawTabletInsertionEvent.getPipeName(),
rawTabletInsertionEvent.getCreationTime(),
- tablet,
- rawTabletInsertionEvent.getTableModelDatabaseName());
+ transformedTablet.getTablet(),
+ transformedTablet.getDatabaseName());
} else {
bufferTreeModelTablet(
rawTabletInsertionEvent.getPipeName(),
rawTabletInsertionEvent.getCreationTime(),
- tablet,
- rawTabletInsertionEvent.isAligned());
+ transformedTablet.getTablet(),
+ transformedTablet.isAligned());
}
return true;
} else {
@@ -183,10 +215,12 @@
return false;
}
- private Tablet pruneTableModelTablet(final Tablet tablet, final String databaseName) {
- return Objects.nonNull(tableModelTabletPruner)
- ? tableModelTabletPruner.apply(databaseName, tablet)
- : tablet;
+ private TabletTransformResult transformTablet(
+ final String databaseName,
+ final Tablet tablet,
+ final boolean isTableModel,
+ final boolean isAligned) {
+ return tabletTransformer.transform(databaseName, tablet, isTableModel, isAligned);
}
private long calculateTabletsSizeInBytes(final List<Tablet> tablets) {
@@ -269,6 +303,48 @@
return new HashMap<>(pipeName2WeightMap);
}
+ @FunctionalInterface
+ public interface TabletTransformer {
+
+ TabletTransformResult transform(
+ String databaseName, Tablet tablet, boolean isTableModel, boolean isAligned);
+ }
+
+ public static final class TabletTransformResult {
+
+ private final Tablet tablet;
+ private final String databaseName;
+ private final boolean tableModel;
+ private final boolean aligned;
+
+ public TabletTransformResult(
+ final Tablet tablet,
+ final String databaseName,
+ final boolean tableModel,
+ final boolean aligned) {
+ this.tablet = tablet;
+ this.databaseName = databaseName;
+ this.tableModel = tableModel;
+ this.aligned = aligned;
+ }
+
+ public Tablet getTablet() {
+ return tablet;
+ }
+
+ public String getDatabaseName() {
+ return databaseName;
+ }
+
+ public boolean isTableModel() {
+ return tableModel;
+ }
+
+ public boolean isAligned() {
+ return aligned;
+ }
+ }
+
/**
* Converts a Tablet to a TSFile and returns the generated TSFile along with its corresponding
* database name.
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java
index e623f81..671c817 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java
@@ -178,6 +178,10 @@
return new DatasetHeader(ColumnHeaderConstant.showTopicColumnHeaders, true);
}
+ public static DatasetHeader getShowTableTopicHeader() {
+ return new DatasetHeader(ColumnHeaderConstant.showTableTopicColumnHeaders, true);
+ }
+
public static DatasetHeader getShowSubscriptionHeader() {
return new DatasetHeader(ColumnHeaderConstant.showSubscriptionColumnHeaders, true);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
index f91969a..033ba02 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
@@ -268,6 +268,7 @@
import org.apache.iotdb.db.queryengine.plan.statement.sys.StopRepairDataStatement;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterParser;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterParser;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
@@ -315,6 +316,7 @@
public static final String DATABASE_NOT_SPECIFIED = "database is not specified";
private static final ColumnFilterParser COLUMN_FILTER_PARSER = new ColumnFilterParser();
+ private static final TagFilterParser TAG_FILTER_PARSER = new TagFilterParser();
private final IClientSession clientSession;
@@ -1650,6 +1652,7 @@
node.getTopicAttributes()
.put(SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TABLE_VALUE);
validateAndNormalizeColumnFilter(node.getTopicAttributes());
+ validateAndNormalizeTagFilter(node.getTopicAttributes());
return new CreateTopicTask(node);
}
@@ -1662,44 +1665,65 @@
node.getTopicAttributes()
.put(SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TABLE_VALUE);
validateAndNormalizeColumnFilter(node.getTopicAttributes());
+ validateAndNormalizeTagFilter(node.getTopicAttributes());
return new AlterTopicTask(node);
}
private static void validateAndNormalizeColumnFilter(final Map<String, String> topicAttributes) {
- String columnFilterKey = null;
- String columnFilter = null;
- boolean hasColumnFilter = false;
+ validateAndNormalizeSubscriptionFilter(
+ topicAttributes, TopicConstant.COLUMN_FILTER_KEY, COLUMN_FILTER_PARSER::parseAndValidate);
+ }
+
+ private static void validateAndNormalizeTagFilter(final Map<String, String> topicAttributes) {
+ validateAndNormalizeSubscriptionFilter(
+ topicAttributes, TopicConstant.TAG_FILTER_KEY, TAG_FILTER_PARSER::parseAndValidate);
+ }
+
+ private static void validateAndNormalizeSubscriptionFilter(
+ final Map<String, String> topicAttributes,
+ final String expectedKey,
+ final SubscriptionFilterValidator validator) {
+ String filterKey = null;
+ String filter = null;
+ boolean hasFilter = false;
for (final Map.Entry<String, String> entry : topicAttributes.entrySet()) {
- if (TopicConstant.COLUMN_FILTER_KEY.equalsIgnoreCase(entry.getKey())) {
- if (hasColumnFilter) {
+ if (expectedKey.equalsIgnoreCase(entry.getKey())) {
+ if (hasFilter) {
throw new SemanticException(
String.format(
- "Failed to create or alter topic, duplicate %s attributes are not allowed",
- TopicConstant.COLUMN_FILTER_KEY));
+ DataNodeQueryMessages
+ .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_DUPLICATE_ARG_ATTRIBUTES_ARE_NOT_ALLOWED_27315578,
+ expectedKey));
}
- hasColumnFilter = true;
- columnFilterKey = entry.getKey();
- columnFilter = entry.getValue();
+ hasFilter = true;
+ filterKey = entry.getKey();
+ filter = entry.getValue();
}
}
- if (!hasColumnFilter) {
+ if (!hasFilter) {
return;
}
- if (!TopicConstant.COLUMN_FILTER_KEY.equals(columnFilterKey)) {
- topicAttributes.remove(columnFilterKey);
- topicAttributes.put(TopicConstant.COLUMN_FILTER_KEY, columnFilter);
+ if (!expectedKey.equals(filterKey)) {
+ topicAttributes.remove(filterKey);
+ topicAttributes.put(expectedKey, filter);
}
try {
- COLUMN_FILTER_PARSER.parseAndValidate(columnFilter);
+ validator.validate(filter);
} catch (final SubscriptionException e) {
throw new SemanticException(e.getMessage());
}
}
+ @FunctionalInterface
+ private interface SubscriptionFilterValidator {
+
+ void validate(String filter) throws SubscriptionException;
+ }
+
@Override
public IConfigTask visitDropTopic(DropTopic node, MPPQueryContext context) {
context.setQueryType(QueryType.OTHER);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java
index d1e363f..269db0c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java
@@ -783,7 +783,7 @@
createTopicStatement
.getTopicAttributes()
.put(SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TREE_VALUE);
- rejectColumnFilterForTreeTopic(createTopicStatement.getTopicAttributes());
+ rejectTableOnlyFiltersForTreeTopic(createTopicStatement.getTopicAttributes());
return new CreateTopicTask(createTopicStatement);
}
@@ -794,18 +794,21 @@
alterTopicStatement
.getTopicAttributes()
.put(SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TREE_VALUE);
- rejectColumnFilterForTreeTopic(alterTopicStatement.getTopicAttributes());
+ rejectTableOnlyFiltersForTreeTopic(alterTopicStatement.getTopicAttributes());
return new AlterTopicTask(alterTopicStatement);
}
- private static void rejectColumnFilterForTreeTopic(final Map<String, String> topicAttributes) {
+ private static void rejectTableOnlyFiltersForTreeTopic(
+ final Map<String, String> topicAttributes) {
for (final String key : topicAttributes.keySet()) {
- if (TopicConstant.COLUMN_FILTER_KEY.equalsIgnoreCase(key)) {
+ if (TopicConstant.COLUMN_FILTER_KEY.equalsIgnoreCase(key)
+ || TopicConstant.TAG_FILTER_KEY.equalsIgnoreCase(key)) {
throw new SemanticException(
String.format(
- "Failed to create or alter topic, %s is only supported for table topics",
- TopicConstant.COLUMN_FILTER_KEY));
+ DataNodeQueryMessages
+ .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_IS_ONLY_SUPPORTED_FOR_TABLE_TOPICS_A5126607,
+ key));
}
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index a995d64..251d3f9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -3313,6 +3313,7 @@
showTopicResp.isSetTopicInfoList()
? showTopicResp.getTopicInfoList()
: Collections.emptyList(),
+ showTopicsStatement.isTableModel(),
future);
} catch (final Exception e) {
future.setException(e);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowTopicsTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowTopicsTask.java
index 1871ef6..e7ffe77 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowTopicsTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowTopicsTask.java
@@ -28,6 +28,8 @@
import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowTopics;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.ShowTopicsStatement;
+import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterMatcher;
import org.apache.iotdb.rpc.TSStatusCode;
import com.google.common.util.concurrent.ListenableFuture;
@@ -37,6 +39,7 @@
import org.apache.tsfile.utils.Binary;
import java.util.List;
+import java.util.function.Function;
import java.util.stream.Collectors;
public class ShowTopicsTask implements IConfigTask {
@@ -60,12 +63,28 @@
}
public static void buildTSBlock(
- final List<TShowTopicInfo> topicInfoList, final SettableFuture<ConfigTaskResult> future) {
+ final List<TShowTopicInfo> topicInfoList,
+ final boolean isTableModel,
+ final SettableFuture<ConfigTaskResult> future) {
+ buildTSBlock(
+ topicInfoList,
+ isTableModel,
+ topicName -> SubscriptionAgent.broker().getTagFilterMatcher(topicName, true),
+ future);
+ }
+
+ static void buildTSBlock(
+ final List<TShowTopicInfo> topicInfoList,
+ final boolean isTableModel,
+ final Function<String, TagFilterMatcher> matcherProvider,
+ final SettableFuture<ConfigTaskResult> future) {
+ final List<ColumnHeader> columnHeaders =
+ isTableModel
+ ? ColumnHeaderConstant.showTableTopicColumnHeaders
+ : ColumnHeaderConstant.showTopicColumnHeaders;
final TsBlockBuilder builder =
new TsBlockBuilder(
- ColumnHeaderConstant.showTopicColumnHeaders.stream()
- .map(ColumnHeader::getColumnType)
- .collect(Collectors.toList()));
+ columnHeaders.stream().map(ColumnHeader::getColumnType).collect(Collectors.toList()));
for (final TShowTopicInfo topicInfo : topicInfoList) {
builder.getTimeColumnBuilder().writeLong(0L);
@@ -75,6 +94,17 @@
builder
.getColumnBuilder(1)
.writeBinary(new Binary(topicInfo.getTopicAttributes(), TSFileConfig.STRING_CHARSET));
+ if (isTableModel) {
+ final TagFilterMatcher matcher;
+ try {
+ matcher = matcherProvider.apply(topicInfo.getTopicName());
+ } catch (final RuntimeException e) {
+ writeTagFilterDiagnostic(builder, TagFilterMatcher.failure(e));
+ builder.declarePosition();
+ continue;
+ }
+ writeTagFilterDiagnostic(builder, matcher);
+ }
builder.declarePosition();
}
@@ -82,6 +112,22 @@
new ConfigTaskResult(
TSStatusCode.SUCCESS_STATUS,
builder.build(),
- DatasetHeaderFactory.getShowTopicHeader()));
+ isTableModel
+ ? DatasetHeaderFactory.getShowTableTopicHeader()
+ : DatasetHeaderFactory.getShowTopicHeader()));
+ }
+
+ private static void writeTagFilterDiagnostic(
+ final TsBlockBuilder builder, final TagFilterMatcher matcher) {
+ builder
+ .getColumnBuilder(2)
+ .writeBinary(new Binary(matcher.getRuntimeStatus().name(), TSFileConfig.STRING_CHARSET));
+ if (matcher.isFailure()) {
+ builder
+ .getColumnBuilder(3)
+ .writeBinary(new Binary(matcher.getFailureMessage(), TSFileConfig.STRING_CHARSET));
+ } else {
+ builder.getColumnBuilder(3).appendNull();
+ }
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java
index 63e94c2..2f8f99e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java
@@ -793,7 +793,7 @@
if (node.hasIfNotExistsCondition()) {
builder.append("IF NOT EXISTS ");
}
- builder.append(node.getTopicName());
+ builder.append(ShowCreateTableTask.getIdentifier(node.getTopicName()));
builder.append(" \n");
if (!node.getTopicAttributes().isEmpty()) {
@@ -805,11 +805,9 @@
.map(
entry ->
indentString(1)
- + "\""
- + entry.getKey()
- + "\" = \""
- + entry.getValue()
- + "\"")
+ + ShowCreateTableTask.getString(entry.getKey())
+ + " = "
+ + ShowCreateTableTask.getString(entry.getValue()))
.collect(joining(", " + "\n")))
.append(")\n");
}
@@ -820,7 +818,7 @@
@Override
public Void visitAlterTopic(AlterTopic node, Integer context) {
builder.append("ALTER TOPIC ");
- builder.append(node.getTopicName());
+ builder.append(ShowCreateTableTask.getIdentifier(node.getTopicName()));
builder.append(" \n");
builder
@@ -831,11 +829,9 @@
.map(
entry ->
indentString(1)
- + "\""
- + entry.getKey()
- + "\" = \""
- + entry.getValue()
- + "\"")
+ + ShowCreateTableTask.getString(entry.getKey())
+ + " = "
+ + ShowCreateTableTask.getString(entry.getValue()))
.collect(joining(", " + "\n")))
.append(")\n");
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 b5bd049..89d8776 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,6 +20,7 @@
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;
@@ -31,6 +32,7 @@
import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
+import org.apache.iotdb.db.schemaengine.table.ITableCache;
import org.apache.iotdb.db.subscription.broker.ConsensusSubscriptionBroker;
import org.apache.iotdb.db.subscription.broker.ISubscriptionBroker;
import org.apache.iotdb.db.subscription.broker.SubscriptionBroker;
@@ -42,10 +44,12 @@
import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher;
import org.apache.iotdb.db.subscription.event.SubscriptionEvent;
import org.apache.iotdb.db.subscription.resource.SubscriptionDataNodeResourceManager;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterMatcher;
import org.apache.iotdb.db.subscription.task.execution.ConsensusSubscriptionPrefetchExecutorManager;
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;
@@ -77,6 +81,7 @@
private static final ColumnFilterMatcher EMPTY_COLUMN_FILTER_MATCHER =
ColumnFilterMatcher.ofSelectedColumnNames(Collections.emptySet());
+ private static final int TAG_FILTER_SCHEMA_SNAPSHOT_MAX_RETRIES = 3;
/** Subscription brokers grouped by consumer group. */
private final Map<String, List<ISubscriptionBroker>> consumerGroupIdToBrokers =
@@ -87,6 +92,9 @@
private final Map<String, ColumnFilterMatcher> topicNameToColumnFilterMatcher =
new ConcurrentHashMap<>();
+ private final Map<String, TagFilterMatcher> topicNameToTagFilterMatcher =
+ new ConcurrentHashMap<>();
+ private final Map<String, Long> topicNameToTagFilterMatcherVersion = new ConcurrentHashMap<>();
//////////////////////////// provided for subscription agent ////////////////////////////
@@ -787,6 +795,157 @@
LOGGER.info(DataNodeMiscMessages.SUBSCRIPTION_DROP_COLUMN_FILTER, topicName);
}
+ public void refreshTagFilter(final String topicName, final TopicConfig topicConfig) {
+ final ITableCache tableCache = DataNodeTableCache.getInstance();
+ if (Objects.isNull(topicConfig)
+ || !topicConfig.isTableTopic()
+ || topicConfig.isTagFilterTrivial()) {
+ cacheTagFilterMatcher(
+ topicName, TagFilterMatcher.matchAll(), tableCache.getInstanceVersion());
+ return;
+ }
+
+ for (int attempt = 0; attempt < TAG_FILTER_SCHEMA_SNAPSHOT_MAX_RETRIES; attempt++) {
+ final long versionBeforeBinding = tableCache.getInstanceVersion();
+ final TagFilterMatcher matcher;
+ try {
+ final Map<String, Map<String, TsTable>> bindingTables =
+ getTagFilterBindingTables(topicConfig);
+ matcher = TagFilterMatcher.fromTopicConfig(topicConfig, bindingTables);
+ } catch (final Exception e) {
+ if (versionBeforeBinding != tableCache.getInstanceVersion()) {
+ continue;
+ }
+ LOGGER.warn(
+ DataNodeMiscMessages
+ .LOG_SUBSCRIPTION_FAILED_TO_REFRESH_TAG_FILTER_MATCHER_FOR_TOPIC_ARG_USE_EMPTY_MATCHER_TO_FAIL_CLOSED_3CB76D70,
+ topicName,
+ e);
+ cacheTagFilterMatcher(topicName, TagFilterMatcher.failure(e), versionBeforeBinding);
+ return;
+ }
+
+ if (versionBeforeBinding != tableCache.getInstanceVersion()) {
+ continue;
+ }
+ cacheTagFilterMatcher(topicName, matcher, versionBeforeBinding);
+ LOGGER.info(
+ DataNodeMiscMessages.LOG_SUBSCRIPTION_REFRESHED_TAG_FILTER_MATCHER_FOR_TOPIC_ARG_71598849,
+ topicName);
+ return;
+ }
+
+ // Leave the matcher uncached after repeated schema changes. The caller fails closed for this
+ // attempt and retries against a fresh schema snapshot on the next access.
+ topicNameToTagFilterMatcher.remove(topicName);
+ topicNameToTagFilterMatcherVersion.remove(topicName);
+ }
+
+ private void cacheTagFilterMatcher(
+ final String topicName, final TagFilterMatcher matcher, final long tableCacheVersion) {
+ topicNameToTagFilterMatcher.put(topicName, matcher);
+ topicNameToTagFilterMatcherVersion.put(topicName, tableCacheVersion);
+ }
+
+ public TagFilterMatcher getTagFilterMatcher(final String topicName) {
+ return getTagFilterMatcher(topicName, true);
+ }
+
+ public TagFilterMatcher getTagFilterMatcher(final String topicName, final boolean isTableModel) {
+ if (!isTableModel) {
+ return TagFilterMatcher.matchAll();
+ }
+
+ final TagFilterMatcher matcher = topicNameToTagFilterMatcher.get(topicName);
+ final long tableCacheVersion = DataNodeTableCache.getInstance().getInstanceVersion();
+ if (Objects.nonNull(matcher)
+ && Objects.equals(topicNameToTagFilterMatcherVersion.get(topicName), tableCacheVersion)) {
+ return matcher;
+ }
+ if (Objects.nonNull(matcher)) {
+ topicNameToTagFilterMatcher.remove(topicName, matcher);
+ topicNameToTagFilterMatcherVersion.remove(topicName);
+ }
+
+ final TopicConfig topicConfig = SubscriptionAgent.topic().getTopicConfig(topicName, true);
+ if (Objects.isNull(topicConfig)) {
+ return TagFilterMatcher.failure(
+ new SubscriptionException(
+ DataNodeMiscMessages
+ .EXCEPTION_TOPIC_CONFIGURATION_IS_NOT_AVAILABLE_FOR_TAG_FILTER_1E023E3A));
+ }
+
+ refreshTagFilter(topicName, topicConfig);
+ return topicNameToTagFilterMatcher.getOrDefault(
+ topicName, getDefaultTagFilterMatcher(topicConfig));
+ }
+
+ private static TagFilterMatcher getDefaultTagFilterMatcher(final TopicConfig topicConfig) {
+ return Objects.nonNull(topicConfig)
+ && topicConfig.isTableTopic()
+ && !topicConfig.isTagFilterTrivial()
+ ? TagFilterMatcher.failure(
+ new SubscriptionException(
+ String.format(
+ DataNodeMiscMessages
+ .EXCEPTION_TABLE_SCHEMA_KEPT_CHANGING_WHILE_BINDING_TAG_FILTER_FOR_DATABASE_PATTERN_ARG_AND_TABLE_PATTERN_ARG_RETRY_AFTER_SCHEMA_SYNCHRONIZATION_77A005E9,
+ topicConfig.getStringOrDefault(
+ TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE),
+ topicConfig.getStringOrDefault(
+ TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE))))
+ : TagFilterMatcher.matchAll();
+ }
+
+ private static Map<String, Map<String, TsTable>> getTagFilterBindingTables(
+ final TopicConfig topicConfig) {
+ if (Objects.isNull(topicConfig)
+ || !topicConfig.isTableTopic()
+ || topicConfig.isTagFilterTrivial()) {
+ 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);
+ if (Objects.isNull(table)) {
+ throw new SubscriptionException(
+ String.format(
+ DataNodeMiscMessages
+ .EXCEPTION_TABLE_SCHEMA_ARG_ARG_IS_ABSENT_FROM_THIS_DATANODE_S_TABLE_CACHE_THE_TABLE_MAY_NOT_EXIST_OR_ITS_SCHEMA_HAS_NOT_BEEN_SYNCHRONIZED_YET_23975BDF,
+ database,
+ tableName));
+ }
+ return 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 dropTagFilter(final String topicName) {
+ topicNameToTagFilterMatcher.remove(topicName);
+ topicNameToTagFilterMatcherVersion.remove(topicName);
+ LOGGER.info(
+ DataNodeMiscMessages.LOG_SUBSCRIPTION_DROPPED_TAG_FILTER_MATCHER_FOR_TOPIC_ARG_9EA52458,
+ topicName);
+ }
+
public void unbindConsensusPrefetchingQueue(
final String consumerGroupId,
final String topicName,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java
index f68e60f..ef655aa 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java
@@ -119,6 +119,9 @@
final boolean isTableModel = metaFromCoordinator.visibleUnderTableModel();
final TopicMeta oldMeta = topicMetaKeeper.getTopicMeta(topicName, isTableModel);
TopicMeta.validateOwnerProgression(oldMeta, metaFromCoordinator);
+ metaFromCoordinator
+ .getConfig()
+ .inheritFilterVersion(Objects.isNull(oldMeta) ? null : oldMeta.getConfig());
final Long recycleIntervalMs =
metaFromCoordinator.getConfig().getRecycleUncommittedEventIntervalMs();
topicMetaKeeper.removeTopicMeta(topicName, isTableModel);
@@ -133,11 +136,16 @@
}
if (shouldRefreshColumnFilter(oldMeta, metaFromCoordinator)) {
SubscriptionAgent.broker().refreshColumnFilter(topicName, metaFromCoordinator.getConfig());
- } else if (!metaFromCoordinator.getConfig().isTableTopic()
+ }
+ if (shouldRefreshTagFilter(oldMeta, metaFromCoordinator)) {
+ SubscriptionAgent.broker().refreshTagFilter(topicName, metaFromCoordinator.getConfig());
+ }
+ if (!metaFromCoordinator.getConfig().isTableTopic()
&& !topicMetaKeeper.containsTopicMeta(topicName, true)) {
// ConfigNode rejects column-filter on tree topics. Drop defensively in case stale or replayed
// topic metadata reaches this DataNode after a table-topic to tree-topic transition.
SubscriptionAgent.broker().dropColumnFilter(topicName);
+ SubscriptionAgent.broker().dropTagFilter(topicName);
}
SubscriptionAgent.broker()
.refreshConsensusQueueOrderMode(
@@ -175,6 +183,35 @@
newConfig, TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE)));
}
+ static boolean shouldRefreshTagFilter(final TopicMeta oldMeta, final TopicMeta newMeta) {
+ if (Objects.isNull(newMeta) || !newMeta.getConfig().isTableTopic()) {
+ return false;
+ }
+ if (Objects.isNull(oldMeta) || !oldMeta.getConfig().isTableTopic()) {
+ return true;
+ }
+
+ final TopicConfig oldConfig = oldMeta.getConfig();
+ final TopicConfig newConfig = newMeta.getConfig();
+ return !Objects.equals(
+ trimFilterBindingValue(oldConfig.getTagFilter()),
+ trimFilterBindingValue(newConfig.getTagFilter()))
+ || !Objects.equals(
+ trimFilterBindingValue(
+ getAttributeIgnoreCase(
+ oldConfig, TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE)),
+ trimFilterBindingValue(
+ getAttributeIgnoreCase(
+ newConfig, TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE)))
+ || !Objects.equals(
+ trimFilterBindingValue(
+ getAttributeIgnoreCase(
+ oldConfig, TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE)),
+ trimFilterBindingValue(
+ getAttributeIgnoreCase(
+ newConfig, TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE)));
+ }
+
private static String getAttributeIgnoreCase(
final TopicConfig topicConfig, final String key, final String defaultValue) {
return topicConfig.getAttribute().entrySet().stream()
@@ -189,6 +226,10 @@
return Objects.nonNull(value) ? value.trim().toLowerCase(Locale.ROOT) : "";
}
+ private static String trimFilterBindingValue(final String value) {
+ return Objects.nonNull(value) ? value.trim() : "";
+ }
+
public TPushTopicMetaRespExceptionMessage handleTopicMetaChanges(
final List<TopicMeta> topicMetasFromCoordinator) {
acquireWriteLock();
@@ -271,6 +312,7 @@
}
if (Objects.nonNull(topicMeta) && topicMeta.visibleUnderTableModel()) {
SubscriptionAgent.broker().dropColumnFilter(topicName);
+ SubscriptionAgent.broker().dropTagFilter(topicName);
}
if (Objects.nonNull(topicMeta) || Objects.nonNull(isTableModel)) {
topicRecycleIntervalsMs.remove(
@@ -346,6 +388,16 @@
return getTopicConfigs(topicNames, false);
}
+ public TopicConfig getTopicConfig(final String topicName, final boolean isTableModel) {
+ acquireReadLock();
+ try {
+ final TopicMeta topicMeta = topicMetaKeeper.getTopicMeta(topicName, isTableModel);
+ return Objects.isNull(topicMeta) ? null : topicMeta.getConfig();
+ } finally {
+ releaseReadLock();
+ }
+ }
+
public Map<String, TopicConfig> getTopicConfigs(
final Set<String> topicNames, final boolean isTableModel) {
acquireReadLock();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java
index 2756f4f..bd0b193 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java
@@ -736,12 +736,11 @@
}
private boolean canPassThroughTsFile(final PipeTsFileInsertionEvent event) {
+ final boolean isTableModelSubscription = SubscriptionAgent.consumer().isTableModel(brokerId);
return PipeEventCollector.canSkipParsing4TsFileEvent(event)
- && (!event.isTableModelEvent()
- || SubscriptionAgent.broker()
- .getColumnFilterMatcher(
- topicName, SubscriptionAgent.consumer().isTableModel(brokerId))
- .isMatchAll());
+ && (!isTableModelSubscription
+ || (SubscriptionAgent.broker().getColumnFilterMatcher(topicName, true).isMatchAll()
+ && SubscriptionAgent.broker().getTagFilterMatcher(topicName, true).isMatchAll()));
}
private RetryableState onRetryableTabletInsertionEvent(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
index 1f132c8..c87d3d4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
@@ -42,6 +42,8 @@
import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher;
import org.apache.iotdb.db.subscription.columnfilter.TabletColumnPruner;
+import org.apache.iotdb.db.subscription.tagfilter.TabletTagFilter;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterMatcher;
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
@@ -71,6 +73,7 @@
private final TablePattern tablePattern;
private final String topicName;
private final ColumnFilterMatcher fallbackColumnFilterMatcher;
+ private final TagFilterMatcher fallbackTagFilterMatcher;
/**
* The actual database name of the DataRegion this converter processes (table-model format without
@@ -83,7 +86,13 @@
final TablePattern tablePattern,
final ColumnFilterMatcher columnFilterMatcher,
final String databaseName) {
- this(treePattern, tablePattern, null, columnFilterMatcher, databaseName);
+ this(
+ treePattern,
+ tablePattern,
+ null,
+ columnFilterMatcher,
+ TagFilterMatcher.matchAll(),
+ databaseName);
}
public ConsensusLogToTabletConverter(
@@ -92,11 +101,38 @@
final String topicName,
final ColumnFilterMatcher columnFilterMatcher,
final String databaseName) {
+ this(
+ treePattern,
+ tablePattern,
+ topicName,
+ columnFilterMatcher,
+ TagFilterMatcher.matchAll(),
+ databaseName);
+ }
+
+ public ConsensusLogToTabletConverter(
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
+ final ColumnFilterMatcher columnFilterMatcher,
+ final TagFilterMatcher tagFilterMatcher,
+ final String databaseName) {
+ this(treePattern, tablePattern, null, columnFilterMatcher, tagFilterMatcher, databaseName);
+ }
+
+ public ConsensusLogToTabletConverter(
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
+ final String topicName,
+ final ColumnFilterMatcher columnFilterMatcher,
+ final TagFilterMatcher tagFilterMatcher,
+ final String databaseName) {
this.treePattern = treePattern;
this.tablePattern = tablePattern;
this.topicName = topicName;
this.fallbackColumnFilterMatcher =
Objects.nonNull(columnFilterMatcher) ? columnFilterMatcher : ColumnFilterMatcher.matchAll();
+ this.fallbackTagFilterMatcher =
+ Objects.nonNull(tagFilterMatcher) ? tagFilterMatcher : TagFilterMatcher.matchAll();
this.databaseName = databaseName;
}
@@ -104,16 +140,17 @@
return databaseName;
}
- /** Freezes the column projection used by a delivered batch for later WAL replay. */
+ /** Freezes the column and TAG filters used by a delivered batch for later WAL replay. */
ConsensusLogToTabletConverter snapshotForReplay() {
return new ConsensusLogToTabletConverter(
- treePattern, tablePattern, getColumnFilterMatcher(), databaseName);
+ treePattern, tablePattern, getColumnFilterMatcher(), getTagFilterMatcher(), databaseName);
}
boolean hasSameReplayProjection(final ConsensusLogToTabletConverter other) {
return treePattern == other.treePattern
&& tablePattern == other.tablePattern
- && fallbackColumnFilterMatcher == other.fallbackColumnFilterMatcher;
+ && fallbackColumnFilterMatcher == other.fallbackColumnFilterMatcher
+ && fallbackTagFilterMatcher == other.fallbackTagFilterMatcher;
}
boolean isTableModel() {
@@ -527,7 +564,10 @@
}
final Tablet prunedTablet =
- TabletColumnPruner.pruneTableModelTablet(tablet, databaseName, getColumnFilterMatcher());
+ TabletColumnPruner.pruneTableModelTablet(
+ TabletTagFilter.filter(tablet, getTagFilterMatcher(), databaseName),
+ databaseName,
+ getColumnFilterMatcher());
return Objects.nonNull(prunedTablet)
? Collections.singletonList(prunedTablet)
: Collections.emptyList();
@@ -601,7 +641,10 @@
rowCount);
final Tablet prunedTablet =
- TabletColumnPruner.pruneTableModelTablet(tablet, databaseName, getColumnFilterMatcher());
+ TabletColumnPruner.pruneTableModelTablet(
+ TabletTagFilter.filter(tablet, getTagFilterMatcher(), databaseName),
+ databaseName,
+ getColumnFilterMatcher());
return Objects.nonNull(prunedTablet)
? Collections.singletonList(prunedTablet)
: Collections.emptyList();
@@ -718,6 +761,12 @@
: fallbackColumnFilterMatcher;
}
+ private TagFilterMatcher getTagFilterMatcher() {
+ return Objects.nonNull(topicName)
+ ? SubscriptionAgent.broker().getTagFilterMatcher(topicName, true)
+ : fallbackTagFilterMatcher;
+ }
+
private boolean isValidColumn(
final String[] measurements,
final TSDataType[] dataTypes,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
index 964bbcb..474deb3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
@@ -716,6 +716,7 @@
if (isTableTopic) {
SubscriptionAgent.broker().refreshColumnFilter(topicName, topicConfig);
+ SubscriptionAgent.broker().refreshTagFilter(topicName, topicConfig);
// Table model: database + table name pattern
tablePattern = buildTablePattern(topicConfig);
return new ConsensusLogToTabletConverter(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterParser.java
index 2e7200a..1770900 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterParser.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterParser.java
@@ -61,7 +61,8 @@
private static final Pattern SINGLE_FIELD_PATTERN =
Pattern.compile("\\s*(?:[A-Za-z_][A-Za-z_0-9]*|\"(?:\"\"|[^\"])*\")\\s*");
private static final Pattern FUNCTION_CALL_START_PATTERN =
- Pattern.compile("\\s*(?:[A-Za-z_][A-Za-z_0-9]*|\"(?:\"\"|[^\"])*\")\\s*\\(.*");
+ // NOT (...) is a logical expression supported by the grammar.
+ Pattern.compile("\\s*(?!(?i:NOT)\\b)(?:[A-Za-z_][A-Za-z_0-9]*|\"(?:\"\"|[^\"])*\")\\s*\\(.*");
private static final Pattern UNQUOTED_COMPARISON_RIGHT_PATTERN =
Pattern.compile("(?is).*(?:!=|<>|=)\\s*[A-Za-z_][A-Za-z_0-9]*\\s*");
@@ -90,7 +91,7 @@
}
}
- Expression parse(final String rawColumnFilter) {
+ public Expression parse(final String rawColumnFilter) {
if (rawColumnFilter == null || rawColumnFilter.trim().isEmpty()) {
throw new ParsingException(
DataNodeMiscMessages.COLUMN_FILTER_SHOULD_NOT_BE_EMPTY, null, 1, 1);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionFilterSnapshot.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionFilterSnapshot.java
new file mode 100644
index 0000000..02dc4ec
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionFilterSnapshot.java
@@ -0,0 +1,125 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.subscription.event.batch;
+
+import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
+import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
+import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingQueue;
+import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterEvaluationException;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterMatcher;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+
+import java.util.Objects;
+
+/** Immutable topic-filter state used by every event in one subscription batch. */
+final class SubscriptionFilterSnapshot {
+
+ private static final int MAX_CAPTURE_ATTEMPTS = 3;
+
+ private final boolean tableModel;
+ private final TopicConfig topicConfig;
+ private final long filterVersion;
+ private final TagFilterMatcher tagFilterMatcher;
+ private final ColumnFilterMatcher columnFilterMatcher;
+
+ private SubscriptionFilterSnapshot(
+ final boolean tableModel,
+ final TopicConfig topicConfig,
+ final TagFilterMatcher tagFilterMatcher,
+ final ColumnFilterMatcher columnFilterMatcher) {
+ this.tableModel = tableModel;
+ this.topicConfig = topicConfig;
+ this.filterVersion = topicConfig.getFilterVersion();
+ this.tagFilterMatcher = tagFilterMatcher;
+ this.columnFilterMatcher = columnFilterMatcher;
+ }
+
+ static SubscriptionFilterSnapshot capture(final SubscriptionPrefetchingQueue queue) {
+ final boolean tableModel =
+ SubscriptionAgent.consumer().isTableModel(queue.getConsumerGroupId());
+ for (int attempt = 0; attempt < MAX_CAPTURE_ATTEMPTS; attempt++) {
+ final TopicConfig topicConfig = getCurrentTopicConfig(queue, tableModel);
+ if (Objects.isNull(topicConfig)) {
+ throw new TagFilterEvaluationException(
+ DataNodeMiscMessages
+ .EXCEPTION_TOPIC_CONFIGURATION_IS_NOT_AVAILABLE_FOR_TAG_FILTER_1E023E3A);
+ }
+
+ final TagFilterMatcher tagFilterMatcher =
+ SubscriptionAgent.broker().getTagFilterMatcher(queue.getTopicName(), tableModel);
+ tagFilterMatcher.throwIfFailure();
+ final ColumnFilterMatcher columnFilterMatcher =
+ SubscriptionAgent.broker().getColumnFilterMatcher(queue.getTopicName(), tableModel);
+ // Column-filter binding failures are represented by a match-none matcher, not an exception
+ // state. Keeping that matcher also fails closed for every event in this batch.
+
+ final TopicConfig currentTopicConfig = getCurrentTopicConfig(queue, tableModel);
+ if (Objects.nonNull(currentTopicConfig)
+ && topicConfig.getFilterVersion() == currentTopicConfig.getFilterVersion()
+ && tagFilterMatcher
+ == SubscriptionAgent.broker().getTagFilterMatcher(queue.getTopicName(), tableModel)
+ && columnFilterMatcher
+ == SubscriptionAgent.broker()
+ .getColumnFilterMatcher(queue.getTopicName(), tableModel)) {
+ return new SubscriptionFilterSnapshot(
+ tableModel, new TopicConfig(topicConfig), tagFilterMatcher, columnFilterMatcher);
+ }
+ }
+
+ throw new TagFilterEvaluationException(
+ DataNodeMiscMessages
+ .EXCEPTION_TOPIC_CONFIGURATION_CHANGED_WHILE_CAPTURING_TAG_FILTER_SNAPSHOT_984CEB1B);
+ }
+
+ boolean isCurrent(final SubscriptionPrefetchingQueue queue) {
+ if (!tableModel) {
+ return true;
+ }
+ final TopicConfig currentTopicConfig = getCurrentTopicConfig(queue, true);
+ return Objects.nonNull(currentTopicConfig)
+ && filterVersion == currentTopicConfig.getFilterVersion()
+ && tagFilterMatcher
+ == SubscriptionAgent.broker().getTagFilterMatcher(queue.getTopicName(), true)
+ && columnFilterMatcher
+ == SubscriptionAgent.broker().getColumnFilterMatcher(queue.getTopicName(), true);
+ }
+
+ TopicConfig getTopicConfig() {
+ return topicConfig;
+ }
+
+ TagFilterMatcher getTagFilterMatcher() {
+ return tagFilterMatcher;
+ }
+
+ ColumnFilterMatcher getColumnFilterMatcher() {
+ return columnFilterMatcher;
+ }
+
+ boolean hasNonTrivialFilter() {
+ return tableModel && (!tagFilterMatcher.isMatchAll() || !columnFilterMatcher.isMatchAll());
+ }
+
+ private static TopicConfig getCurrentTopicConfig(
+ final SubscriptionPrefetchingQueue queue, final boolean tableModel) {
+ return SubscriptionAgent.topic().getTopicConfig(queue.getTopicName(), tableModel);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatch.java
index d9361bf..741baba 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatch.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatch.java
@@ -114,7 +114,7 @@
/////////////////////////////// utility ///////////////////////////////
- protected abstract void onTabletInsertionEvent(final TabletInsertionEvent event);
+ protected abstract void onTabletInsertionEvent(final TabletInsertionEvent event) throws Exception;
protected abstract void onTsFileInsertionEvent(final TsFileInsertionEvent event);
@@ -122,6 +122,10 @@
protected abstract List<SubscriptionEvent> generateSubscriptionEvents() throws Exception;
+ protected boolean isCompatibleWithCurrentTopicConfig() {
+ return true;
+ }
+
//////////////////////////// APIs provided for metric framework ////////////////////////////
public int getPipeEventCount() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatches.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatches.java
index 0cb81ca..d16b357 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatches.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatches.java
@@ -25,6 +25,7 @@
import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingTabletQueue;
import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingTsFileQueue;
import org.apache.iotdb.db.subscription.event.SubscriptionEvent;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
import com.google.common.collect.ImmutableList;
import org.slf4j.Logger;
@@ -77,6 +78,7 @@
}
} catch (final Exception e) {
LOGGER.warn(DataNodeMiscMessages.EXCEPTION_SEALING_EVENTS, batch, e);
+ throw propagate(e);
}
if (hasNew.get()) {
regionIdToBatch.remove(regionId);
@@ -103,33 +105,35 @@
SubscriptionPipeEventBatch batch = regionIdToBatch.get(regionId);
if (Objects.isNull(batch)) {
try {
- batch =
- prefetchingQueue instanceof SubscriptionPrefetchingTabletQueue
- ? new SubscriptionPipeTabletEventBatch(
- regionId,
- (SubscriptionPrefetchingTabletQueue) prefetchingQueue,
- maxDelayInMs,
- maxBatchSizeInBytes)
- : new SubscriptionPipeTsFileEventBatch(
- regionId,
- (SubscriptionPrefetchingTsFileQueue) prefetchingQueue,
- maxDelayInMs,
- maxBatchSizeInBytes);
+ batch = createBatch(regionId);
} catch (final Exception e) {
LOGGER.warn(DataNodeMiscMessages.EXCEPTION_CONSTRUCT_NEW_BATCH, e);
throw e; // rethrow exception for retry
}
}
+ if (!batch.isCompatibleWithCurrentTopicConfig() && batch.getPipeEventCount() > 0) {
+ if (!batch.emit(consumer)) {
+ throw new SubscriptionException(
+ DataNodeMiscMessages.EXCEPTION_FAILED_TO_SEAL_SUBSCRIPTION_EVENT_BATCH_1FB7E92C);
+ }
+ hasNew.set(true);
+ regionIdToBatch.remove(regionId);
+ batch = createBatch(regionId);
+ }
+
+ final boolean emittedCurrentBatch;
try {
- if (batch.onEvent(event, consumer)) {
+ emittedCurrentBatch = batch.onEvent(event, consumer);
+ if (emittedCurrentBatch) {
hasNew.set(true);
}
} catch (final Exception e) {
LOGGER.warn(DataNodeMiscMessages.EXCEPTION_SEALING_EVENTS, batch, e);
+ throw e;
}
- if (hasNew.get()) {
+ if (emittedCurrentBatch) {
regionIdToBatch.remove(regionId);
} else {
regionIdToBatch.put(regionId, batch);
@@ -175,4 +179,26 @@
regionIdToBatch.values().forEach(batch -> batch.cleanUp(true));
regionIdToBatch.clear();
}
+
+ private SubscriptionPipeEventBatch createBatch(final int regionId) {
+ return prefetchingQueue instanceof SubscriptionPrefetchingTabletQueue
+ ? new SubscriptionPipeTabletEventBatch(
+ regionId,
+ (SubscriptionPrefetchingTabletQueue) prefetchingQueue,
+ maxDelayInMs,
+ maxBatchSizeInBytes)
+ : new SubscriptionPipeTsFileEventBatch(
+ regionId,
+ (SubscriptionPrefetchingTsFileQueue) prefetchingQueue,
+ maxDelayInMs,
+ maxBatchSizeInBytes);
+ }
+
+ private static RuntimeException propagate(final Exception exception) {
+ return exception instanceof RuntimeException
+ ? (RuntimeException) exception
+ : new SubscriptionException(
+ DataNodeMiscMessages.EXCEPTION_FAILED_TO_SEAL_SUBSCRIPTION_EVENT_BATCH_1FB7E92C,
+ exception);
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java
index 1d3987a..0f2653a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java
@@ -19,27 +19,24 @@
package org.apache.iotdb.db.subscription.event.batch;
-import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
-import org.apache.iotdb.commons.schema.table.TreeViewSchema;
-import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
+import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
-import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingTabletQueue;
import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher;
import org.apache.iotdb.db.subscription.columnfilter.TabletColumnPruner;
-import org.apache.iotdb.db.subscription.columnfilter.TreeViewTabletProjector;
import org.apache.iotdb.db.subscription.event.SubscriptionEvent;
+import org.apache.iotdb.db.subscription.tagfilter.TabletTagFilter;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterEvaluationException;
import org.apache.iotdb.metrics.core.utils.IoTDBMovingAverage;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
-import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import com.codahale.metrics.Clock;
@@ -50,7 +47,6 @@
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -73,8 +69,8 @@
private final Meter insertNodeTabletInsertionEventSizeEstimator;
private final Meter rawTabletInsertionEventSizeEstimator;
- private volatile boolean treeViewTabletProjectorInitialized;
- private volatile TreeViewTabletProjector treeViewTabletProjector;
+ private volatile SubscriptionFilterSnapshot filterSnapshot;
+ private volatile SubscriptionTreeViewProjector treeViewProjector;
private volatile SubscriptionPipeTabletIterationSnapshot iterationSnapshot;
private final AtomicInteger referenceCount = new AtomicInteger();
@@ -132,6 +128,8 @@
@Override
protected void onTabletInsertionEvent(final TabletInsertionEvent event) {
+ ensureFilterSnapshot();
+
// update processing time
if (firstEventProcessingTime == Long.MIN_VALUE) {
firstEventProcessingTime = System.currentTimeMillis();
@@ -148,6 +146,8 @@
@Override
protected void onTsFileInsertionEvent(final TsFileInsertionEvent event) {
+ ensureFilterSnapshot();
+
// update processing time
if (firstEventProcessingTime == Long.MIN_VALUE) {
firstEventProcessingTime = System.currentTimeMillis();
@@ -163,6 +163,7 @@
@Override
protected List<SubscriptionEvent> generateSubscriptionEvents() {
+ ensureFilterSnapshot();
if (!prepareTreeViewTabletProjectorForEmission()) {
return null;
}
@@ -226,105 +227,46 @@
return tablets;
}
- final TreeViewTabletProjector projector = getTreeViewTabletProjector();
- if (Objects.isNull(projector)) {
- return tablets;
+ if (!prepareTreeViewTabletProjectorForEmission() || !treeViewProjector.isAvailable()) {
+ if (!ensureFilterSnapshot().hasNonTrivialFilter()) {
+ return tablets;
+ }
+ throw new TagFilterEvaluationException(
+ DataNodeMiscMessages
+ .EXCEPTION_TREE_VIEW_PROJECTOR_IS_UNAVAILABLE_FOR_FILTERED_SUBSCRIPTION_DATA_B5F396A5);
}
final List<Tablet> projectedTablets = new ArrayList<>(tablets.right.size());
for (final Tablet tablet : tablets.right) {
- final Tablet projectedTablet = projector.project(tablet);
+ final Tablet projectedTablet = treeViewProjector.project(tablet);
if (Objects.nonNull(projectedTablet)) {
projectedTablets.add(projectedTablet);
}
}
return projectedTablets.isEmpty()
? null
- : new Pair<>(projector.getDatabaseName(), projectedTablets);
- }
-
- private TreeViewTabletProjector getTreeViewTabletProjector() {
- return prepareTreeViewTabletProjectorForEmission() ? treeViewTabletProjector : null;
+ : new Pair<>(treeViewProjector.getDatabaseName(), projectedTablets);
}
private boolean prepareTreeViewTabletProjectorForEmission() {
- if (treeViewTabletProjectorInitialized) {
- return true;
+ final SubscriptionFilterSnapshot snapshot = ensureFilterSnapshot();
+ if (Objects.isNull(treeViewProjector)) {
+ treeViewProjector = new SubscriptionTreeViewProjector(snapshot.getTopicConfig());
}
-
- synchronized (this) {
- if (treeViewTabletProjectorInitialized) {
- return true;
- }
-
- final TopicConfig topicConfig =
- SubscriptionAgent.topic()
- .getTopicConfigs(
- Collections.singleton(prefetchingQueue.getTopicName()),
- SubscriptionAgent.consumer().isTableModel(prefetchingQueue.getConsumerGroupId()))
- .get(prefetchingQueue.getTopicName());
- if (Objects.isNull(topicConfig)) {
- return false;
- }
- if (!topicConfig.isTableTopic()) {
- treeViewTabletProjectorInitialized = true;
- return true;
- }
-
- 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)) {
- treeViewTabletProjectorInitialized = true;
- return true;
- }
-
- if (!isTreeCapturedByTopic(topicConfig) && topicConfig.isColumnFilterTrivial()) {
- treeViewTabletProjectorInitialized = true;
- return true;
- }
-
- final TsTable table = DataNodeTableCache.getInstance().getTable(database, tableName, false);
- if (Objects.isNull(table)) {
- LOGGER.debug(
- DataNodePipeMessages
- .PIPE_LOG_SUBSCRIPTIONPIPETABLETEVENTBATCH_POSTPONE_EMITTING_SUBSCRIPTION_TABLET_BATCH_FOR_TOPIC_ARG_BECAUSE_TABLE_SCHEMA_ARG_ARG_IS_NOT_AVAILABLE_LOCALLY_996C618D,
- prefetchingQueue.getTopicName(),
- database,
- tableName);
- return false;
- }
- if (TreeViewSchema.isTreeViewTable(table)) {
- treeViewTabletProjector = new TreeViewTabletProjector(database, table);
- }
-
- treeViewTabletProjectorInitialized = true;
- return true;
+ final boolean prepared = treeViewProjector.prepare();
+ if (!prepared) {
+ LOGGER.debug(
+ DataNodePipeMessages
+ .PIPE_LOG_SUBSCRIPTIONPIPETABLETEVENTBATCH_POSTPONE_EMITTING_SUBSCRIPTION_TABLET_BATCH_FOR_TOPIC_ARG_BECAUSE_TABLE_SCHEMA_ARG_ARG_IS_NOT_AVAILABLE_LOCALLY_996C618D,
+ prefetchingQueue.getTopicName(),
+ snapshot
+ .getTopicConfig()
+ .getStringOrDefault(TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE),
+ snapshot
+ .getTopicConfig()
+ .getStringOrDefault(TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE));
}
- }
-
- private static boolean isDefaultTopicPattern(final String pattern, final String defaultPattern) {
- return Objects.isNull(pattern) || defaultPattern.equals(pattern.trim());
- }
-
- 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 isTreeCapturedByTopic(final TopicConfig topicConfig) {
- return topicConfig.getBooleanOrDefault(
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_CAPTURE_TREE_KEY,
- PipeSourceConstant.SOURCE_CAPTURE_TREE_KEY),
- false);
+ return prepared;
}
private Pair<String, List<Tablet>> pruneTablets(final Pair<String, List<Tablet>> tablets) {
@@ -332,16 +274,15 @@
return tablets;
}
- final ColumnFilterMatcher matcher =
- SubscriptionAgent.broker()
- .getColumnFilterMatcher(
- prefetchingQueue.getTopicName(),
- SubscriptionAgent.consumer().isTableModel(prefetchingQueue.getConsumerGroupId()));
+ final SubscriptionFilterSnapshot snapshot = ensureFilterSnapshot();
final List<Tablet> prunedTablets = new ArrayList<>(tablets.right.size());
for (final Tablet tablet : tablets.right) {
final Tablet prunedTablet =
- TabletColumnPruner.pruneTableModelTablet(tablet, tablets.left, matcher);
+ TabletColumnPruner.pruneTableModelTablet(
+ TabletTagFilter.filter(tablet, snapshot.getTagFilterMatcher(), tablets.left),
+ tablets.left,
+ snapshot.getColumnFilterMatcher());
if (Objects.nonNull(prunedTablet)) {
prunedTablets.add(prunedTablet);
}
@@ -349,6 +290,18 @@
return prunedTablets.isEmpty() ? null : new Pair<>(tablets.left, prunedTablets);
}
+ @Override
+ protected boolean isCompatibleWithCurrentTopicConfig() {
+ return Objects.isNull(filterSnapshot) || filterSnapshot.isCurrent(prefetchingQueue);
+ }
+
+ private synchronized SubscriptionFilterSnapshot ensureFilterSnapshot() {
+ if (Objects.isNull(filterSnapshot)) {
+ filterSnapshot = SubscriptionFilterSnapshot.capture(prefetchingQueue);
+ }
+ return filterSnapshot;
+ }
+
/////////////////////////////// estimator ///////////////////////////////
private long getEstimatedInsertNodeTabletInsertionEventSize() {
@@ -381,6 +334,10 @@
return result;
}
+ public ColumnFilterMatcher getColumnFilterMatcher() {
+ return ensureFilterSnapshot().getColumnFilterMatcher();
+ }
+
public synchronized void resetForIteration() {
currentEnrichedEventsIterator = enrichedEvents.iterator();
currentTabletInsertionEventsIterator = null;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java
index 7a27f46..5bfc9eb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java
@@ -21,16 +21,19 @@
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventTsFileBatch;
-import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
+import org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventTsFileBatch.TabletTransformResult;
import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingTsFileQueue;
-import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher;
import org.apache.iotdb.db.subscription.columnfilter.TabletColumnPruner;
import org.apache.iotdb.db.subscription.event.SubscriptionEvent;
import org.apache.iotdb.db.subscription.event.pipe.SubscriptionPipeTsFileBatchEvents;
+import org.apache.iotdb.db.subscription.tagfilter.TabletTagFilter;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterEvaluationException;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext;
import org.apache.tsfile.utils.Pair;
@@ -42,6 +45,7 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
public class SubscriptionPipeTsFileEventBatch extends SubscriptionPipeEventBatch {
@@ -51,6 +55,8 @@
private final PipeTabletEventTsFileBatch batch;
private final List<Pair<String, File>> sealedFilePairs = new ArrayList<>();
+ private volatile SubscriptionFilterSnapshot filterSnapshot;
+ private volatile SubscriptionTreeViewProjector treeViewProjector;
public SubscriptionPipeTsFileEventBatch(
final int regionId,
@@ -59,8 +65,7 @@
final long maxBatchSizeInBytes) {
super(regionId, prefetchingQueue, maxDelayInMs, maxBatchSizeInBytes);
this.batch =
- new PipeTabletEventTsFileBatch(
- maxDelayInMs, maxBatchSizeInBytes, this::pruneTableModelTablet);
+ new PipeTabletEventTsFileBatch(maxDelayInMs, maxBatchSizeInBytes, this::transformTablet);
}
@TestOnly
@@ -99,12 +104,11 @@
/////////////////////////////// utility ///////////////////////////////
@Override
- protected void onTabletInsertionEvent(final TabletInsertionEvent event) {
- try {
- batch.onEvent(event);
- } catch (final Exception ignored) {
- // no exceptions will be thrown
- }
+ protected void onTabletInsertionEvent(final TabletInsertionEvent event) throws Exception {
+ ensureFilterSnapshot();
+ // Keep the queue's reference when transformation fails: the prefetching queue retries this
+ // same event without acquiring another reference. Release it only after successful batching.
+ batch.onEvent(event);
((EnrichedEvent) event)
.decreaseReferenceCount(
SubscriptionPipeTsFileEventBatch.class.getName(),
@@ -123,8 +127,7 @@
@Override
protected List<SubscriptionEvent> generateSubscriptionEvents() throws Exception {
if (batch.isEmpty()) {
- enrichedEvents.clear();
- return Collections.emptyList();
+ return discardEmptyBatch();
}
final List<SubscriptionEvent> events = new ArrayList<>();
@@ -132,8 +135,7 @@
if (dbTsFilePairs.isEmpty()) {
batch.decreaseEventsReferenceCount(this.getClass().getName(), true);
batch.onSuccess();
- enrichedEvents.clear();
- return Collections.emptyList();
+ return discardEmptyBatch();
}
sealedFilePairs.addAll(dbTsFilePairs);
final AtomicInteger ackReferenceCount = new AtomicInteger(dbTsFilePairs.size());
@@ -151,17 +153,95 @@
return events;
}
+ /** Closes the inner conversion batch when filtering produced no TsFile payload. */
+ private List<SubscriptionEvent> discardEmptyBatch() {
+ try {
+ batch.close();
+ } finally {
+ enrichedEvents.clear();
+ }
+ return Collections.emptyList();
+ }
+
@Override
protected boolean shouldEmit() {
return (!enrichedEvents.isEmpty() && batch.isEmpty()) || batch.shouldEmit();
}
- private Tablet pruneTableModelTablet(final String databaseName, final Tablet tablet) {
- final ColumnFilterMatcher matcher =
- SubscriptionAgent.broker()
- .getColumnFilterMatcher(
- prefetchingQueue.getTopicName(),
- SubscriptionAgent.consumer().isTableModel(prefetchingQueue.getConsumerGroupId()));
- return TabletColumnPruner.pruneTableModelTablet(tablet, databaseName, matcher);
+ private TabletTransformResult transformTablet(
+ final String databaseName,
+ final Tablet tablet,
+ final boolean isTableModel,
+ final boolean isAligned) {
+ final SubscriptionFilterSnapshot snapshot = ensureFilterSnapshot();
+ if (isTableModel) {
+ return new TabletTransformResult(
+ pruneTableModelTablet(databaseName, tablet, snapshot), databaseName, true, isAligned);
+ }
+ if (!snapshot.getTopicConfig().isTableTopic()) {
+ return new TabletTransformResult(tablet, databaseName, false, isAligned);
+ }
+
+ if (!prepareTreeViewProjector(snapshot)) {
+ throw new TagFilterEvaluationException(
+ DataNodeMiscMessages.EXCEPTION_TABLE_SCHEMA_IS_NOT_AVAILABLE_FOR_TAG_FILTER_993AB728);
+ }
+ if (!treeViewProjector.isAvailable()) {
+ if (snapshot.hasNonTrivialFilter()) {
+ throw new TagFilterEvaluationException(
+ DataNodeMiscMessages
+ .EXCEPTION_TREE_VIEW_PROJECTOR_IS_UNAVAILABLE_FOR_FILTERED_SUBSCRIPTION_DATA_B5F396A5);
+ }
+ return new TabletTransformResult(tablet, databaseName, false, isAligned);
+ }
+
+ final Tablet projectedTablet = treeViewProjector.project(tablet);
+ return Objects.isNull(projectedTablet)
+ ? null
+ : new TabletTransformResult(
+ pruneTableModelTablet(treeViewProjector.getDatabaseName(), projectedTablet, snapshot),
+ treeViewProjector.getDatabaseName(),
+ true,
+ true);
+ }
+
+ private Tablet pruneTableModelTablet(
+ final String databaseName, final Tablet tablet, final SubscriptionFilterSnapshot snapshot) {
+ return TabletColumnPruner.pruneTableModelTablet(
+ TabletTagFilter.filter(tablet, snapshot.getTagFilterMatcher(), databaseName),
+ databaseName,
+ snapshot.getColumnFilterMatcher());
+ }
+
+ private boolean prepareTreeViewProjector(final SubscriptionFilterSnapshot snapshot) {
+ if (Objects.isNull(treeViewProjector)) {
+ treeViewProjector = new SubscriptionTreeViewProjector(snapshot.getTopicConfig());
+ }
+ final boolean prepared = treeViewProjector.prepare();
+ if (!prepared) {
+ LOGGER.debug(
+ DataNodePipeMessages
+ .PIPE_LOG_SUBSCRIPTIONPIPETABLETEVENTBATCH_POSTPONE_EMITTING_SUBSCRIPTION_TABLET_BATCH_FOR_TOPIC_ARG_BECAUSE_TABLE_SCHEMA_ARG_ARG_IS_NOT_AVAILABLE_LOCALLY_996C618D,
+ prefetchingQueue.getTopicName(),
+ snapshot
+ .getTopicConfig()
+ .getStringOrDefault(TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE),
+ snapshot
+ .getTopicConfig()
+ .getStringOrDefault(TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE));
+ }
+ return prepared;
+ }
+
+ @Override
+ protected boolean isCompatibleWithCurrentTopicConfig() {
+ return Objects.isNull(filterSnapshot) || filterSnapshot.isCurrent(prefetchingQueue);
+ }
+
+ private synchronized SubscriptionFilterSnapshot ensureFilterSnapshot() {
+ if (Objects.isNull(filterSnapshot)) {
+ filterSnapshot = SubscriptionFilterSnapshot.capture(prefetchingQueue);
+ }
+ return filterSnapshot;
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionTreeViewProjector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionTreeViewProjector.java
new file mode 100644
index 0000000..140e26a
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionTreeViewProjector.java
@@ -0,0 +1,126 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.subscription.event.batch;
+
+import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
+import org.apache.iotdb.commons.schema.table.TreeViewSchema;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
+import org.apache.iotdb.db.subscription.columnfilter.TreeViewTabletProjector;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterEvaluationException;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+
+import org.apache.tsfile.write.record.Tablet;
+
+import java.util.Arrays;
+import java.util.Objects;
+
+/**
+ * Resolves and applies the table projection for tree-model events captured by a Tree View topic.
+ */
+final class SubscriptionTreeViewProjector {
+
+ private final TopicConfig topicConfig;
+
+ private boolean initialized;
+ private TreeViewTabletProjector projector;
+
+ SubscriptionTreeViewProjector(final TopicConfig topicConfig) {
+ this.topicConfig = topicConfig;
+ }
+
+ synchronized boolean prepare() {
+ if (initialized) {
+ return true;
+ }
+ if (Objects.isNull(topicConfig) || !topicConfig.isTableTopic()) {
+ initialized = true;
+ return true;
+ }
+
+ 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)) {
+ initialized = true;
+ return true;
+ }
+
+ if (!isTreeCapturedByTopic(topicConfig)
+ && topicConfig.isColumnFilterTrivial()
+ && topicConfig.isTagFilterTrivial()) {
+ initialized = true;
+ return true;
+ }
+
+ final TsTable table = DataNodeTableCache.getInstance().getTable(database, tableName, false);
+ if (Objects.isNull(table)) {
+ return false;
+ }
+ if (TreeViewSchema.isTreeViewTable(table)) {
+ projector = new TreeViewTabletProjector(database, table);
+ }
+ initialized = true;
+ return true;
+ }
+
+ Tablet project(final Tablet tablet) {
+ if (Objects.isNull(projector)) {
+ throw new TagFilterEvaluationException(
+ DataNodeMiscMessages
+ .EXCEPTION_TREE_VIEW_PROJECTOR_IS_UNAVAILABLE_FOR_FILTERED_SUBSCRIPTION_DATA_B5F396A5);
+ }
+ // Only a prepared projector can return null to mean that this tablet does not match the view.
+ return projector.project(tablet);
+ }
+
+ boolean isAvailable() {
+ return Objects.nonNull(projector);
+ }
+
+ String getDatabaseName() {
+ return Objects.nonNull(projector) ? projector.getDatabaseName() : null;
+ }
+
+ private static boolean isDefaultTopicPattern(final String pattern, final String defaultPattern) {
+ return Objects.isNull(pattern) || defaultPattern.equals(pattern.trim());
+ }
+
+ 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 isTreeCapturedByTopic(final TopicConfig topicConfig) {
+ return topicConfig.getBooleanOrDefault(
+ Arrays.asList(
+ PipeSourceConstant.EXTRACTOR_CAPTURE_TREE_KEY,
+ PipeSourceConstant.SOURCE_CAPTURE_TREE_KEY),
+ false);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java
index 5854365..3bcbc88 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java
@@ -27,7 +27,6 @@
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
import org.apache.iotdb.db.pipe.resource.memory.PipeTabletMemoryBlock;
-import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingQueue;
import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher;
import org.apache.iotdb.db.subscription.event.SubscriptionEvent;
@@ -96,11 +95,7 @@
this.commitContext = commitContext;
this.rootCommitContext = rootCommitContext;
- this.columnFilterMatcher =
- SubscriptionAgent.broker()
- .getColumnFilterMatcher(
- queue.getTopicName(),
- SubscriptionAgent.consumer().isTableModel(queue.getConsumerGroupId()));
+ this.columnFilterMatcher = batch.getColumnFilterMatcher();
init();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TabletTagFilter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TabletTagFilter.java
new file mode 100644
index 0000000..dcc7ca7
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TabletTagFilter.java
@@ -0,0 +1,251 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.subscription.tagfilter;
+
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Identifier;
+import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
+
+import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+
+import java.lang.reflect.Array;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+
+/** Filters rows of a table-model Tablet according to TAG column values. */
+public class TabletTagFilter {
+
+ private static final String DELIMITED_FIELD_PREFIX = "D:";
+ private static final String UNDELIMITED_FIELD_PREFIX = "U:";
+
+ private TabletTagFilter() {
+ // utility class
+ }
+
+ public static Tablet filter(final Tablet tablet, final TagFilterMatcher matcher) {
+ return filter(tablet, matcher, null);
+ }
+
+ public static Tablet filter(
+ final Tablet tablet, final TagFilterMatcher matcher, final String databaseName) {
+ if (Objects.isNull(tablet)) {
+ return null;
+ }
+
+ final TagFilterMatcher effectiveMatcher =
+ Objects.nonNull(matcher) ? matcher : TagFilterMatcher.matchAll();
+ effectiveMatcher.throwIfFailure();
+ if (effectiveMatcher.isMatchAll()) {
+ return tablet;
+ }
+ if (effectiveMatcher.isMatchNone() || tablet.getRowSize() <= 0) {
+ return null;
+ }
+
+ final List<IMeasurementSchema> schemas = tablet.getSchemas();
+ // Upstream event conversion supplies a structurally valid Tablet. Only check the metadata
+ // needed to bind the predicate here; do not rescan every value array and null bitmap.
+ final List<ColumnCategory> categories = getColumnCategories(tablet, schemas.size());
+ validateBinding(effectiveMatcher, databaseName, tablet.getTableName());
+ final Map<String, Integer> resolvedFields =
+ resolveReferencedTagColumns(effectiveMatcher.getReferencedFields(), schemas, categories);
+
+ final int[] selectedRows = new int[tablet.getRowSize()];
+ int selectedRowCount = 0;
+ try {
+ for (int rowIndex = 0; rowIndex < tablet.getRowSize(); rowIndex++) {
+ final int currentRowIndex = rowIndex;
+ if (effectiveMatcher.matches(
+ identifier ->
+ getTagValue(tablet, currentRowIndex, resolvedFields.get(toFieldKey(identifier))))) {
+ selectedRows[selectedRowCount++] = rowIndex;
+ }
+ }
+ } catch (final RuntimeException e) {
+ if (e instanceof TagFilterEvaluationException) {
+ throw e;
+ }
+ throw new TagFilterEvaluationException(
+ DataNodeMiscMessages.EXCEPTION_FAILED_TO_EVALUATE_A_TABLET_ROW_7E4E94CE, e);
+ }
+
+ if (selectedRowCount == 0) {
+ return null;
+ }
+ if (selectedRowCount == tablet.getRowSize()) {
+ return tablet;
+ }
+ if (selectedRows[selectedRowCount - 1] == selectedRowCount - 1) {
+ // A prefix needs no row compaction. Clone the outer arrays because the TsFile sorter may
+ // replace their entries; the column arrays and bitmaps themselves can be shared.
+ return new Tablet(
+ tablet.getTableName(),
+ schemas,
+ categories,
+ tablet.getTimestamps(),
+ tablet.getValues().clone(),
+ tablet.getBitMaps() == null ? null : tablet.getBitMaps().clone(),
+ selectedRowCount);
+ }
+ return copySelectedRows(tablet, schemas, categories, selectedRows, selectedRowCount);
+ }
+
+ private static void validateBinding(
+ final TagFilterMatcher matcher, final String databaseName, final String tableName) {
+ if (!matcher.isBindingEnforced()) {
+ return;
+ }
+ final TagFilterMatcher.TableBinding binding =
+ matcher.getTableBindings().get(TagFilterMatcher.TableKey.of(databaseName, tableName));
+ if (binding == null) {
+ throw TagFilterEvaluationException.schemaBinding(
+ DataNodeMiscMessages.EXCEPTION_TABLE_BINDING_IS_NOT_AVAILABLE_882C5F2F);
+ }
+ if (binding.isFailed()) {
+ throw TagFilterEvaluationException.schemaBinding(binding.getFailureReason());
+ }
+ for (final Identifier identifier : matcher.getReferencedFields()) {
+ final String normalizedName = identifier.getValue().toLowerCase(Locale.ROOT);
+ if (!binding.getTagNames().contains(normalizedName)) {
+ throw TagFilterEvaluationException.schemaBinding(
+ DataNodeMiscMessages.EXCEPTION_REFERENCED_COLUMN_IS_NOT_A_TAG_COLUMN_33FA34BC);
+ }
+ }
+ }
+
+ private static Map<String, Integer> resolveReferencedTagColumns(
+ final List<Identifier> referencedFields,
+ final List<IMeasurementSchema> schemas,
+ final List<ColumnCategory> categories) {
+ final Map<String, Integer> exactColumns = new HashMap<>();
+ final Map<String, Integer> normalizedColumns = new HashMap<>();
+ for (int i = 0; i < schemas.size(); i++) {
+ final IMeasurementSchema schema = schemas.get(i);
+ exactColumns.put(schema.getMeasurementName(), i);
+ normalizedColumns.put(schema.getMeasurementName().toLowerCase(Locale.ROOT), i);
+ }
+
+ final Map<String, Integer> result = new HashMap<>();
+ for (final Identifier identifier : referencedFields) {
+ final Integer columnIndex =
+ identifier.isDelimited()
+ ? exactColumns.get(identifier.getValue())
+ : normalizedColumns.get(identifier.getValue().toLowerCase(Locale.ROOT));
+ if (Objects.isNull(columnIndex)) {
+ throw TagFilterEvaluationException.schemaBinding(
+ DataNodeMiscMessages.EXCEPTION_REFERENCED_TAG_COLUMN_IS_MISSING_55E07377);
+ }
+ if (categories.get(columnIndex) != ColumnCategory.TAG) {
+ throw TagFilterEvaluationException.schemaBinding(
+ DataNodeMiscMessages.EXCEPTION_REFERENCED_COLUMN_IS_NOT_A_TAG_COLUMN_33FA34BC);
+ }
+ result.put(toFieldKey(identifier), columnIndex);
+ }
+ return result;
+ }
+
+ private static String getTagValue(
+ final Tablet tablet, final int rowIndex, final Integer columnIndex) {
+ if (Objects.isNull(columnIndex) || tablet.isNull(rowIndex, columnIndex)) {
+ return null;
+ }
+ final Object value = tablet.getValue(rowIndex, columnIndex);
+ if (value instanceof Binary) {
+ return ((Binary) value).getStringValue(TSFileConfig.STRING_CHARSET);
+ }
+ return Objects.nonNull(value) ? String.valueOf(value) : null;
+ }
+
+ private static Tablet copySelectedRows(
+ final Tablet tablet,
+ final List<IMeasurementSchema> schemas,
+ final List<ColumnCategory> categories,
+ final int[] selectedRows,
+ final int selectedRowCount) {
+ try {
+ final long[] timestamps = new long[selectedRowCount];
+ for (int targetRow = 0; targetRow < selectedRowCount; targetRow++) {
+ timestamps[targetRow] = tablet.getTimestamp(selectedRows[targetRow]);
+ }
+ final Object[] sourceValues = tablet.getValues();
+ final Object[] values = new Object[schemas.size()];
+ final BitMap[] sourceBitMaps = tablet.getBitMaps();
+ final BitMap[] bitMaps = sourceBitMaps == null ? null : new BitMap[schemas.size()];
+ for (int columnIndex = 0; columnIndex < schemas.size(); columnIndex++) {
+ final Object sourceColumn = sourceValues[columnIndex];
+ if (sourceColumn != null) {
+ values[columnIndex] =
+ Array.newInstance(sourceColumn.getClass().getComponentType(), selectedRowCount);
+ for (int targetRow = 0; targetRow < selectedRowCount; targetRow++) {
+ System.arraycopy(
+ sourceColumn, selectedRows[targetRow], values[columnIndex], targetRow, 1);
+ }
+ }
+ if (sourceBitMaps != null && sourceBitMaps[columnIndex] != null) {
+ bitMaps[columnIndex] = new BitMap(selectedRowCount);
+ for (int targetRow = 0; targetRow < selectedRowCount; targetRow++) {
+ if (sourceBitMaps[columnIndex].isMarked(selectedRows[targetRow])) {
+ bitMaps[columnIndex].mark(targetRow);
+ }
+ }
+ }
+ }
+ // Filtering changes rows only; schema descriptors and column categories stay valid.
+ return new Tablet(
+ tablet.getTableName(),
+ schemas,
+ categories,
+ timestamps,
+ values,
+ bitMaps,
+ selectedRowCount);
+ } catch (final RuntimeException e) {
+ if (e instanceof TagFilterEvaluationException) {
+ throw e;
+ }
+ throw new TagFilterEvaluationException(
+ DataNodeMiscMessages.EXCEPTION_FAILED_TO_COMPACT_FILTERED_TABLET_5D4AD8AA, e);
+ }
+ }
+
+ private static List<ColumnCategory> getColumnCategories(
+ final Tablet tablet, final int columnCount) {
+ final List<ColumnCategory> categories = tablet.getColumnTypes();
+ if (Objects.isNull(categories) || categories.size() < columnCount) {
+ throw TagFilterEvaluationException.schemaBinding(
+ DataNodeMiscMessages.EXCEPTION_TABLET_COLUMN_CATEGORIES_ARE_MISSING_2C660532);
+ }
+ return categories;
+ }
+
+ private static String toFieldKey(final Identifier identifier) {
+ return (identifier.isDelimited() ? DELIMITED_FIELD_PREFIX : UNDELIMITED_FIELD_PREFIX)
+ + (identifier.isDelimited()
+ ? identifier.getValue()
+ : identifier.getValue().toLowerCase(Locale.ROOT));
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterEvaluationException.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterEvaluationException.java
new file mode 100644
index 0000000..4204273
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterEvaluationException.java
@@ -0,0 +1,45 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.subscription.tagfilter;
+
+import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
+
+/** Raised when a TAG filter cannot be evaluated safely for an input event. */
+public class TagFilterEvaluationException extends RuntimeException {
+
+ public TagFilterEvaluationException(final String reason) {
+ super(
+ String.format(
+ DataNodeMiscMessages.EXCEPTION_TAG_FILTER_EVALUATION_FAILED_ARG_1B239B2F, reason));
+ }
+
+ public TagFilterEvaluationException(final String reason, final Throwable cause) {
+ super(
+ String.format(
+ DataNodeMiscMessages.EXCEPTION_TAG_FILTER_EVALUATION_FAILED_ARG_1B239B2F, reason),
+ cause);
+ }
+
+ public static TagFilterEvaluationException schemaBinding(final String reason) {
+ return new TagFilterEvaluationException(
+ String.format(
+ DataNodeMiscMessages.EXCEPTION_TAG_FILTER_SCHEMA_BINDING_FAILED_ARG_7A5D2D47, reason));
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterEvaluator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterEvaluator.java
new file mode 100644
index 0000000..39dd152
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterEvaluator.java
@@ -0,0 +1,228 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.subscription.tagfilter;
+
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.BooleanLiteral;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.CommonQueryAstVisitor;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.ComparisonExpression;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.FunctionCall;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Identifier;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.InListExpression;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.InPredicate;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.IsNullPredicate;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.LikePredicate;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.LogicalExpression;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Node;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.NotExpression;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.StringLiteral;
+import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/** Evaluates a validated tag-filter with SQL three-valued NULL semantics. */
+public class TagFilterEvaluator
+ implements CommonQueryAstVisitor<
+ TagFilterEvaluator.TruthValue, TagFilterEvaluator.ValueProvider> {
+
+ enum TruthValue {
+ TRUE,
+ FALSE,
+ UNKNOWN
+ }
+
+ private final Map<Expression, Pattern> compiledPatterns;
+
+ private TagFilterEvaluator() {
+ this(Collections.emptyMap());
+ }
+
+ private TagFilterEvaluator(final Map<Expression, Pattern> compiledPatterns) {
+ this.compiledPatterns = compiledPatterns;
+ }
+
+ @FunctionalInterface
+ public interface ValueProvider {
+
+ String getValue(Identifier identifier);
+ }
+
+ public static boolean evaluate(final Expression expression, final ValueProvider valueProvider) {
+ final TagFilterValidator.ValidationResult validation =
+ TagFilterValidator.validateAndCompile(expression);
+ return evaluate(expression, valueProvider, validation.getCompiledPatterns());
+ }
+
+ public static boolean evaluate(
+ final Expression expression,
+ final ValueProvider valueProvider,
+ final Map<Expression, Pattern> compiledPatterns) {
+ return new TagFilterEvaluator(compiledPatterns).process(expression, valueProvider)
+ == TruthValue.TRUE;
+ }
+
+ @Override
+ public TruthValue visitNode(final Node node, final ValueProvider context) {
+ throw new IllegalArgumentException(
+ String.format(
+ DataNodeMiscMessages.UNSUPPORTED_EXPRESSION_FMT, node.getClass().getSimpleName()));
+ }
+
+ @Override
+ public TruthValue visitBooleanLiteral(final BooleanLiteral node, final ValueProvider context) {
+ return node.getValue() ? TruthValue.TRUE : TruthValue.FALSE;
+ }
+
+ @Override
+ public TruthValue visitLogicalExpression(
+ final LogicalExpression node, final ValueProvider context) {
+ // UNKNOWN is not decisive: UNKNOWN AND FALSE is FALSE, and UNKNOWN OR TRUE is TRUE.
+ // This also matters under NOT, where returning UNKNOWN early could discard a matching row.
+ boolean hasUnknown = false;
+ if (node.getOperator() == LogicalExpression.Operator.AND) {
+ for (final Expression term : node.getTerms()) {
+ final TruthValue result = process(term, context);
+ if (result == TruthValue.FALSE) {
+ return TruthValue.FALSE;
+ }
+ hasUnknown |= result == TruthValue.UNKNOWN;
+ }
+ return hasUnknown ? TruthValue.UNKNOWN : TruthValue.TRUE;
+ }
+
+ for (final Expression term : node.getTerms()) {
+ final TruthValue result = process(term, context);
+ if (result == TruthValue.TRUE) {
+ return TruthValue.TRUE;
+ }
+ hasUnknown |= result == TruthValue.UNKNOWN;
+ }
+ return hasUnknown ? TruthValue.UNKNOWN : TruthValue.FALSE;
+ }
+
+ @Override
+ public TruthValue visitNotExpression(final NotExpression node, final ValueProvider context) {
+ final TruthValue result = process(node.getValue(), context);
+ if (result == TruthValue.UNKNOWN) {
+ return TruthValue.UNKNOWN;
+ }
+ return result == TruthValue.TRUE ? TruthValue.FALSE : TruthValue.TRUE;
+ }
+
+ @Override
+ public TruthValue visitComparisonExpression(
+ final ComparisonExpression node, final ValueProvider context) {
+ final String left = context.getValue((Identifier) node.getLeft());
+ if (Objects.isNull(left)) {
+ return TruthValue.UNKNOWN;
+ }
+ final boolean equals = left.equals(((StringLiteral) node.getRight()).getValue());
+ return booleanValue(
+ node.getOperator() == ComparisonExpression.Operator.EQUAL ? equals : !equals);
+ }
+
+ @Override
+ public TruthValue visitInPredicate(final InPredicate node, final ValueProvider context) {
+ final String left = context.getValue((Identifier) node.getValue());
+ if (Objects.isNull(left)) {
+ return TruthValue.UNKNOWN;
+ }
+ for (final Expression expression : ((InListExpression) node.getValueList()).getValues()) {
+ if (left.equals(((StringLiteral) expression).getValue())) {
+ return TruthValue.TRUE;
+ }
+ }
+ return TruthValue.FALSE;
+ }
+
+ @Override
+ public TruthValue visitLikePredicate(final LikePredicate node, final ValueProvider context) {
+ final String left = context.getValue((Identifier) node.getValue());
+ if (Objects.isNull(left)) {
+ return TruthValue.UNKNOWN;
+ }
+ final Pattern pattern = compiledPatterns.get(node);
+ if (pattern == null) {
+ throw new IllegalArgumentException(
+ DataNodeMiscMessages.EXCEPTION_UNCOMPILED_LIKE_PREDICATE_01AEE439);
+ }
+ return booleanValue(pattern.matcher(left).matches());
+ }
+
+ @Override
+ public TruthValue visitFunctionCall(final FunctionCall node, final ValueProvider context) {
+ final String left = context.getValue((Identifier) node.getArguments().get(0));
+ if (Objects.isNull(left)) {
+ return TruthValue.UNKNOWN;
+ }
+ final Pattern pattern = compiledPatterns.get(node);
+ if (pattern == null) {
+ throw new IllegalArgumentException(
+ DataNodeMiscMessages.EXCEPTION_UNCOMPILED_REGEXP_PREDICATE_2B0DD646);
+ }
+ return booleanValue(pattern.matcher(left).matches());
+ }
+
+ @Override
+ public TruthValue visitIsNullPredicate(final IsNullPredicate node, final ValueProvider context) {
+ return booleanValue(Objects.isNull(context.getValue((Identifier) node.getValue())));
+ }
+
+ static Pattern compileLikePattern(final String pattern, final String escape) {
+ final Character escapeChar;
+ if (Objects.isNull(escape)) {
+ escapeChar = null;
+ } else if (escape.length() == 1) {
+ escapeChar = escape.charAt(0);
+ } else {
+ throw new IllegalArgumentException(DataNodeMiscMessages.LIKE_ESCAPE_MUST_BE_SINGLE_CHARACTER);
+ }
+
+ final StringBuilder regex = new StringBuilder();
+ boolean escaping = false;
+ for (int i = 0; i < pattern.length(); i++) {
+ final char ch = pattern.charAt(i);
+ if (Objects.nonNull(escapeChar) && ch == escapeChar && !escaping) {
+ escaping = true;
+ continue;
+ }
+ if (!escaping && ch == '%') {
+ regex.append(".*");
+ } else if (!escaping && ch == '_') {
+ regex.append('.');
+ } else {
+ regex.append(Pattern.quote(String.valueOf(ch)));
+ }
+ escaping = false;
+ }
+ if (escaping) {
+ throw new IllegalArgumentException(
+ DataNodeMiscMessages.LIKE_PATTERN_ENDS_WITH_ESCAPE_CHARACTER);
+ }
+ return Pattern.compile(regex.toString(), Pattern.DOTALL);
+ }
+
+ private static TruthValue booleanValue(final boolean value) {
+ return value ? TruthValue.TRUE : TruthValue.FALSE;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterMatcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterMatcher.java
new file mode 100644
index 0000000..7034a21
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterMatcher.java
@@ -0,0 +1,423 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.subscription.tagfilter;
+
+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.queryengine.plan.relational.sql.ast.Identifier;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema;
+import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
+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 java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+/** Immutable parsed tag-filter cached per topic. */
+public class TagFilterMatcher {
+
+ public enum RuntimeStatus {
+ MATCH_ALL,
+ MATCH_NONE,
+ ACTIVE,
+ ERROR
+ }
+
+ private static final TagFilterMatcher MATCH_NONE =
+ new TagFilterMatcher(
+ null, Collections.emptyList(), Collections.emptyMap(), State.MATCH_NONE, null, false);
+
+ private final Expression expression;
+ private final List<Identifier> referencedFields;
+ private final Map<Expression, Pattern> compiledPatterns;
+ private final State state;
+ private final Throwable failure;
+ private final Map<TableKey, TableBinding> tableBindings;
+ private final boolean bindingEnforced;
+
+ private enum State {
+ MATCH_ALL,
+ MATCH_NONE,
+ ACTIVE,
+ FAILURE
+ }
+
+ private TagFilterMatcher(
+ final Expression expression,
+ final List<Identifier> referencedFields,
+ final Map<Expression, Pattern> compiledPatterns,
+ final State state,
+ final Throwable failure,
+ final boolean bindingEnforced) {
+ this(
+ expression,
+ referencedFields,
+ compiledPatterns,
+ state,
+ failure,
+ Collections.emptyMap(),
+ bindingEnforced);
+ }
+
+ private TagFilterMatcher(
+ final Expression expression,
+ final List<Identifier> referencedFields,
+ final Map<Expression, Pattern> compiledPatterns,
+ final State state,
+ final Throwable failure,
+ final Map<TableKey, TableBinding> tableBindings,
+ final boolean bindingEnforced) {
+ this.expression = expression;
+ this.referencedFields = referencedFields;
+ this.compiledPatterns = compiledPatterns;
+ this.state = state;
+ this.failure = failure;
+ this.tableBindings = tableBindings;
+ this.bindingEnforced = bindingEnforced;
+ }
+
+ private static final TagFilterMatcher MATCH_ALL =
+ new TagFilterMatcher(
+ null, Collections.emptyList(), Collections.emptyMap(), State.MATCH_ALL, null, false);
+
+ public static TagFilterMatcher matchAll() {
+ return MATCH_ALL;
+ }
+
+ public static TagFilterMatcher matchNone() {
+ return MATCH_NONE;
+ }
+
+ public static TagFilterMatcher failure(final Throwable failure) {
+ return new TagFilterMatcher(
+ null,
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ State.FAILURE,
+ Objects.nonNull(failure) ? failure : new IllegalStateException(),
+ false);
+ }
+
+ public static TagFilterMatcher fromTopicConfig(final TopicConfig topicConfig)
+ throws SubscriptionException {
+ if (Objects.isNull(topicConfig)
+ || !topicConfig.isTableTopic()
+ || topicConfig.isTagFilterTrivial()) {
+ return matchAll();
+ }
+
+ final Expression expression;
+ final TagFilterValidator.ValidationResult validation;
+ try {
+ expression = new TagFilterParser().parse(topicConfig.getTagFilter());
+ validation = TagFilterValidator.validateAndCompile(expression);
+ } catch (final IllegalArgumentException e) {
+ throw new SubscriptionException(
+ String.format(
+ DataNodeMiscMessages.EXCEPTION_INVALID_TAG_FILTER_ARG_E4B1C1C6, e.getMessage()),
+ e);
+ }
+ return new TagFilterMatcher(
+ expression,
+ validation.getReferencedFields(),
+ validation.getCompiledPatterns(),
+ State.ACTIVE,
+ null,
+ false);
+ }
+
+ /**
+ * Parses a filter and validates every currently visible table selected by the topic pattern. An
+ * empty table map is deliberately allowed for a wildcard topic because the table may be created
+ * after the topic; a null map means that a required concrete schema is unavailable.
+ */
+ public static TagFilterMatcher fromTopicConfig(
+ final TopicConfig topicConfig, final Map<String, Map<String, TsTable>> tables)
+ throws SubscriptionException {
+ final TagFilterMatcher matcher = fromTopicConfig(topicConfig);
+ if (matcher.isMatchAll()) {
+ return matcher;
+ }
+ if (tables == null) {
+ throw new SubscriptionException(
+ DataNodeMiscMessages.EXCEPTION_TABLE_SCHEMA_IS_NOT_AVAILABLE_FOR_TAG_FILTER_993AB728);
+ }
+
+ final TablePattern tablePattern =
+ new TablePattern(
+ true,
+ topicConfig.getStringOrDefault(
+ TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE),
+ topicConfig.getStringOrDefault(
+ TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE));
+ final Map<TableKey, TableBinding> bindings = new HashMap<>();
+ for (final Map.Entry<String, Map<String, TsTable>> databaseEntry : tables.entrySet()) {
+ if (!tablePattern.matchesDatabase(databaseEntry.getKey())) {
+ continue;
+ }
+ for (final TsTable table : databaseEntry.getValue().values()) {
+ if (table == null || !tablePattern.matchesTable(table.getTableName())) {
+ continue;
+ }
+ final TableBinding binding = bindTable(matcher.referencedFields, table);
+ if (binding.isFailed()) {
+ throw new SubscriptionException(binding.getFailureReason());
+ }
+ bindings.put(TableKey.of(databaseEntry.getKey(), table.getTableName()), binding);
+ }
+ }
+ final boolean bindingEnforced =
+ isLiteralTopicPattern(
+ topicConfig.getStringOrDefault(
+ TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE))
+ && isLiteralTopicPattern(
+ topicConfig.getStringOrDefault(
+ TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE));
+ return matcher.withBindings(bindings, bindingEnforced);
+ }
+
+ 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 TableBinding bindTable(
+ final List<Identifier> referencedFields, final TsTable table) {
+ final Set<String> tagNames = new HashSet<>();
+ final Set<String> allNames = new HashSet<>();
+ for (final TsTableColumnSchema schema : table.getColumnList()) {
+ if (schema == null || schema.getColumnName() == null) {
+ continue;
+ }
+ allNames.add(schema.getColumnName().toLowerCase(Locale.ROOT));
+ if (schema.getColumnCategory() == TsTableColumnCategory.TAG) {
+ tagNames.add(schema.getColumnName().toLowerCase(Locale.ROOT));
+ }
+ }
+ for (final Identifier identifier : referencedFields) {
+ final String identifierName = identifier.getValue();
+ TsTableColumnSchema matched = null;
+ for (final TsTableColumnSchema schema : table.getColumnList()) {
+ if (schema == null || schema.getColumnName() == null) {
+ continue;
+ }
+ if (identifier.isDelimited()
+ ? schema.getColumnName().equals(identifierName)
+ : schema.getColumnName().equalsIgnoreCase(identifierName)) {
+ matched = schema;
+ break;
+ }
+ }
+ if (matched == null) {
+ return new TableBinding(
+ true,
+ String.format(
+ DataNodeMiscMessages.EXCEPTION_REFERENCED_TAG_COLUMN_IS_MISSING_ARG_F5300BEA,
+ identifierName),
+ tagNames,
+ allNames);
+ }
+ if (matched.getColumnCategory() != TsTableColumnCategory.TAG) {
+ return new TableBinding(
+ true,
+ String.format(
+ DataNodeMiscMessages.EXCEPTION_REFERENCED_COLUMN_IS_NOT_A_TAG_COLUMN_ARG_34D60881,
+ identifierName),
+ tagNames,
+ allNames);
+ }
+ }
+ return new TableBinding(false, null, tagNames, allNames);
+ }
+
+ TagFilterMatcher withBindings(final Map<TableKey, TableBinding> bindings) {
+ return withBindings(bindings, true);
+ }
+
+ TagFilterMatcher withBindings(
+ final Map<TableKey, TableBinding> bindings, final boolean bindingEnforced) {
+ return new TagFilterMatcher(
+ expression,
+ referencedFields,
+ compiledPatterns,
+ state,
+ failure,
+ Objects.nonNull(bindings)
+ ? Collections.unmodifiableMap(new HashMap<>(bindings))
+ : Collections.emptyMap(),
+ bindingEnforced);
+ }
+
+ public boolean isMatchAll() {
+ return state == State.MATCH_ALL;
+ }
+
+ public boolean isMatchNone() {
+ return state == State.MATCH_NONE;
+ }
+
+ public boolean isFailure() {
+ return state == State.FAILURE;
+ }
+
+ public RuntimeStatus getRuntimeStatus() {
+ switch (state) {
+ case MATCH_ALL:
+ return RuntimeStatus.MATCH_ALL;
+ case MATCH_NONE:
+ return RuntimeStatus.MATCH_NONE;
+ case ACTIVE:
+ return RuntimeStatus.ACTIVE;
+ case FAILURE:
+ default:
+ return RuntimeStatus.ERROR;
+ }
+ }
+
+ public String getFailureMessage() {
+ if (!isFailure() || Objects.isNull(failure)) {
+ return null;
+ }
+ Throwable current = failure;
+ while (Objects.isNull(current.getMessage())
+ && Objects.nonNull(current.getCause())
+ && current != current.getCause()) {
+ current = current.getCause();
+ }
+ return Objects.nonNull(current.getMessage())
+ ? current.getMessage()
+ : current.getClass().getSimpleName();
+ }
+
+ public void throwIfFailure() {
+ if (isFailure()) {
+ throw new TagFilterEvaluationException(getFailureMessage(), failure);
+ }
+ }
+
+ List<Identifier> getReferencedFields() {
+ return referencedFields;
+ }
+
+ Map<TableKey, TableBinding> getTableBindings() {
+ return tableBindings;
+ }
+
+ boolean isBindingEnforced() {
+ return bindingEnforced;
+ }
+
+ boolean matches(final TagFilterEvaluator.ValueProvider valueProvider) {
+ throwIfFailure();
+ return !isMatchNone()
+ && (Objects.isNull(expression)
+ || TagFilterEvaluator.evaluate(expression, valueProvider, compiledPatterns));
+ }
+
+ static final class TableKey {
+
+ private final String database;
+ private final String table;
+
+ private TableKey(final String database, final String table) {
+ this.database = normalize(database);
+ this.table = normalize(table);
+ }
+
+ static TableKey of(final String database, final String table) {
+ return new TableKey(database, table);
+ }
+
+ String getDatabase() {
+ return database;
+ }
+
+ String getTable() {
+ return table;
+ }
+
+ @Override
+ public boolean equals(final Object object) {
+ if (this == object) {
+ return true;
+ }
+ if (!(object instanceof TableKey)) {
+ return false;
+ }
+ final TableKey that = (TableKey) object;
+ return Objects.equals(database, that.database) && Objects.equals(table, that.table);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(database, table);
+ }
+
+ private static String normalize(final String value) {
+ return Objects.nonNull(value) ? value.trim().toLowerCase(Locale.ROOT) : "";
+ }
+ }
+
+ static final class TableBinding {
+
+ private final boolean failed;
+ private final String failureReason;
+ private final Set<String> tagNames;
+ private final Set<String> allNames;
+
+ TableBinding(
+ final boolean failed,
+ final String failureReason,
+ final Set<String> tagNames,
+ final Set<String> allNames) {
+ this.failed = failed;
+ this.failureReason = failureReason;
+ this.tagNames = tagNames;
+ this.allNames = allNames;
+ }
+
+ boolean isFailed() {
+ return failed;
+ }
+
+ String getFailureReason() {
+ return failureReason;
+ }
+
+ Set<String> getTagNames() {
+ return tagNames;
+ }
+
+ Set<String> getAllNames() {
+ return allNames;
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterParser.java
new file mode 100644
index 0000000..1a08394
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterParser.java
@@ -0,0 +1,94 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.subscription.tagfilter;
+
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.parser.ParsingException;
+import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
+import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterParser;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+
+import java.nio.charset.StandardCharsets;
+
+/** Parses the restricted subscription filter expression syntax for table-model TAG values. */
+public class TagFilterParser {
+
+ private static final int MAX_UTF8_BYTES = 4096;
+
+ private final ColumnFilterParser expressionParser = new ColumnFilterParser();
+
+ public Expression parse(final String rawTagFilter) throws SubscriptionException {
+ try {
+ if (rawTagFilter == null || rawTagFilter.trim().isEmpty()) {
+ throw new IllegalArgumentException(
+ DataNodeMiscMessages.EXCEPTION_TAG_FILTER_SHOULD_NOT_BE_EMPTY_507CA5B0);
+ }
+ validateText(rawTagFilter);
+ return expressionParser.parse(rawTagFilter);
+ } catch (final ParsingException | IllegalArgumentException e) {
+ throw new SubscriptionException(
+ String.format(
+ DataNodeMiscMessages.EXCEPTION_INVALID_TAG_FILTER_ARG_E4B1C1C6, e.getMessage()),
+ e);
+ }
+ }
+
+ public Expression parseAndValidate(final String rawTagFilter) throws SubscriptionException {
+ final Expression expression = parse(rawTagFilter);
+ try {
+ TagFilterValidator.validate(expression);
+ } catch (final IllegalArgumentException e) {
+ throw new SubscriptionException(
+ String.format(
+ DataNodeMiscMessages.EXCEPTION_INVALID_TAG_FILTER_ARG_E4B1C1C6, e.getMessage()),
+ e);
+ }
+ return expression;
+ }
+
+ private static void validateText(final String text) {
+ if (text.getBytes(StandardCharsets.UTF_8).length > MAX_UTF8_BYTES) {
+ throw new IllegalArgumentException(
+ String.format(
+ DataNodeMiscMessages
+ .EXCEPTION_TAG_FILTER_EXCEEDS_MAXIMUM_UTF_8_LENGTH_OF_ARG_BYTES_9C8400F2,
+ MAX_UTF8_BYTES));
+ }
+ for (int i = 0; i < text.length(); i++) {
+ final char current = text.charAt(i);
+ if (Character.isISOControl(current) && !Character.isWhitespace(current)) {
+ throw new IllegalArgumentException(
+ String.format(
+ DataNodeMiscMessages.UNEXPECTED_CHARACTER_FMT, Integer.toHexString(current)));
+ }
+ if (Character.isHighSurrogate(current)) {
+ if (i + 1 >= text.length() || !Character.isLowSurrogate(text.charAt(++i))) {
+ throw new IllegalArgumentException(
+ String.format(
+ DataNodeMiscMessages.UNEXPECTED_CHARACTER_FMT, Integer.toHexString(current)));
+ }
+ } else if (Character.isLowSurrogate(current)) {
+ throw new IllegalArgumentException(
+ String.format(
+ DataNodeMiscMessages.UNEXPECTED_CHARACTER_FMT, Integer.toHexString(current)));
+ }
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterValidator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterValidator.java
new file mode 100644
index 0000000..2b53205
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterValidator.java
@@ -0,0 +1,252 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.subscription.tagfilter;
+
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.BooleanLiteral;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.CommonQueryAstVisitor;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.ComparisonExpression;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.FunctionCall;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Identifier;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.InListExpression;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.InPredicate;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.IsNullPredicate;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.LikePredicate;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.LogicalExpression;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Node;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.NotExpression;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.StringLiteral;
+import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+/** Validates the expression shape and collects the TAG columns referenced by a tag-filter. */
+public class TagFilterValidator implements CommonQueryAstVisitor<Void, Void> {
+
+ static final int MAX_AST_DEPTH = 32;
+ static final int MAX_AST_NODES = 256;
+ static final int MAX_IN_VALUES = 256;
+ static final int MAX_REGEXP_LENGTH = 1024;
+
+ private static final String REGEXP_LIKE = "regexp_like";
+
+ private final List<Identifier> referencedFields = new ArrayList<>();
+ private final Map<Expression, Pattern> compiledPatterns = new IdentityHashMap<>();
+ private int depth;
+ private int nodeCount;
+
+ public static List<Identifier> validate(final Expression expression) {
+ return validateAndCompile(expression).getReferencedFields();
+ }
+
+ public static ValidationResult validateAndCompile(final Expression expression) {
+ final TagFilterValidator validator = new TagFilterValidator();
+ validator.process(expression);
+ return new ValidationResult(
+ Collections.unmodifiableList(new ArrayList<>(validator.referencedFields)),
+ Collections.unmodifiableMap(new IdentityHashMap<>(validator.compiledPatterns)));
+ }
+
+ @Override
+ public Void process(final Node node, final Void context) {
+ if (node == null) {
+ throw invalid(DataNodeMiscMessages.UNSUPPORTED_EXPRESSION_FMT.replace("%s", "null"));
+ }
+ accountNode(node);
+ if (++depth > MAX_AST_DEPTH) {
+ throw invalid(
+ String.format(
+ DataNodeMiscMessages.EXCEPTION_TAG_FILTER_AST_DEPTH_EXCEEDS_MAXIMUM_OF_ARG_64C92731,
+ MAX_AST_DEPTH));
+ }
+ try {
+ return CommonQueryAstVisitor.super.process(node, context);
+ } finally {
+ depth--;
+ }
+ }
+
+ @Override
+ public Void visitNode(final Node node, final Void context) {
+ throw invalid(
+ String.format(
+ DataNodeMiscMessages.UNSUPPORTED_EXPRESSION_FMT, node.getClass().getSimpleName()));
+ }
+
+ @Override
+ public Void visitBooleanLiteral(final BooleanLiteral node, final Void context) {
+ return null;
+ }
+
+ @Override
+ public Void visitLogicalExpression(final LogicalExpression node, final Void context) {
+ node.getTerms().forEach(this::process);
+ return null;
+ }
+
+ @Override
+ public Void visitNotExpression(final NotExpression node, final Void context) {
+ process(node.getValue());
+ return null;
+ }
+
+ @Override
+ public Void visitComparisonExpression(final ComparisonExpression node, final Void context) {
+ if (node.getOperator() != ComparisonExpression.Operator.EQUAL
+ && node.getOperator() != ComparisonExpression.Operator.NOT_EQUAL) {
+ throw invalid(
+ DataNodeMiscMessages.EXCEPTION_ONLY_AND_COMPARISONS_ARE_SUPPORTED_IN_TAG_FILTER_19946958);
+ }
+ requireTagField(node.getLeft());
+ requireStringLiteral(
+ node.getRight(), DataNodeMiscMessages.COLUMN_FILTER_COMPARISON_RIGHT_OPERAND);
+ return null;
+ }
+
+ @Override
+ public Void visitInPredicate(final InPredicate node, final Void context) {
+ requireTagField(node.getValue());
+ if (!(node.getValueList() instanceof InListExpression)) {
+ throw invalid(DataNodeMiscMessages.IN_PREDICATE_MUST_USE_STRING_LITERAL_LIST);
+ }
+ final List<Expression> values = ((InListExpression) node.getValueList()).getValues();
+ if (values.isEmpty() || values.size() > MAX_IN_VALUES) {
+ throw invalid(
+ String.format(
+ DataNodeMiscMessages
+ .EXCEPTION_TAG_FILTER_IN_LIST_EXCEEDS_MAXIMUM_OF_ARG_VALUES_07E7FC6C,
+ MAX_IN_VALUES));
+ }
+ accountNode(node.getValueList());
+ for (final Expression expression : values) {
+ accountNode(expression);
+ requireStringLiteral(expression, DataNodeMiscMessages.COLUMN_FILTER_IN_ELEMENT);
+ }
+ return null;
+ }
+
+ @Override
+ public Void visitLikePredicate(final LikePredicate node, final Void context) {
+ requireTagField(node.getValue());
+ final StringLiteral pattern =
+ requireStringLiteral(node.getPattern(), DataNodeMiscMessages.COLUMN_FILTER_LIKE_PATTERN);
+ final String escape =
+ node.getEscape()
+ .map(
+ expression ->
+ requireStringLiteral(expression, DataNodeMiscMessages.COLUMN_FILTER_LIKE_ESCAPE)
+ .getValue())
+ .orElse(null);
+ compiledPatterns.put(node, TagFilterEvaluator.compileLikePattern(pattern.getValue(), escape));
+ return null;
+ }
+
+ @Override
+ public Void visitFunctionCall(final FunctionCall node, final Void context) {
+ if (!REGEXP_LIKE.equalsIgnoreCase(node.getName().toString())
+ || node.isDistinct()
+ || node.getProcessingMode().isPresent()
+ || node.getArguments().size() != 2) {
+ throw invalid(DataNodeMiscMessages.ONLY_REGEXP_SUPPORTED_AS_REGEXP_LIKE);
+ }
+
+ requireTagField(node.getArguments().get(0));
+ final String pattern =
+ requireStringLiteral(
+ node.getArguments().get(1), DataNodeMiscMessages.COLUMN_FILTER_REGEXP_PATTERN)
+ .getValue();
+ if (pattern.length() > MAX_REGEXP_LENGTH) {
+ throw invalid(
+ String.format(
+ DataNodeMiscMessages
+ .EXCEPTION_TAG_FILTER_REGEXP_PATTERN_EXCEEDS_MAXIMUM_LENGTH_OF_ARG_CHARACTERS_67590971,
+ MAX_REGEXP_LENGTH));
+ }
+ try {
+ compiledPatterns.put(node, Pattern.compile(pattern));
+ } catch (final PatternSyntaxException e) {
+ throw invalid(String.format(DataNodeMiscMessages.ILLEGAL_REGEXP_PATTERN_FMT, e.getMessage()));
+ }
+ return null;
+ }
+
+ @Override
+ public Void visitIsNullPredicate(final IsNullPredicate node, final Void context) {
+ requireTagField(node.getValue());
+ return null;
+ }
+
+ private Identifier requireTagField(final Expression expression) {
+ if (!(expression instanceof Identifier)) {
+ throw invalid(DataNodeMiscMessages.EXCEPTION_LEFT_OPERAND_MUST_BE_A_TAG_COLUMN_F9D4548B);
+ }
+ final Identifier identifier = (Identifier) expression;
+ referencedFields.add(identifier);
+ return identifier;
+ }
+
+ private static StringLiteral requireStringLiteral(
+ final Expression expression, final String description) {
+ if (!(expression instanceof StringLiteral)) {
+ throw invalid(String.format(DataNodeMiscMessages.MUST_BE_STRING_LITERAL_FMT, description));
+ }
+ return (StringLiteral) expression;
+ }
+
+ private static IllegalArgumentException invalid(final String message) {
+ return new IllegalArgumentException(message);
+ }
+
+ private void accountNode(final Node node) {
+ if (++nodeCount > MAX_AST_NODES) {
+ throw invalid(
+ String.format(
+ DataNodeMiscMessages
+ .EXCEPTION_TAG_FILTER_AST_NODE_COUNT_EXCEEDS_MAXIMUM_OF_ARG_40CCE694,
+ MAX_AST_NODES));
+ }
+ }
+
+ public static final class ValidationResult {
+
+ private final List<Identifier> referencedFields;
+ private final Map<Expression, Pattern> compiledPatterns;
+
+ private ValidationResult(
+ final List<Identifier> referencedFields, final Map<Expression, Pattern> compiledPatterns) {
+ this.referencedFields = referencedFields;
+ this.compiledPatterns = compiledPatterns;
+ }
+
+ public List<Identifier> getReferencedFields() {
+ return referencedFields;
+ }
+
+ public Map<Expression, Pattern> getCompiledPatterns() {
+ return compiledPatterns;
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatchTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatchTest.java
index 3f70f55..96b665e 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatchTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatchTest.java
@@ -32,6 +32,44 @@
public class PipeTabletEventTsFileBatchTest {
@Test
+ public void testTransformerReceivesModelDatabaseAndAlignment() throws Exception {
+ for (final boolean tableModel : new boolean[] {false, true}) {
+ for (final boolean aligned : new boolean[] {false, true}) {
+ final PipeTabletEventTsFileBatch batch =
+ new PipeTabletEventTsFileBatch(
+ 100_000,
+ 1024 * 1024,
+ (databaseName, tablet, isTableModel, isAligned) -> {
+ Assert.assertEquals(tableModel, isTableModel);
+ Assert.assertEquals(tableModel ? "db" : null, databaseName);
+ Assert.assertEquals(tableModel || aligned, isAligned);
+ return null;
+ });
+ try {
+ final PipeRawTabletInsertionEvent event =
+ new PipeRawTabletInsertionEvent(
+ tableModel,
+ "root.db",
+ "db",
+ "root.db",
+ createTablet(),
+ aligned,
+ "pipe",
+ 1L,
+ null,
+ null,
+ false);
+ Assert.assertEquals(tableModel ? "db" : null, event.getTableModelDatabaseName());
+ batch.onEvent(event);
+ Assert.assertTrue(event.isReleased());
+ } finally {
+ batch.close();
+ }
+ }
+ }
+ }
+
+ @Test
public void testFullyPrunedTableModelTabletIsReleasedAndNotRetained() throws Exception {
final PipeTabletEventTsFileBatch batch =
new PipeTabletEventTsFileBatch(100_000, 1024 * 1024, (databaseName, tablet) -> null);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowTopicsTaskTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowTopicsTaskTest.java
new file mode 100644
index 0000000..0d7ff2e
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowTopicsTaskTest.java
@@ -0,0 +1,98 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription;
+
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
+import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
+import org.apache.iotdb.confignode.rpc.thrift.TShowTopicInfo;
+import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterMatcher;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class ShowTopicsTaskTest {
+
+ @Test
+ public void testBuildTSBlockWritesTagFilterDiagnostics() throws Exception {
+ final Map<String, TagFilterMatcher> matchers = new HashMap<>();
+ matchers.put("all", TagFilterMatcher.matchAll());
+ matchers.put("none", TagFilterMatcher.matchNone());
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TABLE_VALUE);
+ attributes.put(TopicConstant.TAG_FILTER_KEY, "region = \"east\"");
+ matchers.put("active", TagFilterMatcher.fromTopicConfig(new TopicConfig(attributes)));
+ matchers.put("error", TagFilterMatcher.failure(new IllegalStateException("binding failed")));
+ final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+
+ ShowTopicsTask.buildTSBlock(
+ Arrays.asList(topic("all"), topic("none"), topic("active"), topic("error")),
+ true,
+ matchers::get,
+ future);
+
+ final ConfigTaskResult result = future.get();
+ final TsBlock resultSet = result.getResultSet();
+ assertEquals(TSStatusCode.SUCCESS_STATUS, result.getStatusCode());
+ assertEquals(
+ ColumnHeaderConstant.TAG_FILTER_STATUS,
+ result.getResultSetHeader().getRespColumns().get(2));
+ assertEquals(
+ ColumnHeaderConstant.TAG_FILTER_MESSAGE,
+ result.getResultSetHeader().getRespColumns().get(3));
+ assertEquals("MATCH_ALL", resultSet.getColumn(2).getBinary(0).toString());
+ assertEquals("MATCH_NONE", resultSet.getColumn(2).getBinary(1).toString());
+ assertEquals("ACTIVE", resultSet.getColumn(2).getBinary(2).toString());
+ assertEquals("ERROR", resultSet.getColumn(2).getBinary(3).toString());
+ assertTrue(resultSet.getColumn(3).isNull(0));
+ assertTrue(resultSet.getColumn(3).isNull(1));
+ assertTrue(resultSet.getColumn(3).isNull(2));
+ assertEquals("binding failed", resultSet.getColumn(3).getBinary(3).toString());
+ }
+
+ @Test
+ public void testBuildTreeModelTSBlockLeavesTagFilterDiagnosticsNull() throws Exception {
+ final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+
+ ShowTopicsTask.buildTSBlock(
+ Arrays.asList(topic("tree")),
+ false,
+ ignored -> TagFilterMatcher.failure(new IllegalStateException()),
+ future);
+
+ final TsBlock resultSet = future.get().getResultSet();
+ assertEquals(2, resultSet.getValueColumnCount());
+ }
+
+ private static TShowTopicInfo topic(final String name) {
+ return new TShowTopicInfo(name, 1L).setTopicAttributes("{}");
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/TopicStatementTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/TopicStatementTest.java
index d457bdb..5c8686c 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/TopicStatementTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/TopicStatementTest.java
@@ -20,18 +20,76 @@
package org.apache.iotdb.db.queryengine.plan.relational.sql.parser;
import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement;
+import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
+import org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription.ShowCreateTopicTask;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterTopic;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateTopic;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.util.DataNodeSqlFormatter;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterEvaluator;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterParser;
import org.junit.Assert;
import org.junit.Test;
import java.time.ZoneId;
+import java.util.Map;
public class TopicStatementTest {
private final SqlParser sqlParser = new SqlParser();
@Test
+ public void testTagFilterQuotesInCreateAndAlterTopic() throws Exception {
+ final String attributes =
+ "('tag-filter'='region IN (\"O''Reilly\", \"a''''b\", \"a\"\"\"\"b\", \"O''Reilly \"\"A\"\"\"\"B\"\"\")')";
+ final String expectedFilter =
+ "region IN (\"O'Reilly\", \"a''b\", \"a\"\"\"\"b\", \"O'Reilly \"\"A\"\"\"\"B\"\"\")";
+ for (final String command :
+ new String[] {
+ "CREATE TOPIC topic1",
+ "ALTER TOPIC topic1",
+ "CREATE TOPIC IF NOT EXISTS \"topic-1\"",
+ "ALTER TOPIC \"topic-1\""
+ }) {
+ final Statement statement =
+ sqlParser.createStatement(command + " WITH " + attributes, ZoneId.systemDefault(), null);
+ final Statement formattedStatement =
+ sqlParser.createStatement(
+ DataNodeSqlFormatter.formatDataNodeSql(statement), ZoneId.systemDefault(), null);
+ Assert.assertEquals(statement, formattedStatement);
+ final String filter =
+ statement instanceof CreateTopic
+ ? ((CreateTopic) statement).getTopicAttributes().get("tag-filter")
+ : ((AlterTopic) statement).getTopicAttributes().get("tag-filter");
+ Assert.assertEquals(expectedFilter, filter);
+
+ final TagFilterParser parser = new TagFilterParser();
+ for (final String value :
+ new String[] {"O'Reilly", "a''b", "a\"\"b", "O'Reilly \"A\"\"B\""}) {
+ Assert.assertTrue(
+ command + ": " + value,
+ TagFilterEvaluator.evaluate(parser.parseAndValidate(filter), identifier -> value));
+ }
+ Assert.assertFalse(
+ TagFilterEvaluator.evaluate(parser.parseAndValidate(filter), identifier -> "a\"b"));
+ Assert.assertFalse(
+ TagFilterEvaluator.evaluate(parser.parseAndValidate(filter), identifier -> "O''Reilly"));
+ }
+ }
+
+ @Test
+ public void testShowCreateTopicPreservesTagFilterQuotes() {
+ final String filter = "region = \"O'Reilly \"\"A\"\"\"\"B\"\"\"";
+ final TopicMeta topicMeta = new TopicMeta("topic-1", 0L, Map.of("tag-filter", filter));
+ final CreateTopic recreated =
+ (CreateTopic)
+ sqlParser.createStatement(
+ ShowCreateTopicTask.getShowCreateTopicSQL(topicMeta), ZoneId.systemDefault(), null);
+ Assert.assertEquals("topic-1", recreated.getTopicName());
+ Assert.assertEquals(filter, recreated.getTopicAttributes().get("tag-filter"));
+ }
+
+ @Test
public void testAlterTopicStatement() {
final Statement statement =
sqlParser.createStatement(
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java
index 1b09a53..1472351 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java
@@ -182,11 +182,74 @@
SubscriptionTopicAgent.shouldRefreshColumnFilter(null, createTreeTopicMeta()));
}
+ @Test
+ public void testTagFilterRefreshSkippedForOwnerOnlyUpdate() {
+ final TopicMeta oldMeta =
+ createTableTopicMeta("column_name = \"temperature\"", "region = \"north\"", "db", "t1");
+ final Map<String, String> updatedAttributes = new HashMap<>();
+ updatedAttributes.put(TopicConstant.OWNER_ID_KEY, "owner-1");
+ updatedAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "1");
+
+ Assert.assertFalse(
+ SubscriptionTopicAgent.shouldRefreshTagFilter(
+ oldMeta, oldMeta.deepCopyWithUpdatedAttributes(updatedAttributes)));
+ }
+
+ @Test
+ public void testTagFilterRefreshTriggeredForBindingInputUpdate() {
+ final TopicMeta oldMeta =
+ createTableTopicMeta("column_name = \"temperature\"", "region = \"north\"", "db", "t1");
+
+ Assert.assertTrue(
+ SubscriptionTopicAgent.shouldRefreshTagFilter(
+ oldMeta,
+ createTableTopicMeta(
+ "column_name = \"temperature\"", "region = \"north\"", "db2", "t1")));
+ Assert.assertTrue(
+ SubscriptionTopicAgent.shouldRefreshTagFilter(
+ oldMeta,
+ createTableTopicMeta(
+ "column_name = \"temperature\"", "region = \"north\"", "db", "t2")));
+ }
+
+ @Test
+ public void testTagFilterRefreshTriggeredForCaseSensitiveExpressionUpdate() {
+ final TopicMeta oldMeta =
+ createTableTopicMeta("column_name = \"temperature\"", "region = \"north\"", "db", "t1");
+
+ Assert.assertTrue(
+ SubscriptionTopicAgent.shouldRefreshTagFilter(
+ oldMeta,
+ createTableTopicMeta(
+ "column_name = \"temperature\"", "region = \"North\"", "db", "t1")));
+ }
+
+ @Test
+ public void testTagFilterRefreshTriggeredForNewTableTopicOnly() {
+ Assert.assertTrue(
+ SubscriptionTopicAgent.shouldRefreshTagFilter(
+ null,
+ createTableTopicMeta(
+ "column_name = \"temperature\"", "region = \"north\"", "db", "t1")));
+ Assert.assertFalse(SubscriptionTopicAgent.shouldRefreshTagFilter(null, createTreeTopicMeta()));
+ }
+
private static TopicMeta createTableTopicMeta(
final String columnFilter, final String database, final String table) {
+ return createTableTopicMeta(columnFilter, null, database, table);
+ }
+
+ private static TopicMeta createTableTopicMeta(
+ final String columnFilter,
+ final String tagFilter,
+ final String database,
+ final String table) {
final Map<String, String> attributes = new HashMap<>();
attributes.put("__system.sql-dialect", "table");
attributes.put(TopicConstant.COLUMN_FILTER_KEY, columnFilter);
+ if (tagFilter != null) {
+ attributes.put(TopicConstant.TAG_FILTER_KEY, tagFilter);
+ }
attributes.put(TopicConstant.DATABASE_KEY, database);
attributes.put(TopicConstant.TABLE_KEY, table);
return new TopicMeta("topic", 1L, attributes);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java
index 1cb1b05..0d92e9d 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java
@@ -39,6 +39,9 @@
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest;
import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterMatcher;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
@@ -57,6 +60,7 @@
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
public class ConsensusLogToTabletConverterTest {
@@ -393,6 +397,69 @@
}
@Test
+ public void testConvertRelationalInsertTabletFiltersRowsBeforeColumns() throws Exception {
+ final ConsensusLogToTabletConverter converter =
+ createConverterWithTagFilter("id1 IN (\"id:10\", \"id:12\")", "m1");
+
+ final List<Tablet> tablets = converter.convert(StatementTestUtils.genInsertTabletNode(3, 10));
+
+ Assert.assertEquals(1, tablets.size());
+ final Tablet tablet = tablets.get(0);
+ Assert.assertEquals(2, tablet.getRowSize());
+ Assert.assertArrayEquals(new long[] {10L, 12L}, tablet.getTimestamps());
+ Assert.assertEquals("id:10", toUtf8(((Binary[]) tablet.getValues()[0])[0]));
+ Assert.assertEquals("id:12", toUtf8(((Binary[]) tablet.getValues()[0])[1]));
+ Assert.assertArrayEquals(new double[] {10.0, 12.0}, (double[]) tablet.getValues()[1], 0.0);
+ }
+
+ @Test
+ public void testConvertRelationalInsertRowReturnsEmptyWhenTagDoesNotMatch() throws Exception {
+ final ConsensusLogToTabletConverter converter =
+ createConverterWithTagFilter("id1 = \"missing\"", "m1");
+
+ Assert.assertTrue(converter.convert(StatementTestUtils.genInsertRowNode(7)).isEmpty());
+ }
+
+ @Test
+ public void testReplaySnapshotPreservesTagFilter() throws Exception {
+ final ConsensusLogToTabletConverter converter =
+ createConverterWithTagFilter("id1 IN (\"id:10\", \"id:12\")", "m1");
+ final ConsensusLogToTabletConverter snapshot = converter.snapshotForReplay();
+
+ final List<Tablet> tablets = snapshot.convert(StatementTestUtils.genInsertTabletNode(3, 10));
+
+ Assert.assertEquals(1, tablets.size());
+ Assert.assertEquals(2, tablets.get(0).getRowSize());
+ Assert.assertArrayEquals(new long[] {10L, 12L}, tablets.get(0).getTimestamps());
+ Assert.assertTrue(converter.hasSameReplayProjection(snapshot));
+ }
+
+ @Test
+ public void testReplayProjectionChangesWhenTagFilterChanges() throws Exception {
+ final TablePattern tablePattern =
+ new TablePattern(true, DATABASE_NAME, StatementTestUtils.tableName());
+ final ColumnFilterMatcher columnFilterMatcher =
+ ColumnFilterMatcher.ofSelectedColumnNames(new HashSet<>(Arrays.asList("m1")));
+ final ConsensusLogToTabletConverter original =
+ new ConsensusLogToTabletConverter(
+ null, tablePattern, columnFilterMatcher, TagFilterMatcher.matchAll(), DATABASE_NAME);
+ final TagFilterMatcher tagFilterMatcher =
+ TagFilterMatcher.fromTopicConfig(
+ new TopicConfig(
+ Map.of(
+ "__system.sql-dialect",
+ "table",
+ TopicConstant.TAG_FILTER_KEY,
+ "id1 = \"id:10\"")));
+ final ConsensusLogToTabletConverter updated =
+ new ConsensusLogToTabletConverter(
+ null, tablePattern, columnFilterMatcher, tagFilterMatcher, DATABASE_NAME);
+
+ Assert.assertFalse(
+ original.snapshotForReplay().hasSameReplayProjection(updated.snapshotForReplay()));
+ }
+
+ @Test
public void testConvertInsertRowsOfOneDeviceNodeGroupsRowsWithSameSchema()
throws IllegalPathException {
final ConsensusLogToTabletConverter converter = createTreeConverter();
@@ -477,6 +544,19 @@
DATABASE_NAME);
}
+ private static ConsensusLogToTabletConverter createConverterWithTagFilter(
+ final String tagFilter, final String... selectedColumns) throws Exception {
+ final TopicConfig topicConfig =
+ new TopicConfig(
+ Map.of("__system.sql-dialect", "table", TopicConstant.TAG_FILTER_KEY, tagFilter));
+ return new ConsensusLogToTabletConverter(
+ null,
+ new TablePattern(true, DATABASE_NAME, StatementTestUtils.tableName()),
+ ColumnFilterMatcher.ofSelectedColumnNames(new HashSet<>(Arrays.asList(selectedColumns))),
+ TagFilterMatcher.fromTopicConfig(topicConfig),
+ DATABASE_NAME);
+ }
+
private static ConsensusLogToTabletConverter createTreeConverter() {
return new ConsensusLogToTabletConverter(new IoTDBTreePattern("root.sg.**"), null, null, null);
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionFilterSnapshotTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionFilterSnapshotTest.java
new file mode 100644
index 0000000..2215517
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionFilterSnapshotTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.subscription.event.batch;
+
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta;
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta;
+import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
+import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
+import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
+import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingTabletQueue;
+import org.apache.iotdb.db.subscription.tagfilter.TabletTagFilter;
+import org.apache.iotdb.db.subscription.tagfilter.TagFilterEvaluationException;
+import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+
+import org.apache.tsfile.write.record.Tablet;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.Map;
+
+public class SubscriptionFilterSnapshotTest {
+
+ private static final String TOPIC = "filter_snapshot_test_topic";
+ private static final String GROUP = "filter_snapshot_test_group";
+
+ private TopicMeta topicMeta;
+ private SubscriptionPrefetchingTabletQueue queue;
+
+ @Before
+ public void setUp() {
+ topicMeta =
+ new TopicMeta(
+ TOPIC,
+ 1L,
+ Collections.singletonMap(
+ SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TABLE_VALUE));
+ Assert.assertNull(SubscriptionAgent.topic().handleSingleTopicMetaChanges(topicMeta));
+ Assert.assertNull(
+ SubscriptionAgent.consumer()
+ .handleSingleConsumerGroupMetaChanges(
+ new ConsumerGroupMeta(
+ GROUP,
+ 1L,
+ new ConsumerMeta(
+ "consumer",
+ 1L,
+ Collections.singletonMap(
+ ConsumerConstant.SQL_DIALECT_KEY,
+ SystemConstant.SQL_DIALECT_TABLE_VALUE)))));
+ queue = Mockito.mock(SubscriptionPrefetchingTabletQueue.class);
+ Mockito.when(queue.getTopicName()).thenReturn(TOPIC);
+ Mockito.when(queue.getConsumerGroupId()).thenReturn(GROUP);
+ }
+
+ @After
+ public void tearDown() {
+ Assert.assertNull(SubscriptionAgent.consumer().handleDropConsumerGroup(GROUP));
+ Assert.assertNull(SubscriptionAgent.topic().handleDropTopic(TOPIC, true));
+ }
+
+ @Test
+ public void testOwnerAndRetentionUpdatesPreserveBatchButFilterUpdateRotatesIt() {
+ final SubscriptionFilterSnapshot snapshot = SubscriptionFilterSnapshot.capture(queue);
+ updateTopic(
+ Map.of(
+ TopicConstant.OWNER_ID_KEY, "owner-1",
+ TopicConstant.OWNER_EPOCH_KEY, "1",
+ TopicConstant.RETENTION_MS_KEY, "60000"));
+ Assert.assertTrue(snapshot.isCurrent(queue));
+
+ updateTopic(Collections.singletonMap(TopicConstant.TAG_FILTER_KEY, "false"));
+ Assert.assertFalse(snapshot.isCurrent(queue));
+ Assert.assertTrue(snapshot.getTagFilterMatcher().isMatchAll());
+ final SubscriptionFilterSnapshot updated = SubscriptionFilterSnapshot.capture(queue);
+ Assert.assertTrue(updated.isCurrent(queue));
+ final Tablet tablet =
+ new Tablet(
+ "weather",
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ 1);
+ tablet.addTimestamp(0, 1L);
+ Assert.assertNull(TabletTagFilter.filter(tablet, updated.getTagFilterMatcher()));
+ }
+
+ @Test
+ public void testMissingSchemaReportsTargetAndLocalCacheReason() {
+ final String database = "missing_snapshot_database";
+ final String table = "missing_table";
+ updateTopic(
+ Map.of(
+ TopicConstant.DATABASE_KEY, database,
+ TopicConstant.TABLE_KEY, table,
+ TopicConstant.TAG_FILTER_KEY, "region = \"north\""));
+ try {
+ SubscriptionFilterSnapshot.capture(queue);
+ Assert.fail("Expected unavailable table schema to fail closed");
+ } catch (final TagFilterEvaluationException e) {
+ Assert.assertTrue(
+ e.getMessage(),
+ e.getMessage()
+ .contains(
+ String.format(
+ DataNodeMiscMessages
+ .EXCEPTION_TABLE_SCHEMA_ARG_ARG_IS_ABSENT_FROM_THIS_DATANODE_S_TABLE_CACHE_THE_TABLE_MAY_NOT_EXIST_OR_ITS_SCHEMA_HAS_NOT_BEEN_SYNCHRONIZED_YET_23975BDF,
+ database,
+ table)));
+ }
+ }
+
+ @Test
+ public void testColumnBindingFailureRetainsMatchNoneMatcher() {
+ updateTopic(
+ Map.of(
+ TopicConstant.DATABASE_KEY, "missing_snapshot_database",
+ TopicConstant.TABLE_KEY, "missing_table",
+ TopicConstant.COLUMN_FILTER_KEY, "column_name = \"temperature\""));
+ final SubscriptionFilterSnapshot snapshot = SubscriptionFilterSnapshot.capture(queue);
+ Assert.assertFalse(snapshot.getColumnFilterMatcher().isMatchAll());
+ Assert.assertFalse(
+ snapshot
+ .getColumnFilterMatcher()
+ .match("missing_snapshot_database", "missing_table", "temperature"));
+ }
+
+ @Test(expected = TagFilterEvaluationException.class)
+ public void testUnavailableTreeViewProjectorFailsInsteadOfDroppingTablet() {
+ final SubscriptionTreeViewProjector projector =
+ new SubscriptionTreeViewProjector(new TopicConfig());
+ Assert.assertTrue(projector.prepare());
+ Assert.assertFalse(projector.isAvailable());
+ projector.project(null);
+ }
+
+ private void updateTopic(final Map<String, String> attributes) {
+ topicMeta = topicMeta.deepCopyWithUpdatedAttributes(attributes);
+ Assert.assertNull(SubscriptionAgent.topic().handleSingleTopicMetaChanges(topicMeta));
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatchTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatchTest.java
index d0d67f4..d6294ff 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatchTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatchTest.java
@@ -21,15 +21,21 @@
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta;
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta;
+import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager.TsFileParserMemoryReservation;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingTabletQueue;
import org.apache.iotdb.db.subscription.event.pipe.SubscriptionPipeTabletBatchEvents;
+import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
@@ -81,6 +87,26 @@
Mockito.mock(SubscriptionPrefetchingTabletQueue.class);
Mockito.when(queue.getTopicName()).thenReturn("topic");
Mockito.when(queue.getConsumerGroupId()).thenReturn("group");
+ Assert.assertNull(
+ SubscriptionAgent.topic()
+ .handleSingleTopicMetaChanges(
+ new TopicMeta(
+ "topic",
+ 1L,
+ Collections.singletonMap(
+ SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TABLE_VALUE))));
+ Assert.assertNull(
+ SubscriptionAgent.consumer()
+ .handleSingleConsumerGroupMetaChanges(
+ new ConsumerGroupMeta(
+ "group",
+ 1L,
+ new ConsumerMeta(
+ "consumer",
+ 1L,
+ Collections.singletonMap(
+ ConsumerConstant.SQL_DIALECT_KEY,
+ SystemConstant.SQL_DIALECT_TABLE_VALUE)))));
batch =
new SubscriptionPipeTabletEventBatch(Integer.parseInt(REGION_ID), queue, 20, 1024 * 1024);
}
@@ -93,6 +119,8 @@
for (final PipeTsFileInsertionEvent event : tsFileEvents) {
event.close();
}
+ Assert.assertNull(SubscriptionAgent.consumer().handleDropConsumerGroup("group"));
+ Assert.assertNull(SubscriptionAgent.topic().handleDropTopic("topic", true));
config.setPipeTsFileParserMemory(originalParserMemory);
config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatchCleanupTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatchCleanupTest.java
index d4d3609..5723170 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatchCleanupTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatchCleanupTest.java
@@ -33,11 +33,43 @@
import java.io.File;
import java.nio.file.Files;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
public class SubscriptionPipeTsFileEventBatchCleanupTest {
@Test
+ public void testClosesInnerBatchWhenFilteringProducesNoPayload() throws Exception {
+ final PipeTabletEventTsFileBatch innerBatch = Mockito.mock(PipeTabletEventTsFileBatch.class);
+ final SubscriptionPrefetchingTsFileQueue queue =
+ Mockito.mock(SubscriptionPrefetchingTsFileQueue.class);
+ Mockito.when(innerBatch.isEmpty()).thenReturn(true);
+
+ final SubscriptionPipeTsFileEventBatch batch =
+ new SubscriptionPipeTsFileEventBatch(1, queue, 1, 1, innerBatch);
+
+ Assert.assertTrue(batch.generateSubscriptionEvents().isEmpty());
+ Mockito.verify(innerBatch).close();
+ }
+
+ @Test
+ public void testClosesInnerBatchWhenSealingProducesNoFile() throws Exception {
+ final PipeTabletEventTsFileBatch innerBatch = Mockito.mock(PipeTabletEventTsFileBatch.class);
+ final SubscriptionPrefetchingTsFileQueue queue =
+ Mockito.mock(SubscriptionPrefetchingTsFileQueue.class);
+ Mockito.when(innerBatch.isEmpty()).thenReturn(false);
+ Mockito.when(innerBatch.sealTsFiles()).thenReturn(Collections.emptyList());
+
+ final SubscriptionPipeTsFileEventBatch batch =
+ new SubscriptionPipeTsFileEventBatch(1, queue, 1, 1, innerBatch);
+
+ Assert.assertTrue(batch.generateSubscriptionEvents().isEmpty());
+ Mockito.verify(innerBatch).decreaseEventsReferenceCount(batch.getClass().getName(), true);
+ Mockito.verify(innerBatch).onSuccess();
+ Mockito.verify(innerBatch).close();
+ }
+
+ @Test
public void testDeletesSealedFilesAfterAllSharedEventsAreCleaned() throws Exception {
final File temporaryDirectory =
Files.createTempDirectory("subscription-tsfile-cleanup").toFile();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/tagfilter/TabletTagFilterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/tagfilter/TabletTagFilterTest.java
new file mode 100644
index 0000000..f7c8d02
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/tagfilter/TabletTagFilterTest.java
@@ -0,0 +1,182 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.subscription.tagfilter;
+
+import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTableModelTabletEventSorter;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.write.record.Tablet;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class TabletTagFilterTest {
+
+ @Test
+ public void testPrefixSharesColumnDataWithoutChangingSourceWhenSorted() throws Exception {
+ final Tablet tablet = createTablet();
+ tablet.addValue(1, 0, "north");
+ tablet.addValue(1, 1, "d1");
+ tablet.getTimestamps()[0] = 2;
+ tablet.getTimestamps()[1] = 1;
+ final Tablet filtered =
+ TabletTagFilter.filter(tablet, matcher("region IS NOT NULL AND region != \"North\""));
+ Assert.assertNotNull(filtered);
+ Assert.assertEquals(2, filtered.getRowSize());
+ Assert.assertEquals(4, tablet.getRowSize());
+ Assert.assertSame(tablet.getSchemas(), filtered.getSchemas());
+ Assert.assertSame(tablet.getColumnTypes(), filtered.getColumnTypes());
+ Assert.assertSame(tablet.getTimestamps(), filtered.getTimestamps());
+ Assert.assertSame(tablet.getValues()[0], filtered.getValues()[0]);
+ Assert.assertNotSame(tablet.getValues(), filtered.getValues());
+ Assert.assertNotSame(tablet.getBitMaps(), filtered.getBitMaps());
+
+ // Sorting replaces columns and bitmap entries in place, so the outer arrays must be private.
+ final Object originalColumn = tablet.getValues()[0];
+ final BitMap originalBitMap = tablet.getBitMaps()[0];
+ new PipeTableModelTabletEventSorter(filtered).sortAndDeduplicateByDevIdTimestamp();
+ Assert.assertEquals(1, filtered.getTimestamp(0));
+ Assert.assertNotSame(originalColumn, filtered.getValues()[0]);
+ Assert.assertEquals(4, tablet.getRowSize());
+ Assert.assertSame(originalColumn, tablet.getValues()[0]);
+ Assert.assertSame(originalBitMap, tablet.getBitMaps()[0]);
+ Assert.assertEquals("north", tablet.getValue(0, 0).toString());
+ Assert.assertEquals(2, tablet.getTimestamp(0));
+ }
+
+ @Test
+ public void testCompactionRetainsAllNullColumnsWithoutValueArrays() throws Exception {
+ final Tablet tablet = createTablet();
+ tablet.getValues()[2] = null;
+ final BitMap allNull = new BitMap(tablet.getRowSize());
+ allNull.markAll();
+ tablet.getBitMaps()[2] = allNull;
+
+ final Tablet filtered =
+ TabletTagFilter.filter(tablet, matcher("region = \"south\" OR region IS NULL"));
+ Assert.assertNotNull(filtered);
+ Assert.assertEquals(2, filtered.getRowSize());
+ Assert.assertNull(filtered.getValues()[2]);
+ Assert.assertTrue(filtered.isNull(0, 2));
+ Assert.assertTrue(filtered.isNull(1, 2));
+ Assert.assertSame(tablet.getSchemas(), filtered.getSchemas());
+ Assert.assertSame(tablet.getColumnTypes(), filtered.getColumnTypes());
+ }
+
+ @Test
+ public void testFilterRowsAndPreserveValuesAndNulls() throws Exception {
+ final Tablet filtered =
+ TabletTagFilter.filter(createTablet(), matcher("region = \"south\" OR region IS NULL"));
+
+ Assert.assertNotNull(filtered);
+ Assert.assertEquals(2, filtered.getRowSize());
+ Assert.assertEquals(2L, filtered.getTimestamp(0));
+ Assert.assertEquals(3L, filtered.getTimestamp(1));
+ Assert.assertEquals("south", filtered.getValue(0, 0).toString());
+ Assert.assertTrue(filtered.isNull(1, 0));
+ Assert.assertEquals("d2", filtered.getValue(0, 1).toString());
+ Assert.assertEquals("d3", filtered.getValue(1, 1).toString());
+ Assert.assertEquals(20.0, (Double) filtered.getValue(0, 2), 0.0);
+ Assert.assertEquals(30.0, (Double) filtered.getValue(1, 2), 0.0);
+ Assert.assertEquals(createTablet().getColumnTypes(), filtered.getColumnTypes());
+ }
+
+ @Test
+ public void testComparisonIsCaseSensitive() throws Exception {
+ final Tablet filtered = TabletTagFilter.filter(createTablet(), matcher("region = \"north\""));
+
+ Assert.assertNotNull(filtered);
+ Assert.assertEquals(1, filtered.getRowSize());
+ Assert.assertEquals(1L, filtered.getTimestamp(0));
+ }
+
+ @Test
+ public void testUnknownOrNonTagColumnFailsClosed() throws Exception {
+ assertFailsClosed("missing = \"x\"");
+ assertFailsClosed("temperature = \"10\"");
+ }
+
+ @Test
+ public void testNoMatchedRowReturnsNull() throws Exception {
+ Assert.assertNull(TabletTagFilter.filter(createTablet(), matcher("region = \"east\"")));
+ }
+
+ @Test
+ public void testTrivialFilterReturnsOriginalTablet() throws Exception {
+ final Tablet tablet = createTablet();
+
+ Assert.assertSame(tablet, TabletTagFilter.filter(tablet, matcher(" TRUE ")));
+ Assert.assertNull(TabletTagFilter.filter(tablet, TagFilterMatcher.matchNone()));
+ }
+
+ private static TagFilterMatcher matcher(final String filter) throws Exception {
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("__system.sql-dialect", "table");
+ attributes.put(TopicConstant.TAG_FILTER_KEY, filter);
+ return TagFilterMatcher.fromTopicConfig(new TopicConfig(attributes));
+ }
+
+ private static void assertFailsClosed(final String filter) throws Exception {
+ try {
+ TabletTagFilter.filter(createTablet(), matcher(filter));
+ Assert.fail("Expected tag-filter schema binding failure: " + filter);
+ } catch (final TagFilterEvaluationException e) {
+ Assert.assertTrue(e.getMessage(), e.getMessage().contains("Tag-filter"));
+ }
+ }
+
+ private static Tablet createTablet() {
+ final List<String> columnNames = Arrays.asList("region", "device", "temperature");
+ final List<TSDataType> dataTypes =
+ Arrays.asList(TSDataType.STRING, TSDataType.STRING, TSDataType.DOUBLE);
+ final List<ColumnCategory> categories =
+ Arrays.asList(ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.FIELD);
+ final Tablet tablet = new Tablet("weather", columnNames, dataTypes, categories, 4);
+
+ tablet.addTimestamp(0, 1L);
+ tablet.addValue(0, 0, "north");
+ tablet.addValue(0, 1, "d1");
+ tablet.addValue(0, 2, 10.0);
+
+ tablet.addTimestamp(1, 2L);
+ tablet.addValue(1, 0, "south");
+ tablet.addValue(1, 1, "d2");
+ tablet.addValue(1, 2, 20.0);
+
+ tablet.addTimestamp(2, 3L);
+ tablet.addValue(2, 1, "d3");
+ tablet.addValue(2, 2, 30.0);
+
+ tablet.addTimestamp(3, 4L);
+ tablet.addValue(3, 0, "North");
+ tablet.addValue(3, 1, "d4");
+ tablet.addValue(3, 2, 40.0);
+ tablet.setRowSize(4);
+ return tablet;
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterParserTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterParserTest.java
new file mode 100644
index 0000000..c8e49da
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/tagfilter/TagFilterParserTest.java
@@ -0,0 +1,122 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.subscription.tagfilter;
+
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class TagFilterParserTest {
+
+ private static final TagFilterParser PARSER = new TagFilterParser();
+
+ @Test
+ public void testComparisonIsCaseSensitive() throws Exception {
+ Assert.assertTrue(evaluate("region = \"north\"", Map.of("region", "north")));
+ Assert.assertFalse(evaluate("region = \"north\"", Map.of("region", "North")));
+ Assert.assertTrue(evaluate("region != \"north\"", Map.of("region", "North")));
+ }
+
+ @Test
+ public void testQuotesInTagValues() throws Exception {
+ Assert.assertTrue(evaluate("region = \"O'Reilly\"", Map.of("region", "O'Reilly")));
+ Assert.assertFalse(evaluate("region = \"O'Reilly\"", Map.of("region", "O''Reilly")));
+ Assert.assertTrue(evaluate("region = \"a''b\"", Map.of("region", "a''b")));
+ Assert.assertTrue(evaluate("region = \"a\"\"\"\"b\"", Map.of("region", "a\"\"b")));
+ Assert.assertFalse(evaluate("region = \"a\"\"\"\"b\"", Map.of("region", "a\"b")));
+ Assert.assertTrue(evaluate("region = \"\"\"quoted\"\"\"", Map.of("region", "\"quoted\"")));
+ Assert.assertTrue(
+ evaluate(
+ "region = \"O'Reilly \"\"A\"\"\"\"B\"\"\"", Map.of("region", "O'Reilly \"A\"\"B\"")));
+ }
+
+ @Test
+ public void testQuotesInInLikeAndRegexp() throws Exception {
+ final String inFilter = "region IN (\"O'Reilly\", \"a\"\"\"\"b\")";
+ Assert.assertTrue(evaluate(inFilter, Map.of("region", "O'Reilly")));
+ Assert.assertTrue(evaluate(inFilter, Map.of("region", "a\"\"b")));
+ Assert.assertFalse(evaluate(inFilter, Map.of("region", "a\"b")));
+ Assert.assertTrue(
+ evaluate("region LIKE \"O'Reilly \"\"%\"\"\"", Map.of("region", "O'Reilly \"north\"")));
+ Assert.assertTrue(
+ evaluate(
+ "region REGEXP \"^O'Reilly \"\"[A-Z]+\"\"$\"", Map.of("region", "O'Reilly \"NORTH\"")));
+ }
+
+ @Test
+ public void testInLikeRegexpAndBooleanOperators() throws Exception {
+ Assert.assertTrue(
+ evaluate(
+ "region IN (\"north\", \"west\") AND device LIKE \"d_%\"",
+ Map.of("region", "north", "device", "d_1")));
+ Assert.assertTrue(
+ evaluate(
+ "region NOT REGEXP \"south|east\" OR NOT device = \"d2\"",
+ Map.of("region", "north", "device", "d2")));
+ Assert.assertTrue(evaluate("device LIKE \"d!_%\" ESCAPE \"!\"", Map.of("device", "d_sensor")));
+ Assert.assertTrue(evaluate("device LIKE \"d_%\"", Map.of("device", "d\n1")));
+ }
+
+ @Test
+ public void testNullUsesThreeValuedLogic() throws Exception {
+ final Map<String, String> values = new HashMap<>();
+ values.put("region", null);
+
+ Assert.assertTrue(evaluate("region IS NULL", values));
+ Assert.assertFalse(evaluate("region IS NOT NULL", values));
+ Assert.assertFalse(evaluate("region = \"north\"", values));
+ Assert.assertFalse(evaluate("NOT region = \"north\"", values));
+ Assert.assertTrue(evaluate("region = \"north\" OR true", values));
+ Assert.assertFalse(evaluate("region = \"north\" AND true", values));
+ Assert.assertTrue(evaluate("NOT (region = \"north\" AND false)", values));
+ Assert.assertFalse(evaluate("NOT (region = \"north\" OR true)", values));
+ Assert.assertTrue(evaluate("NOT (false AND region = \"north\")", values));
+ Assert.assertFalse(evaluate("NOT (true OR region = \"north\")", values));
+ }
+
+ @Test
+ public void testRejectInvalidExpressions() {
+ assertRejected("", "tag-filter should not be empty");
+ assertRejected("region > \"north\"", "unsupported comparison operator");
+ assertRejected("lower(region) = \"north\"", "expected column predicate operator");
+ assertRejected("region = other_tag", "expected string literal");
+ assertRejected("region REGEXP \"[\"", "illegal REGEXP pattern");
+ }
+
+ private static boolean evaluate(final String filter, final Map<String, String> values)
+ throws SubscriptionException {
+ final Expression expression = PARSER.parseAndValidate(filter);
+ return TagFilterEvaluator.evaluate(expression, identifier -> values.get(identifier.getValue()));
+ }
+
+ private static void assertRejected(final String filter, final String expectedMessagePart) {
+ try {
+ PARSER.parseAndValidate(filter);
+ Assert.fail("Expected tag-filter to be rejected: " + filter);
+ } catch (final SubscriptionException e) {
+ Assert.assertTrue(e.getMessage(), e.getMessage().contains(expectedMessagePart));
+ }
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java
index d50be22..d4a46e4 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java
@@ -134,6 +134,8 @@
// column names for show topics statement
public static final String TOPIC_NAME = "TopicName";
public static final String TOPIC_CONFIGS = "TopicConfigs";
+ public static final String TAG_FILTER_STATUS = "TagFilterStatus";
+ public static final String TAG_FILTER_MESSAGE = "TagFilterMessage";
public static final String TOPIC = "Topic";
public static final String CREATE_TOPIC = "Create Topic";
@@ -696,6 +698,13 @@
new ColumnHeader(TOPIC_NAME, TSDataType.TEXT),
new ColumnHeader(TOPIC_CONFIGS, TSDataType.TEXT));
+ public static final List<ColumnHeader> showTableTopicColumnHeaders =
+ ImmutableList.of(
+ new ColumnHeader(TOPIC_NAME, TSDataType.TEXT),
+ new ColumnHeader(TOPIC_CONFIGS, TSDataType.TEXT),
+ new ColumnHeader(TAG_FILTER_STATUS, TSDataType.TEXT),
+ new ColumnHeader(TAG_FILTER_MESSAGE, TSDataType.TEXT));
+
public static final List<ColumnHeader> showCreateTopicColumnHeaders =
ImmutableList.of(
new ColumnHeader(TOPIC, TSDataType.TEXT),
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java
index c68bcdd..086976d 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java
@@ -25,6 +25,7 @@
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+import org.apache.iotdb.rpc.subscription.config.TopicProperty;
import org.apache.iotdb.rpc.subscription.i18n.SubscriptionMessages;
import org.apache.tsfile.utils.PublicBAOS;
@@ -73,11 +74,11 @@
final String topicName, final long creationTime, final Map<String, String> topicAttributes) {
this.topicName = topicName;
this.creationTime = creationTime;
- this.config = new TopicConfig(topicAttributes);
+ this.config = new TopicConfig(canonicalizeTopicAttributes(topicAttributes));
this.ownerEpoch = -1L;
this.maxOwnerEpoch = -1L;
this.ownerLastTransferTimeMs = -1L;
- initOwnerFromTopicAttributes(topicAttributes);
+ initOwnerFromTopicAttributes(this.config.getAttribute());
this.subscribedConsumerGroupIds = new HashSet<>();
}
@@ -99,12 +100,18 @@
}
public TopicMeta deepCopyWithUpdatedAttributes(final Map<String, String> updatedAttributes) {
- final Map<String, String> copiedAttributes = new HashMap<>(config.getAttribute());
+ final Map<String, String> copiedAttributes = canonicalizeTopicAttributes(config.getAttribute());
if (Objects.nonNull(updatedAttributes)) {
- copiedAttributes.putAll(updatedAttributes);
- if ((updatedAttributes.containsKey(TopicConstant.OWNER_ID_KEY)
- || updatedAttributes.containsKey(TopicConstant.OWNER_EPOCH_KEY))
- && !updatedAttributes.containsKey(TopicConstant.OWNER_LEASE_DURATION_MS_KEY)) {
+ boolean ownerChanged = false;
+ boolean leaseUpdated = false;
+ for (final Map.Entry<String, String> entry : updatedAttributes.entrySet()) {
+ final String key = canonicalizeTopicAttributeKey(entry.getKey());
+ copiedAttributes.put(key, entry.getValue());
+ ownerChanged |=
+ TopicConstant.OWNER_ID_KEY.equals(key) || TopicConstant.OWNER_EPOCH_KEY.equals(key);
+ leaseUpdated |= TopicConstant.OWNER_LEASE_DURATION_MS_KEY.equals(key);
+ }
+ if (ownerChanged && !leaseUpdated) {
copiedAttributes.remove(TopicConstant.OWNER_LEASE_DURATION_MS_KEY);
}
}
@@ -310,6 +317,7 @@
final String value = ReadWriteIOUtils.readString(inputStream);
topicMeta.config.getAttribute().put(key, value);
}
+ canonicalizeTopicAttributesInPlace(topicMeta.config.getAttribute());
size = ReadWriteIOUtils.readInt(inputStream);
for (int i = 0; i < size; i++) {
@@ -333,6 +341,7 @@
final String value = ReadWriteIOUtils.readString(byteBuffer);
topicMeta.config.getAttribute().put(key, value);
}
+ canonicalizeTopicAttributesInPlace(topicMeta.config.getAttribute());
size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; i++) {
@@ -396,6 +405,30 @@
}
}
+ // SQL accepts case-insensitive built-in keys, while runtime getters use canonical keys. Apply
+ // the same normalization on creation, ALTER merges, and deserialization so an ALTER cannot leave
+ // both "Tag-Filter" and "tag-filter" with different values. Extension keys remain unchanged.
+ private static Map<String, String> canonicalizeTopicAttributes(
+ final Map<String, String> topicAttributes) {
+ final Map<String, String> canonicalAttributes = new HashMap<>();
+ if (Objects.isNull(topicAttributes)) {
+ return canonicalAttributes;
+ }
+ topicAttributes.forEach(
+ (key, value) -> canonicalAttributes.put(canonicalizeTopicAttributeKey(key), value));
+ return canonicalAttributes;
+ }
+
+ private static void canonicalizeTopicAttributesInPlace(final Map<String, String> attributes) {
+ final Map<String, String> canonicalAttributes = canonicalizeTopicAttributes(attributes);
+ attributes.clear();
+ attributes.putAll(canonicalAttributes);
+ }
+
+ private static String canonicalizeTopicAttributeKey(final String key) {
+ return TopicProperty.canonicalizeKey(key);
+ }
+
private void initOwnerFromTopicAttributes(final Map<String, String> topicAttributes) {
final TopicConfig topicConfig = new TopicConfig(topicAttributes);
@@ -455,6 +488,8 @@
extractorAttributes.putAll(config.getAttributesWithSourceDatabaseAndTableName());
// column-filter is evaluated by subscription runtime on DataNode.
extractorAttributes.putAll(config.getAttributesWithSourceColumnFilter());
+ // tag-filter is evaluated by subscription runtime on DataNode.
+ extractorAttributes.putAll(config.getAttributesWithSourceTagFilter());
} else {
// tree model: path or pattern
extractorAttributes.putAll(config.getAttributesWithSourcePathOrPattern());
diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/topic/TopicMetaTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/topic/TopicMetaTest.java
new file mode 100644
index 0000000..26af024
--- /dev/null
+++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/topic/TopicMetaTest.java
@@ -0,0 +1,72 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.commons.subscription.topic;
+
+import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.util.Map;
+
+public class TopicMetaTest {
+
+ @Test
+ public void testMixedCaseAlterReplacesBuiltInKeyAndPreservesExtensionKeys() throws Exception {
+ final TopicMeta original =
+ new TopicMeta(
+ "topic",
+ 1L,
+ Map.of("Tag-Filter", "region = \"north\"", "source.CustomOption", "CaseSensitive"));
+ final TopicMeta updated =
+ original.deepCopyWithUpdatedAttributes(Map.of(" TAG-FILTER ", "region = \"North\""));
+ Assert.assertEquals(2, updated.getConfig().getAttribute().size());
+ Assert.assertEquals("region = \"North\"", updated.getConfig().getTagFilter());
+ Assert.assertEquals(
+ "CaseSensitive", updated.getConfig().getAttribute().get("source.CustomOption"));
+ Assert.assertEquals("region = \"north\"", original.getConfig().getTagFilter());
+
+ Assert.assertEquals(updated, TopicMeta.deserialize(updated.serialize()));
+ final ByteArrayOutputStream output = new ByteArrayOutputStream();
+ updated.serialize(output);
+ Assert.assertEquals(
+ updated, TopicMeta.deserialize(new ByteArrayInputStream(output.toByteArray())));
+ }
+
+ @Test
+ public void testDeserializationCanonicalizesLegacyMixedCaseAttributes() throws Exception {
+ final TopicMeta topic = new TopicMeta("topic", 1L, Map.of());
+ // Simulate metadata written before key canonicalization was introduced.
+ topic.getConfig().getAttribute().put("Tag-Filter", "region = \"North\"");
+ final TopicMeta fromBuffer = TopicMeta.deserialize(topic.serialize());
+ final ByteArrayOutputStream output = new ByteArrayOutputStream();
+ topic.serialize(output);
+ final TopicMeta fromStream =
+ TopicMeta.deserialize(new ByteArrayInputStream(output.toByteArray()));
+ for (final TopicMeta restored : new TopicMeta[] {fromBuffer, fromStream}) {
+ Assert.assertEquals(
+ Map.of(TopicConstant.TAG_FILTER_KEY, "region = \"North\""),
+ restored.getConfig().getAttribute());
+ }
+ }
+}
