Caideyipi commented on code in PR #17936:
URL: https://github.com/apache/iotdb/pull/17936#discussion_r3410739880
##########
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:
????????? consensus entry ?????????`pruneTableModelTablet` ???? `null`???
`convert()` ?? empty??? `ConsensusPrefetchingQueue.prepareEntry()` ??? `null`??
`accumulateFromPending()` / WAL catch-up ???? `markMaterializedProgress()`
????? cursor??? entry ?? `SubscriptionEvent`??????? ack/commit ????
alter/rebind/restart ??????????? cursor ???? committed progress ???
?? PR ? CI ?????? `IoTDBConsensusSubscriptionFilterTableIT` ???????Ubuntu ?
`testColumnFilteringAlterRebindsConsensusQueue` ? 202/203?Windows ????
202?`testColumnFilteringWithNoMatchedColumnsReturnsNothing` ? 302??????? entry
???? skip/direct-commit ?????????????????????? materialized/?? cursor?
##########
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);
Review Comment:
?? raw string ??????? quoted string/identifier????????????????grammar ?
`string` ???? `QUOTED_IDENTIFIER`??? column-filter ?? `LIKE`/`REGEXP`???
`column_name REGEXP "s[0-9]+"`?`column_name LIKE "temp+%"`???????? `>`
?????????? ANTLR parse ?????? unsupported operator / unexpected `+` ???
?????? lexer token/AST ? unsupported syntax ?????????????? quote-aware??????
`<`?`>`?`+` ? literal ???
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java:
##########
@@ -91,11 +98,18 @@ protected void onTsFileInsertionEvent(final
TsFileInsertionEvent event) {
@Override
protected List<SubscriptionEvent> generateSubscriptionEvents() throws
Exception {
if (batch.isEmpty()) {
- return null;
+ enrichedEvents.clear();
+ return Collections.emptyList();
Review Comment:
???? `Collections.emptyList()` ??? batch ???????????????
`SubscriptionPipeEventBatch.onEvent()` ?? null ??????? true???
`SubscriptionPipeEventBatches` ???? batch ? `regionIdToBatch` ????????
`cleanUp()`???????? `SubscriptionEvent` ???? cleanup?
`PipeTabletEventTsFileBatch` ??? `PipeTabletEventBatch` ?????
`forceAllocate(requestMaxBatchSizeInBytes)`??? `close()` ????? memory
block?column-filter ? TsFile batch ?????????????????????????
`dbTsFilePairs.isEmpty()` ????? close/cleanup ?? batch?
##########
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);
}
- if (!isConsensusBasedTopicConfig(topicConfig)) {
+ if (!topicConfig.hasColumnFilter()) {
+ return;
+ }
+
+ if (!topicConfig.isTableTopic()) {
final String exceptionMessage =
String.format(
- "Failed to create or alter topic, %s is only supported for
consensus table topics",
- TopicConstant.COLUMN_KEY);
+ "Failed to create or alter topic, %s is only supported for table
topics",
+ TopicConstant.COLUMN_FILTER_KEY);
LOGGER.warn(exceptionMessage);
throw new SubscriptionException(exceptionMessage);
}
- final String columnPattern =
- topicConfig.getStringOrDefault(
- TopicConstant.COLUMN_KEY, TopicConstant.COLUMN_DEFAULT_VALUE);
- try {
- Pattern.compile(columnPattern);
- } catch (final PatternSyntaxException e) {
+ if (topicConfig.getColumnFilter().trim().isEmpty()) {
Review Comment:
ConfigNode ?????????? key??? table topic???????????? `column-filter`
??????SQL ??? DataNode ?
`TableConfigTaskVisitor.validateAndNormalizeColumnFilter()` ???
`ColumnFilterParser.parseAndValidate()`?????? ConfigNode create/alter topic RPC
??????????????????
?? DataNode refresh matcher ????? fail closed ? empty matcher???? topic
??/?????????????????? parser/validator ????????? ConfigNode ??????????
--
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]
