Caideyipi commented on code in PR #17936:
URL: https://github.com/apache/iotdb/pull/17936#discussion_r3464925929
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java:
##########
@@ -512,51 +501,32 @@ private List<Tablet>
convertRelationalInsertTabletNode(final RelationalInsertTab
final String[] measurements = node.getMeasurements();
final TSDataType[] dataTypes = node.getDataTypes();
- final long[] times = node.getTimes();
- final Object[] columns = node.getColumns();
- final BitMap[] bitMaps = node.getBitMaps();
- final int rowCount = node.getRowCount();
- final List<Integer> matchedColumnIndices =
- getMatchedTableColumnIndices(
- measurements, dataTypes, columns, node.getColumnCategories(),
true);
- if (matchedColumnIndices.isEmpty()) {
+ if (Objects.isNull(measurements)
+ || Objects.isNull(dataTypes)
+ || Objects.isNull(node.getColumns())) {
Review Comment:
Fixed. `convertRelationalInsertTabletNode` now validates null/misaligned
measurement/type/column arrays, skips malformed columns with warning logs,
keeps schemas/columns/bitmaps aligned, and returns empty when no valid column
remains. Added a regression test for malformed tablet-node columns.
##########
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:
Removed that binder-only case from the IT. The binding-only coverage stays
in unit tests.
##########
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:
Added the actual subscription check: the IT now creates a consumer, inserts
tree rows, polls record messages, and asserts that only the expected source
column/tag are delivered while the view alias and unrelated field are absent.
##########
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:
Fixed. The serialized fields are no longer declared `transient`.
##########
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:
Kept a defensive immutable copy at the response boundary and normalize
db/table keys while copying. This avoids caller-side mutation after analysis
and lets the consumer side use direct lookups.
--
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]
