This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch delete_with_attribute in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 64105e48cce7ab54d247ec00bda49c1190207c2c Author: Tian Jiang <[email protected]> AuthorDate: Tue Jun 30 18:54:18 2026 +0800 convert attribute filter to DEVICE_IN --- .../iotdb/db/i18n/DataNodeQueryMessages.java | 3 + .../iotdb/db/i18n/DataNodeQueryMessages.java | 2 + .../db/queryengine/plan/analyze/AnalyzeUtils.java | 161 +++++++++++++++++++-- .../relational/analyzer/StatementAnalyzer.java | 2 +- .../plan/relational/sql/ast/DeleteDevice.java | 5 +- .../queryengine/plan/analyze/AnalyzeUtilsTest.java | 5 +- 6 files changed, 163 insertions(+), 15 deletions(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index f870290c09a..9f8c4dd50b8 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -271,6 +271,9 @@ public final class DataNodeQueryMessages { "The table '%s' does not contain a time column"; public static final String THE_OPERATOR_OF_TAG_PREDICATE_MUST_BE_FOR = "The operator of tag predicate must be '=' for "; + public static final String TOO_MANY_DEVICES_MATCHED_BY_ATTRIBUTE_FILTERS_IN_DELETION = + "Too many devices (%d) matched by attribute filters in deletion, limit is %d. " + + "Please remove all attribute filters (%s) or add more attribute filters."; public static final String ONLY_TIME_FILTERS_ARE_SUPPORTED_IN_LAST_QUERY = "Only time filters are supported in LAST query"; public static final String VIEWS_CANNOT_BE_USED_IN_GROUP_BY_TAGS = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index b0dae59e198..6d61d2bef70 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -269,6 +269,8 @@ public final class DataNodeQueryMessages { "表 '%s' 不包含时间列"; public static final String THE_OPERATOR_OF_TAG_PREDICATE_MUST_BE_FOR = "标签谓词的运算符必须为 '='，目标："; + public static final String TOO_MANY_DEVICES_MATCHED_BY_ATTRIBUTE_FILTERS_IN_DELETION = + "删除语句中的属性过滤条件匹配了过多设备（%d），限制为 %d。请移除所有属性过滤条件（%s），或进一步增加属性过滤条件。"; public static final String ONLY_TIME_FILTERS_ARE_SUPPORTED_IN_LAST_QUERY = "LAST 查询中仅支持时间过滤器"; public static final String VIEWS_CANNOT_BE_USED_IN_GROUP_BY_TAGS = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java index 3b5b664def6..aab117c6a0a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java @@ -35,7 +35,9 @@ import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.LogicalExpre import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.LongLiteral; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.NullLiteral; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.StringLiteral; +import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.SymbolReference; 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.service.metric.PerformanceOverviewMetrics; import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeReq; @@ -45,6 +47,10 @@ import org.apache.iotdb.db.protocol.client.ConfigNodeClient; import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager; import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableDeviceSchemaFetcher; +import org.apache.iotdb.db.queryengine.plan.relational.planner.ir.ExpressionRewriter; +import org.apache.iotdb.db.queryengine.plan.relational.planner.ir.ExpressionTreeRewriter; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Delete; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TimeRange; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; @@ -58,6 +64,7 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredica import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.And; +import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.DeviceIn; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.SegmentExactMatch; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -82,6 +89,8 @@ import static org.apache.iotdb.db.queryengine.plan.execution.config.TableConfigT public class AnalyzeUtils { + private static final int ATTRIBUTE_FILTER_DELETE_DEVICE_IN_LIMIT = 1000; + private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS = PerformanceOverviewMetrics.getInstance(); private static final Logger LOGGER = LoggerFactory.getLogger(AnalyzeUtils.class); @@ -372,14 +381,20 @@ public class AnalyzeUtils { // Maybe set by pipe transfer if (Objects.isNull(node.getTableDeletionEntries())) { node.setTableDeletionEntries( - parseExpressions2ModEntries(node.getWhere().orElse(null), table)); + parseExpressions2ModEntries( + node.getWhere().orElse(null), table, databaseName, queryContext)); } } public static List<TableDeletionEntry> parseExpressions2ModEntries( - final Expression expression, final TsTable table) { + final Expression expression, + final TsTable table, + final String databaseName, + final MPPQueryContext queryContext) { return toDisjunctiveNormalForms(expression).stream() - .map(disjunctiveNormalForm -> parsePredicate(disjunctiveNormalForm, table)) + .map( + disjunctiveNormalForm -> + parsePredicate(disjunctiveNormalForm, table, databaseName, queryContext)) .collect(Collectors.toList()); } @@ -447,7 +462,11 @@ public class AnalyzeUtils { return results; } - private static TableDeletionEntry parsePredicate(Expression expression, TsTable table) { + private static TableDeletionEntry parsePredicate( + Expression expression, + TsTable table, + final String databaseName, + final MPPQueryContext queryContext) { if (expression == null) { return new TableDeletionEntry( new DeletionPredicate(table.getTableName()), @@ -458,21 +477,76 @@ public class AnalyzeUtils { expressionQueue.add(expression); DeletionPredicate predicate = new DeletionPredicate(table.getTableName()); TagPredicate tagPredicate = null; + List<Expression> deviceFilterExpressions = null; + List<String> attributeColumns = null; TimeRange timeRange = new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE, true); while (!expressionQueue.isEmpty()) { Expression currExp = expressionQueue.remove(); if (currExp instanceof LogicalExpression) { parseAndPredicate(((LogicalExpression) currExp), expressionQueue); } else if (currExp instanceof ComparisonExpression) { - tagPredicate = + final PredicateParseResult parseResult = parseComparison(((ComparisonExpression) currExp), timeRange, tagPredicate, table); + tagPredicate = parseResult.tagPredicate; + if (parseResult.shouldQueryDevice()) { + if (Objects.isNull(deviceFilterExpressions)) { + deviceFilterExpressions = new ArrayList<>(); + attributeColumns = new ArrayList<>(); + } + deviceFilterExpressions.add(toSymbolReferenceExpression(currExp)); + collectAttributeColumn(attributeColumns, parseResult.attributeColumn); + } else if (parseResult.shouldFilterDevice()) { + if (Objects.isNull(deviceFilterExpressions)) { + deviceFilterExpressions = new ArrayList<>(); + } + deviceFilterExpressions.add(toSymbolReferenceExpression(currExp)); + } } else if (currExp instanceof IsNullPredicate) { - tagPredicate = parseIsNull((IsNullPredicate) currExp, tagPredicate, table); + final PredicateParseResult parseResult = + parseIsNull((IsNullPredicate) currExp, tagPredicate, table); + tagPredicate = parseResult.tagPredicate; + if (parseResult.shouldQueryDevice()) { + if (Objects.isNull(deviceFilterExpressions)) { + deviceFilterExpressions = new ArrayList<>(); + attributeColumns = new ArrayList<>(); + } + deviceFilterExpressions.add(toSymbolReferenceExpression(currExp)); + collectAttributeColumn(attributeColumns, parseResult.attributeColumn); + } else if (parseResult.shouldFilterDevice()) { + if (Objects.isNull(deviceFilterExpressions)) { + deviceFilterExpressions = new ArrayList<>(); + } + deviceFilterExpressions.add(toSymbolReferenceExpression(currExp)); + } } else { throw new SemanticException( DataNodeQueryMessages.UNSUPPORTED_EXPRESSION + currExp + " in " + expression); } } + if (Objects.nonNull(attributeColumns)) { + final Set<IDeviceID> deviceIDs = + TableDeviceSchemaFetcher.getInstance() + .fetchDeviceSchemaForDataQuery( + databaseName, + table.getTableName(), + deviceFilterExpressions, + attributeColumns, + queryContext) + .values() + .stream() + .flatMap(List::stream) + .map(DeviceEntry::getDeviceID) + .collect(Collectors.toSet()); + if (deviceIDs.size() > ATTRIBUTE_FILTER_DELETE_DEVICE_IN_LIMIT) { + throw new SemanticException( + String.format( + DataNodeQueryMessages.TOO_MANY_DEVICES_MATCHED_BY_ATTRIBUTE_FILTERS_IN_DELETION, + deviceIDs.size(), + ATTRIBUTE_FILTER_DELETE_DEVICE_IN_LIMIT, + attributeColumns)); + } + tagPredicate = new DeviceIn(deviceIDs); + } if (tagPredicate != null) { predicate.setTagPredicate(tagPredicate); } @@ -494,7 +568,14 @@ public class AnalyzeUtils { expressionQueue.addAll(expression.getTerms()); } - private static TagPredicate parseIsNull( + private static void collectAttributeColumn( + final List<String> attributeColumns, final String attributeColumn) { + if (Objects.nonNull(attributeColumn) && !attributeColumns.contains(attributeColumn)) { + attributeColumns.add(attributeColumn); + } + } + + private static PredicateParseResult parseIsNull( IsNullPredicate isNullPredicate, TagPredicate oldPredicate, TsTable table) { Expression leftHandExp = isNullPredicate.getValue(); if (!(leftHandExp instanceof Identifier)) { @@ -502,6 +583,11 @@ public class AnalyzeUtils { DataNodeQueryMessages.LEFT_HAND_EXPRESSION_IS_NOT_AN_IDENTIFIER + leftHandExp); } String columnName = ((Identifier) leftHandExp).getValue(); + final TsTableColumnSchema columnSchema = table.getColumnSchema(columnName); + if (Objects.nonNull(columnSchema) + && columnSchema.getColumnCategory().equals(TsTableColumnCategory.ATTRIBUTE)) { + return PredicateParseResult.attribute(columnName, oldPredicate); + } int tagColumnOrdinal = table.getTagColumnOrdinal(columnName); if (tagColumnOrdinal == -1) { throw new SemanticException( @@ -510,7 +596,7 @@ public class AnalyzeUtils { // the first segment is the table name, so + 1 TagPredicate newPredicate = new SegmentExactMatch(null, tagColumnOrdinal + 1); - return combinePredicates(oldPredicate, newPredicate); + return PredicateParseResult.tag(combinePredicates(oldPredicate, newPredicate)); } private static TagPredicate combinePredicates( @@ -525,7 +611,7 @@ public class AnalyzeUtils { return new TagPredicate.And(oldPredicate, newPredicate); } - private static TagPredicate parseComparison( + private static PredicateParseResult parseComparison( ComparisonExpression comparisonExpression, TimeRange timeRange, TagPredicate oldPredicate, @@ -571,10 +657,15 @@ public class AnalyzeUtils { "The operator of time predicate must be <, <=, >, or >=: " + right); } - return oldPredicate; + return PredicateParseResult.time(oldPredicate); } // tag predicate String columnName = identifier.getValue(); + final TsTableColumnSchema columnSchema = table.getColumnSchema(columnName); + if (Objects.nonNull(columnSchema) + && columnSchema.getColumnCategory().equals(TsTableColumnCategory.ATTRIBUTE)) { + return PredicateParseResult.attribute(columnName, oldPredicate); + } int tagColumnOrdinal = table.getTagColumnOrdinal(columnName); if (tagColumnOrdinal == -1) { throw new SemanticException( @@ -582,7 +673,7 @@ public class AnalyzeUtils { } TagPredicate newPredicate = getTagPredicate(comparisonExpression, right, tagColumnOrdinal); - return combinePredicates(oldPredicate, newPredicate); + return PredicateParseResult.tag(combinePredicates(oldPredicate, newPredicate)); } private static String getTimeColumnName(final TsTable table) { @@ -617,6 +708,54 @@ public class AnalyzeUtils { return new SegmentExactMatch(rightHandValue, tagColumnOrdinal + 1); } + private static Expression toSymbolReferenceExpression(final Expression expression) { + return ExpressionTreeRewriter.rewriteWith( + new ExpressionRewriter<>() { + @Override + public Expression rewriteIdentifier( + final Identifier node, + final Void context, + final ExpressionTreeRewriter<Void> treeRewriter) { + return new SymbolReference(node.getValue()); + } + }, + expression); + } + + private static class PredicateParseResult { + private final TagPredicate tagPredicate; + private final String attributeColumn; + private final boolean filterDevice; + + private PredicateParseResult( + final TagPredicate tagPredicate, final String attributeColumn, final boolean filterDevice) { + this.tagPredicate = tagPredicate; + this.attributeColumn = attributeColumn; + this.filterDevice = filterDevice; + } + + private static PredicateParseResult time(final TagPredicate tagPredicate) { + return new PredicateParseResult(tagPredicate, null, false); + } + + private static PredicateParseResult tag(final TagPredicate tagPredicate) { + return new PredicateParseResult(tagPredicate, null, true); + } + + private static PredicateParseResult attribute( + final String attributeColumn, final TagPredicate tagPredicate) { + return new PredicateParseResult(tagPredicate, attributeColumn, true); + } + + private boolean shouldQueryDevice() { + return Objects.nonNull(attributeColumn); + } + + private boolean shouldFilterDevice() { + return filterDevice; + } + } + public interface DataPartitionQueryFunc { DataPartition queryDataPartition( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java index 6bc5e2ab4f2..e9295240b7b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java @@ -686,7 +686,7 @@ public class StatementAnalyzer { final TsTable table = DataNodeTableCache.getInstance().getTable(node.getDatabase(), node.getTableName()); DataNodeTreeViewSchemaUtils.checkTableInWrite(node.getDatabase(), table); - node.parseModEntries(table); + node.parseModEntries(table, node.getDatabase(), queryContext); analyzeTraverseDevice(node, context, node.getWhere().isPresent()); node.parseWhere( null, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/DeleteDevice.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/DeleteDevice.java index d2fa9bbabde..39bc2ecfed5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/DeleteDevice.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/DeleteDevice.java @@ -94,8 +94,9 @@ public class DeleteDevice extends AbstractTraverseDevice { return mayDeleteDevice; } - public void parseModEntries(final TsTable table) { - modEntries = AnalyzeUtils.parseExpressions2ModEntries(where, table); + public void parseModEntries( + final TsTable table, final String databaseName, final MPPQueryContext context) { + modEntries = AnalyzeUtils.parseExpressions2ModEntries(where, table, databaseName, context); } public void serializeModEntries(final DataOutputStream stream) throws IOException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java index bbeaa629d4f..ace46de6d0d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java @@ -34,6 +34,8 @@ import org.apache.iotdb.commons.schema.table.column.TimeColumnSchema; import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeReq; import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeResp; import org.apache.iotdb.db.protocol.client.ConfigNodeClient; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Delete; import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; @@ -66,7 +68,8 @@ public class AnalyzeUtilsTest { new Identifier("ts"), new LongLiteral("100")); - List<TableDeletionEntry> entries = AnalyzeUtils.parseExpressions2ModEntries(expression, table); + List<TableDeletionEntry> entries = + AnalyzeUtils.parseExpressions2ModEntries(expression, table, "db", new MPPQueryContext(new QueryId("1"))); assertEquals(1, entries.size()); assertEquals(Long.MIN_VALUE, entries.get(0).getStartTime());
