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 1ead17632aaf1..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; @@ -2077,7 +2078,9 @@ 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, findTimeOrderedDeviceScan(node, childrenNodes.get(0))); if (childrenNodes.get(0) instanceof UnionNode && physicalAggregation.getAggregations().values().stream() @@ -2246,6 +2249,16 @@ private boolean containsUnorderedCollect(PlanNode node) { private static AggregationNode withRateFunctionInputOrdering( AggregationNode node, OrderingScheme childOrdering) { + 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, DeviceTableScanNode timeOrderedScan) { + PlanNode child = node.getChild(); Map aggregations = new LinkedHashMap<>(); node.getAggregations() .forEach( @@ -2259,8 +2272,13 @@ private static AggregationNode withRateFunctionInputOrdering( aggregation.getFilter(), aggregation.getOrderingScheme(), aggregation.getMask(), - isInputOrderedByTimeAscending( - aggregation, node.getStep(), node.getGroupingKeys(), childOrdering)))); + (timeOrderedScan != null + && isTimeArgumentNativeTime(aggregation, child, timeOrderedScan)) + || isInputOrderedByTimeAscending( + aggregation, + node.getStep(), + node.getGroupingKeys(), + childOrdering)))); return AggregationNode.builderFrom(node).setAggregations(aggregations).build(); } @@ -2269,18 +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 - || !(RATE.equalsIgnoreCase(functionName) - || INCREASE.equalsIgnoreCase(functionName) - || IRATE.equalsIgnoreCase(functionName) - || DELTA.equalsIgnoreCase(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()) { @@ -2296,6 +2313,140 @@ static boolean isInputOrderedByTimeAscending( return true; } + /** + * 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 null; + } + return isDeviceScanTimeOrderedPerGroup( + node.getStep(), + node.getGroupingKeys(), + child, + scan, + analysis.getTableColumnSchema(scan.getQualifiedObjectName())) + ? scan + : null; + } + + 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 groupedColumnNames = + TagColumnUtils.tagColumnNames( + resolveSymbolsToScanLevel(child, groupingKeys), scan.getAssignments()); + return TagColumnUtils.coversAllTagColumns(tableColumnSchema, groupedColumnNames); + } + + static boolean isTimeArgumentNativeTime( + AggregationNode.Aggregation aggregation, PlanNode child, DeviceTableScanNode scan) { + int timeArgumentIndex = + getOrderedInputTimeArgumentIndex( + aggregation.getResolvedFunction().getSignature().getName()); + if (timeArgumentIndex < 0 || aggregation.getArguments().size() <= timeArgumentIndex) { + return false; + } + Expression timeArgument = aggregation.getArguments().get(timeArgumentIndex); + 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; + } + } + } + + /** + * Returns the position of the time argument in an aggregation's argument list, or {@code -1} when + * the function's accumulators do not require time-ordered input. Both ordering checks must use + * this helper: a new family (for example the integral family, whose time argument is {@code + * integral(value, time_unit, time)}) only needs to be registered here, and its time argument + * index then applies to the ordering-based and the scan-based check alike. + */ + private static int getOrderedInputTimeArgumentIndex(String functionName) { + // RATE, INCREASE, IRATE and DELTA take (value, time, ...). + if (RATE.equalsIgnoreCase(functionName) + || INCREASE.equalsIgnoreCase(functionName) + || IRATE.equalsIgnoreCase(functionName) + || DELTA.equalsIgnoreCase(functionName)) { + return 1; + } + return -1; + } + + private static boolean isOrderedInputFunction(String functionName) { + return getOrderedInputTimeArgumentIndex(functionName) >= 0; + } + private boolean prefixMatched(OrderingScheme childOrdering, List preGroupedSymbols) { List orderKeys = childOrdering.getOrderBy(); if (orderKeys.size() < preGroupedSymbols.size()) { 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 new file mode 100644 index 0000000000000..61e802e181253 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedInputTest.java @@ -0,0 +1,284 @@ +/* + * 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.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; +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 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 { + + 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() { + 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 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( + 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)); + 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); + 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)); + tableColumnSchema.put(S2, columnSchema("s2", 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..adad4c3c0205c --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/RateOrderedPlannerTest.java @@ -0,0 +1,152 @@ +/* + * 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.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() { + assertOrdered("SELECT irate(s1, time) AS ir FROM table1" + SINGLE_DEVICE_WHERE); + } + + @Test + 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 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); + } + + @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 planSingleAggregation(String sql) { + PlanTester planTester = new PlanTester(new TestMetadata()); + planTester.createPlan(sql); + for (int fragment = 0; + fragment < planTester.getDistributedPlan().getFragments().size(); + fragment++) { + AggregationNode aggregation = findSingleAggregation(planTester.getFragmentPlan(fragment)); + if (aggregation != null) { + return aggregation; + } + } + return null; + } + + private static AggregationNode findSingleAggregation(PlanNode node) { + if (node instanceof AggregationNode && ((AggregationNode) node).getStep() == SINGLE) { + return (AggregationNode) node; + } + for (PlanNode child : node.getChildren()) { + AggregationNode aggregation = findSingleAggregation(child); + if (aggregation != null) { + return aggregation; + } + } + return null; + } + + private static boolean hasOrderedInput(AggregationNode aggregation) { + return aggregation.getAggregations().values().stream() + .anyMatch(AggregationNode.Aggregation::isInputOrderedByTimeAscending); + } +}