This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 21c9f9d7f8f [To dev/1.3] Fix retry bug in tree model query which
ignores the time filter (#18201)
21c9f9d7f8f is described below
commit 21c9f9d7f8f6809e0a7b782b61e4424c8d631103
Author: Jackie Tien <[email protected]>
AuthorDate: Wed Jul 15 10:41:51 2026 +0800
[To dev/1.3] Fix retry bug in tree model query which ignores the time
filter (#18201)
---
.../queryengine/plan/analyze/AnalyzeVisitor.java | 74 +++---
.../db/queryengine/plan/analyze/Analyzer.java | 16 +-
.../plan/analyze/ConcatPathRewriter.java | 23 +-
.../queryengine/plan/analyze/PredicateUtils.java | 175 ++++++++------
.../plan/analyze/TemplatedAggregationAnalyze.java | 71 ++++--
.../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 | 16 +-
.../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/analyze/PredicateUtilsTest.java | 172 ++++++++++++++
.../plan/analyze/TemplatedAnalyzeRetryTest.java | 127 ++++++++++
.../plan/execution/QueryExecutionRetryTest.java | 218 ++++++++++++++++++
.../plan/planner/LogicalPlanBuilderTest.java | 113 +++++++++
19 files changed, 1309 insertions(+), 164 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 ab81236e9ab..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
@@ -289,7 +294,15 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
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(),
@@ -319,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;
@@ -327,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());
@@ -511,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
@@ -572,24 +595,23 @@ 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();
-
- 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));
}
}
analysis.setGlobalTimePredicate(globalTimePredicate);
@@ -1897,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(
@@ -2067,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);
}
@@ -2387,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 =
@@ -2442,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()
@@ -3400,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..43c810ac808 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,29 @@ 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. 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 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..d348f2ba3d1 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,11 @@ 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. Keep the complete parameter in Analysis
because a downstream
+ // SortNode must use the same implicit keys without reading a mutated
QueryStatement.
+ List<SortItem> sortItemList = new
ArrayList<>(queryStatement.getSortItemList());
if (!queryStatement.isOrderByDevice()) {
sortItemList.add(new SortItem(OrderByKey.DEVICE, Ordering.ASC));
}
@@ -499,6 +499,7 @@ public class LogicalPlanBuilder {
}
OrderByParameter orderByParameter = new OrderByParameter(sortItemList);
+ analysis.setMergeOrderParameter(orderByParameter);
long limitValue =
queryStatement.hasOffset()
@@ -1345,7 +1346,10 @@ public class LogicalPlanBuilder {
Set<Expression> 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 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/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..773da476242
--- /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.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(*)",
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 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..07e7baa99fc
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilderTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.planner.plan.node.process.SortNode;
+import org.apache.iotdb.db.queryengine.plan.statement.component.OrderByKey;
+import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
+import org.apache.iotdb.db.queryengine.plan.statement.component.SortItem;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement;
+
+import org.junit.Test;
+
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+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()));
+ }
+
+ @Test
+ public void testPlanOrderByUsesDeviceViewMergeKeysWithoutMutatingStatement()
{
+ QueryStatement queryStatement =
+ (QueryStatement)
+ StatementGenerator.createStatement(
+ "select s1 from root.sg.* order by s1 desc align by device",
+ ZonedDateTime.now().getOffset());
+ List<SortItem> originalSortItems = new
ArrayList<>(queryStatement.getSortItemList());
+ String originalOrderByClause =
queryStatement.getOrderByComponent().toSQLString();
+ List<SortItem> expectedSortItems = new ArrayList<>(originalSortItems);
+ expectedSortItems.add(new SortItem(OrderByKey.DEVICE, Ordering.ASC));
+ expectedSortItems.add(new SortItem(OrderByKey.TIME, Ordering.ASC));
+
+ Analysis analysis = new Analysis();
+ analysis.setOrderByExpressions(Collections.emptySet());
+ analysis.setSelectExpressions(Collections.emptySet());
+ LogicalPlanBuilder builder =
+ new LogicalPlanBuilder(analysis, new MPPQueryContext(new
QueryId("two_stage_plan")))
+ .planDeviceView(
+ Collections.emptyMap(),
+ Collections.emptySet(),
+ Collections.emptyMap(),
+ Collections.emptySet(),
+ queryStatement,
+ analysis)
+ .planOrderBy(queryStatement, analysis);
+
+ assertTrue(builder.getRoot() instanceof SortNode);
+ assertTrue(
+ ((SortNode) builder.getRoot())
+ .getOrderByParameter()
+ .getSortItemList()
+ .get(0)
+ .isExpression());
+ assertEquals(
+ expectedSortItems, ((SortNode)
builder.getRoot()).getOrderByParameter().getSortItemList());
+ assertEquals(originalSortItems, queryStatement.getSortItemList());
+ assertEquals(originalOrderByClause,
queryStatement.getOrderByComponent().toSQLString());
+ }
+
+ 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);
+ }
+}
