From c220ec5748c5d36a28ad5a5e86e19ff1dc8e2b96 Mon Sep 17 00:00:00 2001 From: zxq <3322351820@qq.com> Date: Thu, 12 Nov 2026 22:19:04 +0800 Subject: [PATCH 1/3] Enable ordered rate accumulators for single-device and tag-grouped scans --- .../TableDistributedPlanGenerator.java | 147 +++++++++++++++- .../distribute/RateOrderedInputTest.java | 157 ++++++++++++++++++ .../distribute/RateOrderedPlannerTest.java | 96 +++++++++++ 3 files changed, 393 insertions(+), 7 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedInputTest.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java index 1ead17632aaf1..b8f2cd2ae71a4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java @@ -2077,7 +2077,12 @@ public List visitAggregation(AggregationNode node, PlanContext context // push down aggregation if the child of aggregation node only has the union Node if (childrenNodes.size() == 1) { node.setChild(childrenNodes.get(0)); - AggregationNode physicalAggregation = withRateFunctionInputOrdering(node, childOrdering); + AggregationNode physicalAggregation = + withRateFunctionInputOrdering( + node, + childOrdering, + isChildTimeOrderedPerGroup(node, childrenNodes.get(0)), + childrenNodes.get(0)); if (childrenNodes.get(0) instanceof UnionNode && physicalAggregation.getAggregations().values().stream() @@ -2246,6 +2251,15 @@ private boolean containsUnorderedCollect(PlanNode node) { private static AggregationNode withRateFunctionInputOrdering( AggregationNode node, OrderingScheme childOrdering) { + return withRateFunctionInputOrdering(node, childOrdering, false, null); + } + + private static AggregationNode withRateFunctionInputOrdering( + AggregationNode node, + OrderingScheme childOrdering, + boolean childTimeOrderedPerGroup, + PlanNode child) { + DeviceTableScanNode scan = childTimeOrderedPerGroup ? findOrderedDeviceScan(child) : null; Map aggregations = new LinkedHashMap<>(); node.getAggregations() .forEach( @@ -2259,8 +2273,15 @@ private static AggregationNode withRateFunctionInputOrdering( aggregation.getFilter(), aggregation.getOrderingScheme(), aggregation.getMask(), - isInputOrderedByTimeAscending( - aggregation, node.getStep(), node.getGroupingKeys(), childOrdering)))); + (childTimeOrderedPerGroup + && isOrderedInputFunction( + aggregation.getResolvedFunction().getSignature().getName()) + && isTimeArgumentNativeTime(aggregation, child, scan)) + || isInputOrderedByTimeAscending( + aggregation, + node.getStep(), + node.getGroupingKeys(), + childOrdering)))); return AggregationNode.builderFrom(node).setAggregations(aggregations).build(); } @@ -2273,10 +2294,7 @@ static boolean isInputOrderedByTimeAscending( if (step != SINGLE || childOrdering == null || aggregation.getArguments().size() < 2 - || !(RATE.equalsIgnoreCase(functionName) - || INCREASE.equalsIgnoreCase(functionName) - || IRATE.equalsIgnoreCase(functionName) - || DELTA.equalsIgnoreCase(functionName))) { + || !isOrderedInputFunction(functionName)) { return false; } @@ -2296,6 +2314,121 @@ static boolean isInputOrderedByTimeAscending( return true; } + private boolean isChildTimeOrderedPerGroup(AggregationNode node, PlanNode child) { + DeviceTableScanNode scan = findOrderedDeviceScan(child); + if (scan == null) { + return false; + } + return isDeviceScanTimeOrderedPerGroup( + node.getStep(), + node.getGroupingKeys(), + child, + scan, + analysis.getTableColumnSchema(scan.getQualifiedObjectName())); + } + + static boolean isDeviceScanTimeOrderedPerGroup( + AggregationNode.Step step, + List groupingKeys, + PlanNode child, + DeviceTableScanNode scan, + Map tableColumnSchema) { + if (step != AggregationNode.Step.SINGLE || scan.getScanOrder() != Ordering.ASC) { + return false; + } + int deviceEntryCount = scan.getDeviceEntryCount(); + if (deviceEntryCount <= 0) { + return false; + } + if (deviceEntryCount == 1) { + return true; + } + if (tableColumnSchema == null) { + return false; + } + Set tagColumnNames = new HashSet<>(); + for (ColumnSchema columnSchema : tableColumnSchema.values()) { + if (columnSchema.getColumnCategory() == TsTableColumnCategory.TAG) { + tagColumnNames.add(columnSchema.getName()); + } + } + Map scanColumnSchema = scan.getAssignments(); + Set groupedColumnNames = new HashSet<>(); + for (Symbol symbol : resolveSymbolsToScanLevel(child, groupingKeys)) { + ColumnSchema columnSchema = scanColumnSchema.get(symbol); + if (columnSchema != null) { + groupedColumnNames.add(columnSchema.getName()); + } + } + return groupedColumnNames.containsAll(tagColumnNames); + } + + private static boolean isTimeArgumentNativeTime( + AggregationNode.Aggregation aggregation, PlanNode child, DeviceTableScanNode scan) { + if (scan == null || aggregation.getArguments().size() < 2) { + return false; + } + Expression timeArgument = aggregation.getArguments().get(1); + if (!(timeArgument instanceof SymbolReference)) { + return false; + } + for (Symbol symbol : + resolveSymbolsToScanLevel(child, Collections.singletonList(Symbol.from(timeArgument)))) { + if (scan.isTimeColumn(symbol)) { + return true; + } + } + return false; + } + + static DeviceTableScanNode findOrderedDeviceScan(PlanNode node) { + PlanNode cursor = node; + while (cursor instanceof ProjectNode || cursor instanceof FilterNode) { + if (cursor.getChildren().size() != 1) { + return null; + } + cursor = cursor.getChildren().get(0); + } + if (cursor instanceof AggregationTableScanNode + || cursor instanceof TreeDeviceViewScanNode + || cursor instanceof ExternalTsFileScanNode) { + return null; + } + return cursor instanceof DeviceTableScanNode ? (DeviceTableScanNode) cursor : null; + } + + private static Set resolveSymbolsToScanLevel(PlanNode node, Collection symbols) { + Set current = new HashSet<>(symbols); + PlanNode cursor = node; + while (true) { + if (cursor instanceof ProjectNode) { + Map assignments = ((ProjectNode) cursor).getAssignments().getMap(); + Set mapped = new HashSet<>(); + for (Symbol symbol : current) { + Expression expression = assignments.get(symbol); + if (expression instanceof SymbolReference) { + mapped.add(Symbol.from(expression)); + } + } + current = mapped; + cursor = cursor.getChildren().get(0); + } else if (cursor instanceof FilterNode) { + cursor = cursor.getChildren().get(0); + } else { + return current; + } + } + } + + private static boolean isOrderedInputFunction(String functionName) { + // Aggregation families whose accumulators require strictly ascending time input. New families + // (for example the integral family) must be added here as well. + return RATE.equalsIgnoreCase(functionName) + || INCREASE.equalsIgnoreCase(functionName) + || IRATE.equalsIgnoreCase(functionName) + || DELTA.equalsIgnoreCase(functionName); + } + private boolean prefixMatched(OrderingScheme childOrdering, List preGroupedSymbols) { List orderKeys = childOrdering.getOrderBy(); if (orderKeys.size() < preGroupedSymbols.size()) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedInputTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedInputTest.java new file mode 100644 index 0000000000000..a42b67090b66e --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedInputTest.java @@ -0,0 +1,157 @@ +/* + * 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.relational.planner.distribute; + +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnMetadata; +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.Assignments; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.Symbol; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.AggregationNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.ProjectNode; +import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.SymbolReference; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode; +import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; + +import org.apache.tsfile.read.common.type.StringType; +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.iotdb.commons.queryengine.plan.relational.planner.node.AggregationNode.Step.PARTIAL; +import static org.apache.iotdb.commons.queryengine.plan.relational.planner.node.AggregationNode.Step.SINGLE; +import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.FIELD; +import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.TAG; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +public class RateOrderedInputTest { + + private static final Symbol TAG1 = new Symbol("tag1"); + private static final Symbol TAG2 = new Symbol("tag2"); + private static final Symbol S1 = new Symbol("s1"); + + @Test + public void testSingleDeviceIsOrderedPerGroup() { + assertTrue(isOrderedPerGroup(SINGLE, Collections.emptyList(), scan(1))); + } + + @Test + public void testGroupedByAllTagsIsOrderedPerGroup() { + assertTrue(isOrderedPerGroup(SINGLE, Arrays.asList(TAG1, TAG2), scan(2))); + } + + @Test + public void testGroupedBySubsetOfTagsIsNotOrderedPerGroup() { + assertFalse(isOrderedPerGroup(SINGLE, Collections.singletonList(TAG1), scan(2))); + } + + @Test + public void testDescendingScanIsNotOrderedPerGroup() { + DeviceTableScanNode scan = scan(1); + scan.setScanOrder(Ordering.DESC); + assertFalse(isOrderedPerGroup(SINGLE, Collections.emptyList(), scan)); + } + + @Test + public void testNonSingleStepIsNotOrderedPerGroup() { + assertFalse(isOrderedPerGroup(PARTIAL, Collections.emptyList(), scan(1))); + } + + @Test + public void testGroupingKeysAreResolvedThroughProjections() { + DeviceTableScanNode scan = scan(2); + Symbol tag1Out = new Symbol("tag1_out"); + Symbol tag2Out = new Symbol("tag2_out"); + ProjectNode project = + new ProjectNode( + new PlanNodeId("project"), + scan, + Assignments.builder() + .put(tag1Out, new SymbolReference("tag1")) + .put(tag2Out, new SymbolReference("tag2")) + .build()); + assertTrue(isOrderedPerGroup(SINGLE, Arrays.asList(tag1Out, tag2Out), project)); + } + + @Test + public void testOrderChangingChildIsRejected() { + assertNull(TableDistributedPlanGenerator.findOrderedDeviceScan(Mockito.mock(PlanNode.class))); + } + + private static boolean isOrderedPerGroup( + AggregationNode.Step step, List groupingKeys, PlanNode child) { + DeviceTableScanNode scan = TableDistributedPlanGenerator.findOrderedDeviceScan(child); + if (scan == null) { + return false; + } + return TableDistributedPlanGenerator.isDeviceScanTimeOrderedPerGroup( + step, groupingKeys, child, scan, fullTableColumnSchema()); + } + + private static DeviceTableScanNode scan(int deviceCount) { + Map assignments = new LinkedHashMap<>(); + assignments.put(TAG1, columnSchema("tag1", TAG)); + assignments.put(TAG2, columnSchema("tag2", TAG)); + assignments.put(S1, columnSchema("s1", FIELD)); + Map tagAndAttributeIndexMap = new HashMap<>(); + tagAndAttributeIndexMap.put(TAG1, 0); + tagAndAttributeIndexMap.put(TAG2, 1); + DeviceTableScanNode scan = + new DeviceTableScanNode( + new PlanNodeId("scan"), + new QualifiedObjectName("testdb", "t"), + new ArrayList<>(assignments.keySet()), + assignments, + tagAndAttributeIndexMap); + List deviceEntries = new ArrayList<>(deviceCount); + for (int i = 0; i < deviceCount; i++) { + deviceEntries.add(Mockito.mock(DeviceEntry.class)); + } + scan.setDeviceEntries(deviceEntries); + return scan; + } + + private static Map fullTableColumnSchema() { + Map tableColumnSchema = new LinkedHashMap<>(); + tableColumnSchema.put(TAG1, columnSchema("tag1", TAG)); + tableColumnSchema.put(TAG2, columnSchema("tag2", TAG)); + tableColumnSchema.put(S1, columnSchema("s1", FIELD)); + return tableColumnSchema; + } + + private static ColumnSchema columnSchema(String name, TsTableColumnCategory category) { + return ColumnSchema.builder(new ColumnMetadata(name, StringType.STRING)) + .setColumnCategory(category) + .build(); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java new file mode 100644 index 0000000000000..b4735411eb563 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java @@ -0,0 +1,96 @@ +/* + * 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.relational.planner.distribute; + +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.AggregationNode; +import org.apache.iotdb.db.queryengine.plan.relational.analyzer.TestMetadata; +import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester; + +import org.junit.Test; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +public class RateOrderedPlannerTest { + + @Test + public void testSingleDeviceQueryEnablesOrderedInput() { + AggregationNode aggregation = + planAggregation( + "SELECT irate(s1, time) AS ir FROM table1 " + + "WHERE tag1='shanghai' AND tag2='A3' AND tag3='YY'"); + assertNotNull(aggregation); + assertTrue(hasOrderedInput(aggregation)); + } + + @Test + public void testMultiDeviceQueryKeepsNaiveInput() { + AggregationNode aggregation = planAggregation("SELECT irate(s1, time) AS ir FROM table1"); + assertNotNull(aggregation); + assertFalse(hasOrderedInput(aggregation)); + } + + @Test + public void testNonTimeArgumentKeepsNaiveInput() { + AggregationNode aggregation = + planAggregation( + "SELECT irate(s1, s2) AS ir FROM table1 " + + "WHERE tag1='shanghai' AND tag2='A3' AND tag3='YY'"); + assertNotNull(aggregation); + assertFalse(hasOrderedInput(aggregation)); + } + + private static AggregationNode planAggregation(String sql) { + PlanTester planTester = new PlanTester(new TestMetadata()); + planTester.createPlan(sql); + for (int fragment = 0; fragment < 3; fragment++) { + AggregationNode aggregation; + try { + aggregation = findAggregation(planTester.getFragmentPlan(fragment)); + } catch (RuntimeException e) { + break; + } + if (aggregation != null) { + return aggregation; + } + } + return null; + } + + private static AggregationNode findAggregation(PlanNode node) { + if (node instanceof AggregationNode) { + return (AggregationNode) node; + } + for (PlanNode child : node.getChildren()) { + AggregationNode aggregation = findAggregation(child); + if (aggregation != null) { + return aggregation; + } + } + return null; + } + + private static boolean hasOrderedInput(AggregationNode aggregation) { + return aggregation.getAggregations().values().stream() + .anyMatch(AggregationNode.Aggregation::isInputOrderedByTimeAscending); + } +} From f86deb9f84a11dc2d8ed7b5bbb3c8c6a58b281d1 Mon Sep 17 00:00:00 2001 From: zxq <3322351820@qq.com> Date: Fri, 13 Nov 2026 11:05:57 +0800 Subject: [PATCH 2/3] Address review feedback on ordered rate input detection --- .../relational/planner/TagColumnUtils.java | 67 +++++++++ .../TableDistributedPlanGenerator.java | 118 +++++++++------- .../optimizations/ParallelizeGrouping.java | 19 ++- .../TransformSortToStreamSort.java | 18 +-- .../distribute/RateOrderedInputTest.java | 131 +++++++++++++++++- .../distribute/RateOrderedPlannerTest.java | 94 +++++++++---- 6 files changed, 345 insertions(+), 102 deletions(-) create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TagColumnUtils.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TagColumnUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TagColumnUtils.java new file mode 100644 index 0000000000000..7fa695b7242a5 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TagColumnUtils.java @@ -0,0 +1,67 @@ +/* + * 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.relational.planner; + +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.Symbol; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.TAG; + +/** + * Shared checks for whether a set of columns covers every TAG column of a table. + * + *

A scan produces one device per distinct TAG combination, so "every TAG column is covered" + * implies that each group of the covering columns belongs to at most one device. Distribution, + * grouping parallelization and stream sort all rely on this property, so they must agree on its + * definition. + */ +public final class TagColumnUtils { + + private TagColumnUtils() {} + + /** Returns the names of all TAG columns in the given table schema. */ + public static Set tagColumnNames(Map tableSchema) { + return tagColumnNames(tableSchema.keySet(), tableSchema); + } + + /** Returns the names of the TAG columns among the given symbols. */ + public static Set tagColumnNames( + Collection symbols, Map columnSchema) { + Set tagColumnNames = new HashSet<>(); + for (Symbol symbol : symbols) { + ColumnSchema schema = columnSchema.get(symbol); + if (schema != null && schema.getColumnCategory() == TAG) { + tagColumnNames.add(schema.getName()); + } + } + return tagColumnNames; + } + + /** Whether the given column names cover every TAG column of the table. */ + public static boolean coversAllTagColumns( + Map tableSchema, Collection coveredColumnNames) { + return coveredColumnNames.containsAll(tagColumnNames(tableSchema)); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java index b8f2cd2ae71a4..0aa8c59f5740b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java @@ -105,6 +105,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntrySortedMaterializer; import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.SpilledDeviceEntryDataSet; import org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator; +import org.apache.iotdb.db.queryengine.plan.relational.planner.TagColumnUtils; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTreeDeviceViewScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AlignedAggregationTreeDeviceViewScanNode; @@ -2079,10 +2080,7 @@ public List visitAggregation(AggregationNode node, PlanContext context node.setChild(childrenNodes.get(0)); AggregationNode physicalAggregation = withRateFunctionInputOrdering( - node, - childOrdering, - isChildTimeOrderedPerGroup(node, childrenNodes.get(0)), - childrenNodes.get(0)); + node, childOrdering, findTimeOrderedDeviceScan(node, childrenNodes.get(0))); if (childrenNodes.get(0) instanceof UnionNode && physicalAggregation.getAggregations().values().stream() @@ -2251,15 +2249,16 @@ private boolean containsUnorderedCollect(PlanNode node) { private static AggregationNode withRateFunctionInputOrdering( AggregationNode node, OrderingScheme childOrdering) { - return withRateFunctionInputOrdering(node, childOrdering, false, null); + return withRateFunctionInputOrdering(node, childOrdering, null); } + /** + * @param timeOrderedScan the child's scan when the scan input is already time ordered per group, + * or {@code null} when no such guarantee was established + */ private static AggregationNode withRateFunctionInputOrdering( - AggregationNode node, - OrderingScheme childOrdering, - boolean childTimeOrderedPerGroup, - PlanNode child) { - DeviceTableScanNode scan = childTimeOrderedPerGroup ? findOrderedDeviceScan(child) : null; + AggregationNode node, OrderingScheme childOrdering, DeviceTableScanNode timeOrderedScan) { + PlanNode child = node.getChild(); Map aggregations = new LinkedHashMap<>(); node.getAggregations() .forEach( @@ -2273,10 +2272,8 @@ private static AggregationNode withRateFunctionInputOrdering( aggregation.getFilter(), aggregation.getOrderingScheme(), aggregation.getMask(), - (childTimeOrderedPerGroup - && isOrderedInputFunction( - aggregation.getResolvedFunction().getSignature().getName()) - && isTimeArgumentNativeTime(aggregation, child, scan)) + (timeOrderedScan != null + && isTimeArgumentNativeTime(aggregation, child, timeOrderedScan)) || isInputOrderedByTimeAscending( aggregation, node.getStep(), @@ -2290,15 +2287,17 @@ static boolean isInputOrderedByTimeAscending( AggregationNode.Step step, List groupingKeys, OrderingScheme childOrdering) { - String functionName = aggregation.getResolvedFunction().getSignature().getName(); - if (step != SINGLE - || childOrdering == null - || aggregation.getArguments().size() < 2 - || !isOrderedInputFunction(functionName)) { + if (step != SINGLE || childOrdering == null) { + return false; + } + int timeArgumentIndex = + getOrderedInputTimeArgumentIndex( + aggregation.getResolvedFunction().getSignature().getName()); + if (timeArgumentIndex < 0 || aggregation.getArguments().size() <= timeArgumentIndex) { return false; } - Symbol timeSymbol = Symbol.from(aggregation.getArguments().get(1)); + Symbol timeSymbol = Symbol.from(aggregation.getArguments().get(timeArgumentIndex)); List orderBy = childOrdering.getOrderBy(); int timeIndex = orderBy.indexOf(timeSymbol); if (timeIndex < 0 || !childOrdering.getOrdering(timeSymbol).isAscending()) { @@ -2314,17 +2313,31 @@ static boolean isInputOrderedByTimeAscending( return true; } - private boolean isChildTimeOrderedPerGroup(AggregationNode node, PlanNode child) { + /** + * Returns the child's scan when its input is time ordered per output group, or {@code null}. Only + * ordered-input families benefit from the guarantee, so the table schema is not loaded for plain + * aggregations such as {@code count} or {@code sum}. + */ + private DeviceTableScanNode findTimeOrderedDeviceScan(AggregationNode node, PlanNode child) { + if (node.getAggregations().values().stream() + .noneMatch( + aggregation -> + isOrderedInputFunction( + aggregation.getResolvedFunction().getSignature().getName()))) { + return null; + } DeviceTableScanNode scan = findOrderedDeviceScan(child); if (scan == null) { - return false; + return null; } return isDeviceScanTimeOrderedPerGroup( - node.getStep(), - node.getGroupingKeys(), - child, - scan, - analysis.getTableColumnSchema(scan.getQualifiedObjectName())); + node.getStep(), + node.getGroupingKeys(), + child, + scan, + analysis.getTableColumnSchema(scan.getQualifiedObjectName())) + ? scan + : null; } static boolean isDeviceScanTimeOrderedPerGroup( @@ -2346,29 +2359,21 @@ static boolean isDeviceScanTimeOrderedPerGroup( if (tableColumnSchema == null) { return false; } - Set tagColumnNames = new HashSet<>(); - for (ColumnSchema columnSchema : tableColumnSchema.values()) { - if (columnSchema.getColumnCategory() == TsTableColumnCategory.TAG) { - tagColumnNames.add(columnSchema.getName()); - } - } - Map scanColumnSchema = scan.getAssignments(); - Set groupedColumnNames = new HashSet<>(); - for (Symbol symbol : resolveSymbolsToScanLevel(child, groupingKeys)) { - ColumnSchema columnSchema = scanColumnSchema.get(symbol); - if (columnSchema != null) { - groupedColumnNames.add(columnSchema.getName()); - } - } - return groupedColumnNames.containsAll(tagColumnNames); + Set groupedColumnNames = + TagColumnUtils.tagColumnNames( + resolveSymbolsToScanLevel(child, groupingKeys), scan.getAssignments()); + return TagColumnUtils.coversAllTagColumns(tableColumnSchema, groupedColumnNames); } - private static boolean isTimeArgumentNativeTime( + static boolean isTimeArgumentNativeTime( AggregationNode.Aggregation aggregation, PlanNode child, DeviceTableScanNode scan) { - if (scan == null || aggregation.getArguments().size() < 2) { + int timeArgumentIndex = + getOrderedInputTimeArgumentIndex( + aggregation.getResolvedFunction().getSignature().getName()); + if (timeArgumentIndex < 0 || aggregation.getArguments().size() <= timeArgumentIndex) { return false; } - Expression timeArgument = aggregation.getArguments().get(1); + Expression timeArgument = aggregation.getArguments().get(timeArgumentIndex); if (!(timeArgument instanceof SymbolReference)) { return false; } @@ -2420,13 +2425,26 @@ private static Set resolveSymbolsToScanLevel(PlanNode node, Collection= 0; } private boolean prefixMatched(OrderingScheme childOrdering, List preGroupedSymbols) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/ParallelizeGrouping.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/ParallelizeGrouping.java index 406780fd8204c..ec1d714011245 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/ParallelizeGrouping.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/ParallelizeGrouping.java @@ -37,6 +37,7 @@ import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.UnionNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor; import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis; +import org.apache.iotdb.db.queryengine.plan.relational.planner.TagColumnUtils; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExternalTsFileAggregationScanNode; @@ -276,11 +277,6 @@ public PlanNode visitDeviceTableScan(DeviceTableScanNode node, Context context) tableColumnSchema = analysis.getTableColumnSchema(node.getQualifiedObjectName()); } // check there are no field in sortKey and all tags in sortKey - Set tagSymbols = - tableColumnSchema.entrySet().stream() - .filter(entry -> entry.getValue().getColumnCategory() == TAG) - .map(Map.Entry::getKey) - .collect(Collectors.toSet()); for (int i = 0; i < context.partitionKeyCount; i++) { Symbol symbol = sortKey.getOrderBy().get(i); if (!tableColumnSchema.containsKey(symbol)) { @@ -289,7 +285,6 @@ public PlanNode visitDeviceTableScan(DeviceTableScanNode node, Context context) } switch (tableColumnSchema.get(symbol).getColumnCategory()) { case TAG: - tagSymbols.remove(symbol); break; case ATTRIBUTE: // If all tags in partition key, attributes must be the same in one partition. @@ -299,11 +294,13 @@ public PlanNode visitDeviceTableScan(DeviceTableScanNode node, Context context) return node; } } - if (!tagSymbols.isEmpty()) { - context.canOptimized = TO_SORT; - } else { - context.canOptimized = ENABLE_PARALLEL; - } + Set tagColumnNames = + TagColumnUtils.tagColumnNames( + sortKey.getOrderBy().subList(0, context.partitionKeyCount), tableColumnSchema); + context.canOptimized = + TagColumnUtils.coversAllTagColumns(tableColumnSchema, tagColumnNames) + ? ENABLE_PARALLEL + : TO_SORT; } return node; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TransformSortToStreamSort.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TransformSortToStreamSort.java index 90c2c91931d13..2c20a8714fa07 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TransformSortToStreamSort.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TransformSortToStreamSort.java @@ -33,6 +33,7 @@ import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor; import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis; +import org.apache.iotdb.db.queryengine.plan.relational.planner.TagColumnUtils; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CteScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode; @@ -224,20 +225,9 @@ public static boolean isOrderByAllIdsAndTime( Map nodeColumnSchema, OrderingScheme orderingScheme, int streamSortIndex) { - int tagCount = 0; - for (ColumnSchema columnSchema : tableColumnSchema.values()) { - if (columnSchema.getColumnCategory() == TsTableColumnCategory.TAG) { - tagCount++; - } - } - - for (Symbol orderBy : orderingScheme.getOrderBy()) { - ColumnSchema columnSchema = nodeColumnSchema.get(orderBy); - if (columnSchema != null && columnSchema.getColumnCategory() == TsTableColumnCategory.TAG) { - tagCount--; - } - } - return tagCount == 0 + Set orderedTagColumnNames = + TagColumnUtils.tagColumnNames(orderingScheme.getOrderBy(), nodeColumnSchema); + return TagColumnUtils.coversAllTagColumns(tableColumnSchema, orderedTagColumnNames) && (orderingScheme.getOrderings().size() == streamSortIndex + 1 || isTimeColumn( orderingScheme.getOrderBy().get(streamSortIndex + 1), tableColumnSchema)); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedInputTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedInputTest.java index a42b67090b66e..61e802e181253 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedInputTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedInputTest.java @@ -21,17 +21,28 @@ import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.queryengine.plan.relational.function.BoundSignature; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnMetadata; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName; +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ResolvedFunction; import org.apache.iotdb.commons.queryengine.plan.relational.planner.Assignments; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.OrderingScheme; import org.apache.iotdb.commons.queryengine.plan.relational.planner.Symbol; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.AggregationNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.CollectNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.FilterNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.LimitNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.ProjectNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.SortNode; +import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.SymbolReference; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExternalTsFileScanNode; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode; import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.tsfile.read.common.type.StringType; @@ -45,13 +56,17 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Optional; +import static org.apache.iotdb.commons.queryengine.plan.relational.planner.SortOrder.ASC_NULLS_LAST; import static org.apache.iotdb.commons.queryengine.plan.relational.planner.node.AggregationNode.Step.PARTIAL; import static org.apache.iotdb.commons.queryengine.plan.relational.planner.node.AggregationNode.Step.SINGLE; import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.FIELD; import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.TAG; +import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.TIME; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; public class RateOrderedInputTest { @@ -59,6 +74,8 @@ public class RateOrderedInputTest { private static final Symbol TAG1 = new Symbol("tag1"); private static final Symbol TAG2 = new Symbol("tag2"); private static final Symbol S1 = new Symbol("s1"); + private static final Symbol S2 = new Symbol("s2"); + private static final Symbol TIME_SYMBOL = new Symbol("time"); @Test public void testSingleDeviceIsOrderedPerGroup() { @@ -104,8 +121,115 @@ public void testGroupingKeysAreResolvedThroughProjections() { } @Test - public void testOrderChangingChildIsRejected() { - assertNull(TableDistributedPlanGenerator.findOrderedDeviceScan(Mockito.mock(PlanNode.class))); + public void testProjectAndFilterPreserveScan() { + DeviceTableScanNode scan = scan(1); + ProjectNode project = + new ProjectNode( + new PlanNodeId("project"), + scan, + Assignments.builder().put(S1, new SymbolReference("s1")).build()); + assertSame(scan, TableDistributedPlanGenerator.findOrderedDeviceScan(project)); + + FilterNode filter = + new FilterNode(new PlanNodeId("filter"), scan, Mockito.mock(Expression.class)); + assertSame(scan, TableDistributedPlanGenerator.findOrderedDeviceScan(filter)); + } + + @Test + public void testSortChildIsRejected() { + DeviceTableScanNode scan = scan(1); + SortNode sort = + new SortNode( + new PlanNodeId("sort"), + scan, + new OrderingScheme( + Collections.singletonList(TIME_SYMBOL), + Collections.singletonMap(TIME_SYMBOL, ASC_NULLS_LAST)), + false, + false); + assertNull(TableDistributedPlanGenerator.findOrderedDeviceScan(sort)); + assertFalse(isOrderedPerGroup(SINGLE, Collections.emptyList(), sort)); + } + + @Test + public void testLimitChildIsRejected() { + DeviceTableScanNode scan = scan(1); + LimitNode limit = new LimitNode(new PlanNodeId("limit"), scan, 10, Optional.empty()); + assertNull(TableDistributedPlanGenerator.findOrderedDeviceScan(limit)); + } + + @Test + public void testSingleChildCollectIsRejected() { + DeviceTableScanNode scan = scan(1); + CollectNode collect = + new CollectNode( + new PlanNodeId("collect"), + Collections.singletonList(scan), + Collections.singletonList(S1)); + assertNull(TableDistributedPlanGenerator.findOrderedDeviceScan(collect)); + assertFalse(isOrderedPerGroup(SINGLE, Collections.emptyList(), collect)); + } + + @Test + public void testUnsupportedScanTypesAreRejected() { + assertNull( + TableDistributedPlanGenerator.findOrderedDeviceScan( + Mockito.mock(AggregationTableScanNode.class))); + assertNull( + TableDistributedPlanGenerator.findOrderedDeviceScan( + Mockito.mock(TreeDeviceViewScanNode.class))); + assertNull( + TableDistributedPlanGenerator.findOrderedDeviceScan( + Mockito.mock(ExternalTsFileScanNode.class))); + } + + @Test + public void testTimeArgumentResolvesToNativeTime() { + DeviceTableScanNode scan = scan(1); + assertTrue( + TableDistributedPlanGenerator.isTimeArgumentNativeTime( + aggregation("irate", S1, TIME_SYMBOL), scan, scan)); + } + + @Test + public void testTimeArgumentResolvesThroughAliasToNativeTime() { + DeviceTableScanNode scan = scan(1); + Symbol value = new Symbol("v"); + Symbol time = new Symbol("t"); + ProjectNode project = + new ProjectNode( + new PlanNodeId("project"), + scan, + Assignments.builder() + .put(value, new SymbolReference("s1")) + .put(time, new SymbolReference("time")) + .build()); + assertTrue( + TableDistributedPlanGenerator.isTimeArgumentNativeTime( + aggregation("irate", value, time), project, scan)); + } + + @Test + public void testNonTimeColumnTimeArgumentIsRejected() { + DeviceTableScanNode scan = scan(1); + assertFalse( + TableDistributedPlanGenerator.isTimeArgumentNativeTime( + aggregation("irate", S1, S2), scan, scan)); + } + + private static AggregationNode.Aggregation aggregation( + String functionName, Symbol value, Symbol time) { + ResolvedFunction function = Mockito.mock(ResolvedFunction.class); + BoundSignature signature = Mockito.mock(BoundSignature.class); + Mockito.when(function.getSignature()).thenReturn(signature); + Mockito.when(signature.getName()).thenReturn(functionName); + return new AggregationNode.Aggregation( + function, + Arrays.asList(value.toSymbolReference(), time.toSymbolReference()), + false, + Optional.empty(), + Optional.empty(), + Optional.empty()); } private static boolean isOrderedPerGroup( @@ -123,6 +247,8 @@ private static DeviceTableScanNode scan(int deviceCount) { assignments.put(TAG1, columnSchema("tag1", TAG)); assignments.put(TAG2, columnSchema("tag2", TAG)); assignments.put(S1, columnSchema("s1", FIELD)); + assignments.put(S2, columnSchema("s2", FIELD)); + assignments.put(TIME_SYMBOL, columnSchema("time", TIME)); Map tagAndAttributeIndexMap = new HashMap<>(); tagAndAttributeIndexMap.put(TAG1, 0); tagAndAttributeIndexMap.put(TAG2, 1); @@ -146,6 +272,7 @@ private static Map fullTableColumnSchema() { tableColumnSchema.put(TAG1, columnSchema("tag1", TAG)); tableColumnSchema.put(TAG2, columnSchema("tag2", TAG)); tableColumnSchema.put(S1, columnSchema("s1", FIELD)); + tableColumnSchema.put(S2, columnSchema("s2", FIELD)); return tableColumnSchema; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java index b4735411eb563..eb08c29bafa79 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java @@ -26,49 +26,93 @@ import org.junit.Test; +import static org.apache.iotdb.commons.queryengine.plan.relational.planner.node.AggregationNode.Step.SINGLE; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; public class RateOrderedPlannerTest { + private static final String SINGLE_DEVICE_WHERE = + " WHERE tag1='shanghai' AND tag2='A3' AND tag3='YY'"; + // DEVICE_3 and DEVICE_4 are both in DATA_REGION_GROUP_2, so this filter still produces a SINGLE + // aggregation over a two-device scan. + private static final String TWO_DEVICE_WHERE = " WHERE tag1='shanghai'"; + @Test public void testSingleDeviceQueryEnablesOrderedInput() { - AggregationNode aggregation = - planAggregation( - "SELECT irate(s1, time) AS ir FROM table1 " - + "WHERE tag1='shanghai' AND tag2='A3' AND tag3='YY'"); - assertNotNull(aggregation); - assertTrue(hasOrderedInput(aggregation)); + assertOrdered("SELECT irate(s1, time) AS ir FROM table1" + SINGLE_DEVICE_WHERE); } @Test - public void testMultiDeviceQueryKeepsNaiveInput() { - AggregationNode aggregation = planAggregation("SELECT irate(s1, time) AS ir FROM table1"); - assertNotNull(aggregation); - assertFalse(hasOrderedInput(aggregation)); + public void testTwoDevicesGroupedByAllTagsEnableOrderedInput() { + assertOrdered( + "SELECT irate(s1, time) AS ir FROM table1" + + TWO_DEVICE_WHERE + + " GROUP BY tag1, tag2, tag3"); + } + + @Test + public void testTwoDevicesGroupedBySubsetOfTagsKeepNaiveInput() { + assertNotOrdered( + "SELECT irate(s1, time) AS ir FROM table1" + TWO_DEVICE_WHERE + " GROUP BY tag1"); + } + + @Test + public void testGlobalAggregationOverTwoDevicesKeepsNaiveInput() { + assertNotOrdered("SELECT irate(s1, time) AS ir FROM table1" + TWO_DEVICE_WHERE); + } + + @Test + public void testDescendingSubqueryKeepsNaiveInput() { + assertNotOrdered( + "SELECT irate(s1, time) AS ir FROM (SELECT s1, time FROM table1" + + TWO_DEVICE_WHERE + + " ORDER BY time DESC LIMIT 10)"); } @Test public void testNonTimeArgumentKeepsNaiveInput() { - AggregationNode aggregation = - planAggregation( - "SELECT irate(s1, s2) AS ir FROM table1 " - + "WHERE tag1='shanghai' AND tag2='A3' AND tag3='YY'"); + assertNotOrdered("SELECT irate(s1, s2) AS ir FROM table1" + SINGLE_DEVICE_WHERE); + } + + @Test + public void testFourArgumentRateFunctionsEnableOrderedInput() { + assertOrdered("SELECT rate(s1, time, s2, s2) AS r FROM table1" + SINGLE_DEVICE_WHERE); + assertOrdered("SELECT increase(s1, time, s2, s2) AS r FROM table1" + SINGLE_DEVICE_WHERE); + assertOrdered("SELECT delta(s1, time, s2, s2) AS r FROM table1" + SINGLE_DEVICE_WHERE); + } + + @Test + public void testTimeArgumentThroughAliasEnablesOrderedInput() { + assertOrdered( + "SELECT irate(v, t) AS ir FROM (SELECT s1 AS v, time AS t FROM table1" + + SINGLE_DEVICE_WHERE + + ")"); + } + + private static void assertOrdered(String sql) { + AggregationNode aggregation = planSingleAggregation(sql); + assertNotNull(aggregation); + assertEquals(SINGLE, aggregation.getStep()); + assertTrue(hasOrderedInput(aggregation)); + } + + private static void assertNotOrdered(String sql) { + AggregationNode aggregation = planSingleAggregation(sql); assertNotNull(aggregation); + assertEquals(SINGLE, aggregation.getStep()); assertFalse(hasOrderedInput(aggregation)); } - private static AggregationNode planAggregation(String sql) { + private static AggregationNode planSingleAggregation(String sql) { PlanTester planTester = new PlanTester(new TestMetadata()); planTester.createPlan(sql); - for (int fragment = 0; fragment < 3; fragment++) { - AggregationNode aggregation; - try { - aggregation = findAggregation(planTester.getFragmentPlan(fragment)); - } catch (RuntimeException e) { - break; - } + for (int fragment = 0; + fragment < planTester.getDistributedPlan().getFragments().size(); + fragment++) { + AggregationNode aggregation = findSingleAggregation(planTester.getFragmentPlan(fragment)); if (aggregation != null) { return aggregation; } @@ -76,12 +120,12 @@ private static AggregationNode planAggregation(String sql) { return null; } - private static AggregationNode findAggregation(PlanNode node) { - if (node instanceof AggregationNode) { + private static AggregationNode findSingleAggregation(PlanNode node) { + if (node instanceof AggregationNode && ((AggregationNode) node).getStep() == SINGLE) { return (AggregationNode) node; } for (PlanNode child : node.getChildren()) { - AggregationNode aggregation = findAggregation(child); + AggregationNode aggregation = findSingleAggregation(child); if (aggregation != null) { return aggregation; } From d8f13295272e97e3cf4967facf099defe50f0b5b Mon Sep 17 00:00:00 2001 From: zxq <3322351820@qq.com> Date: Fri, 13 Nov 2026 11:51:01 +0800 Subject: [PATCH 3/3] Cover the descending scan-order check at the planner level --- .../planner/distribute/RateOrderedPlannerTest.java | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java index eb08c29bafa79..adad4c3c0205c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java @@ -65,13 +65,25 @@ public void testGlobalAggregationOverTwoDevicesKeepsNaiveInput() { } @Test - public void testDescendingSubqueryKeepsNaiveInput() { + public void testDescendingSubqueryWithTwoDevicesKeepsNaiveInput() { + // With two devices the TopK above the descending scan cannot be eliminated, so this covers + // rejection of an intermediate node that does not preserve the scan order. assertNotOrdered( "SELECT irate(s1, time) AS ir FROM (SELECT s1, time FROM table1" + TWO_DEVICE_WHERE + " ORDER BY time DESC LIMIT 10)"); } + @Test + public void testDescendingSingleDeviceScanKeepsNaiveInput() { + // The limit of a single-device scan is pushed into the scan, so the plan is + // Aggregation <- DeviceTableScan(DESC) and the descending scan-order check is exercised. + assertNotOrdered( + "SELECT irate(s1, time) AS ir FROM (SELECT s1, time FROM table1" + + SINGLE_DEVICE_WHERE + + " ORDER BY time DESC LIMIT 10)"); + } + @Test public void testNonTimeArgumentKeepsNaiveInput() { assertNotOrdered("SELECT irate(s1, s2) AS ir FROM table1" + SINGLE_DEVICE_WHERE);