jt2594838 commented on code in PR #17936:
URL: https://github.com/apache/iotdb/pull/17936#discussion_r3425483731
##########
iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/SubscriptionPollResponse.java:
##########
@@ -77,6 +110,8 @@ private void serialize(final DataOutputStream stream) throws
IOException {
ReadWriteIOUtils.write(responseType, stream);
payload.serialize(stream);
commitContext.serialize(stream);
+ ReadWriteIOUtils.write(timeSelected, stream);
+ serializeTimeSelectedByTable(stream);
Review Comment:
They are serialized, then why do you call them transient?
##########
integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java:
##########
@@ -0,0 +1,1523 @@
+/*
+ * 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.subscription.it.dual.tablemodel;
+
+import org.apache.iotdb.commons.schema.table.TreeViewSchema;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TagColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TimeColumnSchema;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.db.subscription.columnfilter.BoundColumnFilter;
+import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterBinder;
+import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import
org.apache.iotdb.itbase.category.MultiClusterIT2SubscriptionTableArchVerification;
+import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+import org.apache.iotdb.session.subscription.ISubscriptionTableSession;
+import org.apache.iotdb.session.subscription.SubscriptionTableSessionBuilder;
+import
org.apache.iotdb.session.subscription.consumer.ISubscriptionTablePullConsumer;
+import
org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumerBuilder;
+import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
+import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
+import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler;
+import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
+import org.apache.iotdb.subscription.it.dual.AbstractSubscriptionDualIT;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.RowRecord;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.Set;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2SubscriptionTableArchVerification.class})
+public class IoTDBSubscriptionColumnFilterIT extends
AbstractSubscriptionDualIT {
+
+ private static final String TABLE_NAME = "t_column_filter";
+ private static final String TABLE_SCHEMA =
+ "tag1 STRING TAG, s1 INT64 FIELD, s2 DOUBLE FIELD, s3 BOOLEAN FIELD";
+ private static final String COLUMN_FILTER = "column_name IN (\"time\",
\"tag1\", \"s2\")";
+ private static final String FIELD_CATEGORY_FILTER = "category = \"FIELD\"";
+ private static final String FIELD_CATEGORY_REBIND_FILTER = "category IN
(\"FIELD\")";
+ private static final String MIXED_COLUMN_FILTER =
+ "( CATEGORY = \"field\" AND datatype IN (\"int64\", \"double\") )"
+ + " OR \"column_name\" REGEXP \"tag.*\"";
+ private static final String COMPLEX_OPERATOR_FILTER =
+ "datatype IS NOT NULL"
+ + " AND column_name != \"s1\""
+ + " AND column_name NOT IN (\"s1\", \"s3\")"
+ + " AND column_name NOT LIKE \"unknown%\""
+ + " AND column_name NOT REGEXP \"unknown.*\""
+ + " AND (column_name LIKE \"s%\" OR column_name = \"tag1\")";
+ private static final Set<String> EXPECTED_COLUMNS =
+ new LinkedHashSet<>(Arrays.asList("time", "tag1", "s2"));
+ private static final Set<String> EXPECTED_ALL_COLUMNS =
+ new LinkedHashSet<>(Arrays.asList("time", "tag1", "s1", "s2", "s3"));
+ private static final Set<String> EXPECTED_MIXED_COLUMNS =
+ new LinkedHashSet<>(Arrays.asList("tag1", "s1", "s2"));
+ private static final Set<String> EXPECTED_COMPLEX_OPERATOR_COLUMNS =
+ new LinkedHashSet<>(Arrays.asList("tag1", "s2"));
+
+ @Override
+ protected void setUpConfig() {
+ super.setUpConfig();
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(30);
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setPipeMetaSyncerInitialSyncDelayMinutes(1)
+ .setPipeMemoryManagementEnabled(false)
+ .setIsPipeEnableMemoryCheck(false);
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setPipeMetaSyncerSyncIntervalMinutes(1)
+ .setPipeMemoryManagementEnabled(false)
+ .setIsPipeEnableMemoryCheck(false);
+ }
+
+ @Test
+ public void testLiveRecordHandlerColumnFilter() throws Exception {
+ final String database = databaseName("record");
+ final String topicName = topicName("record");
+ final String consumerId = consumerName("record");
+ final String consumerGroupId = consumerGroupName("record");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ COLUMN_FILTER);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ insertRows(senderEnv, database, TABLE_NAME, 1, 4);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(1L, 2L, 3L));
+ final ConsumedRecordStats stats =
+ pollRecordMessagesForTimestamps(consumer, expectedTimestamps,
true);
+
+ Assert.assertEquals(expectedTimestamps, stats.timestamps);
+ Assert.assertEquals(EXPECTED_COLUMNS, stats.columnNames);
+ Assert.assertFalse(stats.columnNames.contains("s1"));
+ Assert.assertFalse(stats.columnNames.contains("s3"));
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void testLiveRecordHandlerWithoutColumnFilterPreservesAllColumns()
throws Exception {
+ final String database = databaseName("default");
+ final String topicName = topicName("default");
+ final String consumerId = consumerName("default");
+ final String consumerGroupId = consumerGroupName("default");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createTopicWithoutColumnFilter(
+ topicName, database, TABLE_NAME,
TopicConstant.FORMAT_RECORD_HANDLER_VALUE);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ insertRows(senderEnv, database, TABLE_NAME, 110, 113);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(110L, 111L, 112L));
+ final ConsumedRecordStats stats =
+ pollRecordMessagesForTimestamps(consumer, expectedTimestamps,
true);
+
+ Assert.assertEquals(expectedTimestamps, stats.timestamps);
+ Assert.assertEquals(EXPECTED_ALL_COLUMNS, stats.columnNames);
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void
testLiveRecordHandlerColumnFilterKeepsHistoricalAndRealtimeConsistent()
+ throws Exception {
+ final String database = databaseName("live_both");
+ final String topicName = topicName("live_both");
+ final String consumerId = consumerName("live_both");
+ final String consumerGroupId = consumerGroupName("live_both");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ insertRows(senderEnv, database, TABLE_NAME, 180, 183);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ COLUMN_FILTER);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+
+ final Set<Long> historicalTimestamps = new
LinkedHashSet<>(Arrays.asList(180L, 181L, 182L));
+ final ConsumedRecordStats historicalStats =
+ pollRecordMessagesForTimestamps(consumer, historicalTimestamps,
true);
+ Assert.assertEquals(EXPECTED_COLUMNS, historicalStats.columnNames);
+
+ insertRows(senderEnv, database, TABLE_NAME, 190, 193);
+
+ final Set<Long> realtimeTimestamps = new
LinkedHashSet<>(Arrays.asList(190L, 191L, 192L));
+ final ConsumedRecordStats realtimeStats =
+ pollRecordMessagesForTimestamps(consumer, realtimeTimestamps,
true);
+ Assert.assertEquals(EXPECTED_COLUMNS, realtimeStats.columnNames);
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void
testLiveRecordHandlerColumnFilterAppliesToMultipleConsumerGroups() throws
Exception {
+ final String database = databaseName("multi_group");
+ final String topicName = topicName("multi_group");
+ final String firstConsumerId = consumerName("multi_group_a");
+ final String firstConsumerGroupId = consumerGroupName("multi_group_a");
+ final String secondConsumerId = consumerName("multi_group_b");
+ final String secondConsumerGroupId = consumerGroupName("multi_group_b");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ COLUMN_FILTER);
+
+ try (final ISubscriptionTablePullConsumer firstConsumer =
+ createConsumer(firstConsumerId, firstConsumerGroupId);
+ final ISubscriptionTablePullConsumer secondConsumer =
+ createConsumer(secondConsumerId, secondConsumerGroupId)) {
+ firstConsumer.subscribe(topicName);
+ secondConsumer.subscribe(topicName);
+ insertRows(senderEnv, database, TABLE_NAME, 170, 173);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(170L, 171L, 172L));
+ final ConsumedRecordStats firstStats =
+ pollRecordMessagesForTimestamps(firstConsumer, expectedTimestamps,
true);
+ final ConsumedRecordStats secondStats =
+ pollRecordMessagesForTimestamps(secondConsumer,
expectedTimestamps, true);
+
+ Assert.assertEquals(EXPECTED_COLUMNS, firstStats.columnNames);
+ Assert.assertEquals(EXPECTED_COLUMNS, secondStats.columnNames);
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void testSnapshotRecordHandlerColumnFilter() throws Exception {
+ final String database = databaseName("snapshot");
+ final String topicName = topicName("snapshot");
+ final String consumerId = consumerName("snapshot");
+ final String consumerGroupId = consumerGroupName("snapshot");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ insertRows(senderEnv, database, TABLE_NAME, 20, 23);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.MODE_SNAPSHOT_VALUE,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ COLUMN_FILTER);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(20L, 21L, 22L));
+ 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 testLiveRecordHandlerMixedCaseInsensitiveColumnFilter() throws
Exception {
+ final String database = databaseName("mixed");
+ final String topicName = topicName("mixed");
+ final String consumerId = consumerName("mixed");
+ final String consumerGroupId = consumerGroupName("mixed");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ MIXED_COLUMN_FILTER);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ insertRows(senderEnv, database, TABLE_NAME, 30, 33);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(30L, 31L, 32L));
+ final ConsumedRecordStats stats =
+ pollRecordMessagesForTimestamps(consumer, expectedTimestamps,
false);
+
+ Assert.assertEquals(expectedTimestamps, stats.timestamps);
+ Assert.assertEquals(EXPECTED_MIXED_COLUMNS, stats.columnNames);
+ Assert.assertFalse(stats.columnNames.contains("time"));
+ Assert.assertFalse(stats.columnNames.contains("s3"));
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void testLiveRecordHandlerComplexOperatorColumnFilter() throws
Exception {
+ final String database = databaseName("operators");
+ final String topicName = topicName("operators");
+ final String consumerId = consumerName("operators");
+ final String consumerGroupId = consumerGroupName("operators");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ COMPLEX_OPERATOR_FILTER);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ insertRows(senderEnv, database, TABLE_NAME, 100, 103);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(100L, 101L, 102L));
+ final ConsumedRecordStats stats =
+ pollRecordMessagesForTimestamps(consumer, expectedTimestamps,
false);
+
+ Assert.assertEquals(expectedTimestamps, stats.timestamps);
+ Assert.assertEquals(EXPECTED_COMPLEX_OPERATOR_COLUMNS,
stats.columnNames);
+ Assert.assertFalse(stats.columnNames.contains("time"));
+ Assert.assertFalse(stats.columnNames.contains("s1"));
+ Assert.assertFalse(stats.columnNames.contains("s3"));
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void testLiveRecordHandlerColumnFilterDropsMatchedAttributeColumn()
throws Exception {
+ final String database = databaseName("attribute");
+ final String topicName = topicName("attribute");
+ final String consumerId = consumerName("attribute");
+ final String consumerGroupId = consumerGroupName("attribute");
+ final String schemaWithAttribute =
+ "tag1 STRING TAG, attr1 STRING ATTRIBUTE, s1 INT64 FIELD, s2 DOUBLE
FIELD";
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME,
schemaWithAttribute);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ "category = \"ATTRIBUTE\" OR column_name = \"s2\"");
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ insertRowsWithAttribute(senderEnv, database, TABLE_NAME, 160, 163);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(160L, 161L, 162L));
+ final ConsumedRecordStats stats =
+ pollRecordMessagesForTimestamps(consumer, expectedTimestamps,
false);
+
+ Assert.assertEquals(expectedTimestamps, stats.timestamps);
+ Assert.assertEquals(new LinkedHashSet<>(Arrays.asList("tag1", "s2")),
stats.columnNames);
+ Assert.assertFalse(stats.columnNames.contains("attr1"));
+ Assert.assertFalse(stats.columnNames.contains("s1"));
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void testTreeViewColumnFilterBindingUsesSourceFieldName() throws
Exception {
+ final String database = databaseName("view");
+ final String viewName = TABLE_NAME + "_view";
+ final Map<String, String> attributes = new LinkedHashMap<>();
+ attributes.put("__system.sql-dialect", "table");
+ attributes.put(TopicConstant.DATABASE_KEY, database);
+ attributes.put(TopicConstant.TABLE_KEY, viewName);
+ attributes.put(TopicConstant.COLUMN_FILTER_KEY, "column_name =
\"s_view\"");
+
+ final BoundColumnFilter boundColumnFilter =
+ new ColumnFilterBinder()
+ .bind(
+ new TopicConfig(attributes),
+ Map.of(database, Map.of(viewName,
createTreeViewSchema(viewName))));
+ final ColumnFilterMatcher matcher =
+ ColumnFilterMatcher.fromBoundColumnFilter(boundColumnFilter);
+
+ Assert.assertTrue(matcher.match(database, viewName, "tag1"));
+ Assert.assertTrue(matcher.match(database, viewName, "s_src"));
+ Assert.assertFalse(matcher.match(database, viewName, "s_view"));
+ }
+
+ @Test
+ public void testCreateTopicWithTreeViewColumnFilterUsesViewFieldName()
throws Exception {
+ final String database = databaseName("view_topic");
+ final String treeDatabase = database + "_tree";
+ final String viewName = TABLE_NAME + "_view";
+ final String topicName = topicName("view_topic");
+
+ try {
+ createTreeView(senderEnv, database, treeDatabase, viewName);
+ createTopic(
+ topicName,
+ database,
+ viewName,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ "column_name = \"s_view\"");
+ } finally {
+ cleanup(topicName, database);
+ dropTreeDatabase(senderEnv, treeDatabase);
+ }
+ }
Review Comment:
Where is the check?
##########
iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java:
##########
@@ -89,12 +106,57 @@ public void removeUserData() {
resultSets.forEach(SubscriptionResultSet::removeUserData);
}
+ private static boolean resolveTimeSelected(
+ final Map<String, Map<String, Boolean>> timeSelectedByTable,
+ final boolean defaultTimeSelected,
+ final String databaseName,
+ final Tablet tablet) {
+ if (Objects.isNull(timeSelectedByTable) || timeSelectedByTable.isEmpty()) {
+ return defaultTimeSelected;
+ }
+ final Map<String, Boolean> tableMap =
timeSelectedByTable.get(databaseName);
+ if (Objects.isNull(tablet)) {
+ return defaultTimeSelected;
+ }
+ final Map<String, Boolean> resolvedTableMap =
+ Objects.nonNull(tableMap)
+ ? tableMap
+ : getIgnoreCase(timeSelectedByTable, databaseName, null);
Review Comment:
It would be better to store dbNames and tableNames in lowercase in the maps.
Then you may avoid iterating the map.
##########
integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java:
##########
@@ -0,0 +1,1523 @@
+/*
+ * 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.subscription.it.dual.tablemodel;
+
+import org.apache.iotdb.commons.schema.table.TreeViewSchema;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TagColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TimeColumnSchema;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.db.subscription.columnfilter.BoundColumnFilter;
+import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterBinder;
+import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import
org.apache.iotdb.itbase.category.MultiClusterIT2SubscriptionTableArchVerification;
+import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+import org.apache.iotdb.session.subscription.ISubscriptionTableSession;
+import org.apache.iotdb.session.subscription.SubscriptionTableSessionBuilder;
+import
org.apache.iotdb.session.subscription.consumer.ISubscriptionTablePullConsumer;
+import
org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumerBuilder;
+import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
+import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
+import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler;
+import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
+import org.apache.iotdb.subscription.it.dual.AbstractSubscriptionDualIT;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.RowRecord;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.Set;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2SubscriptionTableArchVerification.class})
+public class IoTDBSubscriptionColumnFilterIT extends
AbstractSubscriptionDualIT {
+
+ private static final String TABLE_NAME = "t_column_filter";
+ private static final String TABLE_SCHEMA =
+ "tag1 STRING TAG, s1 INT64 FIELD, s2 DOUBLE FIELD, s3 BOOLEAN FIELD";
+ private static final String COLUMN_FILTER = "column_name IN (\"time\",
\"tag1\", \"s2\")";
+ private static final String FIELD_CATEGORY_FILTER = "category = \"FIELD\"";
+ private static final String FIELD_CATEGORY_REBIND_FILTER = "category IN
(\"FIELD\")";
+ private static final String MIXED_COLUMN_FILTER =
+ "( CATEGORY = \"field\" AND datatype IN (\"int64\", \"double\") )"
+ + " OR \"column_name\" REGEXP \"tag.*\"";
+ private static final String COMPLEX_OPERATOR_FILTER =
+ "datatype IS NOT NULL"
+ + " AND column_name != \"s1\""
+ + " AND column_name NOT IN (\"s1\", \"s3\")"
+ + " AND column_name NOT LIKE \"unknown%\""
+ + " AND column_name NOT REGEXP \"unknown.*\""
+ + " AND (column_name LIKE \"s%\" OR column_name = \"tag1\")";
+ private static final Set<String> EXPECTED_COLUMNS =
+ new LinkedHashSet<>(Arrays.asList("time", "tag1", "s2"));
+ private static final Set<String> EXPECTED_ALL_COLUMNS =
+ new LinkedHashSet<>(Arrays.asList("time", "tag1", "s1", "s2", "s3"));
+ private static final Set<String> EXPECTED_MIXED_COLUMNS =
+ new LinkedHashSet<>(Arrays.asList("tag1", "s1", "s2"));
+ private static final Set<String> EXPECTED_COMPLEX_OPERATOR_COLUMNS =
+ new LinkedHashSet<>(Arrays.asList("tag1", "s2"));
+
+ @Override
+ protected void setUpConfig() {
+ super.setUpConfig();
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(30);
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setPipeMetaSyncerInitialSyncDelayMinutes(1)
+ .setPipeMemoryManagementEnabled(false)
+ .setIsPipeEnableMemoryCheck(false);
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setPipeMetaSyncerSyncIntervalMinutes(1)
+ .setPipeMemoryManagementEnabled(false)
+ .setIsPipeEnableMemoryCheck(false);
+ }
+
+ @Test
+ public void testLiveRecordHandlerColumnFilter() throws Exception {
+ final String database = databaseName("record");
+ final String topicName = topicName("record");
+ final String consumerId = consumerName("record");
+ final String consumerGroupId = consumerGroupName("record");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ COLUMN_FILTER);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ insertRows(senderEnv, database, TABLE_NAME, 1, 4);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(1L, 2L, 3L));
+ final ConsumedRecordStats stats =
+ pollRecordMessagesForTimestamps(consumer, expectedTimestamps,
true);
+
+ Assert.assertEquals(expectedTimestamps, stats.timestamps);
+ Assert.assertEquals(EXPECTED_COLUMNS, stats.columnNames);
+ Assert.assertFalse(stats.columnNames.contains("s1"));
+ Assert.assertFalse(stats.columnNames.contains("s3"));
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void testLiveRecordHandlerWithoutColumnFilterPreservesAllColumns()
throws Exception {
+ final String database = databaseName("default");
+ final String topicName = topicName("default");
+ final String consumerId = consumerName("default");
+ final String consumerGroupId = consumerGroupName("default");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createTopicWithoutColumnFilter(
+ topicName, database, TABLE_NAME,
TopicConstant.FORMAT_RECORD_HANDLER_VALUE);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ insertRows(senderEnv, database, TABLE_NAME, 110, 113);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(110L, 111L, 112L));
+ final ConsumedRecordStats stats =
+ pollRecordMessagesForTimestamps(consumer, expectedTimestamps,
true);
+
+ Assert.assertEquals(expectedTimestamps, stats.timestamps);
+ Assert.assertEquals(EXPECTED_ALL_COLUMNS, stats.columnNames);
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void
testLiveRecordHandlerColumnFilterKeepsHistoricalAndRealtimeConsistent()
+ throws Exception {
+ final String database = databaseName("live_both");
+ final String topicName = topicName("live_both");
+ final String consumerId = consumerName("live_both");
+ final String consumerGroupId = consumerGroupName("live_both");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ insertRows(senderEnv, database, TABLE_NAME, 180, 183);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ COLUMN_FILTER);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+
+ final Set<Long> historicalTimestamps = new
LinkedHashSet<>(Arrays.asList(180L, 181L, 182L));
+ final ConsumedRecordStats historicalStats =
+ pollRecordMessagesForTimestamps(consumer, historicalTimestamps,
true);
+ Assert.assertEquals(EXPECTED_COLUMNS, historicalStats.columnNames);
+
+ insertRows(senderEnv, database, TABLE_NAME, 190, 193);
+
+ final Set<Long> realtimeTimestamps = new
LinkedHashSet<>(Arrays.asList(190L, 191L, 192L));
+ final ConsumedRecordStats realtimeStats =
+ pollRecordMessagesForTimestamps(consumer, realtimeTimestamps,
true);
+ Assert.assertEquals(EXPECTED_COLUMNS, realtimeStats.columnNames);
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void
testLiveRecordHandlerColumnFilterAppliesToMultipleConsumerGroups() throws
Exception {
+ final String database = databaseName("multi_group");
+ final String topicName = topicName("multi_group");
+ final String firstConsumerId = consumerName("multi_group_a");
+ final String firstConsumerGroupId = consumerGroupName("multi_group_a");
+ final String secondConsumerId = consumerName("multi_group_b");
+ final String secondConsumerGroupId = consumerGroupName("multi_group_b");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ COLUMN_FILTER);
+
+ try (final ISubscriptionTablePullConsumer firstConsumer =
+ createConsumer(firstConsumerId, firstConsumerGroupId);
+ final ISubscriptionTablePullConsumer secondConsumer =
+ createConsumer(secondConsumerId, secondConsumerGroupId)) {
+ firstConsumer.subscribe(topicName);
+ secondConsumer.subscribe(topicName);
+ insertRows(senderEnv, database, TABLE_NAME, 170, 173);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(170L, 171L, 172L));
+ final ConsumedRecordStats firstStats =
+ pollRecordMessagesForTimestamps(firstConsumer, expectedTimestamps,
true);
+ final ConsumedRecordStats secondStats =
+ pollRecordMessagesForTimestamps(secondConsumer,
expectedTimestamps, true);
+
+ Assert.assertEquals(EXPECTED_COLUMNS, firstStats.columnNames);
+ Assert.assertEquals(EXPECTED_COLUMNS, secondStats.columnNames);
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void testSnapshotRecordHandlerColumnFilter() throws Exception {
+ final String database = databaseName("snapshot");
+ final String topicName = topicName("snapshot");
+ final String consumerId = consumerName("snapshot");
+ final String consumerGroupId = consumerGroupName("snapshot");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ insertRows(senderEnv, database, TABLE_NAME, 20, 23);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.MODE_SNAPSHOT_VALUE,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ COLUMN_FILTER);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(20L, 21L, 22L));
+ 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 testLiveRecordHandlerMixedCaseInsensitiveColumnFilter() throws
Exception {
+ final String database = databaseName("mixed");
+ final String topicName = topicName("mixed");
+ final String consumerId = consumerName("mixed");
+ final String consumerGroupId = consumerGroupName("mixed");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ MIXED_COLUMN_FILTER);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ insertRows(senderEnv, database, TABLE_NAME, 30, 33);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(30L, 31L, 32L));
+ final ConsumedRecordStats stats =
+ pollRecordMessagesForTimestamps(consumer, expectedTimestamps,
false);
+
+ Assert.assertEquals(expectedTimestamps, stats.timestamps);
+ Assert.assertEquals(EXPECTED_MIXED_COLUMNS, stats.columnNames);
+ Assert.assertFalse(stats.columnNames.contains("time"));
+ Assert.assertFalse(stats.columnNames.contains("s3"));
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void testLiveRecordHandlerComplexOperatorColumnFilter() throws
Exception {
+ final String database = databaseName("operators");
+ final String topicName = topicName("operators");
+ final String consumerId = consumerName("operators");
+ final String consumerGroupId = consumerGroupName("operators");
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME, TABLE_SCHEMA);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ COMPLEX_OPERATOR_FILTER);
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ insertRows(senderEnv, database, TABLE_NAME, 100, 103);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(100L, 101L, 102L));
+ final ConsumedRecordStats stats =
+ pollRecordMessagesForTimestamps(consumer, expectedTimestamps,
false);
+
+ Assert.assertEquals(expectedTimestamps, stats.timestamps);
+ Assert.assertEquals(EXPECTED_COMPLEX_OPERATOR_COLUMNS,
stats.columnNames);
+ Assert.assertFalse(stats.columnNames.contains("time"));
+ Assert.assertFalse(stats.columnNames.contains("s1"));
+ Assert.assertFalse(stats.columnNames.contains("s3"));
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void testLiveRecordHandlerColumnFilterDropsMatchedAttributeColumn()
throws Exception {
+ final String database = databaseName("attribute");
+ final String topicName = topicName("attribute");
+ final String consumerId = consumerName("attribute");
+ final String consumerGroupId = consumerGroupName("attribute");
+ final String schemaWithAttribute =
+ "tag1 STRING TAG, attr1 STRING ATTRIBUTE, s1 INT64 FIELD, s2 DOUBLE
FIELD";
+
+ try {
+ createDatabaseAndTable(senderEnv, database, TABLE_NAME,
schemaWithAttribute);
+ createTopic(
+ topicName,
+ database,
+ TABLE_NAME,
+ TopicConstant.FORMAT_RECORD_HANDLER_VALUE,
+ "category = \"ATTRIBUTE\" OR column_name = \"s2\"");
+
+ try (final ISubscriptionTablePullConsumer consumer =
+ createConsumer(consumerId, consumerGroupId)) {
+ consumer.subscribe(topicName);
+ insertRowsWithAttribute(senderEnv, database, TABLE_NAME, 160, 163);
+
+ final Set<Long> expectedTimestamps = new
LinkedHashSet<>(Arrays.asList(160L, 161L, 162L));
+ final ConsumedRecordStats stats =
+ pollRecordMessagesForTimestamps(consumer, expectedTimestamps,
false);
+
+ Assert.assertEquals(expectedTimestamps, stats.timestamps);
+ Assert.assertEquals(new LinkedHashSet<>(Arrays.asList("tag1", "s2")),
stats.columnNames);
+ Assert.assertFalse(stats.columnNames.contains("attr1"));
+ Assert.assertFalse(stats.columnNames.contains("s1"));
+ }
+ } finally {
+ cleanup(topicName, database);
+ }
+ }
+
+ @Test
+ public void testTreeViewColumnFilterBindingUsesSourceFieldName() throws
Exception {
+ final String database = databaseName("view");
+ final String viewName = TABLE_NAME + "_view";
+ final Map<String, String> attributes = new LinkedHashMap<>();
+ attributes.put("__system.sql-dialect", "table");
+ attributes.put(TopicConstant.DATABASE_KEY, database);
+ attributes.put(TopicConstant.TABLE_KEY, viewName);
+ attributes.put(TopicConstant.COLUMN_FILTER_KEY, "column_name =
\"s_view\"");
+
+ final BoundColumnFilter boundColumnFilter =
+ new ColumnFilterBinder()
+ .bind(
+ new TopicConfig(attributes),
+ Map.of(database, Map.of(viewName,
createTreeViewSchema(viewName))));
+ final ColumnFilterMatcher matcher =
+ ColumnFilterMatcher.fromBoundColumnFilter(boundColumnFilter);
+
+ Assert.assertTrue(matcher.match(database, viewName, "tag1"));
+ Assert.assertTrue(matcher.match(database, viewName, "s_src"));
+ Assert.assertFalse(matcher.match(database, viewName, "s_view"));
+ }
Review Comment:
This does not seem to be an IT?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java:
##########
@@ -93,14 +94,62 @@ public TPushTopicMetaRespExceptionMessage
handleSingleTopicMetaChanges(
private void handleSingleTopicMetaChangesInternal(final TopicMeta
metaFromCoordinator) {
final String topicName = metaFromCoordinator.getTopicName();
- TopicMeta.validateOwnerProgression(
- topicMetaKeeper.getTopicMeta(topicName), metaFromCoordinator);
+ final TopicMeta oldMeta = topicMetaKeeper.getTopicMeta(topicName);
+ TopicMeta.validateOwnerProgression(oldMeta, metaFromCoordinator);
topicMetaKeeper.removeTopicMeta(topicName);
topicMetaKeeper.addTopicMeta(topicName, metaFromCoordinator);
+ if (shouldRefreshColumnFilter(oldMeta, metaFromCoordinator)) {
+ SubscriptionAgent.broker().refreshColumnFilter(topicName,
metaFromCoordinator.getConfig());
+ } else if (!metaFromCoordinator.getConfig().isTableTopic()) {
+ SubscriptionAgent.broker().dropColumnFilter(topicName);
+ }
Review Comment:
Why would a tree topic be added with ColumnFilter in the first place.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java:
##########
@@ -1813,6 +1819,25 @@ private boolean createAndEnqueueEvent(
return true;
}
+ private Map<String, Map<String, Boolean>> getTimeSelectedByTable(
+ final String databaseName, final List<Tablet> tablets) {
+ if (Objects.isNull(databaseName) || Objects.isNull(tablets) ||
tablets.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ final ColumnFilterMatcher matcher =
+ SubscriptionAgent.broker().getColumnFilterMatcher(topicName);
+ final Map<String, Boolean> tableMap = new HashMap<>();
+ for (final Tablet tablet : tablets) {
+ if (Objects.nonNull(tablet) && Objects.nonNull(tablet.getTableName())) {
+ tableMap.put(
+ tablet.getTableName(), matcher.isTimeSelected(databaseName,
tablet.getTableName()));
+ }
+ }
+ return tableMap.isEmpty()
+ ? Collections.emptyMap()
+ : Collections.singletonMap(databaseName, tableMap);
+ }
+
Review Comment:
Why not pass down the matcher directly?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterEvaluator.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.columnfilter;
+
+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.commons.subscription.columnfilter.ColumnMetadata;
+
+import java.util.Locale;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+public class ColumnFilterEvaluator implements CommonQueryAstVisitor<Boolean,
ColumnMetadata> {
+
+ public static boolean evaluate(final Expression expression, final
ColumnMetadata metadata) {
+ return Boolean.TRUE.equals(new ColumnFilterEvaluator().process(expression,
metadata));
+ }
+
+ @Override
+ public Boolean visitNode(final Node node, final ColumnMetadata context) {
+ throw new IllegalArgumentException(
+ "unsupported expression: " + node.getClass().getSimpleName());
+ }
+
+ @Override
+ public Boolean visitBooleanLiteral(final BooleanLiteral node, final
ColumnMetadata context) {
+ return node.getValue();
+ }
+
+ @Override
+ public Boolean visitLogicalExpression(
+ final LogicalExpression node, final ColumnMetadata context) {
+ if (node.getOperator() == LogicalExpression.Operator.AND) {
+ for (final Expression term : node.getTerms()) {
+ if (!process(term, context)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ for (final Expression term : node.getTerms()) {
+ if (process(term, context)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public Boolean visitNotExpression(final NotExpression node, final
ColumnMetadata context) {
+ return !process(node.getValue(), context);
+ }
+
+ @Override
+ public Boolean visitComparisonExpression(
+ final ComparisonExpression node, final ColumnMetadata context) {
+ final String left = fieldValue((Identifier) node.getLeft(), context);
+ final String right = ((StringLiteral) node.getRight()).getValue();
+ final boolean equals = equalsIgnoreCase(left, right);
+ return node.getOperator() == ComparisonExpression.Operator.EQUAL ? equals
: !equals;
+ }
+
+ @Override
+ public Boolean visitInPredicate(final InPredicate node, final ColumnMetadata
context) {
+ final String left = fieldValue((Identifier) node.getValue(), context);
+ for (final Expression expression : ((InListExpression)
node.getValueList()).getValues()) {
+ if (equalsIgnoreCase(left, ((StringLiteral) expression).getValue())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public Boolean visitLikePredicate(final LikePredicate node, final
ColumnMetadata context) {
+ final String left = fieldValue((Identifier) node.getValue(), context);
+ final String pattern = ((StringLiteral) node.getPattern()).getValue();
+ final String escape =
+ node.getEscape().map(expression -> ((StringLiteral)
expression).getValue()).orElse(null);
+ return compileLikePattern(pattern, escape).matcher(left).matches();
+ }
+
+ @Override
+ public Boolean visitFunctionCall(final FunctionCall node, final
ColumnMetadata context) {
+ final String left = fieldValue((Identifier) node.getArguments().get(0),
context);
+ final String pattern = ((StringLiteral)
node.getArguments().get(1)).getValue();
+ return Pattern.compile(pattern,
Pattern.CASE_INSENSITIVE).matcher(left).matches();
+ }
+
+ @Override
+ public Boolean visitIsNullPredicate(final IsNullPredicate node, final
ColumnMetadata context) {
+ return Objects.isNull(fieldValue((Identifier) node.getValue(), context));
+ }
+
+ 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("LIKE escape must be a 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("LIKE pattern ends with escape
character");
+ }
+ return Pattern.compile(regex.toString(), Pattern.CASE_INSENSITIVE);
+ }
Review Comment:
How is LIKE processed by the query engine? Is it possible to reuse it?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterParser.java:
##########
@@ -0,0 +1,274 @@
+/*
+ * 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.columnfilter;
+
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.BooleanLiteral;
+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.NotExpression;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.QualifiedName;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.StringLiteral;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.parser.ParsingException;
+import org.apache.iotdb.db.relational.grammar.sql.ColumnFilterBaseVisitor;
+import org.apache.iotdb.db.relational.grammar.sql.ColumnFilterLexer;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+
+import org.antlr.v4.runtime.BaseErrorListener;
+import org.antlr.v4.runtime.CharStreams;
+import org.antlr.v4.runtime.CommonTokenStream;
+import org.antlr.v4.runtime.DefaultErrorStrategy;
+import org.antlr.v4.runtime.InputMismatchException;
+import org.antlr.v4.runtime.Parser;
+import org.antlr.v4.runtime.RecognitionException;
+import org.antlr.v4.runtime.Recognizer;
+import org.antlr.v4.runtime.Token;
+import org.antlr.v4.runtime.atn.PredictionMode;
+import org.antlr.v4.runtime.tree.TerminalNode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+public class ColumnFilterParser {
+
+ 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*\\(.*");
+ private static final Pattern UNQUOTED_COMPARISON_RIGHT_PATTERN =
+ Pattern.compile("(?is).*(?:!=|<>|=)\\s*[A-Za-z_][A-Za-z_0-9]*\\s*");
+
+ private static final BaseErrorListener ERROR_LISTENER =
+ new BaseErrorListener() {
+ @Override
+ public void syntaxError(
+ final Recognizer<?, ?> recognizer,
+ final Object offendingSymbol,
+ final int line,
+ final int charPositionInLine,
+ final String message,
+ final RecognitionException e) {
+ throw new ParsingException(message, e, line, charPositionInLine + 1);
+ }
+ };
+
+ public Expression parseAndValidate(final String rawColumnFilter) throws
SubscriptionException {
+ try {
+ final Expression expression = parse(rawColumnFilter);
+ ColumnFilterValidator.validate(expression);
+ return expression;
+ } catch (final ParsingException | IllegalArgumentException e) {
+ throw new SubscriptionException(
+ String.format("Invalid column-filter: %s", e.getMessage()), e);
+ }
+ }
+
+ Expression parse(final String rawColumnFilter) {
+ if (rawColumnFilter == null || rawColumnFilter.trim().isEmpty()) {
+ throw new ParsingException("column-filter should not be empty", null, 1,
1);
+ }
+ validateUnsupportedSyntax(rawColumnFilter);
+
+ final ColumnFilterLexer lexer = new
ColumnFilterLexer(CharStreams.fromString(rawColumnFilter));
+ final CommonTokenStream tokenStream = new CommonTokenStream(lexer);
+ final org.apache.iotdb.db.relational.grammar.sql.ColumnFilterParser parser
=
+ new
org.apache.iotdb.db.relational.grammar.sql.ColumnFilterParser(tokenStream);
Review Comment:
ColumnFilterParser -> AntlrColumnFilterParser to avoid name conflict?
##########
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterBinderTest.java:
##########
@@ -0,0 +1,234 @@
+/*
+ * 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.columnfilter;
+
+import org.apache.iotdb.commons.schema.table.TreeViewSchema;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.column.AttributeColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TagColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TimeColumnSchema;
+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.write.record.Tablet;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class ColumnFilterBinderTest {
+
+ @Test
+ public void testBindSnapshotKeepsOnlyBoundTagsAtRuntime() {
+ final BoundColumnFilter boundFilter =
+ new ColumnFilterBinder()
+ .bind(
+ createTableTopicConfig("column_name = \"temperature\""),
+ Collections.singletonMap(
+ "db", Collections.singletonMap("sensors",
createTableSchema())));
+ final ColumnFilterMatcher matcher =
ColumnFilterMatcher.fromBoundColumnFilter(boundFilter);
+
+ final Tablet prunedTablet =
+
TabletColumnPruner.pruneTableModelTablet(createRuntimeTabletWithNewTag(), "db",
matcher);
+
+ Assert.assertNotNull(prunedTablet);
+ Assert.assertEquals(2, prunedTablet.getSchemas().size());
+ Assert.assertEquals("device",
prunedTablet.getSchemas().get(0).getMeasurementName());
+ Assert.assertEquals("temperature",
prunedTablet.getSchemas().get(1).getMeasurementName());
+ }
+
+ @Test
+ public void testAttributeMatchBindsTagsButNotAttribute() {
+ final BoundColumnFilter boundFilter =
+ new ColumnFilterBinder()
+ .bind(
+ createTableTopicConfig("category = \"ATTRIBUTE\""),
+ Collections.singletonMap(
+ "db", Collections.singletonMap("sensors",
createTableSchema())));
Review Comment:
Should we directly reject this during parsing?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/TabletColumnPruner.java:
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.columnfilter;
+
+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.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/** Prunes table-model Tablets according to a subscription column matcher. */
+public class TabletColumnPruner {
+
+ private TabletColumnPruner() {
+ // utility class
+ }
+
+ public static Tablet pruneTableModelTablet(
+ final Tablet tablet, final String databaseName, final
ColumnFilterMatcher matcher) {
+ if (Objects.isNull(tablet) || Objects.isNull(databaseName)) {
+ return tablet;
+ }
+
+ final ColumnFilterMatcher effectiveMatcher =
+ Objects.nonNull(matcher) ? matcher : ColumnFilterMatcher.matchAll();
+ final List<IMeasurementSchema> schemas = tablet.getSchemas();
+ final Object[] values = tablet.getValues();
+ if (Objects.isNull(schemas) || schemas.isEmpty() ||
Objects.isNull(values)) {
+ return null;
+ }
+
+ final List<ColumnCategory> categories = getColumnCategories(tablet,
schemas.size());
+ final boolean[] selectedColumns = new boolean[schemas.size()];
+ boolean hasMatchedColumn = false;
+
+ for (int i = 0; i < schemas.size(); i++) {
+ if (!isValidColumn(schemas, values, i)) {
+ continue;
+ }
+ final IMeasurementSchema schema = schemas.get(i);
+ final ColumnCategory category = categories.get(i);
+ if (effectiveMatcher.match(
+ databaseName,
+ tablet.getTableName(),
+ schema.getMeasurementName(),
+ schema.getType(),
+ category)) {
+ hasMatchedColumn = true;
+ if (category != ColumnCategory.ATTRIBUTE) {
+ selectedColumns[i] = true;
+ }
+ }
+ }
+
+ if (!hasMatchedColumn) {
+ return null;
+ }
+
+ if (effectiveMatcher.shouldAutoRetainTagsAtRuntime()) {
+ for (int i = 0; i < schemas.size(); i++) {
+ if (isValidColumn(schemas, values, i)
+ && categories.get(i) == ColumnCategory.TAG
+ && hasValueArray(values, i)) {
+ selectedColumns[i] = true;
+ }
+ }
+ }
+
+ final List<Integer> selectedIndices = getSelectedIndices(selectedColumns);
+ if (selectedIndices.isEmpty()) {
+ return null;
+ }
+ if (selectedIndices.size() == schemas.size()) {
+ return tablet;
+ }
+
+ final List<IMeasurementSchema> prunedSchemas = new
ArrayList<>(selectedIndices.size());
+ final List<ColumnCategory> prunedCategories = new
ArrayList<>(selectedIndices.size());
+ final Object[] prunedValues = new Object[selectedIndices.size()];
+ final BitMap[] bitMaps = tablet.getBitMaps();
+ final BitMap[] prunedBitMaps =
+ Objects.nonNull(bitMaps) ? new BitMap[selectedIndices.size()] : null;
+
+ for (int i = 0; i < selectedIndices.size(); i++) {
+ final int originalIndex = selectedIndices.get(i);
+ final IMeasurementSchema originalSchema = schemas.get(originalIndex);
+ prunedSchemas.add(
+ new MeasurementSchema(originalSchema.getMeasurementName(),
originalSchema.getType()));
+ prunedCategories.add(categories.get(originalIndex));
+ prunedValues[i] = values[originalIndex];
+ if (Objects.nonNull(bitMaps) && originalIndex < bitMaps.length) {
+ prunedBitMaps[i] = bitMaps[originalIndex];
+ }
+ }
+
+ return new Tablet(
+ tablet.getTableName(),
+ prunedSchemas,
+ prunedCategories,
+ tablet.getTimestamps(),
+ prunedValues,
+ prunedBitMaps,
+ tablet.getRowSize());
+ }
Review Comment:
May add Tablet.pruneByIndex() for future reusing.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/TabletColumnPruner.java:
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.columnfilter;
+
+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.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/** Prunes table-model Tablets according to a subscription column matcher. */
+public class TabletColumnPruner {
+
+ private TabletColumnPruner() {
+ // utility class
+ }
+
+ public static Tablet pruneTableModelTablet(
+ final Tablet tablet, final String databaseName, final
ColumnFilterMatcher matcher) {
+ if (Objects.isNull(tablet) || Objects.isNull(databaseName)) {
+ return tablet;
+ }
+
+ final ColumnFilterMatcher effectiveMatcher =
+ Objects.nonNull(matcher) ? matcher : ColumnFilterMatcher.matchAll();
+ final List<IMeasurementSchema> schemas = tablet.getSchemas();
+ final Object[] values = tablet.getValues();
+ if (Objects.isNull(schemas) || schemas.isEmpty() ||
Objects.isNull(values)) {
+ return null;
+ }
+
+ final List<ColumnCategory> categories = getColumnCategories(tablet,
schemas.size());
+ final boolean[] selectedColumns = new boolean[schemas.size()];
+ boolean hasMatchedColumn = false;
+
+ for (int i = 0; i < schemas.size(); i++) {
+ if (!isValidColumn(schemas, values, i)) {
+ continue;
+ }
+ final IMeasurementSchema schema = schemas.get(i);
+ final ColumnCategory category = categories.get(i);
+ if (effectiveMatcher.match(
+ databaseName,
+ tablet.getTableName(),
+ schema.getMeasurementName(),
+ schema.getType(),
+ category)) {
+ hasMatchedColumn = true;
+ if (category != ColumnCategory.ATTRIBUTE) {
+ selectedColumns[i] = true;
+ }
+ }
+ }
+
+ if (!hasMatchedColumn) {
+ return null;
+ }
+
+ if (effectiveMatcher.shouldAutoRetainTagsAtRuntime()) {
+ for (int i = 0; i < schemas.size(); i++) {
+ if (isValidColumn(schemas, values, i)
+ && categories.get(i) == ColumnCategory.TAG
+ && hasValueArray(values, i)) {
+ selectedColumns[i] = true;
+ }
+ }
+ }
Review Comment:
Merge the last if into the first one?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java:
##########
@@ -93,14 +94,62 @@ public TPushTopicMetaRespExceptionMessage
handleSingleTopicMetaChanges(
private void handleSingleTopicMetaChangesInternal(final TopicMeta
metaFromCoordinator) {
final String topicName = metaFromCoordinator.getTopicName();
- TopicMeta.validateOwnerProgression(
- topicMetaKeeper.getTopicMeta(topicName), metaFromCoordinator);
+ final TopicMeta oldMeta = topicMetaKeeper.getTopicMeta(topicName);
+ TopicMeta.validateOwnerProgression(oldMeta, metaFromCoordinator);
topicMetaKeeper.removeTopicMeta(topicName);
topicMetaKeeper.addTopicMeta(topicName, metaFromCoordinator);
+ if (shouldRefreshColumnFilter(oldMeta, metaFromCoordinator)) {
+ SubscriptionAgent.broker().refreshColumnFilter(topicName,
metaFromCoordinator.getConfig());
+ } else if (!metaFromCoordinator.getConfig().isTableTopic()) {
+ SubscriptionAgent.broker().dropColumnFilter(topicName);
+ }
SubscriptionAgent.broker()
.refreshConsensusQueueOrderMode(topicName,
metaFromCoordinator.getConfig().getOrderMode());
}
+ static boolean shouldRefreshColumnFilter(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(
+ normalizeColumnFilterBindingValue(oldConfig.getColumnFilter()),
+ normalizeColumnFilterBindingValue(newConfig.getColumnFilter()))
+ || !Objects.equals(
+ normalizeColumnFilterBindingValue(
+ getAttributeIgnoreCase(
+ oldConfig, TopicConstant.DATABASE_KEY,
TopicConstant.DATABASE_DEFAULT_VALUE)),
+ normalizeColumnFilterBindingValue(
+ getAttributeIgnoreCase(
+ newConfig, TopicConstant.DATABASE_KEY,
TopicConstant.DATABASE_DEFAULT_VALUE)))
+ || !Objects.equals(
+ normalizeColumnFilterBindingValue(
+ getAttributeIgnoreCase(
+ oldConfig, TopicConstant.TABLE_KEY,
TopicConstant.TABLE_DEFAULT_VALUE)),
+ normalizeColumnFilterBindingValue(
+ 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()
+ .filter(entry -> key.equalsIgnoreCase(entry.getKey()))
+ .map(Map.Entry::getValue)
+ .filter(Objects::nonNull)
+ .findFirst()
+ .orElse(defaultValue);
+ }
Review Comment:
May translate cases before constructing attributes.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterBinder.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.columnfilter;
+
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.commons.schema.table.TreeViewSchema;
+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.commons.subscription.columnfilter.ColumnMetadata;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+public class ColumnFilterBinder {
+
+ private final ColumnFilterParser parser = new ColumnFilterParser();
+
+ public BoundColumnFilter bind(
+ final TopicConfig topicConfig, final Map<String, Map<String, TsTable>>
tables)
+ throws SubscriptionException {
+ if (Objects.isNull(topicConfig)
+ || !topicConfig.isTableTopic()
+ || topicConfig.isColumnFilterTrivial()) {
+ return BoundColumnFilter.matchAll();
+ }
+
+ final Expression expression =
parser.parseAndValidate(topicConfig.getColumnFilter());
+ final TablePattern tablePattern = buildTablePattern(topicConfig);
+ final Map<BoundColumnFilter.TableKey, Set<String>> selectedColumnsByTable
= new HashMap<>();
+ final Map<BoundColumnFilter.TableKey, Boolean> timeSelectedByTable = new
HashMap<>();
+ boolean timeSelected = false;
+
+ for (final Map.Entry<String, Map<String, TsTable>> databaseEntry :
tables.entrySet()) {
+ final String database = databaseEntry.getKey();
+ if (!tablePattern.matchesDatabase(database)) {
+ continue;
+ }
+ for (final TsTable table : databaseEntry.getValue().values()) {
+ if (Objects.isNull(table) ||
!tablePattern.matchesTable(table.getTableName())) {
+ continue;
+ }
+ final BindTableResult tableResult = bindTable(expression, database,
table);
+ timeSelected |= tableResult.timeSelected;
+ timeSelectedByTable.put(
+ BoundColumnFilter.TableKey.of(database, table.getTableName()),
+ tableResult.timeSelected);
+ if (!tableResult.selectedColumnNames.isEmpty()) {
+ selectedColumnsByTable.put(
+ BoundColumnFilter.TableKey.of(database, table.getTableName()),
+ tableResult.selectedColumnNames);
+ }
+ }
+ }
+
+ return BoundColumnFilter.of(selectedColumnsByTable, timeSelected,
timeSelectedByTable);
+ }
+
+ private static BindTableResult bindTable(
+ final Expression expression, final String database, final TsTable table)
{
+ boolean hasMatchedColumn = false;
+ boolean timeSelected = false;
+ final Set<String> selectedColumnNames = new HashSet<>();
+ final Set<String> tagColumnNames = new HashSet<>();
+
+ for (final TsTableColumnSchema columnSchema : table.getColumnList()) {
+ if (Objects.isNull(columnSchema)) {
+ continue;
+ }
+ if (columnSchema.getColumnCategory() == TsTableColumnCategory.TAG) {
+ tagColumnNames.add(BoundColumnFilter.normalize(sourceColumnName(table,
columnSchema)));
+ }
Review Comment:
Continue？
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterEvaluator.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.columnfilter;
+
+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.commons.subscription.columnfilter.ColumnMetadata;
+
+import java.util.Locale;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+public class ColumnFilterEvaluator implements CommonQueryAstVisitor<Boolean,
ColumnMetadata> {
+
+ public static boolean evaluate(final Expression expression, final
ColumnMetadata metadata) {
+ return Boolean.TRUE.equals(new ColumnFilterEvaluator().process(expression,
metadata));
+ }
+
+ @Override
+ public Boolean visitNode(final Node node, final ColumnMetadata context) {
+ throw new IllegalArgumentException(
+ "unsupported expression: " + node.getClass().getSimpleName());
+ }
+
+ @Override
+ public Boolean visitBooleanLiteral(final BooleanLiteral node, final
ColumnMetadata context) {
+ return node.getValue();
+ }
+
+ @Override
+ public Boolean visitLogicalExpression(
+ final LogicalExpression node, final ColumnMetadata context) {
+ if (node.getOperator() == LogicalExpression.Operator.AND) {
+ for (final Expression term : node.getTerms()) {
+ if (!process(term, context)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ for (final Expression term : node.getTerms()) {
+ if (process(term, context)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public Boolean visitNotExpression(final NotExpression node, final
ColumnMetadata context) {
+ return !process(node.getValue(), context);
+ }
+
+ @Override
+ public Boolean visitComparisonExpression(
+ final ComparisonExpression node, final ColumnMetadata context) {
+ final String left = fieldValue((Identifier) node.getLeft(), context);
+ final String right = ((StringLiteral) node.getRight()).getValue();
+ final boolean equals = equalsIgnoreCase(left, right);
+ return node.getOperator() == ComparisonExpression.Operator.EQUAL ? equals
: !equals;
+ }
+
+ @Override
+ public Boolean visitInPredicate(final InPredicate node, final ColumnMetadata
context) {
+ final String left = fieldValue((Identifier) node.getValue(), context);
+ for (final Expression expression : ((InListExpression)
node.getValueList()).getValues()) {
+ if (equalsIgnoreCase(left, ((StringLiteral) expression).getValue())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public Boolean visitLikePredicate(final LikePredicate node, final
ColumnMetadata context) {
+ final String left = fieldValue((Identifier) node.getValue(), context);
+ final String pattern = ((StringLiteral) node.getPattern()).getValue();
+ final String escape =
+ node.getEscape().map(expression -> ((StringLiteral)
expression).getValue()).orElse(null);
+ return compileLikePattern(pattern, escape).matcher(left).matches();
+ }
+
+ @Override
+ public Boolean visitFunctionCall(final FunctionCall node, final
ColumnMetadata context) {
+ final String left = fieldValue((Identifier) node.getArguments().get(0),
context);
+ final String pattern = ((StringLiteral)
node.getArguments().get(1)).getValue();
+ return Pattern.compile(pattern,
Pattern.CASE_INSENSITIVE).matcher(left).matches();
+ }
Review Comment:
Is this for a regex?
Is the function name check?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/BoundColumnFilter.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.columnfilter;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+public class BoundColumnFilter {
+
+ private static final BoundColumnFilter MATCH_ALL = new
BoundColumnFilter(true, false, null, null);
+
+ private final boolean matchAll;
+ private final boolean timeSelected;
+ private final Map<TableKey, Set<String>> selectedColumnsByTable;
+ private final Map<TableKey, Boolean> timeSelectedByTable;
+
+ private BoundColumnFilter(
+ final boolean matchAll,
+ final boolean timeSelected,
+ final Map<TableKey, Set<String>> selectedColumnsByTable,
+ final Map<TableKey, Boolean> timeSelectedByTable) {
+ this.matchAll = matchAll;
+ this.timeSelected = timeSelected;
+ this.selectedColumnsByTable = selectedColumnsByTable;
+ this.timeSelectedByTable = timeSelectedByTable;
+ }
+
+ public static BoundColumnFilter matchAll() {
+ return MATCH_ALL;
+ }
+
+ public static BoundColumnFilter of(
+ final Map<TableKey, Set<String>> selectedColumnsByTable,
+ final boolean timeSelected,
+ final Map<TableKey, Boolean> timeSelectedByTable) {
+ final Map<TableKey, Set<String>> copied = new HashMap<>();
+ selectedColumnsByTable.forEach(
+ (key, value) -> copied.put(key, Collections.unmodifiableSet(new
HashSet<>(value))));
Review Comment:
Check the necessity of copying
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterMatcher.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.columnfilter;
+
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.commons.subscription.columnfilter.ColumnMetadata;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+public class ColumnFilterMatcher {
+
+ private static final ColumnFilterMatcher MATCH_ALL = new
ColumnFilterMatcher(null, null);
+
+ private final BoundColumnFilter boundColumnFilter;
+ private final Set<String> selectedColumnNames;
+ private final Expression expression;
+
+ private ColumnFilterMatcher(final Set<String> selectedColumnNames, final
Expression expression) {
+ this(null, selectedColumnNames, expression);
+ }
+
+ private ColumnFilterMatcher(
+ final BoundColumnFilter boundColumnFilter,
+ final Set<String> selectedColumnNames,
+ final Expression expression) {
+ this.boundColumnFilter = boundColumnFilter;
+ this.selectedColumnNames = selectedColumnNames;
+ this.expression = expression;
+ }
+
+ public static ColumnFilterMatcher matchAll() {
+ return MATCH_ALL;
+ }
+
+ public static ColumnFilterMatcher fromBoundColumnFilter(
+ final BoundColumnFilter boundColumnFilter) {
+ if (Objects.isNull(boundColumnFilter) || boundColumnFilter.isMatchAll()) {
+ return matchAll();
+ }
+ return new ColumnFilterMatcher(boundColumnFilter, null, null);
+ }
+
+ public static ColumnFilterMatcher fromTopicConfig(final TopicConfig
topicConfig)
+ throws SubscriptionException {
+ if (Objects.isNull(topicConfig)
+ || !topicConfig.isTableTopic()
+ || topicConfig.isColumnFilterTrivial()) {
+ return matchAll();
+ }
+ return new ColumnFilterMatcher(
+ null, new
ColumnFilterParser().parseAndValidate(topicConfig.getColumnFilter()));
+ }
+
+ public static ColumnFilterMatcher ofSelectedColumnNames(final Set<String>
selectedColumnNames) {
+ if (Objects.isNull(selectedColumnNames)) {
+ return matchAll();
+ }
+
+ final Set<String> normalizedColumnNames = new HashSet<>();
+ selectedColumnNames.stream()
+ .filter(Objects::nonNull)
+ .map(ColumnFilterMatcher::normalize)
+ .forEach(normalizedColumnNames::add);
+ return new
ColumnFilterMatcher(Collections.unmodifiableSet(normalizedColumnNames), null);
+ }
+
+ public boolean isMatchAll() {
+ return Objects.isNull(boundColumnFilter)
+ && Objects.isNull(selectedColumnNames)
+ && Objects.isNull(expression);
+ }
+
+ public boolean shouldAutoRetainTagsAtRuntime() {
+ return Objects.isNull(boundColumnFilter);
+ }
+
+ public boolean isTimeSelected() {
+ if (isMatchAll()) {
+ return true;
+ }
+ if (Objects.nonNull(boundColumnFilter)) {
+ return boundColumnFilter.isTimeSelected();
+ }
+ if (Objects.nonNull(selectedColumnNames)) {
+ return selectedColumnNames.contains("time");
+ }
Review Comment:
Is the renamed time column properly processed here?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/TabletColumnPruner.java:
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.columnfilter;
+
+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.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/** Prunes table-model Tablets according to a subscription column matcher. */
+public class TabletColumnPruner {
+
+ private TabletColumnPruner() {
+ // utility class
+ }
+
+ public static Tablet pruneTableModelTablet(
+ final Tablet tablet, final String databaseName, final
ColumnFilterMatcher matcher) {
+ if (Objects.isNull(tablet) || Objects.isNull(databaseName)) {
+ return tablet;
+ }
+
+ final ColumnFilterMatcher effectiveMatcher =
+ Objects.nonNull(matcher) ? matcher : ColumnFilterMatcher.matchAll();
+ final List<IMeasurementSchema> schemas = tablet.getSchemas();
Review Comment:
If matchAll, return the tablet immediately.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java:
##########
@@ -458,43 +472,18 @@ private List<Tablet> convertRelationalInsertRowNode(final
RelationalInsertRowNod
final String[] measurements = node.getMeasurements();
final TSDataType[] dataTypes = node.getDataTypes();
final Object[] values = node.getValues();
- final List<Integer> matchedColumnIndices =
- getMatchedTableColumnIndices(
- measurements, dataTypes, values, node.getColumnCategories(),
false);
- if (matchedColumnIndices.isEmpty()) {
- return Collections.emptyList();
- }
-
- final int columnCount = matchedColumnIndices.size();
- final List<String> columnNames = new ArrayList<>(columnCount);
- final List<TSDataType> columnDataTypes = new ArrayList<>(columnCount);
- final List<ColumnCategory> columnTypes = new ArrayList<>(columnCount);
- for (final int originalColIdx : matchedColumnIndices) {
- columnNames.add(measurements[originalColIdx]);
- columnDataTypes.add(dataTypes[originalColIdx]);
- columnTypes.add(toTsFileColumnCategory(node.getColumnCategories(),
originalColIdx));
- }
-
final Tablet tablet =
- new Tablet(
- tableName != null ? tableName : "", columnNames, columnDataTypes,
columnTypes, 1);
- tablet.addTimestamp(0, time);
-
- for (int i = 0; i < columnCount; i++) {
- final int originalColIdx = matchedColumnIndices.get(i);
- final Object value = values[originalColIdx];
- if (value == null) {
- if (tablet.getBitMaps() == null) {
- tablet.initBitMaps();
- }
- tablet.getBitMaps()[i].mark(0);
- } else {
- addValueToTablet(tablet, 0, i, dataTypes[originalColIdx], value);
- }
+ buildTableModelTabletFromRow(
+ tableName, time, measurements, dataTypes, values,
node.getColumnCategories());
+ if (Objects.isNull(tablet)) {
+ return Collections.emptyList();
}
- tablet.setRowSize(1);
- return Collections.singletonList(tablet);
+ final Tablet prunedTablet =
+ TabletColumnPruner.pruneTableModelTablet(tablet, databaseName,
getColumnFilterMatcher());
Review Comment:
Is it possible to prune during the first construction so that no unnecessary
objects will be created and copied.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterMatcher.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.columnfilter;
+
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.commons.subscription.columnfilter.ColumnMetadata;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+public class ColumnFilterMatcher {
+
+ private static final ColumnFilterMatcher MATCH_ALL = new
ColumnFilterMatcher(null, null);
+
+ private final BoundColumnFilter boundColumnFilter;
+ private final Set<String> selectedColumnNames;
+ private final Expression expression;
+
+ private ColumnFilterMatcher(final Set<String> selectedColumnNames, final
Expression expression) {
+ this(null, selectedColumnNames, expression);
+ }
+
+ private ColumnFilterMatcher(
+ final BoundColumnFilter boundColumnFilter,
+ final Set<String> selectedColumnNames,
+ final Expression expression) {
+ this.boundColumnFilter = boundColumnFilter;
+ this.selectedColumnNames = selectedColumnNames;
+ this.expression = expression;
+ }
+
+ public static ColumnFilterMatcher matchAll() {
+ return MATCH_ALL;
+ }
+
+ public static ColumnFilterMatcher fromBoundColumnFilter(
+ final BoundColumnFilter boundColumnFilter) {
+ if (Objects.isNull(boundColumnFilter) || boundColumnFilter.isMatchAll()) {
+ return matchAll();
+ }
+ return new ColumnFilterMatcher(boundColumnFilter, null, null);
+ }
+
+ public static ColumnFilterMatcher fromTopicConfig(final TopicConfig
topicConfig)
+ throws SubscriptionException {
+ if (Objects.isNull(topicConfig)
+ || !topicConfig.isTableTopic()
+ || topicConfig.isColumnFilterTrivial()) {
+ return matchAll();
+ }
+ return new ColumnFilterMatcher(
+ null, new
ColumnFilterParser().parseAndValidate(topicConfig.getColumnFilter()));
+ }
+
+ public static ColumnFilterMatcher ofSelectedColumnNames(final Set<String>
selectedColumnNames) {
+ if (Objects.isNull(selectedColumnNames)) {
+ return matchAll();
+ }
+
+ final Set<String> normalizedColumnNames = new HashSet<>();
+ selectedColumnNames.stream()
+ .filter(Objects::nonNull)
+ .map(ColumnFilterMatcher::normalize)
+ .forEach(normalizedColumnNames::add);
+ return new
ColumnFilterMatcher(Collections.unmodifiableSet(normalizedColumnNames), null);
+ }
+
+ public boolean isMatchAll() {
+ return Objects.isNull(boundColumnFilter)
+ && Objects.isNull(selectedColumnNames)
+ && Objects.isNull(expression);
+ }
+
+ public boolean shouldAutoRetainTagsAtRuntime() {
+ return Objects.isNull(boundColumnFilter);
+ }
+
+ public boolean isTimeSelected() {
+ if (isMatchAll()) {
+ return true;
+ }
+ if (Objects.nonNull(boundColumnFilter)) {
+ return boundColumnFilter.isTimeSelected();
+ }
+ if (Objects.nonNull(selectedColumnNames)) {
+ return selectedColumnNames.contains("time");
+ }
+ return isTimeSelected("", "");
+ }
+
+ public boolean isTimeSelected(final String databaseName, final String
tableName) {
+ if (isMatchAll()) {
+ return true;
+ }
+ if (Objects.nonNull(boundColumnFilter)) {
+ return boundColumnFilter.isTimeSelected(databaseName, tableName);
+ }
+ if (Objects.nonNull(selectedColumnNames)) {
+ return selectedColumnNames.contains("time");
+ }
+ return ColumnFilterEvaluator.evaluate(
+ expression,
+ new ColumnMetadata(
+ normalizeNullable(databaseName),
+ normalizeNullable(tableName),
+ "time",
+ TSDataType.TIMESTAMP.name(),
+ ColumnCategory.TIME.name()));
+ }
+
+ public Map<String, Map<String, Boolean>> getTimeSelectedByTable(final String
databaseName) {
+ if (Objects.isNull(boundColumnFilter) ||
boundColumnFilter.getTimeSelectedByTable().isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final String normalizedDatabaseName =
+ Objects.nonNull(databaseName) ? normalize(databaseName) : null;
+ final Map<String, Map<String, Boolean>> result = new HashMap<>();
+ boundColumnFilter
+ .getTimeSelectedByTable()
+ .forEach(
+ (tableKey, timeSelected) -> {
+ if (Objects.nonNull(normalizedDatabaseName)
+ && !Objects.equals(normalizedDatabaseName,
tableKey.getDatabaseName())) {
+ return;
+ }
+ result
+ .computeIfAbsent(tableKey.getDatabaseName(), ignored -> new
HashMap<>())
+ .put(tableKey.getTableName(), timeSelected);
+ });
+ if (result.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ final Map<String, Map<String, Boolean>> copied = new HashMap<>();
+ result.forEach(
+ (database, tableMap) -> copied.put(database,
Collections.unmodifiableMap(tableMap)));
+ return Collections.unmodifiableMap(copied);
+ }
Review Comment:
Result is already a local variable, why copy it?
##########
iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java:
##########
@@ -89,12 +106,57 @@ public void removeUserData() {
resultSets.forEach(SubscriptionResultSet::removeUserData);
}
+ private static boolean resolveTimeSelected(
+ final Map<String, Map<String, Boolean>> timeSelectedByTable,
+ final boolean defaultTimeSelected,
+ final String databaseName,
+ final Tablet tablet) {
+ if (Objects.isNull(timeSelectedByTable) || timeSelectedByTable.isEmpty()) {
+ return defaultTimeSelected;
+ }
+ final Map<String, Boolean> tableMap =
timeSelectedByTable.get(databaseName);
+ if (Objects.isNull(tablet)) {
+ return defaultTimeSelected;
+ }
Review Comment:
Check tablet before calling map.get()
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java:
##########
@@ -374,41 +372,41 @@ private void validateConsensusProtocolSupport(final
TopicConfig topicConfig)
throw new SubscriptionException(exceptionMessage);
}
- private void validateConsensusTableColumnPattern(final TopicConfig
topicConfig)
- throws SubscriptionException {
- if (!topicConfig.hasAttribute(TopicConstant.COLUMN_KEY)) {
- return;
+ private void validateColumnFilter(final TopicConfig topicConfig) throws
SubscriptionException {
+ int columnFilterKeyCount = 0;
+ for (final String key : topicConfig.getAttribute().keySet()) {
+ if (TopicConstant.COLUMN_FILTER_KEY.equalsIgnoreCase(key)) {
+ columnFilterKeyCount++;
+ }
}
-
- if (!topicConfig.isTableTopic()) {
+ if (columnFilterKeyCount > 1) {
final String exceptionMessage =
String.format(
- "Failed to create or alter topic, %s is only supported for table
topics",
- TopicConstant.COLUMN_KEY);
+ "Failed to create or alter topic, duplicate %s attributes are
not allowed",
+ TopicConstant.COLUMN_FILTER_KEY);
LOGGER.warn(exceptionMessage);
throw new SubscriptionException(exceptionMessage);
}
Review Comment:
Do other keys support duplications?
Better to have consistent behavior.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterParser.java:
##########
@@ -0,0 +1,274 @@
+/*
+ * 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.columnfilter;
+
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.BooleanLiteral;
+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.NotExpression;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.QualifiedName;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.StringLiteral;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.parser.ParsingException;
+import org.apache.iotdb.db.relational.grammar.sql.ColumnFilterBaseVisitor;
+import org.apache.iotdb.db.relational.grammar.sql.ColumnFilterLexer;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+
+import org.antlr.v4.runtime.BaseErrorListener;
+import org.antlr.v4.runtime.CharStreams;
+import org.antlr.v4.runtime.CommonTokenStream;
+import org.antlr.v4.runtime.DefaultErrorStrategy;
+import org.antlr.v4.runtime.InputMismatchException;
+import org.antlr.v4.runtime.Parser;
+import org.antlr.v4.runtime.RecognitionException;
+import org.antlr.v4.runtime.Recognizer;
+import org.antlr.v4.runtime.Token;
+import org.antlr.v4.runtime.atn.PredictionMode;
+import org.antlr.v4.runtime.tree.TerminalNode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+public class ColumnFilterParser {
+
+ 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*\\(.*");
+ private static final Pattern UNQUOTED_COMPARISON_RIGHT_PATTERN =
+ Pattern.compile("(?is).*(?:!=|<>|=)\\s*[A-Za-z_][A-Za-z_0-9]*\\s*");
+
+ private static final BaseErrorListener ERROR_LISTENER =
+ new BaseErrorListener() {
+ @Override
+ public void syntaxError(
+ final Recognizer<?, ?> recognizer,
+ final Object offendingSymbol,
+ final int line,
+ final int charPositionInLine,
+ final String message,
+ final RecognitionException e) {
+ throw new ParsingException(message, e, line, charPositionInLine + 1);
+ }
+ };
+
+ public Expression parseAndValidate(final String rawColumnFilter) throws
SubscriptionException {
+ try {
+ final Expression expression = parse(rawColumnFilter);
+ ColumnFilterValidator.validate(expression);
+ return expression;
+ } catch (final ParsingException | IllegalArgumentException e) {
+ throw new SubscriptionException(
+ String.format("Invalid column-filter: %s", e.getMessage()), e);
+ }
+ }
+
+ Expression parse(final String rawColumnFilter) {
+ if (rawColumnFilter == null || rawColumnFilter.trim().isEmpty()) {
+ throw new ParsingException("column-filter should not be empty", null, 1,
1);
+ }
+ validateUnsupportedSyntax(rawColumnFilter);
+
+ final ColumnFilterLexer lexer = new
ColumnFilterLexer(CharStreams.fromString(rawColumnFilter));
+ final CommonTokenStream tokenStream = new CommonTokenStream(lexer);
+ final org.apache.iotdb.db.relational.grammar.sql.ColumnFilterParser parser
=
+ new
org.apache.iotdb.db.relational.grammar.sql.ColumnFilterParser(tokenStream);
+
+ lexer.removeErrorListeners();
+ lexer.addErrorListener(ERROR_LISTENER);
+ parser.removeErrorListeners();
+ parser.addErrorListener(ERROR_LISTENER);
+ parser.setErrorHandler(
+ new DefaultErrorStrategy() {
+ @Override
+ public Token recoverInline(final Parser recognizer) throws
RecognitionException {
+ if (nextTokensContext == null) {
+ throw new InputMismatchException(recognizer);
+ }
+ throw new InputMismatchException(recognizer, nextTokensState,
nextTokensContext);
+ }
+ });
+
+ try {
+ parser.getInterpreter().setPredictionMode(PredictionMode.SLL);
+ return new AstBuilder().visit(parser.columnFilter());
+ } catch (final ParsingException e) {
+ tokenStream.seek(0);
+ parser.reset();
+ parser.getInterpreter().setPredictionMode(PredictionMode.LL);
+ return new AstBuilder().visit(parser.columnFilter());
+ }
+ }
+
+ private static void validateUnsupportedSyntax(final String rawColumnFilter) {
+ final String trimmedColumnFilter = rawColumnFilter.trim();
+ if ((SINGLE_FIELD_PATTERN.matcher(rawColumnFilter).matches()
+ && !"true".equalsIgnoreCase(trimmedColumnFilter)
+ && !"false".equalsIgnoreCase(trimmedColumnFilter))
+ || FUNCTION_CALL_START_PATTERN.matcher(rawColumnFilter).matches()) {
+ throw new ParsingException("expected column predicate operator", null,
1, 1);
+ }
+ if (UNQUOTED_COMPARISON_RIGHT_PATTERN.matcher(rawColumnFilter).matches()) {
+ throw new ParsingException("expected string literal", null, 1, 1);
+ }
+ for (int i = 0; i < rawColumnFilter.length(); i++) {
+ final char ch = rawColumnFilter.charAt(i);
+ if (ch == '<') {
+ if (i + 1 < rawColumnFilter.length() && rawColumnFilter.charAt(i + 1)
== '>') {
+ i++;
+ continue;
+ }
+ throw new ParsingException("unsupported comparison operator '<'",
null, 1, i + 1);
+ }
+ if (ch == '>') {
+ throw new ParsingException("unsupported comparison operator '>'",
null, 1, i + 1);
+ }
+ if (ch == '+') {
+ throw new ParsingException("unexpected character '+'", null, 1, i + 1);
+ }
+ }
+ }
+
+ private static class AstBuilder extends ColumnFilterBaseVisitor<Expression> {
+
+ @Override
+ public Expression visitColumnFilter(
+ final
org.apache.iotdb.db.relational.grammar.sql.ColumnFilterParser.ColumnFilterContext
+ context) {
+ return visit(context.booleanExpression());
+ }
+
+ @Override
+ public Expression visitPredicateExpression(
+ final org.apache.iotdb.db.relational.grammar.sql.ColumnFilterParser
+ .PredicateExpressionContext
+ context) {
+ return visit(context.predicate());
+ }
+
+ @Override
+ public Expression visitLogicalNot(
+ final
org.apache.iotdb.db.relational.grammar.sql.ColumnFilterParser.LogicalNotContext
+ context) {
+ return new NotExpression(visit(context.booleanExpression()));
+ }
+
+ @Override
+ public Expression visitLogicalBinary(
+ final
org.apache.iotdb.db.relational.grammar.sql.ColumnFilterParser.LogicalBinaryContext
+ context) {
+ final Expression left = visit(context.booleanExpression(0));
+ final Expression right = visit(context.booleanExpression(1));
+ return Objects.nonNull(context.AND())
+ ? LogicalExpression.and(left, right)
+ : LogicalExpression.or(left, right);
+ }
+
+ @Override
+ public Expression visitPredicate(
+ final
org.apache.iotdb.db.relational.grammar.sql.ColumnFilterParser.PredicateContext
+ context) {
+ if (Objects.nonNull(context.booleanValue())) {
+ return visit(context.booleanValue());
+ }
+ if (Objects.nonNull(context.booleanExpression())) {
+ return visit(context.booleanExpression());
+ }
+
+ final Identifier field = toIdentifier(context.field());
+ if (Objects.nonNull(context.comparisonOperator())) {
+ return new ComparisonExpression(
+ Objects.nonNull(context.comparisonOperator().EQ())
+ ? ComparisonExpression.Operator.EQUAL
+ : ComparisonExpression.Operator.NOT_EQUAL,
+ field,
+ toStringLiteral(context.string(0)));
+ }
Review Comment:
What does `validateUnsupportedSyntax` exactly do?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterParser.java:
##########
@@ -0,0 +1,274 @@
+/*
+ * 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.columnfilter;
+
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.BooleanLiteral;
+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.NotExpression;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.QualifiedName;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.StringLiteral;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.parser.ParsingException;
+import org.apache.iotdb.db.relational.grammar.sql.ColumnFilterBaseVisitor;
+import org.apache.iotdb.db.relational.grammar.sql.ColumnFilterLexer;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+
+import org.antlr.v4.runtime.BaseErrorListener;
+import org.antlr.v4.runtime.CharStreams;
+import org.antlr.v4.runtime.CommonTokenStream;
+import org.antlr.v4.runtime.DefaultErrorStrategy;
+import org.antlr.v4.runtime.InputMismatchException;
+import org.antlr.v4.runtime.Parser;
+import org.antlr.v4.runtime.RecognitionException;
+import org.antlr.v4.runtime.Recognizer;
+import org.antlr.v4.runtime.Token;
+import org.antlr.v4.runtime.atn.PredictionMode;
+import org.antlr.v4.runtime.tree.TerminalNode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+public class ColumnFilterParser {
+
+ 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*\\(.*");
+ private static final Pattern UNQUOTED_COMPARISON_RIGHT_PATTERN =
+ Pattern.compile("(?is).*(?:!=|<>|=)\\s*[A-Za-z_][A-Za-z_0-9]*\\s*");
+
+ private static final BaseErrorListener ERROR_LISTENER =
+ new BaseErrorListener() {
+ @Override
+ public void syntaxError(
+ final Recognizer<?, ?> recognizer,
+ final Object offendingSymbol,
+ final int line,
+ final int charPositionInLine,
+ final String message,
+ final RecognitionException e) {
+ throw new ParsingException(message, e, line, charPositionInLine + 1);
+ }
+ };
+
+ public Expression parseAndValidate(final String rawColumnFilter) throws
SubscriptionException {
+ try {
+ final Expression expression = parse(rawColumnFilter);
+ ColumnFilterValidator.validate(expression);
+ return expression;
+ } catch (final ParsingException | IllegalArgumentException e) {
+ throw new SubscriptionException(
+ String.format("Invalid column-filter: %s", e.getMessage()), e);
Review Comment:
i18n, check other places
##########
iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/ColumnFilter.g4:
##########
Review Comment:
ColumnFilter -> SubscriptionColumnFilter
Or add "Subscription" as one package level.
##########
iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/SubscriptionPollResponse.java:
##########
@@ -114,7 +149,71 @@ public static SubscriptionPollResponse deserialize(final
ByteBuffer buffer) {
}
final SubscriptionCommitContext commitContext =
SubscriptionCommitContext.deserialize(buffer);
- return new SubscriptionPollResponse(responseType, payload, commitContext);
+ final boolean timeSelected = buffer.hasRemaining() ?
ReadWriteIOUtils.readBool(buffer) : true;
+ final Map<String, Map<String, Boolean>> timeSelectedByTable =
+ buffer.hasRemaining() ? deserializeTimeSelectedByTable(buffer) :
Collections.emptyMap();
+ return new SubscriptionPollResponse(
+ responseType, payload, commitContext, timeSelected,
timeSelectedByTable);
+ }
+
+ private void serializeTimeSelectedByTable(final DataOutputStream stream)
throws IOException {
+ ReadWriteIOUtils.write(timeSelectedByTable.size(), stream);
+ for (final Map.Entry<String, Map<String, Boolean>> databaseEntry :
+ timeSelectedByTable.entrySet()) {
+ ReadWriteIOUtils.write(databaseEntry.getKey(), stream);
+ ReadWriteIOUtils.write(databaseEntry.getValue().size(), stream);
+ for (final Map.Entry<String, Boolean> tableEntry :
databaseEntry.getValue().entrySet()) {
+ ReadWriteIOUtils.write(tableEntry.getKey(), stream);
+ ReadWriteIOUtils.write(tableEntry.getValue(), stream);
+ }
+ }
+ }
+
+ private static Map<String, Map<String, Boolean>>
deserializeTimeSelectedByTable(
+ final ByteBuffer buffer) {
+ final int databaseSize = ReadWriteIOUtils.readInt(buffer);
+ if (databaseSize <= 0) {
+ return Collections.emptyMap();
+ }
+ final Map<String, Map<String, Boolean>> result = new HashMap<>();
+ for (int i = 0; i < databaseSize; ++i) {
+ final String databaseName = ReadWriteIOUtils.readString(buffer);
+ final int tableSize = ReadWriteIOUtils.readInt(buffer);
+ final Map<String, Boolean> tableMap = new HashMap<>();
+ for (int j = 0; j < tableSize; ++j) {
+ tableMap.put(ReadWriteIOUtils.readString(buffer),
ReadWriteIOUtils.readBool(buffer));
+ }
+ result.put(databaseName, tableMap);
+ }
+ return copyTimeSelectedByTable(result);
Review Comment:
Why copy here and there?
Will the map be modified after analysis?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java:
##########
@@ -313,6 +313,28 @@ public long getInstanceVersion() {
return instanceVersion.get();
}
+ public Map<String, Map<String, TsTable>> getTableSnapshot() {
+ readWriteLock.readLock().lock();
+ try {
+ return databaseTableMap.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry ->
+ entry.getValue().entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ tableEntry -> new
TsTable(tableEntry.getValue()),
+ (left, right) -> right,
+ ConcurrentHashMap::new)),
+ (left, right) -> right,
+ ConcurrentHashMap::new));
+ } finally {
Review Comment:
Is it necessary to use concurrentMap for snapshot?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterParser.java:
##########
@@ -0,0 +1,274 @@
+/*
+ * 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.columnfilter;
+
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.BooleanLiteral;
+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.NotExpression;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.QualifiedName;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.StringLiteral;
+import
org.apache.iotdb.commons.queryengine.plan.relational.sql.parser.ParsingException;
+import org.apache.iotdb.db.relational.grammar.sql.ColumnFilterBaseVisitor;
+import org.apache.iotdb.db.relational.grammar.sql.ColumnFilterLexer;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+
+import org.antlr.v4.runtime.BaseErrorListener;
+import org.antlr.v4.runtime.CharStreams;
+import org.antlr.v4.runtime.CommonTokenStream;
+import org.antlr.v4.runtime.DefaultErrorStrategy;
+import org.antlr.v4.runtime.InputMismatchException;
+import org.antlr.v4.runtime.Parser;
+import org.antlr.v4.runtime.RecognitionException;
+import org.antlr.v4.runtime.Recognizer;
+import org.antlr.v4.runtime.Token;
+import org.antlr.v4.runtime.atn.PredictionMode;
+import org.antlr.v4.runtime.tree.TerminalNode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+public class ColumnFilterParser {
+
+ 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*\\(.*");
+ private static final Pattern UNQUOTED_COMPARISON_RIGHT_PATTERN =
+ Pattern.compile("(?is).*(?:!=|<>|=)\\s*[A-Za-z_][A-Za-z_0-9]*\\s*");
+
+ private static final BaseErrorListener ERROR_LISTENER =
+ new BaseErrorListener() {
+ @Override
+ public void syntaxError(
+ final Recognizer<?, ?> recognizer,
+ final Object offendingSymbol,
+ final int line,
+ final int charPositionInLine,
+ final String message,
+ final RecognitionException e) {
+ throw new ParsingException(message, e, line, charPositionInLine + 1);
+ }
+ };
+
+ public Expression parseAndValidate(final String rawColumnFilter) throws
SubscriptionException {
+ try {
+ final Expression expression = parse(rawColumnFilter);
+ ColumnFilterValidator.validate(expression);
+ return expression;
+ } catch (final ParsingException | IllegalArgumentException e) {
+ throw new SubscriptionException(
+ String.format("Invalid column-filter: %s", e.getMessage()), e);
+ }
+ }
+
+ Expression parse(final String rawColumnFilter) {
+ if (rawColumnFilter == null || rawColumnFilter.trim().isEmpty()) {
+ throw new ParsingException("column-filter should not be empty", null, 1,
1);
+ }
+ validateUnsupportedSyntax(rawColumnFilter);
+
+ final ColumnFilterLexer lexer = new
ColumnFilterLexer(CharStreams.fromString(rawColumnFilter));
+ final CommonTokenStream tokenStream = new CommonTokenStream(lexer);
+ final org.apache.iotdb.db.relational.grammar.sql.ColumnFilterParser parser
=
+ new
org.apache.iotdb.db.relational.grammar.sql.ColumnFilterParser(tokenStream);
+
+ lexer.removeErrorListeners();
+ lexer.addErrorListener(ERROR_LISTENER);
+ parser.removeErrorListeners();
+ parser.addErrorListener(ERROR_LISTENER);
+ parser.setErrorHandler(
+ new DefaultErrorStrategy() {
+ @Override
+ public Token recoverInline(final Parser recognizer) throws
RecognitionException {
+ if (nextTokensContext == null) {
+ throw new InputMismatchException(recognizer);
+ }
+ throw new InputMismatchException(recognizer, nextTokensState,
nextTokensContext);
+ }
+ });
+
+ try {
+ parser.getInterpreter().setPredictionMode(PredictionMode.SLL);
+ return new AstBuilder().visit(parser.columnFilter());
+ } catch (final ParsingException e) {
+ tokenStream.seek(0);
+ parser.reset();
+ parser.getInterpreter().setPredictionMode(PredictionMode.LL);
+ return new AstBuilder().visit(parser.columnFilter());
+ }
+ }
+
+ private static void validateUnsupportedSyntax(final String rawColumnFilter) {
+ final String trimmedColumnFilter = rawColumnFilter.trim();
+ if ((SINGLE_FIELD_PATTERN.matcher(rawColumnFilter).matches()
+ && !"true".equalsIgnoreCase(trimmedColumnFilter)
+ && !"false".equalsIgnoreCase(trimmedColumnFilter))
+ || FUNCTION_CALL_START_PATTERN.matcher(rawColumnFilter).matches()) {
+ throw new ParsingException("expected column predicate operator", null,
1, 1);
+ }
+ if (UNQUOTED_COMPARISON_RIGHT_PATTERN.matcher(rawColumnFilter).matches()) {
+ throw new ParsingException("expected string literal", null, 1, 1);
+ }
+ for (int i = 0; i < rawColumnFilter.length(); i++) {
+ final char ch = rawColumnFilter.charAt(i);
+ if (ch == '<') {
+ if (i + 1 < rawColumnFilter.length() && rawColumnFilter.charAt(i + 1)
== '>') {
+ i++;
+ continue;
+ }
+ throw new ParsingException("unsupported comparison operator '<'",
null, 1, i + 1);
+ }
+ if (ch == '>') {
+ throw new ParsingException("unsupported comparison operator '>'",
null, 1, i + 1);
+ }
+ if (ch == '+') {
+ throw new ParsingException("unexpected character '+'", null, 1, i + 1);
+ }
+ }
Review Comment:
What if these are part of the column name?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/columnfilter/ColumnFilterValidator.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.columnfilter;
+
+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 java.util.Locale;
+import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+public class ColumnFilterValidator implements CommonQueryAstVisitor<Void,
Void> {
+
+ private static final Set<String> LEGAL_FIELDS =
+ Set.of("database", "table_name", "column_name", "datatype", "category");
+ private static final String REGEXP_LIKE = "regexp_like";
+
+ public static void validate(final Expression expression) {
+ new ColumnFilterValidator().process(expression);
+ }
+
+ @Override
+ public Void visitNode(final Node node, final Void context) {
+ throw invalid("unsupported expression: " +
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("only =, !=, and <> comparisons are supported in
column-filter");
+ }
+ requireField(node.getLeft());
+ requireStringLiteral(node.getRight(), "comparison right operand");
+ return null;
+ }
+
+ @Override
+ public Void visitInPredicate(final InPredicate node, final Void context) {
+ requireField(node.getValue());
+ if (!(node.getValueList() instanceof InListExpression)) {
+ throw invalid("IN predicate must use a string literal list");
+ }
+ for (final Expression expression : ((InListExpression)
node.getValueList()).getValues()) {
+ requireStringLiteral(expression, "IN element");
+ }
+ return null;
+ }
+
+ @Override
+ public Void visitLikePredicate(final LikePredicate node, final Void context)
{
+ requireField(node.getValue());
+ final StringLiteral pattern = requireStringLiteral(node.getPattern(),
"LIKE pattern");
+ final String escape =
+ node.getEscape()
+ .map(expression -> requireStringLiteral(expression, "LIKE
escape").getValue())
+ .orElse(null);
+ ColumnFilterEvaluator.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("only REGEXP is supported as regexp_like(field, pattern)");
+ }
Review Comment:
Is it necessary to perform multi-pass validation and parsing? Can we squash
them into one pass?
##########
iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/ColumnFilter.g4:
##########
@@ -0,0 +1,88 @@
+/*
+ * 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.
+ */
+
+grammar ColumnFilter;
+
+options { caseInsensitive = true; }
+
+columnFilter
+ : booleanExpression EOF
+ ;
+
+booleanExpression
+ : NOT booleanExpression #logicalNot
+ | booleanExpression AND booleanExpression #logicalBinary
+ | booleanExpression OR booleanExpression #logicalBinary
+ | predicate
#predicateExpression
+ ;
+
+predicate
+ : booleanValue
+ | field comparisonOperator string
+ | field NOT? IN '(' string (',' string)* ')'
+ | field NOT? LIKE string (ESCAPE string)?
+ | field NOT? REGEXP string
+ | field IS NOT? NULL
+ | '(' booleanExpression ')'
+ ;
+
Review Comment:
Function call is not present in the list.
Check and remove useless logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
