Caideyipi commented on code in PR #17936: URL: https://github.com/apache/iotdb/pull/17936#discussion_r3464928614
########## 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: The copy is intentional. `BoundColumnFilter` is cached by the broker and reused by matchers, so it should be an immutable snapshot of the binder result rather than retaining caller-owned mutable maps/sets. ########## 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: Yes, this is intentional. Null column schemas are skipped so one malformed/unknown column does not fail binding for the whole table; only valid matched columns contribute to the bound result. ########## 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: `REGEXP` is represented internally as a `FunctionCall(regexp_like, ...)`. The validator checks the function name, argument count, and pattern before evaluation, so the evaluator only sees validated `regexp_like` calls. ########## 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: The query engine LIKE path has execution/runtime dependencies, while this evaluator only matches a few in-memory metadata strings. I kept a small SQL-like pattern compiler here to avoid coupling subscription metadata filtering to query execution internals. ########## 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: Yes. Binding evaluates the filter against each table schema, including the visible time column name, and records per-table time selection. The existing binder test covers `time` versus a renamed `event_time` table. -- 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]
