This is an automated email from the ASF dual-hosted git repository. JackieTien97 pushed a commit to branch codex/backport-pr-18179-dev-1.3 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d26bd28fb6ff72d696371cbebe97dbd6eba34c4a Author: JackieTien97 <[email protected]> AuthorDate: Sat Jul 11 11:35:08 2026 +0800 Fix query analysis state handling on retry --- .../queryengine/plan/analyze/AnalyzeVisitor.java | 81 ++++--- .../db/queryengine/plan/analyze/Analyzer.java | 16 +- .../plan/analyze/ConcatPathRewriter.java | 23 +- .../queryengine/plan/analyze/PredicateUtils.java | 175 ++++++++------ .../plan/analyze/TemplatedAggregationAnalyze.java | 66 ++++-- .../queryengine/plan/analyze/TemplatedAnalyze.java | 20 +- .../plan/analyze/TreeAnalysisMutationJournal.java | 255 +++++++++++++++++++++ .../queryengine/plan/execution/QueryExecution.java | 42 +++- .../visitor/predicate/PredicateSimplifier.java | 83 ++++--- .../db/queryengine/plan/planner/IPlanner.java | 6 + .../plan/planner/LogicalPlanBuilder.java | 9 +- .../queryengine/plan/planner/TreeModelPlanner.java | 20 +- .../plan/statement/component/GroupByComponent.java | 5 + .../plan/statement/component/OrderByComponent.java | 30 +++ .../plan/statement/component/WhereCondition.java | 7 + .../plan/statement/crud/QueryStatement.java | 13 -- .../plan/analyze/PredicateUtilsTest.java | 172 ++++++++++++++ .../plan/analyze/TemplatedAnalyzeRetryTest.java | 126 ++++++++++ .../plan/execution/QueryExecutionRetryTest.java | 218 ++++++++++++++++++ .../plan/planner/LogicalPlanBuilderTest.java | 66 ++++++ 20 files changed, 1250 insertions(+), 183 deletions(-) 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 973b3553c78..1536c3cc865 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 @@ -244,14 +244,19 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> private final IPartitionFetcher partitionFetcher; private final ISchemaFetcher schemaFetcher; private final IModelFetcher modelFetcher; + private final TreeAnalysisMutationJournal mutationJournal; private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS = PerformanceOverviewMetrics.getInstance(); - 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 @@ -284,15 +289,20 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> @Override public Analysis visitQuery(QueryStatement queryStatement, MPPQueryContext context) { - // used for retry - queryStatement.setWhereConditionBackToOriginal(); - Analysis analysis = new Analysis(); analysis.setLastLevelUseWildcard(queryStatement.isLastLevelUseWildcard()); try { // check for semantic errors - queryStatement.semanticCheck(); + boolean originalCountTimeAggregation = queryStatement.isCountTimeAggregation(); + try { + queryStatement.semanticCheck(); + } finally { + if (queryStatement.isCountTimeAggregation() != originalCountTimeAggregation) { + mutationJournal.recordCountTimeAggregationChange( + queryStatement, originalCountTimeAggregation); + } + } context.initResultSetColumnMemoryTracking( queryStatement.getSeriesLimit(), @@ -322,7 +332,13 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> 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; @@ -330,6 +346,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> if (canPushDownLimitOffsetInGroupByTimeForDevice(queryStatement)) { // remove the device which won't appear in resultSet after limit/offset + mutationJournal.recordLimitOffsetPushDown(queryStatement); deviceList = pushDownLimitOffsetInGroupByTimeForDevice( deviceList, queryStatement, context.getZoneId()); @@ -514,7 +531,10 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> queryStatement = (QueryStatement) concatPathRewriter.rewrite( - queryStatement, new PathPatternTree(queryStatement.useWildcard()), context); + queryStatement, + new PathPatternTree(queryStatement.useWildcard()), + context, + mutationJournal); analysis.setStatement(queryStatement); // request schema fetch API @@ -575,29 +595,24 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> Expression globalTimePredicate = null; boolean hasValueFilter = false; if (queryStatement.getWhereCondition() != null) { - WhereCondition whereCondition = queryStatement.getWhereCondition(); - Expression predicate = whereCondition.getPredicate(); - - WhereCondition originalWhereCondition = new WhereCondition(predicate); - - Pair<Expression, Boolean> 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); + hasValueFilter = extractionResult.hasValueFilter(); + Expression residualPredicate = + PredicateUtils.simplifyPredicate(extractionResult.getResidualPredicate()); // 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); + if (!hasValueFilter || residualPredicate.equals(ConstantOperand.TRUE)) { + mutationJournal.replaceWhereCondition(queryStatement, null); + } else if (residualPredicate != predicate) { + mutationJournal.replaceWhereCondition( + queryStatement, WhereCondition.fromNormalizedPredicate(residualPredicate)); } - - queryStatement.setOriginalWhereCondition(originalWhereCondition); } analysis.setGlobalTimePredicate(globalTimePredicate); analysis.setHasValueFilter(hasValueFilter); @@ -1904,7 +1919,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> orderByExpressions.add(orderByExpression); } analysis.setOrderByExpressions(orderByExpressions); - queryStatement.updateSortItems(orderByExpressions); + mutationJournal.updateSortItems(queryStatement, orderByExpressions); } private void analyzeOrderBy( @@ -2074,7 +2089,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> } analysis.setOrderByExpressions(deviceViewOrderByExpression); - queryStatement.updateSortItems(deviceViewOrderByExpression); + mutationJournal.updateSortItems(queryStatement, deviceViewOrderByExpression); analysis.setDeviceToSortItems(deviceToSortItems); analysis.setDeviceToOrderByExpressions(deviceToOrderByExpressions); } @@ -2394,7 +2409,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> if (!queryStatement.isSelectInto()) { return; } - queryStatement.setOrderByComponent(null); + mutationJournal.clearOrderByComponent(queryStatement); List<PartialPath> sourceDevices = new ArrayList<>(deviceSet); List<Expression> sourceColumns = @@ -2449,7 +2464,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> if (!queryStatement.isSelectInto()) { return; } - queryStatement.setOrderByComponent(null); + mutationJournal.clearOrderByComponent(queryStatement); List<Expression> sourceColumns = outputExpressions.stream() @@ -3407,17 +3422,17 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext> private void analyzeGlobalTimeConditionInShowMetaData( WhereCondition timeCondition, Analysis analysis) { Expression predicate = timeCondition.getPredicate(); - Pair<Expression, Boolean> resultPair = - PredicateUtils.extractGlobalTimePredicate(predicate, true, true); - if (resultPair.right) { + PredicateUtils.GlobalTimePredicateExtractionResult extractionResult = + PredicateUtils.extractGlobalTimePredicate(predicate); + if (extractionResult.hasValueFilter()) { throw new SemanticException( "Value Filter can't exist in the condition of SHOW/COUNT clause, only time condition supported"); } - if (resultPair.left == null) { + if (extractionResult.getGlobalTimePredicate() == null) { throw new SemanticException( "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); } 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 3dfaa0a1137..3f4f692bbfa 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 class Analyzer { }); 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 041f08e1b5c..a1dd315517e 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 @@ -50,7 +50,10 @@ public class ConcatPathRewriter { } 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; @@ -78,20 +81,20 @@ public class ConcatPathRewriter { // concat SELECT with FROM List<ResultColumn> 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<Expression> sortItemExpressions = queryStatement.getExpressionSortItemList(); + List<Expression> 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 4a65ebbfe72..5eb4d4f4334 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 @@ -44,7 +44,6 @@ import org.apache.iotdb.db.queryengine.plan.expression.visitor.predicate.Reverse 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; @@ -61,88 +60,85 @@ public class PredicateUtils { } /** - * Extract global time predicate from query predicate. + * Extracts the global time predicate and the predicate that still needs to be evaluated. + * + * <p>This method never modifies {@code predicate}. A {@code null} global time predicate + * represents {@code TRUE}. The returned predicates satisfy the following invariant: + * + * <pre> + * predicate == globalTimePredicate AND residualPredicate + * </pre> + * + * 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<Expression, Boolean> extractGlobalTimePredicate( - Expression predicate, boolean canRewrite, boolean isFirstOr) { + public static GlobalTimePredicateExtractionResult extractGlobalTimePredicate( + Expression predicate) { if (predicate.getExpressionType().equals(ExpressionType.LOGIC_AND)) { - Pair<Expression, Boolean> leftResultPair = - extractGlobalTimePredicate( - ((BinaryExpression) predicate).getLeftExpression(), canRewrite, isFirstOr); - Pair<Expression, Boolean> 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")); - } - } + GlobalTimePredicateExtractionResult leftResult = + extractGlobalTimePredicate(((BinaryExpression) predicate).getLeftExpression()); + GlobalTimePredicateExtractionResult rightResult = + extractGlobalTimePredicate(((BinaryExpression) predicate).getRightExpression()); - 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); + 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<Expression, Boolean> leftResultPair = - extractGlobalTimePredicate( - ((BinaryExpression) predicate).getLeftExpression(), false, false); - Pair<Expression, Boolean> 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<Expression, Boolean> 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(); @@ -157,31 +153,68 @@ public class PredicateUtils { 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 67837a6b78c..b7ac5792079 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 @@ -39,6 +39,7 @@ import org.apache.tsfile.write.schema.IMeasurementSchema; 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.Set; @@ -68,6 +69,26 @@ public class TemplatedAggregationAnalyze { MPPQueryContext context, Template template, List<PartialPath> 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<PartialPath> deviceList, + TreeAnalysisMutationJournal mutationJournal) { // not support order by expression and non-aligned template if (queryStatement.hasOrderByExpression() || !template.isDirectAligned()) { @@ -76,13 +97,6 @@ public class TemplatedAggregationAnalyze { 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<Pair<Expression, String>> outputExpressions = new ArrayList<>(); boolean valid = analyzeSelect(queryStatement, analysis, outputExpressions, template); if (!valid) { @@ -110,6 +124,21 @@ public class TemplatedAggregationAnalyze { 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); @@ -149,13 +178,24 @@ public class TemplatedAggregationAnalyze { 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. + FunctionExpression templatedCountTimeExpression = + new FunctionExpression( + countTimeExpression.getFunctionName(), + new LinkedHashMap<>(countTimeExpression.getFunctionAttributes()), + 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 157ddd723cd..a1a69aa2ae5 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 @@ -97,7 +97,8 @@ public class TemplatedAnalyze { IPartitionFetcher partitionFetcher, ISchemaTree schemaTree, MPPQueryContext context, - List<PartialPath> deviceList) { + List<PartialPath> deviceList, + TreeAnalysisMutationJournal mutationJournal) { if (queryStatement.getGroupByComponent() != null || queryStatement.isSelectInto() || queryStatement.hasFill() @@ -114,7 +115,14 @@ public class TemplatedAnalyze { if (queryStatement.isAggregationQuery()) { return canBuildAggregationPlanUseTemplate( - analysis, queryStatement, partitionFetcher, schemaTree, context, template, deviceList); + analysis, + queryStatement, + partitionFetcher, + schemaTree, + context, + template, + deviceList, + mutationJournal); } List<Pair<Expression, String>> outputExpressions = new ArrayList<>(); @@ -189,7 +197,8 @@ public class TemplatedAnalyze { } analysis.setDeviceList(deviceList); - analyzeDeviceToOrderBy(analysis, queryStatement, schemaTree, deviceList, context); + analyzeDeviceToOrderBy( + analysis, queryStatement, schemaTree, deviceList, context, mutationJournal); analyzeDeviceToSourceTransform(analysis); analyzeDeviceToSource(analysis); @@ -277,7 +286,8 @@ public class TemplatedAnalyze { QueryStatement queryStatement, ISchemaTree schemaTree, List<PartialPath> deviceSet, - MPPQueryContext queryContext) { + MPPQueryContext queryContext, + TreeAnalysisMutationJournal mutationJournal) { if (!queryStatement.hasOrderByExpression()) { return; } @@ -322,7 +332,7 @@ public class TemplatedAnalyze { } 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 00000000000..767afd2d215 --- /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. + * + * <p>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. + * + * <p>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<ResultColumn> 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. + * + * <p>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. + * + * <p>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<ResultColumn> 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<Expression> 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 5af143bfca8..75598e46791 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 @@ -124,7 +124,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); @@ -159,6 +165,15 @@ public class QueryExecution implements IQueryExecution { } public void start() { + try { + startInternal(); + } catch (RuntimeException | Error e) { + finishAnalysisAttemptForCurrentState(); + throw e; + } + } + + private void startInternal() { final long startTime = System.nanoTime(); if (skipExecute()) { LOGGER.debug("[SkipExecute]"); @@ -168,6 +183,7 @@ public class QueryExecution implements IQueryExecution { constructResultForMemorySource(); stateMachine.transitionToRunning(); } + finishAnalysisAttemptForCurrentState(); return; } @@ -200,6 +216,7 @@ public class QueryExecution implements IQueryExecution { if (context.getQueryType() == QueryType.WRITE && analysis.isFailed()) { stateMachine.transitionToFailed(analysis.getFailStatus()); } + finishAnalysisAttemptForCurrentState(); } private void checkTimeOutForQuery() { @@ -217,12 +234,15 @@ public class QueryExecution implements IQueryExecution { private ExecutionResult retry() { if (retryCount >= MAX_RETRY_COUNT) { LOGGER.warn("[ReachMaxRetryCount]"); + this.stopAndCleanup(stateMachine.getFailureException()); + planner.rollbackAnalysisAttempt(); stateMachine.transitionToFailed(); return getStatus(); } LOGGER.warn("error when executing query. {}", stateMachine.getFailureMessage()); // stop and clean up resources the QueryExecution used this.stopAndCleanup(stateMachine.getFailureException()); + planner.rollbackAnalysisAttempt(); LOGGER.info("[WaitBeforeRetry] wait {}ms.", RETRY_INTERVAL_IN_MS); try { Thread.sleep(RETRY_INTERVAL_IN_MS); @@ -241,12 +261,30 @@ public class QueryExecution implements IQueryExecution { 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 be81a33587c..b2518f7a7e6 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 @@ package org.apache.iotdb.db.queryengine.plan.expression.visitor.predicate; 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 org.apache.iotdb.db.queryengine.plan.expression.visitor.ExpressionAnalyze 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<Expression, Void> { @@ -59,8 +64,7 @@ public class PredicateSimplifier extends ExpressionAnalyzeVisitor<Expression, Vo return ConstantOperand.TRUE; } } - isNullExpression.setExpression(simplifiedInputExpression); - return isNullExpression; + return reconstructUnaryIfNeeded(isNullExpression, simplifiedInputExpression); } @Override @@ -83,8 +87,7 @@ public class PredicateSimplifier extends ExpressionAnalyzeVisitor<Expression, Vo if (simplifiedInputExpression.getExpressionType().equals(ExpressionType.NULL)) { return ConstantOperand.FALSE; } - unaryExpression.setExpression(simplifiedInputExpression); - return unaryExpression; + return reconstructUnaryIfNeeded(unaryExpression, simplifiedInputExpression); } @Override @@ -96,8 +99,7 @@ public class PredicateSimplifier extends ExpressionAnalyzeVisitor<Expression, Vo if (simplifiedInputExpression.equals(ConstantOperand.FALSE)) { return ConstantOperand.TRUE; } - logicNotExpression.setExpression(simplifiedInputExpression); - return logicNotExpression; + return reconstructUnaryIfNeeded(logicNotExpression, simplifiedInputExpression); } @Override @@ -106,8 +108,7 @@ public class PredicateSimplifier extends ExpressionAnalyzeVisitor<Expression, Vo if (simplifiedInputExpression.getExpressionType().equals(ExpressionType.NULL)) { return new NullOperand(); } - negationExpression.setExpression(simplifiedInputExpression); - return negationExpression; + return reconstructUnaryIfNeeded(negationExpression, simplifiedInputExpression); } @Override @@ -121,9 +122,7 @@ public class PredicateSimplifier extends ExpressionAnalyzeVisitor<Expression, Vo // 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 class PredicateSimplifier extends ExpressionAnalyzeVisitor<Expression, Vo 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 class PredicateSimplifier extends ExpressionAnalyzeVisitor<Expression, 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 class PredicateSimplifier extends ExpressionAnalyzeVisitor<Expression, Vo // 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 @@ public class PredicateSimplifier extends ExpressionAnalyzeVisitor<Expression, Vo 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 <T extends Comparable<T>> boolean lessThan( @@ -243,16 +242,42 @@ public class PredicateSimplifier extends ExpressionAnalyzeVisitor<Expression, Vo @Override public Expression visitFunctionExpression(FunctionExpression functionExpression, Void context) { - List<Expression> simplifiedInputExpressions = new ArrayList<>(); - for (Expression inputExpression : functionExpression.getExpressions()) { + List<Expression> inputExpressions = functionExpression.getExpressions(); + List<Expression> 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 b9b239b5e26..9273203cc4d 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(IAnalysis analysis, LogicalQueryPlan logicalPlan); 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 32d7b312257..32d663c84e2 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 @@ -486,11 +486,10 @@ public class LogicalPlanBuilder { .map(Expression::getExpressionString) .collect(Collectors.toList()); - List<SortItem> 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. + List<SortItem> sortItemList = new ArrayList<>(queryStatement.getSortItemList()); if (!queryStatement.isOrderByDevice()) { sortItemList.add(new SortItem(OrderByKey.DEVICE, Ordering.ASC)); } 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 2812d81cf27..62aa7385707 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 @@ -30,6 +30,7 @@ import org.apache.iotdb.db.queryengine.plan.analyze.Analysis; 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; @@ -53,6 +54,7 @@ import java.util.concurrent.ScheduledExecutorService; public class TreeModelPlanner implements IPlanner { private final Statement statement; + private final TreeAnalysisMutationJournal mutationJournal = new TreeAnalysisMutationJournal(); private final ScheduledExecutorService scheduledExecutor; @@ -84,7 +86,23 @@ public class TreeModelPlanner implements IPlanner { @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/statement/component/GroupByComponent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/component/GroupByComponent.java index ec45a90a4b8..b1d5fefde52 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 abstract class GroupByComponent extends StatementNode { 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 8774db378f5..777410c8b0c 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 @@ -48,6 +48,36 @@ public class OrderByComponent extends StatementNode { this.sortItemExpressionList = new ArrayList<>(); } + /** + * Creates a structurally independent component for analysis-time copy-on-write updates. + * + * <p>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 7352c1e96fc..d80677fd830 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 class WhereCondition extends StatementNode { 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 39055f716d6..58f48cd12df 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 @@ -87,9 +87,6 @@ public class QueryStatement extends AuthorityInformationStatement { private FromComponent fromComponent; private WhereCondition whereCondition; - // used for retry - private WhereCondition originalWhereCondition; - private HavingCondition havingCondition; // row limit for result set. The default value is 0, which means no limit @@ -262,16 +259,6 @@ public class QueryStatement extends AuthorityInformationStatement { this.whereCondition = whereCondition; } - public void setWhereConditionBackToOriginal() { - if (originalWhereCondition != null) { - this.whereCondition = originalWhereCondition; - } - } - - public void setOriginalWhereCondition(WhereCondition originalWhereCondition) { - this.originalWhereCondition = originalWhereCondition; - } - public boolean hasHaving() { return havingCondition != null; } 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 00000000000..bc9d252d30e --- /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 00000000000..9a45ac82fb1 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedAnalyzeRetryTest.java @@ -0,0 +1,126 @@ +/* + * 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.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.common.SessionInfo; +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.apache.iotdb.db.schemaengine.template.Template; + +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(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 00000000000..139d9ee74f6 --- /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.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.common.SessionInfo; +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.WRITE); + 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<String> 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) { + 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<String> 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 00000000000..bc3f7b8018a --- /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); + } +}
