diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java index d8f1781bdaf6d..64df0f56e5cd2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java @@ -232,11 +232,16 @@ public class AnalyzeVisitor extends StatementVisitor private final IPartitionFetcher partitionFetcher; private final ISchemaFetcher schemaFetcher; private final IModelFetcher modelFetcher; + private final TreeAnalysisMutationJournal mutationJournal; - public AnalyzeVisitor(IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher) { + public AnalyzeVisitor( + IPartitionFetcher partitionFetcher, + ISchemaFetcher schemaFetcher, + TreeAnalysisMutationJournal mutationJournal) { this.partitionFetcher = partitionFetcher; this.schemaFetcher = schemaFetcher; this.modelFetcher = ModelFetcher.getInstance(); + this.mutationJournal = mutationJournal; } @Override @@ -280,7 +285,18 @@ public Analysis visitQuery(QueryStatement queryStatement, MPPQueryContext contex try { // check for semantic errors - queryStatement.semanticCheck(); + boolean originalCountTimeAggregation = queryStatement.isCountTimeAggregation(); + try { + queryStatement.semanticCheck(); + } finally { + // semanticCheck derives this flag from SELECT and writes it into QueryStatement. Record the + // rare actual change so a failed attempt or dispatch retry restores the parser-owned state; + // ordinary queries pay no journal entry or copy cost. + if (queryStatement.isCountTimeAggregation() != originalCountTimeAggregation) { + mutationJournal.recordCountTimeAggregationChange( + queryStatement, originalCountTimeAggregation); + } + } context.initResultSetColumnMemoryTracking( queryStatement.getSeriesLimit(), @@ -311,7 +327,13 @@ public Analysis visitQuery(QueryStatement queryStatement, MPPQueryContext contex if (deviceList.size() > 1 && TemplatedAnalyze.canBuildPlanUseTemplate( - analysis, queryStatement, partitionFetcher, schemaTree, context, deviceList)) { + analysis, + queryStatement, + partitionFetcher, + schemaTree, + context, + deviceList, + mutationJournal)) { // when device size is less than 1, there is no need to use template optimization, i.e. no // need to extract common variables return analysis; @@ -319,6 +341,7 @@ public Analysis visitQuery(QueryStatement queryStatement, MPPQueryContext contex if (canPushDownLimitOffsetInGroupByTimeForDevice(queryStatement)) { // remove the device which won't appear in resultSet after limit/offset + mutationJournal.recordLimitOffsetPushDown(queryStatement); deviceList = pushDownLimitOffsetInGroupByTimeForDevice( deviceList, queryStatement, context.getZoneId()); @@ -460,7 +483,10 @@ private ISchemaTree analyzeSchema( queryStatement = (QueryStatement) concatPathRewriter.rewrite( - queryStatement, new PathPatternTree(queryStatement.useWildcard()), context); + queryStatement, + new PathPatternTree(queryStatement.useWildcard()), + context, + mutationJournal); analysis.setRealStatement(queryStatement); // request schema fetch API @@ -521,24 +547,26 @@ private void analyzeGlobalTimeFilter(Analysis analysis, QueryStatement queryStat Expression globalTimePredicate = null; boolean hasValueFilter = false; if (queryStatement.getWhereCondition() != null) { - WhereCondition whereCondition = queryStatement.getWhereCondition(); - Expression predicate = whereCondition.getPredicate(); - - Pair resultPair = - PredicateUtils.extractGlobalTimePredicate(predicate, true, true); - globalTimePredicate = resultPair.left; + Expression predicate = queryStatement.getWhereCondition().getPredicate(); + PredicateUtils.GlobalTimePredicateExtractionResult extractionResult = + PredicateUtils.extractGlobalTimePredicate(predicate); + globalTimePredicate = extractionResult.getGlobalTimePredicate(); if (globalTimePredicate != null) { globalTimePredicate = PredicateUtils.predicateRemoveNot(globalTimePredicate); } - hasValueFilter = resultPair.right; - - predicate = PredicateUtils.simplifyPredicate(predicate); - - // set where condition to null if predicate is true or time filter. - if (!hasValueFilter || predicate.equals(ConstantOperand.TRUE)) { - queryStatement.setWhereCondition(null); - } else { - whereCondition.setPredicate(predicate); + hasValueFilter = extractionResult.hasValueFilter(); + + Expression residualPredicate = + PredicateUtils.simplifyPredicate(extractionResult.getResidualPredicate()); + + // Keep the parser-owned predicate untouched. The journal only saves its wrapper reference, + // while the analyzer installs a residual working predicate built by path-copying the nodes + // changed by time-filter extraction and simplification. + if (!hasValueFilter || residualPredicate.equals(ConstantOperand.TRUE)) { + mutationJournal.replaceWhereCondition(queryStatement, null); + } else if (residualPredicate != predicate) { + mutationJournal.replaceWhereCondition( + queryStatement, WhereCondition.fromNormalizedPredicate(residualPredicate)); } } analysis.setGlobalTimePredicate(globalTimePredicate); @@ -1810,7 +1838,7 @@ private void analyzeOrderByBase( orderByExpressions.add(orderByExpression); } analysis.setOrderByExpressions(orderByExpressions); - queryStatement.updateSortItems(orderByExpressions); + mutationJournal.updateSortItems(queryStatement, orderByExpressions); } private void analyzeOrderBy( @@ -1986,7 +2014,7 @@ private void analyzeDeviceToOrderBy( } analysis.setOrderByExpressions(deviceViewOrderByExpression); - queryStatement.updateSortItems(deviceViewOrderByExpression); + mutationJournal.updateSortItems(queryStatement, deviceViewOrderByExpression); analysis.setDeviceToSortItems(deviceToSortItems); analysis.setDeviceToOrderByExpressions(deviceToOrderByExpressions); } @@ -2315,7 +2343,7 @@ private void analyzeInto( if (!queryStatement.isSelectInto()) { return; } - queryStatement.setOrderByComponent(null); + mutationJournal.clearOrderByComponent(queryStatement); List sourceDevices = new ArrayList<>(deviceSet); List sourceColumns = @@ -2373,7 +2401,7 @@ private void analyzeInto( if (!queryStatement.isSelectInto()) { return; } - queryStatement.setOrderByComponent(null); + mutationJournal.clearOrderByComponent(queryStatement); List sourceColumns = outputExpressions.stream() @@ -3267,19 +3295,19 @@ public Analysis visitShowTTL(ShowTTLStatement showTTLStatement, MPPQueryContext private void analyzeGlobalTimeConditionInShowMetaData( WhereCondition timeCondition, Analysis analysis) { Expression predicate = timeCondition.getPredicate(); - Pair resultPair = - PredicateUtils.extractGlobalTimePredicate(predicate, true, true); - if (resultPair.right) { + PredicateUtils.GlobalTimePredicateExtractionResult extractionResult = + PredicateUtils.extractGlobalTimePredicate(predicate); + if (extractionResult.hasValueFilter()) { throw new SemanticException( DataNodeQueryMessages .VALUE_FILTER_CAN_T_EXIST_IN_THE_CONDITION_OF_SHOW_COUNT_CLAUSE_ONLY_TIME_CONDITION); } - if (resultPair.left == null) { + if (extractionResult.getGlobalTimePredicate() == null) { throw new SemanticException( DataNodeQueryMessages .TIME_CONDITION_CAN_T_BE_EMPTY_IN_THE_CONDITION_OF_SHOW_COUNT_CLAUSE); } - Expression globalTimePredicate = resultPair.left; + Expression globalTimePredicate = extractionResult.getGlobalTimePredicate(); globalTimePredicate = PredicateUtils.predicateRemoveNot(globalTimePredicate); analysis.setGlobalTimePredicate(globalTimePredicate); } @@ -3859,12 +3887,10 @@ public Analysis visitShowQueries( analysis.setSourceExpressions(sourceExpressions); sourceExpressions.forEach(expression -> analyzeExpressionType(analysis, expression)); - if (!analyzeWhereForShowStatements( + analyzeWhereForShowStatements( analysis, showQueriesStatement.getWhereCondition(), - ColumnHeaderConstant.showQueriesColumnHeaders)) { - showQueriesStatement.setWhereCondition(null); - } + ColumnHeaderConstant.showQueriesColumnHeaders); analysis.setMergeOrderParameter(new OrderByParameter(showQueriesStatement.getSortItemList())); @@ -3894,12 +3920,10 @@ public Analysis visitShowDiskUsage( analysis.setSourceExpressions(sourceExpressions); sourceExpressions.forEach(expression -> analyzeExpressionType(analysis, expression)); - if (!analyzeWhereForShowStatements( + analyzeWhereForShowStatements( analysis, showDiskUsageStatement.getWhereCondition(), - ColumnHeaderConstant.showDiskUsageColumnHeaders)) { - showDiskUsageStatement.setWhereCondition(null); - } + ColumnHeaderConstant.showDiskUsageColumnHeaders); analysis.setMergeOrderParameter(new OrderByParameter(showDiskUsageStatement.getSortItemList())); @@ -3927,17 +3951,15 @@ private boolean analyzeWhereForShowStatements( Expression predicate = ExpressionAnalyzer.bindTypeForTimeSeriesOperand( whereCondition.getPredicate(), statementColumnHeaders); - Pair resultPair = - PredicateUtils.extractGlobalTimePredicate(predicate, true, true); - boolean hasValueFilter = resultPair.getRight(); + PredicateUtils.GlobalTimePredicateExtractionResult extractionResult = + PredicateUtils.extractGlobalTimePredicate(predicate); + boolean hasValueFilter = extractionResult.hasValueFilter(); - predicate = PredicateUtils.simplifyPredicate(predicate); + predicate = PredicateUtils.simplifyPredicate(extractionResult.getResidualPredicate()); // set where condition to null if predicate is true or don't have value filter if (!hasValueFilter || predicate.equals(ConstantOperand.TRUE)) { return false; - } else { - whereCondition.setPredicate(predicate); } TSDataType outputType = analyzeExpressionType(analysis, predicate); if (outputType != TSDataType.BOOLEAN) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analyzer.java index b3517a6086f0d..4ef11925f26ce 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analyzer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analyzer.java @@ -33,17 +33,28 @@ public class Analyzer { private final IPartitionFetcher partitionFetcher; private final ISchemaFetcher schemaFetcher; + private final TreeAnalysisMutationJournal mutationJournal; public Analyzer( MPPQueryContext context, IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher) { + this(context, partitionFetcher, schemaFetcher, new TreeAnalysisMutationJournal()); + } + + public Analyzer( + MPPQueryContext context, + IPartitionFetcher partitionFetcher, + ISchemaFetcher schemaFetcher, + TreeAnalysisMutationJournal mutationJournal) { this.context = context; this.partitionFetcher = partitionFetcher; this.schemaFetcher = schemaFetcher; + this.mutationJournal = mutationJournal; } public Analysis analyze(Statement statement) { long startTime = System.nanoTime(); - AnalyzeVisitor visitor = new AnalyzeVisitor(partitionFetcher, schemaFetcher); + mutationJournal.prepareForAnalyze(); + AnalyzeVisitor visitor = new AnalyzeVisitor(partitionFetcher, schemaFetcher, mutationJournal); Analysis analysis = null; context.setReserveMemoryForSchemaTreeFunc( mem -> { @@ -53,6 +64,9 @@ public Analysis analyze(Statement statement) { }); try { analysis = visitor.process(statement, context); + } catch (RuntimeException | Error e) { + mutationJournal.rollback(); + throw e; } finally { if (analysis != null && context.releaseSchemaTreeAfterAnalyzing()) { analysis.setSchemaTree(null); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ConcatPathRewriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ConcatPathRewriter.java index d7c3505c678fe..a072bbcd8ec82 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ConcatPathRewriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ConcatPathRewriter.java @@ -51,7 +51,10 @@ public PathPatternTree getPatternTree() { } public Statement rewrite( - Statement statement, PathPatternTree patternTree, MPPQueryContext queryContext) + Statement statement, + PathPatternTree patternTree, + MPPQueryContext queryContext, + TreeAnalysisMutationJournal mutationJournal) throws StatementAnalyzeException { QueryStatement queryStatement = (QueryStatement) statement; this.patternTree = patternTree; @@ -79,20 +82,20 @@ public Statement rewrite( // concat SELECT with FROM List resultColumns = concatSelectWithFrom(queryStatement.getSelectComponent(), prefixPaths, queryContext); - queryStatement.getSelectComponent().setResultColumns(resultColumns); + mutationJournal.replaceResultColumns(queryStatement, resultColumns); // concat GROUP BY with FROM if (queryStatement.hasGroupByExpression()) { - queryStatement - .getGroupByComponent() - .setControlColumnExpression( - contactGroupByWithFrom( - queryStatement.getGroupByComponent().getControlColumnExpression(), - prefixPaths, - queryContext)); + mutationJournal.replaceGroupByControlExpression( + queryStatement, + contactGroupByWithFrom( + queryStatement.getGroupByComponent().getControlColumnExpression(), + prefixPaths, + queryContext)); } if (queryStatement.hasOrderByExpression()) { - List sortItemExpressions = queryStatement.getExpressionSortItemList(); + List sortItemExpressions = + mutationJournal.getMutableOrderByComponent(queryStatement).getExpressionSortItemList(); sortItemExpressions.replaceAll( expression -> contactOrderByWithFrom(expression, prefixPaths, queryContext)); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/PredicateUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/PredicateUtils.java index fc5e9329f8a1b..edf52063588a5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/PredicateUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/PredicateUtils.java @@ -47,7 +47,6 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.read.filter.basic.Filter; -import org.apache.tsfile.utils.Pair; import java.time.ZoneId; import java.util.ArrayList; @@ -66,88 +65,85 @@ private PredicateUtils() { } /** - * Extract global time predicate from query predicate. + * Extracts the global time predicate and the predicate that still needs to be evaluated. + * + *

This method never modifies {@code predicate}. A {@code null} global time predicate + * represents {@code TRUE}. The returned predicates satisfy the following invariant: + * + * 

+   * predicate == globalTimePredicate AND residualPredicate
+   *
+ * + * where a {@code null} global time predicate is omitted from the conjunction. * * @param predicate raw query predicate - * @param canRewrite determined by the father of current expression - * @param isFirstOr whether it is the first LogicOrExpression encountered - * @return global time predicate + * @return the extracted global time predicate, residual predicate and value-filter flag * @throws UnknownExpressionTypeException unknown expression type */ - public static Pair extractGlobalTimePredicate( - Expression predicate, boolean canRewrite, boolean isFirstOr) { + public static GlobalTimePredicateExtractionResult extractGlobalTimePredicate( + Expression predicate) { if (predicate.getExpressionType().equals(ExpressionType.LOGIC_AND)) { - Pair leftResultPair = - extractGlobalTimePredicate( - ((BinaryExpression) predicate).getLeftExpression(), canRewrite, isFirstOr); - Pair rightResultPair = - extractGlobalTimePredicate( - ((BinaryExpression) predicate).getRightExpression(), canRewrite, isFirstOr); - - // rewrite predicate to avoid duplicate calculation on time filter - // If Left-child or Right-child does not contain value filter - // We can set it to true in Predicate Tree - if (canRewrite) { - if (leftResultPair.left != null && !leftResultPair.right) { - ((BinaryExpression) predicate) - .setLeftExpression(new ConstantOperand(TSDataType.BOOLEAN, "true")); - } - if (rightResultPair.left != null && !rightResultPair.right) { - ((BinaryExpression) predicate) - .setRightExpression(new ConstantOperand(TSDataType.BOOLEAN, "true")); - } - } - - if (leftResultPair.left != null && rightResultPair.left != null) { - return new Pair<>( - ExpressionFactory.and(leftResultPair.left, rightResultPair.left), - leftResultPair.right || rightResultPair.right); - } else if (leftResultPair.left != null) { - return new Pair<>(leftResultPair.left, true); - } else if (rightResultPair.left != null) { - return new Pair<>(rightResultPair.left, true); - } - return new Pair<>(null, true); + GlobalTimePredicateExtractionResult leftResult = + extractGlobalTimePredicate(((BinaryExpression) predicate).getLeftExpression()); + GlobalTimePredicateExtractionResult rightResult = + extractGlobalTimePredicate(((BinaryExpression) predicate).getRightExpression()); + + return new GlobalTimePredicateExtractionResult( + combineConjuncts( + leftResult.getGlobalTimePredicate(), rightResult.getGlobalTimePredicate(), null), + combineConjuncts( + leftResult.getResidualPredicate(), + rightResult.getResidualPredicate(), + ConstantOperand.TRUE), + leftResult.hasValueFilter() || rightResult.hasValueFilter()); } else if (predicate.getExpressionType().equals(ExpressionType.LOGIC_OR)) { - Pair leftResultPair = - extractGlobalTimePredicate( - ((BinaryExpression) predicate).getLeftExpression(), false, false); - Pair rightResultPair = - extractGlobalTimePredicate( - ((BinaryExpression) predicate).getRightExpression(), false, false); - - if (leftResultPair.left != null && rightResultPair.left != null) { - if (Boolean.TRUE.equals(isFirstOr && !leftResultPair.right && !rightResultPair.right)) { - ((BinaryExpression) predicate) - .setLeftExpression(new ConstantOperand(TSDataType.BOOLEAN, "true")); - ((BinaryExpression) predicate) - .setRightExpression(new ConstantOperand(TSDataType.BOOLEAN, "true")); + GlobalTimePredicateExtractionResult leftResult = + extractGlobalTimePredicate(((BinaryExpression) predicate).getLeftExpression()); + GlobalTimePredicateExtractionResult rightResult = + extractGlobalTimePredicate(((BinaryExpression) predicate).getRightExpression()); + + if (leftResult.getGlobalTimePredicate() != null + && rightResult.getGlobalTimePredicate() != null) { + Expression globalTimePredicate = + ExpressionFactory.or( + leftResult.getGlobalTimePredicate(), rightResult.getGlobalTimePredicate()); + boolean hasValueFilter = leftResult.hasValueFilter() || rightResult.hasValueFilter(); + if (!hasValueFilter) { + return new GlobalTimePredicateExtractionResult( + globalTimePredicate, ConstantOperand.TRUE, false); } - return new Pair<>( - ExpressionFactory.or(leftResultPair.left, rightResultPair.left), - leftResultPair.right || rightResultPair.right); + + // (G1 AND R1) OR (G2 AND R2) implies G1 OR G2, but R1 OR R2 is not an + // equivalent residual. Keeping the original OR predicate preserves + // P == (G1 OR G2) AND P without copying or mutating its subtrees. + return new GlobalTimePredicateExtractionResult(globalTimePredicate, predicate, true); } - return new Pair<>(null, true); + return new GlobalTimePredicateExtractionResult(null, predicate, true); } else if (predicate.getExpressionType().equals(ExpressionType.LOGIC_NOT)) { - Pair childResultPair = - extractGlobalTimePredicate( - ((UnaryExpression) predicate).getExpression(), canRewrite, isFirstOr); - if (childResultPair.left != null) { - return new Pair<>(ExpressionFactory.not(childResultPair.left), childResultPair.right); + GlobalTimePredicateExtractionResult childResult = + extractGlobalTimePredicate(((UnaryExpression) predicate).getExpression()); + if (childResult.getGlobalTimePredicate() != null && !childResult.hasValueFilter()) { + return new GlobalTimePredicateExtractionResult( + ExpressionFactory.not(childResult.getGlobalTimePredicate()), + ConstantOperand.TRUE, + false); } - return new Pair<>(null, true); + // NOT(G AND R) cannot in general be represented as NOT(G) AND NOT(R). Do + // not extract a time predicate when the negated subtree contains a value + // filter. + return new GlobalTimePredicateExtractionResult(null, predicate, true); } else if (predicate.isCompareBinaryExpression()) { Expression leftExpression = ((BinaryExpression) predicate).getLeftExpression(); Expression rightExpression = ((BinaryExpression) predicate).getRightExpression(); if (checkIsTimeFilter(leftExpression, rightExpression) || checkIsTimeFilter(rightExpression, leftExpression)) { - return new Pair<>(predicate, false); + return new GlobalTimePredicateExtractionResult(predicate, ConstantOperand.TRUE, false); } - return new Pair<>(null, true); + return new GlobalTimePredicateExtractionResult(null, predicate, true); } else if (predicate.getExpressionType().equals(ExpressionType.LIKE) || predicate.getExpressionType().equals(ExpressionType.REGEXP)) { // time filter don't support LIKE and REGEXP - return new Pair<>(null, true); + return new GlobalTimePredicateExtractionResult(null, predicate, true); } else if (predicate.getExpressionType().equals(ExpressionType.BETWEEN)) { Expression firstExpression = ((TernaryExpression) predicate).getFirstExpression(); Expression secondExpression = ((TernaryExpression) predicate).getSecondExpression(); @@ -162,31 +158,68 @@ public static Pair extractGlobalTimePredicate( isTimeFilter = checkBetweenConstantSatisfy(secondExpression, firstExpression); } if (isTimeFilter) { - return new Pair<>(predicate, false); + return new GlobalTimePredicateExtractionResult(predicate, ConstantOperand.TRUE, false); } - return new Pair<>(null, true); + return new GlobalTimePredicateExtractionResult(null, predicate, true); } else if (predicate.getExpressionType().equals(ExpressionType.IS_NULL)) { // time filter don't support IS_NULL - return new Pair<>(null, true); + return new GlobalTimePredicateExtractionResult(null, predicate, true); } else if (predicate.getExpressionType().equals(ExpressionType.IN)) { Expression timeExpression = ((InExpression) predicate).getExpression(); if (timeExpression.getExpressionType().equals(ExpressionType.TIMESTAMP)) { - return new Pair<>(predicate, false); + return new GlobalTimePredicateExtractionResult(predicate, ConstantOperand.TRUE, false); } - return new Pair<>(null, true); + return new GlobalTimePredicateExtractionResult(null, predicate, true); } else if (predicate.getExpressionType().equals(ExpressionType.TIMESERIES) || predicate.getExpressionType().equals(ExpressionType.CONSTANT) || predicate.getExpressionType().equals(ExpressionType.NULL)) { - return new Pair<>(null, true); + return new GlobalTimePredicateExtractionResult(null, predicate, true); } else if (predicate.getExpressionType().equals(ExpressionType.CASE_WHEN_THEN)) { - return new Pair<>(null, true); + return new GlobalTimePredicateExtractionResult(null, predicate, true); } else if (ExpressionType.FUNCTION.equals(predicate.getExpressionType())) { - return new Pair<>(null, true); + return new GlobalTimePredicateExtractionResult(null, predicate, true); } else { throw new UnknownExpressionTypeException(predicate.getExpressionType()); } } + private static Expression combineConjuncts( + Expression leftExpression, Expression rightExpression, Expression identity) { + if (leftExpression == null || leftExpression.equals(identity)) { + return rightExpression; + } + if (rightExpression == null || rightExpression.equals(identity)) { + return leftExpression; + } + return ExpressionFactory.and(leftExpression, rightExpression); + } + + public static final class GlobalTimePredicateExtractionResult { + + private final Expression globalTimePredicate; + private final Expression residualPredicate; + private final boolean hasValueFilter; + + private GlobalTimePredicateExtractionResult( + Expression globalTimePredicate, Expression residualPredicate, boolean hasValueFilter) { + this.globalTimePredicate = globalTimePredicate; + this.residualPredicate = residualPredicate; + this.hasValueFilter = hasValueFilter; + } + + public Expression getGlobalTimePredicate() { + return globalTimePredicate; + } + + public Expression getResidualPredicate() { + return residualPredicate; + } + + public boolean hasValueFilter() { + return hasValueFilter; + } + } + private static boolean checkIsTimeFilter(Expression timeExpression, Expression valueExpression) { return timeExpression.getExpressionType().equals(ExpressionType.TIMESTAMP) && valueExpression instanceof ConstantOperand diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAggregationAnalyze.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAggregationAnalyze.java index e1bef50380fa3..0de197964e245 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAggregationAnalyze.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAggregationAnalyze.java @@ -40,6 +40,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; import java.util.Map; @@ -70,6 +71,26 @@ static boolean canBuildAggregationPlanUseTemplate( MPPQueryContext context, Template template, List deviceList) { + return canBuildAggregationPlanUseTemplate( + analysis, + queryStatement, + partitionFetcher, + schemaTree, + context, + template, + deviceList, + new TreeAnalysisMutationJournal()); + } + + static boolean canBuildAggregationPlanUseTemplate( + Analysis analysis, + QueryStatement queryStatement, + IPartitionFetcher partitionFetcher, + ISchemaTree schemaTree, + MPPQueryContext context, + Template template, + List deviceList, + TreeAnalysisMutationJournal mutationJournal) { // not support order by expression and non-aligned template if (queryStatement.hasOrderByExpression() || !template.isDirectAligned()) { @@ -78,13 +99,6 @@ static boolean canBuildAggregationPlanUseTemplate( analysis.setNoWhereAndAggregation(false); - if (canPushDownLimitOffsetInGroupByTimeForDevice(queryStatement)) { - // remove the device which won't appear in resultSet after limit/offset - deviceList = - pushDownLimitOffsetInGroupByTimeForDevice( - deviceList, queryStatement, context.getZoneId()); - } - List> outputExpressions = new ArrayList<>(); boolean valid = analyzeSelect(queryStatement, analysis, outputExpressions, template); if (!valid) { @@ -112,6 +126,21 @@ static boolean canBuildAggregationPlanUseTemplate( return false; } + if (canPushDownLimitOffsetInGroupByTimeForDevice(queryStatement)) { + // Pushdown rewrites LIMIT/OFFSET and the GROUP BY time range in the statement. Do it only + // after every template eligibility check succeeds, because the regular analyzer consumes the + // same statement when this optimization falls back. + mutationJournal.recordLimitOffsetPushDown(queryStatement); + deviceList = + pushDownLimitOffsetInGroupByTimeForDevice( + deviceList, queryStatement, context.getZoneId()); + if (deviceList.isEmpty()) { + analysis.setFinishQueryAfterAnalyze(true); + return true; + } + analysis.setDeviceList(deviceList); + } + analyzeDeviceToExpressions(analysis); analyzeDeviceViewOutput(analysis, queryStatement); @@ -151,13 +180,29 @@ private static boolean analyzeSelect( if (selectExpression instanceof FunctionExpression && COUNT_TIME.equalsIgnoreCase( ((FunctionExpression) selectExpression).getFunctionName())) { - outputExpressions.add(new Pair<>(selectExpression, resultColumn.getAlias())); - selectExpressions.add(selectExpression); - aggregationExpressions.add(selectExpression); + FunctionExpression countTimeExpression = (FunctionExpression) selectExpression; + // The parsed count_time(*) expression belongs to the statement and may be analyzed again + // after a dispatch failure. Build a template-only expression instead of replacing its + // wildcard child in place. Prime the copy's expression string before replacing the child: + // the result header must remain count_time(*), while the planning output symbol must be + // count_time(Time), just as it was when the parsed expression was mutated in place. + FunctionExpression templatedCountTimeExpression = + new FunctionExpression( + countTimeExpression.getFunctionName(), + new LinkedHashMap<>(countTimeExpression.getFunctionAttributes()), + new ArrayList<>(countTimeExpression.getExpressions())); + templatedCountTimeExpression.getExpressionString(); + templatedCountTimeExpression.setExpressions( + Collections.singletonList(new TimestampOperand())); + templatedCountTimeExpression.setFunctionType(countTimeExpression.getFunctionType()); + + outputExpressions.add(new Pair<>(templatedCountTimeExpression, resultColumn.getAlias())); + selectExpressions.add(templatedCountTimeExpression); + aggregationExpressions.add(templatedCountTimeExpression); - analysis.getExpressionTypes().put(NodeRef.of(selectExpression), TSDataType.INT64); - ((FunctionExpression) selectExpression) - .setExpressions(Collections.singletonList(new TimestampOperand())); + analysis + .getExpressionTypes() + .put(NodeRef.of(templatedCountTimeExpression), TSDataType.INT64); continue; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyze.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyze.java index f3eefda1446c4..350cc5d090ea5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyze.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyze.java @@ -98,7 +98,8 @@ public static boolean canBuildPlanUseTemplate( IPartitionFetcher partitionFetcher, ISchemaTree schemaTree, MPPQueryContext context, - List deviceList) { + List deviceList, + TreeAnalysisMutationJournal mutationJournal) { if (queryStatement.getGroupByComponent() != null || queryStatement.isSelectInto() || queryStatement.hasFill() @@ -115,7 +116,14 @@ public static boolean canBuildPlanUseTemplate( if (queryStatement.isAggregationQuery()) { return canBuildAggregationPlanUseTemplate( - analysis, queryStatement, partitionFetcher, schemaTree, context, template, deviceList); + analysis, + queryStatement, + partitionFetcher, + schemaTree, + context, + template, + deviceList, + mutationJournal); } List> outputExpressions = new ArrayList<>(); @@ -189,7 +197,8 @@ public static boolean canBuildPlanUseTemplate( } analysis.setDeviceList(deviceList); - analyzeDeviceToOrderBy(analysis, queryStatement, schemaTree, deviceList, context); + analyzeDeviceToOrderBy( + analysis, queryStatement, schemaTree, deviceList, context, mutationJournal); analyzeDeviceToSourceTransform(analysis); analyzeDeviceToSource(analysis); @@ -277,7 +286,8 @@ static void analyzeDeviceToOrderBy( QueryStatement queryStatement, ISchemaTree schemaTree, List deviceSet, - MPPQueryContext queryContext) { + MPPQueryContext queryContext, + TreeAnalysisMutationJournal mutationJournal) { if (!queryStatement.hasOrderByExpression()) { return; } @@ -325,7 +335,7 @@ static void analyzeDeviceToOrderBy( } analysis.setOrderByExpressions(deviceViewOrderByExpression); - queryStatement.updateSortItems(deviceViewOrderByExpression); + mutationJournal.updateSortItems(queryStatement, deviceViewOrderByExpression); analysis.setDeviceToSortItems(deviceToSortItems); analysis.setDeviceToOrderByExpressions(deviceToOrderByExpressions); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TreeAnalysisMutationJournal.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TreeAnalysisMutationJournal.java new file mode 100644 index 0000000000000..767afd2d2155c --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TreeAnalysisMutationJournal.java @@ -0,0 +1,255 @@ +/* + * 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.queryengine.plan.analyze; + +import org.apache.iotdb.db.queryengine.plan.expression.Expression; +import org.apache.iotdb.db.queryengine.plan.statement.component.GroupByTimeComponent; +import org.apache.iotdb.db.queryengine.plan.statement.component.OrderByComponent; +import org.apache.iotdb.db.queryengine.plan.statement.component.ResultColumn; +import org.apache.iotdb.db.queryengine.plan.statement.component.WhereCondition; +import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement; + +import java.util.List; +import java.util.Set; + +/** + * Records the parser-owned state changed by one tree-query analysis attempt. + * + *

A retryable dispatch failure causes {@code QueryExecution} to analyze the same statement + * object again. The journal records only the references and primitive values that an analysis + * attempt replaces. Objects reachable from a recorded reference must therefore remain untouched; + * callers that need to change a nested mutable object must first obtain a copy-on-write working + * object from this class. + * + *

The journal belongs to the planner for one query execution. It is not a general AST snapshot + * and must not be used by cached relational prepared-statement ASTs. + */ +public final class TreeAnalysisMutationJournal { + + private static final int SELECT_RESULT_COLUMNS = 1; + private static final int WHERE_CONDITION = 1 << 1; + private static final int GROUP_BY_CONTROL_EXPRESSION = 1 << 2; + private static final int ORDER_BY_COMPONENT = 1 << 3; + private static final int LIMIT_OFFSET = 1 << 4; + private static final int COUNT_TIME_AGGREGATION = 1 << 5; + + private boolean active; + private int recordedFields; + private QueryStatement statement; + + private List originalResultColumns; + private WhereCondition originalWhereCondition; + private Expression originalGroupByControlExpression; + private OrderByComponent originalOrderByComponent; + private boolean orderByDetached; + + private long originalRowLimit; + private long originalRowOffset; + private boolean originalResultSetEmpty; + private GroupByTimeComponent originalGroupByTimeComponent; + private long originalGroupByStartTime; + private long originalGroupByEndTime; + private boolean originalCountTimeAggregation; + + /** Starts a new attempt, rolling back an unfinished preceding attempt if necessary. */ + public void begin() { + if (active) { + rollback(); + } + active = true; + } + + /** + * Makes repeated direct Analyzer invocations behave like QueryExecution retry attempts. + * + *

Production code starts and finishes attempts through the planner lifecycle. Unit tests and a + * few internal callers invoke Analyzer directly, so Analyzer calls this method before walking the + * statement. + */ + public void prepareForAnalyze() { + if (!active) { + active = true; + } else if (recordedFields != 0) { + rollback(); + active = true; + } + } + + /** Restores the parser-owned state and releases all references held by this attempt. */ + public void rollback() { + if (!active) { + return; + } + if (statement != null) { + if (isRecorded(SELECT_RESULT_COLUMNS)) { + statement.getSelectComponent().setResultColumns(originalResultColumns); + } + if (isRecorded(WHERE_CONDITION)) { + statement.setWhereCondition(originalWhereCondition); + } + if (isRecorded(GROUP_BY_CONTROL_EXPRESSION)) { + statement + .getGroupByComponent() + .setControlColumnExpressionForAnalyze(originalGroupByControlExpression); + } + if (isRecorded(ORDER_BY_COMPONENT)) { + statement.setOrderByComponent(originalOrderByComponent); + } + if (isRecorded(LIMIT_OFFSET)) { + statement.setRowLimit(originalRowLimit); + statement.setRowOffset(originalRowOffset); + statement.setResultSetEmpty(originalResultSetEmpty); + originalGroupByTimeComponent.setStartTime(originalGroupByStartTime); + originalGroupByTimeComponent.setEndTime(originalGroupByEndTime); + } + if (isRecorded(COUNT_TIME_AGGREGATION)) { + statement.setCountTimeAggregation(originalCountTimeAggregation); + } + } + clear(); + } + + /** + * Commits an attempt by discarding its undo information. + * + *

The working statement is still referenced by Analysis and the generated plans, so a + * successful attempt is not restored after the dispatch retry window has closed. + */ + public void commit() { + clear(); + } + + public void replaceResultColumns( + QueryStatement queryStatement, List resultColumns) { + ensureActive(queryStatement); + if (!isRecorded(SELECT_RESULT_COLUMNS)) { + originalResultColumns = queryStatement.getSelectComponent().getResultColumns(); + recordedFields |= SELECT_RESULT_COLUMNS; + } + queryStatement.getSelectComponent().setResultColumns(resultColumns); + } + + public void replaceWhereCondition(QueryStatement queryStatement, WhereCondition whereCondition) { + ensureActive(queryStatement); + if (!isRecorded(WHERE_CONDITION)) { + originalWhereCondition = queryStatement.getWhereCondition(); + recordedFields |= WHERE_CONDITION; + } + queryStatement.setWhereCondition(whereCondition); + } + + public void replaceGroupByControlExpression( + QueryStatement queryStatement, Expression controlExpression) { + ensureActive(queryStatement); + if (!isRecorded(GROUP_BY_CONTROL_EXPRESSION)) { + originalGroupByControlExpression = + queryStatement.getGroupByComponent().getControlColumnExpression(); + recordedFields |= GROUP_BY_CONTROL_EXPRESSION; + } + queryStatement.getGroupByComponent().setControlColumnExpressionForAnalyze(controlExpression); + } + + /** + * Returns an attempt-local ORDER BY component whose nested lists and SortItems may be changed. + */ + public OrderByComponent getMutableOrderByComponent(QueryStatement queryStatement) { + ensureActive(queryStatement); + if (queryStatement.getOrderByComponent() == null) { + return null; + } + recordOrderByComponent(queryStatement); + if (!orderByDetached) { + queryStatement.setOrderByComponent( + OrderByComponent.copyOf(queryStatement.getOrderByComponent())); + orderByDetached = true; + } + return queryStatement.getOrderByComponent(); + } + + public void clearOrderByComponent(QueryStatement queryStatement) { + ensureActive(queryStatement); + recordOrderByComponent(queryStatement); + queryStatement.setOrderByComponent(null); + } + + public void updateSortItems(QueryStatement queryStatement, Set orderByExpressions) { + if (getMutableOrderByComponent(queryStatement) != null) { + queryStatement.updateSortItems(orderByExpressions); + } + } + + /** Records all values changed together by LIMIT/OFFSET pushdown. */ + public void recordLimitOffsetPushDown(QueryStatement queryStatement) { + ensureActive(queryStatement); + if (isRecorded(LIMIT_OFFSET)) { + return; + } + originalRowLimit = queryStatement.getRowLimit(); + originalRowOffset = queryStatement.getRowOffset(); + originalResultSetEmpty = queryStatement.isResultSetEmpty(); + originalGroupByTimeComponent = queryStatement.getGroupByTimeComponent(); + originalGroupByStartTime = originalGroupByTimeComponent.getStartTime(); + originalGroupByEndTime = originalGroupByTimeComponent.getEndTime(); + recordedFields |= LIMIT_OFFSET; + } + + /** Records the derived flag that semantic validation sets for {@code count_time(*)}. */ + public void recordCountTimeAggregationChange( + QueryStatement queryStatement, boolean originalValue) { + ensureActive(queryStatement); + if (!isRecorded(COUNT_TIME_AGGREGATION)) { + originalCountTimeAggregation = originalValue; + recordedFields |= COUNT_TIME_AGGREGATION; + } + } + + private void recordOrderByComponent(QueryStatement queryStatement) { + if (!isRecorded(ORDER_BY_COMPONENT)) { + originalOrderByComponent = queryStatement.getOrderByComponent(); + recordedFields |= ORDER_BY_COMPONENT; + } + } + + private boolean isRecorded(int field) { + return (recordedFields & field) != 0; + } + + private void ensureActive(QueryStatement queryStatement) { + if (!active) { + active = true; + } + if (statement == null) { + statement = queryStatement; + } else if (statement != queryStatement) { + throw new IllegalStateException(); + } + } + + private void clear() { + active = false; + recordedFields = 0; + statement = null; + originalResultColumns = null; + originalWhereCondition = null; + originalGroupByControlExpression = null; + originalOrderByComponent = null; + orderByDetached = false; + originalGroupByTimeComponent = null; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java index 96da558b81a04..037ea8688bfa4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java @@ -128,7 +128,13 @@ public class QueryExecution implements IQueryExecution { public QueryExecution(IPlanner planner, MPPQueryContext context, ExecutorService executor) { this.context = context; this.planner = planner; - this.analysis = analyze(context); + planner.beginAnalysisAttempt(); + try { + this.analysis = analyze(context); + } catch (RuntimeException | Error e) { + planner.rollbackAnalysisAttempt(); + throw e; + } context.setNeedSetHighestPriority(analysis.needSetHighestPriority()); this.stateMachine = new QueryStateMachine(context.getQueryId(), executor); @@ -166,6 +172,15 @@ interface ISourceHandleSupplier { @Override public void start() { + try { + startInternal(); + } catch (RuntimeException | Error e) { + finishAnalysisAttemptForCurrentState(); + throw e; + } + } + + private void startInternal() { final long startTime = System.nanoTime(); if (skipExecute()) { LOGGER.debug(DataNodeQueryMessages.SKIP_EXECUTE); @@ -175,6 +190,7 @@ public void start() { constructResultForMemorySource(); stateMachine.transitionToRunning(); } + finishAnalysisAttemptForCurrentState(); return; } @@ -191,6 +207,7 @@ public void start() { constructResultForMemorySource(); stateMachine.transitionToRunning(); } + finishAnalysisAttemptForCurrentState(); return; } @@ -216,6 +233,7 @@ public void start() { if (!context.isQuery() && analysis.isFailed()) { stateMachine.transitionToFailed(analysis.getFailStatus()); } + finishAnalysisAttemptForCurrentState(); } private void checkTimeOutForQuery() { @@ -229,12 +247,19 @@ private void checkTimeOutForQuery() { private ExecutionResult retry() { if (retryCount >= MAX_RETRY_COUNT) { LOGGER.warn(DataNodeQueryMessages.REACHMAXRETRYCOUNT); + // Keep the same ownership order as an ordinary retry: plans and scheduler stop using the + // analysis working state before the journal restores parser-owned objects. + this.stopAndCleanup(stateMachine.getFailureException()); + planner.rollbackAnalysisAttempt(); stateMachine.transitionToFailed(); return getStatus(); } LOGGER.warn(DataNodeQueryMessages.ERROR_WHEN_EXECUTING_QUERY, stateMachine.getFailureMessage()); // stop and clean up resources the QueryExecution used this.stopAndCleanup(stateMachine.getFailureException()); + // The generated plans no longer read the statement after cleanup. Restore the parser-owned + // state before the next analysis attempt so retry observes the same SQL as the first attempt. + planner.rollbackAnalysisAttempt(); LOGGER.info(DataNodeQueryMessages.WAITBEFORERETRY_WAIT_MS, RETRY_INTERVAL_IN_MS); try { Thread.sleep(RETRY_INTERVAL_IN_MS); @@ -253,12 +278,30 @@ private ExecutionResult retry() { this.stopped.compareAndSet(true, false); this.resultHandleCleanUp.compareAndSet(true, false); // re-analyze the query - this.analysis = analyze(context); + planner.beginAnalysisAttempt(); + try { + this.analysis = analyze(context); + } catch (RuntimeException | Error e) { + planner.rollbackAnalysisAttempt(); + throw e; + } // re-start the QueryExecution this.start(); return getStatus(); } + private void finishAnalysisAttemptForCurrentState() { + QueryState state = stateMachine.getState(); + if (state == QueryState.PENDING_RETRY) { + return; + } + if (state == QueryState.RUNNING || state == QueryState.FINISHED) { + planner.commitAnalysisAttempt(); + } else { + planner.rollbackAnalysisAttempt(); + } + } + private boolean skipExecute() { return analysis.canSkipExecute(context); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/predicate/PredicateSimplifier.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/predicate/PredicateSimplifier.java index be81a33587c40..b2518f7a7e6d1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/predicate/PredicateSimplifier.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/predicate/PredicateSimplifier.java @@ -22,6 +22,7 @@ import org.apache.iotdb.db.queryengine.plan.expression.Expression; import org.apache.iotdb.db.queryengine.plan.expression.ExpressionType; import org.apache.iotdb.db.queryengine.plan.expression.binary.ArithmeticBinaryExpression; +import org.apache.iotdb.db.queryengine.plan.expression.binary.BinaryExpression; import org.apache.iotdb.db.queryengine.plan.expression.binary.CompareBinaryExpression; import org.apache.iotdb.db.queryengine.plan.expression.binary.LogicAndExpression; import org.apache.iotdb.db.queryengine.plan.expression.binary.LogicOrExpression; @@ -45,6 +46,10 @@ import java.util.ArrayList; import java.util.List; +import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.reconstructBinaryExpression; +import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.reconstructFunctionExpression; +import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.reconstructTernaryExpression; +import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionUtils.reconstructUnaryExpression; import static org.apache.iotdb.db.queryengine.plan.expression.visitor.predicate.ConvertPredicateToFilterVisitor.getValue; public class PredicateSimplifier extends ExpressionAnalyzeVisitor { @@ -59,8 +64,7 @@ public Expression visitIsNullExpression(IsNullExpression isNullExpression, Void return ConstantOperand.TRUE; } } - isNullExpression.setExpression(simplifiedInputExpression); - return isNullExpression; + return reconstructUnaryIfNeeded(isNullExpression, simplifiedInputExpression); } @Override @@ -83,8 +87,7 @@ private Expression processFilterUnaryExpression(UnaryExpression unaryExpression, if (simplifiedInputExpression.getExpressionType().equals(ExpressionType.NULL)) { return ConstantOperand.FALSE; } - unaryExpression.setExpression(simplifiedInputExpression); - return unaryExpression; + return reconstructUnaryIfNeeded(unaryExpression, simplifiedInputExpression); } @Override @@ -96,8 +99,7 @@ public Expression visitLogicNotExpression(LogicNotExpression logicNotExpression, if (simplifiedInputExpression.equals(ConstantOperand.FALSE)) { return ConstantOperand.TRUE; } - logicNotExpression.setExpression(simplifiedInputExpression); - return logicNotExpression; + return reconstructUnaryIfNeeded(logicNotExpression, simplifiedInputExpression); } @Override @@ -106,8 +108,7 @@ public Expression visitNegationExpression(NegationExpression negationExpression, if (simplifiedInputExpression.getExpressionType().equals(ExpressionType.NULL)) { return new NullOperand(); } - negationExpression.setExpression(simplifiedInputExpression); - return negationExpression; + return reconstructUnaryIfNeeded(negationExpression, simplifiedInputExpression); } @Override @@ -121,9 +122,7 @@ public Expression visitArithmeticBinaryExpression( // null calculation will return null return new NullOperand(); } - arithmeticBinaryExpression.setLeftExpression(simplifiedLeft); - arithmeticBinaryExpression.setRightExpression(simplifiedRight); - return arithmeticBinaryExpression; + return reconstructBinaryIfNeeded(arithmeticBinaryExpression, simplifiedLeft, simplifiedRight); } @Override @@ -144,9 +143,7 @@ public Expression visitLogicAndExpression(LogicAndExpression logicAndExpression, if (isLeftFalse || isRightFalse) { return ConstantOperand.FALSE; } - logicAndExpression.setLeftExpression(simplifiedLeft); - logicAndExpression.setRightExpression(simplifiedRight); - return logicAndExpression; + return reconstructBinaryIfNeeded(logicAndExpression, simplifiedLeft, simplifiedRight); } @Override @@ -167,9 +164,7 @@ public Expression visitLogicOrExpression(LogicOrExpression logicOrExpression, Vo } else if (isRightFalse) { return simplifiedLeft; } - logicOrExpression.setLeftExpression(simplifiedLeft); - logicOrExpression.setRightExpression(simplifiedRight); - return logicOrExpression; + return reconstructBinaryIfNeeded(logicOrExpression, simplifiedLeft, simplifiedRight); } @Override @@ -183,9 +178,7 @@ public Expression visitCompareBinaryExpression( // null comparison will return false return ConstantOperand.FALSE; } - compareBinaryExpression.setLeftExpression(simplifiedLeft); - compareBinaryExpression.setRightExpression(simplifiedRight); - return compareBinaryExpression; + return reconstructBinaryIfNeeded(compareBinaryExpression, simplifiedLeft, simplifiedRight); } @Override @@ -228,10 +221,16 @@ && lessThan( return ConstantOperand.FALSE; } } - betweenExpression.setFirstExpression(simplifiedFirstExpression); - betweenExpression.setSecondExpression(simplifiedSecondExpression); - betweenExpression.setThirdExpression(simplifiedThirdExpression); - return betweenExpression; + if (simplifiedFirstExpression == betweenExpression.getFirstExpression() + && simplifiedSecondExpression == betweenExpression.getSecondExpression() + && simplifiedThirdExpression == betweenExpression.getThirdExpression()) { + return betweenExpression; + } + return reconstructTernaryExpression( + betweenExpression, + simplifiedFirstExpression, + simplifiedSecondExpression, + simplifiedThirdExpression); } private > boolean lessThan( @@ -243,16 +242,42 @@ private > boolean lessThan( @Override public Expression visitFunctionExpression(FunctionExpression functionExpression, Void context) { - List simplifiedInputExpressions = new ArrayList<>(); - for (Expression inputExpression : functionExpression.getExpressions()) { + List inputExpressions = functionExpression.getExpressions(); + List simplifiedInputExpressions = null; + for (int i = 0; i < inputExpressions.size(); i++) { + Expression inputExpression = inputExpressions.get(i); Expression simplifiedInputExpression = process(inputExpression, context); if (simplifiedInputExpression.getExpressionType().equals(ExpressionType.NULL)) { return new NullOperand(); } - simplifiedInputExpressions.add(simplifiedInputExpression); + if (simplifiedInputExpression != inputExpression) { + if (simplifiedInputExpressions == null) { + simplifiedInputExpressions = new ArrayList<>(inputExpressions); + } + simplifiedInputExpressions.set(i, simplifiedInputExpression); + } + } + if (simplifiedInputExpressions == null) { + return functionExpression; + } + return reconstructFunctionExpression(functionExpression, simplifiedInputExpressions); + } + + private Expression reconstructUnaryIfNeeded( + UnaryExpression originalExpression, Expression simplifiedChild) { + if (simplifiedChild == originalExpression.getExpression()) { + return originalExpression; + } + return reconstructUnaryExpression(originalExpression, simplifiedChild); + } + + private Expression reconstructBinaryIfNeeded( + BinaryExpression originalExpression, Expression simplifiedLeft, Expression simplifiedRight) { + if (simplifiedLeft == originalExpression.getLeftExpression() + && simplifiedRight == originalExpression.getRightExpression()) { + return originalExpression; } - functionExpression.setExpressions(simplifiedInputExpressions); - return functionExpression; + return reconstructBinaryExpression(originalExpression, simplifiedLeft, simplifiedRight); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/IPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/IPlanner.java index 501aa1e09d42e..99cd988bbb7b6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/IPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/IPlanner.java @@ -34,6 +34,12 @@ public interface IPlanner { IAnalysis analyze(MPPQueryContext context); + default void beginAnalysisAttempt() {} + + default void rollbackAnalysisAttempt() {} + + default void commitAnalysisAttempt() {} + LogicalQueryPlan doLogicalPlan(IAnalysis analysis, MPPQueryContext context); DistributedQueryPlan doDistributionPlan( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java index ce450b5005259..f2c0011b8011b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java @@ -497,11 +497,11 @@ public LogicalPlanBuilder planDeviceView( .map(Expression::getExpressionString) .collect(Collectors.toList()); - List sortItemList = queryStatement.getSortItemList(); - - if (sortItemList.isEmpty()) { - sortItemList = new ArrayList<>(); - } + // DEVICE and TIME are implicit merge keys for DeviceView. Append them to a planning-owned copy + // so rebuilding the plan after a dispatch failure does not mutate the parsed statement or + // accumulate duplicate keys. Keep the complete parameter in Analysis because a downstream + // SortNode must use the same implicit keys without reading a mutated QueryStatement. + List sortItemList = new ArrayList<>(queryStatement.getSortItemList()); if (!queryStatement.isOrderByDevice()) { sortItemList.add(new SortItem(OrderByKey.DEVICE, Ordering.ASC)); } @@ -510,6 +510,7 @@ public LogicalPlanBuilder planDeviceView( } OrderByParameter orderByParameter = new OrderByParameter(sortItemList); + analysis.setMergeOrderParameter(orderByParameter); long limitValue = queryStatement.hasOffset() @@ -1431,7 +1432,10 @@ public LogicalPlanBuilder planOrderBy(QueryStatement queryStatement, Analysis an Set orderByExpressions = analysis.getOrderByExpressions(); updateTypeProvider(orderByExpressions); - OrderByParameter orderByParameter = new OrderByParameter(queryStatement.getSortItemList()); + OrderByParameter orderByParameter = + queryStatement.isAlignByDevice() + ? analysis.getMergeOrderParameter() + : new OrderByParameter(queryStatement.getSortItemList()); if (orderByParameter.isEmpty()) { return this; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TreeModelPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TreeModelPlanner.java index 756f17148cad3..8a614809559ec 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TreeModelPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TreeModelPlanner.java @@ -31,6 +31,7 @@ import org.apache.iotdb.db.queryengine.plan.analyze.Analyzer; import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis; import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher; +import org.apache.iotdb.db.queryengine.plan.analyze.TreeAnalysisMutationJournal; import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher; import org.apache.iotdb.db.queryengine.plan.planner.distribution.DistributionPlanner; import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan; @@ -57,6 +58,7 @@ public class TreeModelPlanner implements IPlanner { private final Statement statement; + private final TreeAnalysisMutationJournal mutationJournal = new TreeAnalysisMutationJournal(); private final ExecutorService executor; private final ScheduledExecutorService scheduledExecutor; @@ -91,7 +93,23 @@ public TreeModelPlanner( @Override public IAnalysis analyze(MPPQueryContext context) { - return new Analyzer(context, partitionFetcher, schemaFetcher).analyze(statement); + return new Analyzer(context, partitionFetcher, schemaFetcher, mutationJournal) + .analyze(statement); + } + + @Override + public void beginAnalysisAttempt() { + mutationJournal.begin(); + } + + @Override + public void rollbackAnalysisAttempt() { + mutationJournal.rollback(); + } + + @Override + public void commitAnalysisAttempt() { + mutationJournal.commit(); } @Override 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 80e8bae032abf..741116c844c53 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 @@ -5400,20 +5400,29 @@ public Scope visitFetchDevice(final FetchDevice node, final Optional cont @Override public Scope visitShowDevice(final ShowDevice node, final Optional context) { - analyzeQueryDevice(node, context); + // SQL SHOW DEVICES statements retain their parser-owned AST and therefore need an isolated + // analysis copy. Schema fetching also uses ShowDevice internally, but that form has already + // been populated with tag, fuzzy, partition, and attribute filters; those fields are the + // input of the internal request rather than derived state and must not be discarded. + final ShowDevice workingNode = + Objects.nonNull(node.getTable()) ? node.copyForAnalysis() : node; + analysis.setStatement(workingNode); + analyzeQueryDevice(workingNode, context); // TODO: use real scope when parameter in offset and limit is supported - if (Objects.nonNull(node.getOffset())) { - analyzeOffset(node.getOffset(), null); + if (Objects.nonNull(workingNode.getOffset())) { + analyzeOffset(workingNode.getOffset(), null); } - if (Objects.nonNull(node.getLimit())) { - analyzeLimit(node.getLimit(), null); + if (Objects.nonNull(workingNode.getLimit())) { + analyzeLimit(workingNode.getLimit(), null); } return null; } @Override public Scope visitCountDevice(final CountDevice node, final Optional context) { - analyzeQueryDevice(node, context); + final CountDevice workingNode = node.copyForAnalysis(); + analysis.setStatement(workingNode); + analyzeQueryDevice(workingNode, context); return null; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractQueryDeviceWithCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractQueryDeviceWithCache.java index 8c3d46b2cca41..f0422e8787103 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractQueryDeviceWithCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractQueryDeviceWithCache.java @@ -62,6 +62,23 @@ protected AbstractQueryDeviceWithCache(final String database, final String table super(database, tableName); } + /** + * Copies only the parsed input of a device query into a fresh analysis working statement. + * + *

The analyzer and schema fetcher populate {@link AbstractTraverseDevice} with resolved names, + * translated predicates, cache results, and partition filters. A parsed statement, however, may + * be retained by a prepared statement and analyzed again by another execution or by a dispatch + * retry. Keeping those derived fields out of this copy constructor ensures that every execution + * starts from the same raw SQL AST without copying the expression tree. + */ + protected AbstractQueryDeviceWithCache(final AbstractQueryDeviceWithCache source) { + super(source.getLocation().orElse(null), source.table, source.where); + this.database = source.database; + this.tableName = source.tableName; + } + + public abstract AbstractQueryDeviceWithCache copyForAnalysis(); + public boolean parseRawExpression( final TsTable tableInstance, final List attributeColumns, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/CountDevice.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/CountDevice.java index b8800a31f4da9..f6a7199264070 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/CountDevice.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/CountDevice.java @@ -46,6 +46,15 @@ public CountDevice( super(location, table, rawExpression); } + private CountDevice(final CountDevice source) { + super(source); + } + + @Override + public CountDevice copyForAnalysis() { + return new CountDevice(this); + } + @Override public DatasetHeader getDataSetHeader() { return new DatasetHeader( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/ShowDevice.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/ShowDevice.java index 2a929cec5410b..0020ecf35aa5c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/ShowDevice.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/ShowDevice.java @@ -66,6 +66,17 @@ public ShowDevice(final String database, final String tableName) { super(database, tableName); } + private ShowDevice(final ShowDevice source) { + super(source); + this.offset = source.offset; + this.limit = source.limit; + } + + @Override + public ShowDevice copyForAnalysis() { + return new ShowDevice(this); + } + public Offset getOffset() { return offset; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/GroupByComponent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/GroupByComponent.java index ec45a90a4b827..b1d5fefde523c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/GroupByComponent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/GroupByComponent.java @@ -45,6 +45,11 @@ public void setControlColumnExpression(Expression controlColumnExpression) { ExpressionAnalyzer.toLowerCaseExpression(controlColumnExpression); } + /** Sets an expression that has already been normalized by the analyzer. */ + public void setControlColumnExpressionForAnalyze(Expression controlColumnExpression) { + this.controlColumnExpression = controlColumnExpression; + } + public Expression getControlColumnExpression() { return controlColumnExpression; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/OrderByComponent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/OrderByComponent.java index 1014e32df1d4b..8f07439b76e55 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/OrderByComponent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/OrderByComponent.java @@ -49,6 +49,36 @@ public OrderByComponent() { this.sortItemExpressionList = new ArrayList<>(); } + /** + * Creates a structurally independent component for analysis-time copy-on-write updates. + * + *

Expression nodes are shared because analysis replaces them instead of mutating them. Lists + * and SortItems are copied because both are changed in place while resolving ORDER BY + * expressions. + */ + public static OrderByComponent copyOf(OrderByComponent source) { + OrderByComponent result = new OrderByComponent(); + int expressionIndex = 0; + for (SortItem sortItem : source.sortItemList) { + if (sortItem.isExpression()) { + SortItem copiedSortItem = + new SortItem( + sortItem.getExpression(), sortItem.getOrdering(), sortItem.getNullOrdering()); + result.sortItemList.add(copiedSortItem); + // SortItem intentionally keeps the parser-facing expression for SQL rendering, while the + // parallel expression list contains the lower-case normalized AST consumed by analysis. + // They are not interchangeable, so preserve both references independently in the COW + // component. + result.sortItemExpressionList.add(source.sortItemExpressionList.get(expressionIndex++)); + } else { + result.addSortItem( + new SortItem( + sortItem.getSortKey(), sortItem.getOrdering(), sortItem.getNullOrdering())); + } + } + return result; + } + public void addSortItem(SortItem sortItem) { this.sortItemList.add(sortItem); switch (sortItem.getSortKey()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/WhereCondition.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/WhereCondition.java index 7352c1e96fcb6..d80677fd83062 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/WhereCondition.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/WhereCondition.java @@ -35,6 +35,13 @@ public WhereCondition(Expression predicate) { this.predicate = ExpressionAnalyzer.toLowerCaseExpression(predicate); } + /** Wraps an expression that has already been normalized without rebuilding its AST. */ + public static WhereCondition fromNormalizedPredicate(Expression predicate) { + WhereCondition whereCondition = new WhereCondition(); + whereCondition.predicate = predicate; + return whereCondition; + } + public Expression getPredicate() { return predicate; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java index 8c199f6c19c42..70a209bc4be8b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java @@ -82,6 +82,7 @@ public class QueryStatement extends AuthorityInformationStatement { private SelectComponent selectComponent; private FromComponent fromComponent; private WhereCondition whereCondition; + private HavingCondition havingCondition; // row limit for result set. The default value is 0, which means no limit diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeTest.java index 2cc9b8faf7e64..12191b6297018 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeTest.java @@ -22,6 +22,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot; import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.common.SessionInfo; import org.apache.iotdb.commons.schema.column.ColumnHeader; @@ -648,6 +649,218 @@ public void testTimeFilterRewrite() throws IllegalPathException { } } + @Test + public void testTimeFilterPreservedWhenQueryAnalysisIsRetried() throws Exception { + // In the reported incident, the first dispatch failed because the coordinator's internal + // client pool for the leader DataNode endpoint had reached its 1000-client limit. + // QueryExecution handles that transient failure by analyzing the same Statement object again. + // Repeating analysis directly makes the regression deterministic and independent of the + // resource exhaustion and dispatch timing. + String[] sqls = + new String[] { + "select s1 from root.sg.d1 where time > 1", + "select s1 from root.sg.d1 where time > 1 and time < 3", + "select s1 from root.sg.d1 where time > 1 and time < 3 and s1 > 1" + }; + boolean[] expectedHasValueFilter = new boolean[] {false, false, true}; + + for (int i = 0; i < sqls.length; i++) { + String sql = sqls[i]; + Statement statement = + StatementGenerator.createStatement(sql, ZonedDateTime.now().getOffset()); + MPPQueryContext context = + new MPPQueryContext( + "", + new QueryId("test_query_retry"), + new SessionInfo(0, "test", ZoneId.systemDefault()), + new TEndPoint(), + new TEndPoint()); + Analyzer analyzer = + new Analyzer(context, new FakePartitionFetcherImpl(), new FakeSchemaFetcherImpl()); + + Analysis initialAnalysis = analyzer.analyze(statement); + + Assert.assertNotNull(sql, initialAnalysis.getGlobalTimePredicate()); + assertEquals(sql, expectedHasValueFilter[i], initialAnalysis.hasValueFilter()); + + // A second retry verifies that the same untouched snapshot can be restored repeatedly. + for (int retry = 0; retry < 2; retry++) { + context.prepareForRetry(); + Analysis retriedAnalysis = analyzer.analyze(statement); + + assertEquals( + sql, + initialAnalysis.getGlobalTimePredicate(), + retriedAnalysis.getGlobalTimePredicate()); + assertEquals( + sql, initialAnalysis.getWhereExpression(), retriedAnalysis.getWhereExpression()); + assertEquals(sql, expectedHasValueFilter[i], retriedAnalysis.hasValueFilter()); + } + } + } + + @Test + public void testCountTimeWithMultipleFromPathsPreservedWhenQueryAnalysisIsRetried() { + Pair analyses = + analyzeSQLTwice("select count_time(*) from root.sg.d1, root.sg.d2"); + + assertEquals(analyses.left.getRespDatasetHeader(), analyses.right.getRespDatasetHeader()); + assertEquals(analyses.left.getSelectExpressions(), analyses.right.getSelectExpressions()); + assertEquals( + analyses.left.getAggregationExpressions(), analyses.right.getAggregationExpressions()); + } + + @Test + public void testFailedSemanticCheckRestoresCountTimeFlag() { + QueryStatement statement = + (QueryStatement) + StatementGenerator.createStatement( + "select count_time(*) from root.sg.d1 having count_time(*) > 0", + ZonedDateTime.now().getOffset()); + MPPQueryContext context = + new MPPQueryContext( + "", + new QueryId("test_count_time_semantic_failure"), + new SessionInfo(0, "test", ZoneId.systemDefault()), + new TEndPoint(), + new TEndPoint()); + + Assert.assertFalse(statement.isCountTimeAggregation()); + try { + new Analyzer(context, new FakePartitionFetcherImpl(), new FakeSchemaFetcherImpl()) + .analyze(statement); + fail("count_time with HAVING should fail semantic validation"); + } catch (SemanticException expected) { + Assert.assertFalse(statement.isCountTimeAggregation()); + } + } + + @Test + public void testGroupByLevelHavingPreservedWhenQueryAnalysisIsRetried() { + Pair analyses = + analyzeSQLTwice("select count(s1) from root.sg.d1 group by level = 1 having count(s1) > 0"); + + assertEquals(analyses.left.getRespDatasetHeader(), analyses.right.getRespDatasetHeader()); + assertEquals(analyses.left.getHavingExpression(), analyses.right.getHavingExpression()); + assertEquals( + analyses.left.getCrossGroupByExpressions(), analyses.right.getCrossGroupByExpressions()); + } + + @Test + public void testLimitOffsetPushDownPreservedWhenQueryAnalysisIsRetried() + throws IllegalPathException { + Pair analyses = + analyzeSQLTwice( + "select count(s1) from root.sg.* group by ([0, 100), 10ms) " + + "limit 10 offset 10 align by device"); + + assertEquals( + Collections.singletonList(new PartialPath("root.sg.d2")), analyses.left.getDeviceList()); + assertEquals(analyses.left.getDeviceList(), analyses.right.getDeviceList()); + assertEquals(analyses.left.getGroupByTimeParameter(), analyses.right.getGroupByTimeParameter()); + } + + @Test + public void testSelectIntoDeviceOrderPreservedWhenQueryAnalysisIsRetried() + throws IllegalPathException { + Pair analyses = + analyzeSQLTwice( + "select s1 into root.copy.t1(x), root.copy.t2(x) from root.sg.* " + + "order by device desc align by device"); + + Map>> expectedMapping = new HashMap<>(); + expectedMapping.put( + "root.sg.d2", + Collections.singletonList(new Pair<>("s1", new PartialPath("root.copy.t1.x")))); + expectedMapping.put( + "root.sg.d1", + Collections.singletonList(new Pair<>("s1", new PartialPath("root.copy.t2.x")))); + assertEquals( + expectedMapping, + analyses.left.getDeviceViewIntoPathDescriptor().getDeviceToSourceTargetPathPairListMap()); + assertEquals( + analyses.left.getDeviceViewIntoPathDescriptor().getDeviceToSourceTargetPathPairListMap(), + analyses.right.getDeviceViewIntoPathDescriptor().getDeviceToSourceTargetPathPairListMap()); + } + + @Test + public void testNormalizedOrderByExpressionPreservedWhenQueryAnalysisIsRetried() { + Pair analyses = + analyzeSQLTwice("select s1 from root.sg.* order by Sin(s1) align by device"); + + assertEquals(analyses.left.getOrderByExpressions(), analyses.right.getOrderByExpressions()); + assertEquals( + analyses.left.getDeviceToOrderByExpressions(), + analyses.right.getDeviceToOrderByExpressions()); + assertEquals(analyses.left.getDeviceToSortItems(), analyses.right.getDeviceToSortItems()); + } + + @Test + public void testNormalizedOrderByExpressionPreservedForNonAlignQueryRetry() { + Pair analyses = + analyzeSQLTwice("select s1 from root.sg.d1 order by Sin(s1)"); + + assertEquals(analyses.left.getOrderByExpressions(), analyses.right.getOrderByExpressions()); + assertEquals(analyses.left.getMergeOrderParameter(), analyses.right.getMergeOrderParameter()); + } + + @Test + public void testGroupByControlExpressionPreservedWhenQueryAnalysisIsRetried() { + Pair analyses = + analyzeSQLTwice("select count(s1) from root.sg.d1 group by variation(s1)"); + + assertEquals(analyses.left.getGroupByExpression(), analyses.right.getGroupByExpression()); + assertEquals( + analyses.left.getAggregationExpressions(), analyses.right.getAggregationExpressions()); + } + + @Test + public void testMixedOrPredicatePreservedWhenQueryAnalysisIsRetried() { + Pair analyses = + analyzeSQLTwice( + "select s1 from root.sg.d1 where " + + "(time > 1 and s1 > 1) or (time > 10 and s1 < 10)"); + + Assert.assertNotNull(analyses.left.getGlobalTimePredicate()); + Assert.assertTrue(analyses.left.hasValueFilter()); + assertEquals(analyses.left.getGlobalTimePredicate(), analyses.right.getGlobalTimePredicate()); + assertEquals(analyses.left.getWhereExpression(), analyses.right.getWhereExpression()); + } + + @Test + public void testMixedNotPredicatePreservedWhenQueryAnalysisIsRetried() { + Pair analyses = + analyzeSQLTwice("select s1 from root.sg.d1 where not(time > 1 and s1 > 1)"); + + Assert.assertNull(analyses.left.getGlobalTimePredicate()); + Assert.assertTrue(analyses.left.hasValueFilter()); + assertEquals(analyses.left.getWhereExpression(), analyses.right.getWhereExpression()); + } + + @Test + public void testShowTimeSeriesTimeFilterPreservedWhenAnalysisIsRetried() { + assertMetadataTimeFilterPreservedWhenAnalysisIsRetried( + "show timeseries root.sg.** where time > 1 and time < 3"); + } + + @Test + public void testShowDevicesTimeFilterPreservedWhenAnalysisIsRetried() { + assertMetadataTimeFilterPreservedWhenAnalysisIsRetried( + "show devices root.sg.** where time > 1 and time < 3"); + } + + @Test + public void testCountTimeSeriesTimeFilterPreservedWhenAnalysisIsRetried() { + assertMetadataTimeFilterPreservedWhenAnalysisIsRetried( + "count timeseries root.sg.** where time > 1 and time < 3"); + } + + @Test + public void testCountDevicesTimeFilterPreservedWhenAnalysisIsRetried() { + assertMetadataTimeFilterPreservedWhenAnalysisIsRetried( + "count devices root.sg.** where time > 1 and time < 3"); + } + @Test public void testDataPartitionAnalyze() { Analysis analysis = analyzeSQL("insert into root.sg.d1(timestamp,s) values(1,10),(86401,11)"); @@ -1070,6 +1283,31 @@ private Analysis analyzeSQL(String sql) { return null; } + private Pair analyzeSQLTwice(String sql) { + Statement statement = StatementGenerator.createStatement(sql, ZonedDateTime.now().getOffset()); + MPPQueryContext context = + new MPPQueryContext( + "", + new QueryId("test_query_retry"), + new SessionInfo(0, "test", ZoneId.systemDefault()), + new TEndPoint(), + new TEndPoint()); + Analyzer analyzer = + new Analyzer(context, new FakePartitionFetcherImpl(), new FakeSchemaFetcherImpl()); + + Analysis initialAnalysis = analyzer.analyze(statement); + context.prepareForRetry(); + Analysis retriedAnalysis = analyzer.analyze(statement); + return new Pair<>(initialAnalysis, retriedAnalysis); + } + + private void assertMetadataTimeFilterPreservedWhenAnalysisIsRetried(String sql) { + Pair analyses = analyzeSQLTwice(sql); + + Assert.assertNotNull(analyses.left.getGlobalTimePredicate()); + assertEquals(analyses.left.getGlobalTimePredicate(), analyses.right.getGlobalTimePredicate()); + } + private void alignByTimeAnalysisEqualTest(Analysis actualAnalysis, Analysis expectedAnalysis) { commonAnalysisEqualTest(expectedAnalysis, actualAnalysis); assertEquals(expectedAnalysis.getSourceExpressions(), actualAnalysis.getSourceExpressions()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/PredicateUtilsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/PredicateUtilsTest.java new file mode 100644 index 0000000000000..bc9d252d30e32 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/PredicateUtilsTest.java @@ -0,0 +1,172 @@ +/* + * 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.queryengine.plan.analyze; + +import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils.GlobalTimePredicateExtractionResult; +import org.apache.iotdb.db.queryengine.plan.expression.Expression; +import org.apache.iotdb.db.queryengine.plan.expression.binary.BinaryExpression; +import org.apache.iotdb.db.queryengine.plan.expression.leaf.ConstantOperand; +import org.apache.iotdb.db.queryengine.plan.expression.multi.FunctionExpression; +import org.apache.iotdb.db.queryengine.plan.expression.unary.LogicNotExpression; + +import org.junit.Test; + +import static org.apache.iotdb.db.queryengine.plan.expression.ExpressionFactory.and; +import static org.apache.iotdb.db.queryengine.plan.expression.ExpressionFactory.function; +import static org.apache.iotdb.db.queryengine.plan.expression.ExpressionFactory.gt; +import static org.apache.iotdb.db.queryengine.plan.expression.ExpressionFactory.intValue; +import static org.apache.iotdb.db.queryengine.plan.expression.ExpressionFactory.longValue; +import static org.apache.iotdb.db.queryengine.plan.expression.ExpressionFactory.lt; +import static org.apache.iotdb.db.queryengine.plan.expression.ExpressionFactory.not; +import static org.apache.iotdb.db.queryengine.plan.expression.ExpressionFactory.or; +import static org.apache.iotdb.db.queryengine.plan.expression.ExpressionFactory.time; +import static org.apache.iotdb.db.queryengine.plan.expression.ExpressionFactory.timeSeries; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +public class PredicateUtilsTest { + + @Test + public void testExtractTimePredicateFromAndWithoutMutatingInput() throws IllegalPathException { + Expression timePredicate = gt(time(), longValue(1)); + Expression valuePredicate = gt(timeSeries("root.sg.d.s1"), intValue("1")); + BinaryExpression predicate = and(timePredicate, valuePredicate); + String originalExpressionString = predicate.getExpressionString(); + + GlobalTimePredicateExtractionResult result = + PredicateUtils.extractGlobalTimePredicate(predicate); + + assertSame(timePredicate, result.getGlobalTimePredicate()); + assertSame(valuePredicate, result.getResidualPredicate()); + assertTrue(result.hasValueFilter()); + assertSame(timePredicate, predicate.getLeftExpression()); + assertSame(valuePredicate, predicate.getRightExpression()); + assertEquals(originalExpressionString, predicate.getExpressionString()); + } + + @Test + public void testExtractPureTimeOrWithoutMutatingInput() { + Expression leftTimePredicate = gt(time(), longValue(1)); + Expression rightTimePredicate = lt(time(), longValue(10)); + BinaryExpression predicate = or(leftTimePredicate, rightTimePredicate); + + GlobalTimePredicateExtractionResult result = + PredicateUtils.extractGlobalTimePredicate(predicate); + + assertEquals(predicate, result.getGlobalTimePredicate()); + assertSame(ConstantOperand.TRUE, result.getResidualPredicate()); + assertFalse(result.hasValueFilter()); + assertSame(leftTimePredicate, predicate.getLeftExpression()); + assertSame(rightTimePredicate, predicate.getRightExpression()); + } + + @Test + public void testMixedOrKeepsOriginalPredicateAsResidual() throws IllegalPathException { + Expression leftTimePredicate = gt(time(), longValue(1)); + Expression rightTimePredicate = gt(time(), longValue(10)); + Expression leftValuePredicate = gt(timeSeries("root.sg.d.s1"), intValue("1")); + Expression rightValuePredicate = gt(timeSeries("root.sg.d.s2"), intValue("2")); + BinaryExpression left = and(leftTimePredicate, leftValuePredicate); + BinaryExpression right = and(rightTimePredicate, rightValuePredicate); + BinaryExpression predicate = or(left, right); + + GlobalTimePredicateExtractionResult result = + PredicateUtils.extractGlobalTimePredicate(predicate); + + assertEquals(or(leftTimePredicate, rightTimePredicate), result.getGlobalTimePredicate()); + assertSame(predicate, result.getResidualPredicate()); + assertTrue(result.hasValueFilter()); + assertSame(left, predicate.getLeftExpression()); + assertSame(right, predicate.getRightExpression()); + } + + @Test + public void testMixedNotDoesNotExtractTimePredicate() throws IllegalPathException { + Expression timePredicate = gt(time(), longValue(1)); + Expression valuePredicate = gt(timeSeries("root.sg.d.s1"), intValue("1")); + BinaryExpression child = and(timePredicate, valuePredicate); + LogicNotExpression predicate = not(child); + + GlobalTimePredicateExtractionResult result = + PredicateUtils.extractGlobalTimePredicate(predicate); + + assertNull(result.getGlobalTimePredicate()); + assertSame(predicate, result.getResidualPredicate()); + assertTrue(result.hasValueFilter()); + assertSame(child, predicate.getExpression()); + } + + @Test + public void testPureTimeNotCanBeExtracted() { + Expression timePredicate = gt(time(), longValue(1)); + LogicNotExpression predicate = not(timePredicate); + + GlobalTimePredicateExtractionResult result = + PredicateUtils.extractGlobalTimePredicate(predicate); + + assertEquals(predicate, result.getGlobalTimePredicate()); + assertSame(ConstantOperand.TRUE, result.getResidualPredicate()); + assertFalse(result.hasValueFilter()); + assertSame(timePredicate, predicate.getExpression()); + } + + @Test + public void testSimplifierUsesCopyOnWriteForUnaryExpression() throws IllegalPathException { + Expression valuePredicate = gt(timeSeries("root.sg.d.s1"), intValue("1")); + BinaryExpression reducibleChild = and(ConstantOperand.TRUE, valuePredicate); + LogicNotExpression predicate = not(reducibleChild); + + Expression simplified = PredicateUtils.simplifyPredicate(predicate); + + assertNotSame(predicate, simplified); + assertTrue(simplified instanceof LogicNotExpression); + assertSame(valuePredicate, ((LogicNotExpression) simplified).getExpression()); + assertSame(reducibleChild, predicate.getExpression()); + assertSame(ConstantOperand.TRUE, reducibleChild.getLeftExpression()); + assertSame(valuePredicate, reducibleChild.getRightExpression()); + } + + @Test + public void testSimplifierUsesCopyOnWriteForFunctionExpression() throws IllegalPathException { + Expression valuePredicate = gt(timeSeries("root.sg.d.s1"), intValue("1")); + BinaryExpression reducibleChild = and(ConstantOperand.TRUE, valuePredicate); + FunctionExpression functionExpression = function("test", reducibleChild); + + Expression simplified = PredicateUtils.simplifyPredicate(functionExpression); + + assertNotSame(functionExpression, simplified); + assertTrue(simplified instanceof FunctionExpression); + assertSame(valuePredicate, ((FunctionExpression) simplified).getExpressions().get(0)); + assertSame(reducibleChild, functionExpression.getExpressions().get(0)); + assertSame(ConstantOperand.TRUE, reducibleChild.getLeftExpression()); + } + + @Test + public void testSimplifierReusesUnchangedExpression() throws IllegalPathException { + Expression predicate = gt(timeSeries("root.sg.d.s1"), intValue("1")); + + assertSame(predicate, PredicateUtils.simplifyPredicate(predicate)); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyzeRetryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyzeRetryTest.java new file mode 100644 index 0000000000000..aea763bf29177 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyzeRetryTest.java @@ -0,0 +1,127 @@ +/* + * 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.queryengine.plan.analyze; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.commons.schema.template.Template; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.plan.expression.multi.FunctionExpression; +import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator; +import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement; + +import org.junit.Test; + +import java.time.ZoneId; +import java.time.ZonedDateTime; +import java.util.Arrays; +import java.util.Collections; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertTrue; + +public class TemplatedAnalyzeRetryTest { + + @Test + public void testCountTimeTemplateAnalysisDoesNotMutateStatementForRetry() throws Exception { + QueryStatement queryStatement = + (QueryStatement) + StatementGenerator.createStatement( + "select count_time(*) from root.sg.* align by device", + ZonedDateTime.now().getOffset()); + queryStatement.semanticCheck(); + + Analysis analysis = new Analysis(); + boolean canBuildPlan = + TemplatedAggregationAnalyze.canBuildAggregationPlanUseTemplate( + analysis, + queryStatement, + null, + null, + createQueryContext(), + createAlignedTemplate(), + Collections.emptyList()); + + assertTrue(canBuildPlan); + FunctionExpression countTimeExpression = + (FunctionExpression) + queryStatement.getSelectComponent().getResultColumns().get(0).getExpression(); + assertEquals("*", countTimeExpression.getExpressions().get(0).getOutputSymbol()); + FunctionExpression analyzedCountTimeExpression = + (FunctionExpression) analysis.getAggregationExpressions().iterator().next(); + assertNotSame(countTimeExpression, analyzedCountTimeExpression); + assertEquals("count_time(*)", analyzedCountTimeExpression.getExpressionString()); + assertEquals("count_time(Time)", analyzedCountTimeExpression.getOutputSymbol()); + + // QueryExecution analyzes the same statement again when retrying a dispatch failure. + queryStatement.semanticCheck(); + } + + @Test + public void testFailedTemplateAttemptDoesNotConsumeLimitOffset() throws Exception { + QueryStatement queryStatement = + (QueryStatement) + StatementGenerator.createStatement( + "select __endTime, count(s1) from root.sg.* " + + "group by ([0, 100), 10ms) limit 5 offset 12 align by device", + ZonedDateTime.now().getOffset()); + + boolean canBuildPlan = + TemplatedAggregationAnalyze.canBuildAggregationPlanUseTemplate( + new Analysis(), + queryStatement, + null, + null, + createQueryContext(), + createAlignedTemplate(), + Arrays.asList(new PartialPath("root.sg.d1"), new PartialPath("root.sg.d2"))); + + assertFalse(canBuildPlan); + assertEquals( + Arrays.asList(5L, 12L, 0L, 100L), + Arrays.asList( + queryStatement.getRowLimit(), + queryStatement.getRowOffset(), + queryStatement.getGroupByTimeComponent().getStartTime(), + queryStatement.getGroupByTimeComponent().getEndTime())); + } + + private static MPPQueryContext createQueryContext() { + return new MPPQueryContext( + "", + new QueryId("test_query"), + new SessionInfo(0, "test", ZoneId.systemDefault()), + new TEndPoint(), + new TEndPoint()); + } + + private static Template createAlignedTemplate() throws Exception { + return new Template( + "template", + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyList(), + true); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecutionRetryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecutionRetryTest.java new file mode 100644 index 0000000000000..1d717d0f7d9e2 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecutionRetryTest.java @@ -0,0 +1,218 @@ +/* + * 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.queryengine.plan.execution; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.execution.QueryStateMachine; +import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis; +import org.apache.iotdb.db.queryengine.plan.analyze.QueryType; +import org.apache.iotdb.db.queryengine.plan.planner.IPlanner; +import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan; +import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan; +import org.apache.iotdb.db.queryengine.plan.scheduler.IScheduler; +import org.apache.iotdb.rpc.RpcUtils; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.lang.reflect.Field; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class QueryExecutionRetryTest { + + private ExecutorService stateMachineExecutor; + + @Before + public void setUp() { + stateMachineExecutor = Executors.newSingleThreadExecutor(); + } + + @After + public void tearDown() { + stateMachineExecutor.shutdownNow(); + } + + @Test + public void testAnalysisAttemptIsRolledBackOnlyWhenRetryStarts() { + RecordingPlanner planner = new RecordingPlanner(false); + QueryExecution execution = + new QueryExecution(planner, createQueryContext("retry_succeeds"), stateMachineExecutor); + + assertEquals(Arrays.asList("begin", "analyze"), planner.getAnalysisEvents()); + + execution.start(); + + assertEquals(1, planner.getScheduleAttempts()); + // PENDING_RETRY must keep the journal alive until the plans from this attempt have been + // stopped. In particular, returning from start() must not commit or roll back the attempt. + assertEquals(Arrays.asList("begin", "analyze"), planner.getAnalysisEvents()); + + ExecutionResult result = execution.getStatus(); + + assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), result.status.code); + assertEquals(2, planner.getScheduleAttempts()); + assertEquals( + Arrays.asList("begin", "analyze", "rollback", "begin", "analyze", "commit"), + planner.getAnalysisEvents()); + } + + @Test + public void testAnalysisAttemptIsRolledBackWhenRetryLimitIsReached() throws Exception { + RecordingPlanner planner = new RecordingPlanner(true); + QueryExecution execution = + new QueryExecution(planner, createQueryContext("retry_exhausted"), stateMachineExecutor); + + execution.start(); + assertEquals(Arrays.asList("begin", "analyze"), planner.getAnalysisEvents()); + + setRetryCount(execution, Integer.MAX_VALUE); + ExecutionResult result = execution.getStatus(); + + assertEquals(TSStatusCode.DISPATCH_ERROR.getStatusCode(), result.status.code); + assertEquals(Arrays.asList("begin", "analyze", "rollback"), planner.getAnalysisEvents()); + assertFalse(planner.getAnalysisEvents().contains("commit")); + } + + private static MPPQueryContext createQueryContext(String queryId) { + MPPQueryContext context = + new MPPQueryContext( + "", + new QueryId(queryId), + new SessionInfo(0, "test", ZoneId.systemDefault()), + new TEndPoint(), + new TEndPoint()); + // Keeping this as a non-query avoids constructing a result exchange handle. The fake scheduler + // still drives the same QueryStateMachine transitions used by a retryable query dispatch. + context.setQueryType(QueryType.OTHER); + return context; + } + + private static void setRetryCount(QueryExecution execution, int retryCount) throws Exception { + Field retryCountField = QueryExecution.class.getDeclaredField("retryCount"); + retryCountField.setAccessible(true); + retryCountField.setInt(execution, retryCount); + } + + private static class RecordingPlanner implements IPlanner { + + private final IAnalysis analysis = mock(IAnalysis.class); + private final IScheduler scheduler = mock(IScheduler.class); + private final LogicalQueryPlan logicalPlan = mock(LogicalQueryPlan.class); + private final DistributedQueryPlan distributedPlan = mock(DistributedQueryPlan.class); + private final List analysisEvents = new ArrayList<>(); + private final boolean alwaysFailDispatch; + + private int scheduleAttempts; + + private RecordingPlanner(boolean alwaysFailDispatch) { + this.alwaysFailDispatch = alwaysFailDispatch; + when(distributedPlan.getInstances()).thenReturn(Collections.emptyList()); + } + + @Override + public void beginAnalysisAttempt() { + analysisEvents.add("begin"); + } + + @Override + public IAnalysis analyze(MPPQueryContext context) { + analysisEvents.add("analyze"); + return analysis; + } + + @Override + public void rollbackAnalysisAttempt() { + analysisEvents.add("rollback"); + } + + @Override + public void commitAnalysisAttempt() { + analysisEvents.add("commit"); + } + + @Override + public LogicalQueryPlan doLogicalPlan(IAnalysis analysis, MPPQueryContext context) { + return logicalPlan; + } + + @Override + public DistributedQueryPlan doDistributionPlan( + IAnalysis analysis, LogicalQueryPlan logicalPlan, MPPQueryContext context) { + return distributedPlan; + } + + @Override + public IScheduler doSchedule( + IAnalysis analysis, + DistributedQueryPlan distributedPlan, + MPPQueryContext context, + QueryStateMachine stateMachine) { + scheduleAttempts++; + stateMachine.transitionToDispatching(); + if (alwaysFailDispatch || scheduleAttempts == 1) { + TSStatus dispatchFailure = RpcUtils.getStatus(TSStatusCode.DISPATCH_ERROR); + stateMachine.transitionToPendingRetry(dispatchFailure); + } else { + stateMachine.transitionToRunning(); + } + return scheduler; + } + + @Override + public void invalidatePartitionCache() { + // No cache is used by this fake planner. + } + + @Override + public ScheduledExecutorService getScheduledExecutorService() { + return null; + } + + @Override + public void setRedirectInfo(IAnalysis analysis, TEndPoint localEndPoint, TSStatus status) { + // Redirect information is irrelevant to the retry journal lifecycle. + } + + private List getAnalysisEvents() { + return analysisEvents; + } + + private int getScheduleAttempts() { + return scheduleAttempts; + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilderTest.java new file mode 100644 index 0000000000000..bc3f7b8018a9c --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilderTest.java @@ -0,0 +1,66 @@ +/* + * 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.queryengine.plan.planner; + +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.plan.analyze.Analysis; +import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator; +import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement; + +import org.junit.Test; + +import java.time.ZonedDateTime; +import java.util.Arrays; +import java.util.Collections; + +import static org.junit.Assert.assertEquals; + +public class LogicalPlanBuilderTest { + + @Test + public void testPlanDeviceViewDoesNotAppendDefaultSortItemsToStatement() { + QueryStatement queryStatement = + (QueryStatement) + StatementGenerator.createStatement( + "select s1 from root.sg.* order by time desc align by device", + ZonedDateTime.now().getOffset()); + int originalSortItemCount = queryStatement.getSortItemList().size(); + + planDeviceView(queryStatement, "first_plan"); + int sortItemCountAfterFirstPlan = queryStatement.getSortItemList().size(); + planDeviceView(queryStatement, "retry_plan"); + + assertEquals( + Arrays.asList(originalSortItemCount, originalSortItemCount), + Arrays.asList(sortItemCountAfterFirstPlan, queryStatement.getSortItemList().size())); + } + + private static void planDeviceView(QueryStatement queryStatement, String queryId) { + Analysis analysis = new Analysis(); + new LogicalPlanBuilder(analysis, new MPPQueryContext(new QueryId(queryId))) + .planDeviceView( + Collections.emptyMap(), + Collections.emptySet(), + Collections.emptyMap(), + Collections.emptySet(), + queryStatement, + analysis); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java index 107070425e58b..5b5a384f756af 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java @@ -47,8 +47,11 @@ import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.LogicalExpression; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement; import org.apache.iotdb.commons.queryengine.plan.relational.type.InternalTypeManager; +import org.apache.iotdb.commons.schema.filter.SchemaFilter; import org.apache.iotdb.commons.schema.table.InsertNodeMeasurementInfo; import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.AttributeColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TagColumnSchema; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.protocol.session.InternalClientSession; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; @@ -73,6 +76,8 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExchangeNode; import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl; import org.apache.iotdb.db.queryengine.plan.relational.security.AllowAllAccessControl; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AbstractQueryDeviceWithCache; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDevice; import org.apache.iotdb.db.queryengine.plan.relational.sql.parser.SqlParser; import org.apache.iotdb.db.queryengine.plan.relational.sql.rewrite.StatementRewriteFactory; import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; @@ -81,6 +86,8 @@ import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import com.google.common.collect.ImmutableSet; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.IDeviceID.Factory; import org.apache.tsfile.utils.Binary; import org.junit.BeforeClass; @@ -127,7 +134,9 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.eq; @@ -1103,6 +1112,114 @@ public void duplicateProjectionsTest() { analysis.getRespDatasetHeader().getColumnNameIndexMap().get("TAG1").intValue()); } + @Test + public void testShowDeviceWithWhereCanBeAnalyzedAgain() { + assertDeviceQueryCanBeAnalyzedAgain("SHOW DEVICES FROM table1 WHERE tag1 = 'shanghai'"); + } + + @Test + public void testCountDeviceWithWhereCanBeAnalyzedAgain() { + assertDeviceQueryCanBeAnalyzedAgain("COUNT DEVICES FROM table1 WHERE tag1 = 'shanghai'"); + } + + @Test + public void testInternalShowDeviceKeepsPreparedTraversalState() { + final String database = "testdb"; + final TsTable tsTable = new TsTable(table); + tsTable.addColumnSchema(new TagColumnSchema("tag1", TSDataType.STRING)); + tsTable.addColumnSchema(new AttributeColumnSchema("attr1", TSDataType.STRING)); + DataNodeTableCache.getInstance().preUpdateTable(database, tsTable, null); + DataNodeTableCache.getInstance().commitUpdateTable(database, table, null); + + try { + final ShowDevice statement = new ShowDevice(database, table); + final List> tagFilters = + Collections.singletonList(Collections.emptyList()); + final List partitionKeys = + Collections.singletonList( + Factory.DEFAULT_FACTORY.create(new String[] {table, "shanghai"})); + final List attributeColumns = Collections.singletonList("attr1"); + statement.setTagDeterminedFilterList(tagFilters); + statement.setPartitionKeyList(partitionKeys); + statement.setAttributeColumns(attributeColumns); + + final SessionInfo session = + new SessionInfo(0, "test", ZoneId.systemDefault(), database, SqlDialect.TABLE); + final MPPQueryContext queryContext = + new MPPQueryContext("", new QueryId("internal_show_device"), session, null, null); + final Analysis analysis = + analyzeStatement(statement, TEST_MATADATA, queryContext, new SqlParser(), session); + + final ShowDevice analyzedStatement = (ShowDevice) analysis.getStatement(); + assertSame(statement, analyzedStatement); + assertSame(tagFilters, analyzedStatement.getTagDeterminedFilterList()); + assertSame(partitionKeys, analyzedStatement.getPartitionKeyList()); + assertSame(attributeColumns, analyzedStatement.getAttributeColumns()); + } finally { + DataNodeTableCache.getInstance().invalid(database); + } + } + + private void assertDeviceQueryCanBeAnalyzedAgain(final String sql) { + final String database = "testdb"; + final TsTable tsTable = new TsTable(table); + tsTable.addColumnSchema(new TagColumnSchema("tag1", TSDataType.STRING)); + tsTable.addColumnSchema(new TagColumnSchema("tag2", TSDataType.STRING)); + tsTable.addColumnSchema(new TagColumnSchema("tag3", TSDataType.STRING)); + tsTable.addColumnSchema(new AttributeColumnSchema("attr1", TSDataType.STRING)); + tsTable.addColumnSchema(new AttributeColumnSchema("attr2", TSDataType.STRING)); + DataNodeTableCache.getInstance().preUpdateTable(database, tsTable, null); + DataNodeTableCache.getInstance().commitUpdateTable(database, table, null); + + try { + final SqlParser sqlParser = new SqlParser(); + final Statement statement = + sqlParser.createStatement( + sql, ZoneId.systemDefault(), new InternalClientSession("testClient")); + final SessionInfo session = + new SessionInfo(0, "test", ZoneId.systemDefault(), database, SqlDialect.TABLE); + final MPPQueryContext queryContext = + new MPPQueryContext(sql, new QueryId("device_query_retry"), session, null, null); + + final AbstractQueryDeviceWithCache parsedStatement = (AbstractQueryDeviceWithCache) statement; + final Expression parsedWhere = parsedStatement.getWhere().orElseThrow(AssertionError::new); + final String parsedStatementText = parsedStatement.toString(); + final long parsedStatementSize = parsedStatement.ramBytesUsed(); + + final Analysis firstAnalysis = + analyzeStatement(statement, TEST_MATADATA, queryContext, sqlParser, session); + assertParsedDeviceStatementUnchanged( + parsedStatement, parsedWhere, parsedStatementText, parsedStatementSize); + assertNotSame(statement, firstAnalysis.getStatement()); + + queryContext.prepareForRetry(); + final Analysis retryAnalysis = + analyzeStatement(statement, TEST_MATADATA, queryContext, sqlParser, session); + assertParsedDeviceStatementUnchanged( + parsedStatement, parsedWhere, parsedStatementText, parsedStatementSize); + assertNotSame(statement, retryAnalysis.getStatement()); + assertNotSame(firstAnalysis.getStatement(), retryAnalysis.getStatement()); + } finally { + DataNodeTableCache.getInstance().invalid(database); + } + } + + private void assertParsedDeviceStatementUnchanged( + final AbstractQueryDeviceWithCache parsedStatement, + final Expression parsedWhere, + final String parsedStatementText, + final long parsedStatementSize) { + assertSame(parsedWhere, parsedStatement.getWhere().orElseThrow(AssertionError::new)); + assertEquals(parsedStatementText, parsedStatement.toString()); + assertEquals(parsedStatementSize, parsedStatement.ramBytesUsed()); + assertNull(parsedStatement.getDatabase()); + assertNull(parsedStatement.getTableName()); + assertNull(parsedStatement.getTagFuzzyPredicate()); + assertNull(parsedStatement.getPartitionKeyList()); + assertNull(parsedStatement.getAttributeColumns()); + assertNull(parsedStatement.getColumnHeaderList()); + } + private Metadata mockMetadataForInsertion() { final TsTable tsTable = StatementTestUtils.genTsTable(); DataNodeTableCache.getInstance().preUpdateTable(database, tsTable, null); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/statement/QueryStatementTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/statement/QueryStatementTest.java index ad812e74a0c6e..7fec82a77af3c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/statement/QueryStatementTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/statement/QueryStatementTest.java @@ -21,7 +21,9 @@ import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.db.queryengine.plan.expression.Expression; import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator; +import org.apache.iotdb.db.queryengine.plan.statement.component.OrderByComponent; import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement; import org.apache.tsfile.utils.Pair; @@ -35,6 +37,8 @@ import static org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement.RAW_AGGREGATION_HYBRID_QUERY_ERROR_MSG; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; import static org.junit.Assert.fail; public class QueryStatementTest { @@ -155,6 +159,28 @@ public void getPathsTest() { assertEquals("root.sg.d1.s3", fullPaths.get(1).getFullPath()); } + @Test + public void testOrderByCopyPreservesRawAndNormalizedExpressions() { + QueryStatement statement = + (QueryStatement) + StatementGenerator.createStatement( + "SELECT s1 FROM root.sg.d1 ORDER BY Sin(s1)", ZonedDateTime.now().getOffset()); + OrderByComponent source = statement.getOrderByComponent(); + Expression rawExpression = source.getSortItemList().get(0).getExpression(); + Expression normalizedExpression = source.getExpressionSortItemList().get(0); + + assertEquals("Sin(s1)", rawExpression.getExpressionString()); + assertEquals("sin(s1)", normalizedExpression.getExpressionString()); + assertNotSame(rawExpression, normalizedExpression); + + OrderByComponent copy = OrderByComponent.copyOf(source); + + assertNotSame(source.getSortItemList(), copy.getSortItemList()); + assertNotSame(source.getSortItemList().get(0), copy.getSortItemList().get(0)); + assertSame(rawExpression, copy.getSortItemList().get(0).getExpression()); + assertSame(normalizedExpression, copy.getExpressionSortItemList().get(0)); + } + private void checkErrorQuerySql(String sql) { QueryStatement statement = (QueryStatement) StatementGenerator.createStatement(sql, ZonedDateTime.now().getOffset());