From 36ea1eee8751e10207591b44937b2c1f30104d73 Mon Sep 17 00:00:00 2001 From: Weihao Li <18110526956@163.com> Date: Fri, 6 Nov 2026 08:11:28 +0800 Subject: [PATCH] fix Signed-off-by: Weihao Li <18110526956@163.com> --- .../relational/it/db/it/IoTDBWindowTVFIT.java | 94 +++++++++ .../optimizations/ParallelizeGrouping.java | 34 +-- .../analyzer/TableFunctionTest.java | 194 ++++++++++++++++++ .../planner/assertions/MergeSortMatcher.java | 67 ++++++ .../planner/assertions/PlanMatchPattern.java | 4 + 5 files changed, 381 insertions(+), 12 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/MergeSortMatcher.java diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java index 0adddc0cf69ec..7134eadcde971 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java @@ -251,6 +251,100 @@ public void testSessionFunction() { DATABASE_NAME); } + @Test + public void testNestedSessionWithPermutedPartitionColumns() { + String[] expectedHeader = + new String[] { + "window_start", + "window_end", + "inner_start", + "inner_end", + "time", + "factory_id", + "device_id", + "s1" + }; + String[] retArray = + new String[] { + "1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.001Z,F1,device_1,10.0,", + "1970-01-01T00:00:00.005Z,1970-01-01T00:00:00.005Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.005Z,F1,device_1,15.0,", + "1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.009Z,F1,device_1,5.0,", + "1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,F1,device_2,20.0,", + "1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,F1,device_2,25.0,", + "1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,F2,device_1,30.0," + }; + // The outer PARTITION BY lists the same columns as the inner one in a different order; the + // result must be identical to the same-order variant (next test). + String sql = + "SELECT window_start, window_end, inner_start, inner_end, time, factory_id, device_id, s1 " + + "FROM SESSION(DATA => (" + + " SELECT window_start AS inner_start, window_end AS inner_end, time, factory_id, device_id, s1 " + + " FROM SESSION(DATA => table5 PARTITION BY (factory_id, device_id) ORDER BY time, GAP => 5ms)" + + ") PARTITION BY (device_id, factory_id) ORDER BY time, GAP => 3ms) " + + "ORDER BY factory_id, device_id, time"; + tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME); + } + + @Test + public void testNestedSessionWithSamePartitionColumnOrder() { + String[] expectedHeader = + new String[] { + "window_start", + "window_end", + "inner_start", + "inner_end", + "time", + "factory_id", + "device_id", + "s1" + }; + String[] retArray = + new String[] { + "1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.001Z,F1,device_1,10.0,", + "1970-01-01T00:00:00.005Z,1970-01-01T00:00:00.005Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.005Z,F1,device_1,15.0,", + "1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.009Z,F1,device_1,5.0,", + "1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,F1,device_2,20.0,", + "1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,F1,device_2,25.0,", + "1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,F2,device_1,30.0," + }; + // Reference case: both levels list the PARTITION BY columns in the same order. + String sql = + "SELECT window_start, window_end, inner_start, inner_end, time, factory_id, device_id, s1 " + + "FROM SESSION(DATA => (" + + " SELECT window_start AS inner_start, window_end AS inner_end, time, factory_id, device_id, s1 " + + " FROM SESSION(DATA => table5 PARTITION BY (factory_id, device_id) ORDER BY time, GAP => 5ms)" + + ") PARTITION BY (factory_id, device_id) ORDER BY time, GAP => 3ms) " + + "ORDER BY factory_id, device_id, time"; + tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME); + } + + @Test + public void testThreeLevelNestedSessionWithPermutedPartitionColumns() { + String[] expectedHeader = + new String[] {"window_start", "window_end", "time", "factory_id", "device_id", "s1"}; + String[] retArray = + new String[] { + "1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.001Z,F1,device_1,10.0,", + "1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.005Z,F1,device_1,15.0,", + "1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.009Z,F1,device_1,5.0,", + "1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,F1,device_2,20.0,", + "1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,F1,device_2,25.0,", + "1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,F2,device_1,30.0," + }; + // Three levels list the same PARTITION BY columns in three different orders. + String sql = + "SELECT window_start, window_end, time, factory_id, device_id, s1 " + + "FROM SESSION(DATA => (" + + " SELECT window_start AS mid_start, window_end AS mid_end, inner_start, inner_end, time, factory_id, device_id, s1 " + + " FROM SESSION(DATA => (" + + " SELECT window_start AS inner_start, window_end AS inner_end, time, factory_id, device_id, s1 " + + " FROM SESSION(DATA => table5 PARTITION BY (factory_id, device_id) ORDER BY time, GAP => 5ms)" + + " ) PARTITION BY (device_id, factory_id) ORDER BY time, GAP => 3ms)" + + ") PARTITION BY (factory_id, device_id) ORDER BY time, GAP => 5ms) " + + "ORDER BY factory_id, device_id, time"; + tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME); + } + @Test public void testVariationFunction() { String[] expectedHeader = new String[] {"window_index", "time", "stock_id", "price", "s1"}; 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 ac3fc65fe6683..37f0b8b262b33 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 @@ -42,6 +42,8 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExternalTsFileAggregationScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExternalTsFileScanNode; +import com.google.common.collect.ImmutableSet; + import java.util.List; import java.util.Map; import java.util.Optional; @@ -117,25 +119,33 @@ public PlanNode visitUnion(UnionNode node, Context context) { .collect(Collectors.toList())); } - /** We need to make sure: context#partitionKey can match the prefix of childOrderSchema */ + /** + * We need to make sure: context#partitionKey can be satisfied by the prefix of childOrder. + * + *

The prefix is compared as a column set: partition columns are identified by their values, + * so a different writing order still clusters rows into the same partitions. This lets nested + * set-semantic table functions reuse their input clustering when the same PARTITION BY columns + * are listed in a different order. + * + *

The ORDER BY part of the requirement needs no check here. A group node sorts within every + * partition run by its full ordering scheme, so the input only has to be clustered; whatever + * order rows have inside a partition is re-established locally by the group's own sort. + */ private void checkPrefixMatch(Context context, List childOrder) { if (context.canSkip()) { return; } - if (context.partitionKeyCount > childOrder.size()) { + int partitionKeyCount = context.partitionKeyCount; + if (partitionKeyCount > childOrder.size()) { context.canOptimized = TO_SORT; return; } - OrderingScheme prefix = context.sortKey; - for (int i = 0; i < context.partitionKeyCount; i++) { - Symbol lhs = prefix.getOrderBy().get(i); - Symbol rhs = childOrder.get(i); - if (!lhs.equals(rhs)) { - context.canOptimized = TO_SORT; - return; - } - } - context.canOptimized = ENABLE_PARALLEL; + List requiredPrefix = context.sortKey.getOrderBy().subList(0, partitionKeyCount); + List actualPrefix = childOrder.subList(0, partitionKeyCount); + context.canOptimized = + ImmutableSet.copyOf(requiredPrefix).equals(ImmutableSet.copyOf(actualPrefix)) + ? ENABLE_PARALLEL + : TO_SORT; } @Override diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java index 637113026f4f0..fc287b52c17dc 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java @@ -36,6 +36,7 @@ import org.junit.Test; import java.util.Collections; +import java.util.List; import java.util.function.Consumer; import static org.apache.iotdb.commons.queryengine.plan.relational.function.tvf.ForecastTableFunction.DEFAULT_OUTPUT_INTERVAL; @@ -226,6 +227,199 @@ public void testSimpleSetSemantic() { assertPlan(planTester.getFragmentPlan(5), tableScan); } + private static final String NESTED_SESSION_SQL = + "SELECT * FROM SESSION(DATA => (SELECT * FROM SESSION(DATA => table1 " + + "PARTITION BY (tag1, tag2, tag3) ORDER BY time, GAP => 1m)) " + + "PARTITION BY (tag3, tag2, tag1) ORDER BY time, GAP => 2m)"; + + private static Consumer sessionMatcher( + String windowStart, String windowEnd, long gap) { + return builder -> + builder + .name("session") + .properOutputs(windowStart, windowEnd) + .requiredSymbols("time") + .handle(new MapTableFunctionHandle.Builder().addProperty("GAP", gap).build()); + } + + private static PlanMatchPattern scanAllColumns() { + return tableScan( + "testdb.table1", + ImmutableList.of("time", "tag1", "tag2", "tag3", "attr1", "attr2", "s1", "s2", "s3"), + ImmutableSet.of("time", "tag1", "tag2", "tag3", "attr1", "attr2", "s1", "s2", "s3")); + } + + private static List orderBy(String... fields) { + ImmutableList.Builder result = ImmutableList.builder(); + for (String field : fields) { + result.add(sort(field, ASCENDING, LAST)); + } + return result.build(); + } + + @Test + public void testNestedSetSemanticPermutedPartitionsKeepBothGroups() { + PlanTester planTester = new PlanTester(); + LogicalQueryPlan logicalQueryPlan = planTester.createPlan(NESTED_SESSION_SQL); + Consumer outer = + sessionMatcher("window_start", "window_end", 120_000L); + Consumer inner = + sessionMatcher("window_start_0", "window_end_1", 60_000L); + // PARTITION BY columns are the same set in a different order and ORDER BY matches, so both + // grouping nodes survive: the outer keeps its declared order, the inner keeps its own. + assertPlan( + logicalQueryPlan, + anyTree( + tableFunctionProcessor( + outer, + group( + orderBy("tag3", "tag2", "tag1", "time"), + 3, + tableFunctionProcessor( + inner, + group(orderBy("tag1", "tag2", "tag3", "time"), 3, scanAllColumns())))))); + // Both levels are split per region; only regions with several devices need a merged stream. + assertPlan(planTester.getFragmentPlan(0), output(collect(exchange(), exchange(), exchange()))); + assertPlan( + planTester.getFragmentPlan(1), + tableFunctionProcessor(outer, tableFunctionProcessor(inner, scanAllColumns()))); + assertPlan( + planTester.getFragmentPlan(2), + tableFunctionProcessor(outer, tableFunctionProcessor(inner, scanAllColumns()))); + assertPlan( + planTester.getFragmentPlan(3), + tableFunctionProcessor( + outer, + tableFunctionProcessor( + inner, + mergeSort(orderBy("tag3", "tag2", "tag1", "time"), exchange(), exchange())))); + assertPlan(planTester.getFragmentPlan(4), scanAllColumns()); + assertPlan(planTester.getFragmentPlan(5), scanAllColumns()); + } + + @Test + public void testNestedSetSemanticDifferentOrderByKeepsOuterGroup() { + PlanTester planTester = new PlanTester(); + String sql = + "SELECT * FROM SESSION(DATA => (SELECT * FROM SESSION(DATA => table1 " + + "PARTITION BY (tag1, tag2, tag3) ORDER BY time, GAP => 1m)) " + + "PARTITION BY (tag3, tag2, tag1) ORDER BY s1, GAP => 2m)"; + LogicalQueryPlan logicalQueryPlan = planTester.createPlan(sql); + Consumer outer = + sessionMatcher("window_start", "window_end", 120_000L); + Consumer inner = + sessionMatcher("window_start_0", "window_end_1", 60_000L); + // Only the partition prefix has to be clustered: the outer group re-establishes its s1 order + // inside every partition, so a different ORDER BY does not degrade it to a full sort. + assertPlan( + logicalQueryPlan, + anyTree( + tableFunctionProcessor( + outer, + group( + orderBy("tag3", "tag2", "tag1", "s1"), + 3, + tableFunctionProcessor( + inner, + group(orderBy("tag1", "tag2", "tag3", "time"), 3, scanAllColumns())))))); + // Both levels are still split per region. + assertPlan(planTester.getFragmentPlan(0), output(collect(exchange(), exchange(), exchange()))); + } + + @Test + public void testNestedSetSemanticGlobalOrderByDeclaredOrder() { + PlanTester planTester = new PlanTester(); + LogicalQueryPlan logicalQueryPlan = + planTester.createPlan(NESTED_SESSION_SQL + " ORDER BY tag3, tag2, tag1, time"); + Consumer outer = + sessionMatcher("window_start", "window_end", 120_000L); + Consumer inner = + sessionMatcher("window_start_0", "window_end_1", 60_000L); + // The requested global order equals the outer declared order, so no per-region sort is needed + // and the cross-region merge uses exactly that order. + assertPlan( + planTester.getFragmentPlan(0), + output( + mergeSort( + orderBy("tag3", "tag2", "tag1", "time"), exchange(), exchange(), exchange()))); + assertPlan( + planTester.getFragmentPlan(1), + tableFunctionProcessor(outer, tableFunctionProcessor(inner, scanAllColumns()))); + assertPlan( + planTester.getFragmentPlan(3), + tableFunctionProcessor( + outer, + tableFunctionProcessor( + inner, + mergeSort(orderBy("tag3", "tag2", "tag1", "time"), exchange(), exchange())))); + } + + @Test + public void testNestedSetSemanticGlobalOrderByPermutesPartitionColumns() { + PlanTester planTester = new PlanTester(); + // The global order permutes the partition columns again; the distributed planner must prove it + // by rewriting grouping/device comparators instead of reusing the TVF declared order. + LogicalQueryPlan logicalQueryPlan = + planTester.createPlan(NESTED_SESSION_SQL + " ORDER BY tag2, tag3, tag1, time"); + Consumer outer = + sessionMatcher("window_start", "window_end", 120_000L); + Consumer inner = + sessionMatcher("window_start_0", "window_end_1", 60_000L); + assertPlan( + planTester.getFragmentPlan(0), + output( + mergeSort( + orderBy("tag2", "tag3", "tag1", "time"), exchange(), exchange(), exchange()))); + assertPlan( + planTester.getFragmentPlan(1), + tableFunctionProcessor(outer, tableFunctionProcessor(inner, scanAllColumns()))); + assertPlan( + planTester.getFragmentPlan(3), + tableFunctionProcessor( + outer, + tableFunctionProcessor( + inner, + mergeSort(orderBy("tag2", "tag3", "tag1", "time"), exchange(), exchange())))); + } + + @Test + public void testThreeLevelNestedSetSemanticPermutedPartitions() { + PlanTester planTester = new PlanTester(); + String sql = + "SELECT * FROM SESSION(DATA => (SELECT * FROM SESSION(DATA => (SELECT * FROM SESSION(DATA => table1 " + + "PARTITION BY (tag1, tag2, tag3) ORDER BY time, GAP => 1m)) " + + "PARTITION BY (tag3, tag2, tag1) ORDER BY time, GAP => 2m)) " + + "PARTITION BY (tag2, tag1, tag3) ORDER BY time, GAP => 3m)"; + LogicalQueryPlan logicalQueryPlan = planTester.createPlan(sql); + Consumer outerMost = + sessionMatcher("window_start", "window_end", 180_000L); + Consumer middle = + sessionMatcher("window_start_0", "window_end_1", 120_000L); + Consumer inner = + sessionMatcher("window_start_2", "window_end_3", 60_000L); + // Every level lists the same PARTITION BY columns in a different order, so all three groups + // survive and are executed in parallel per region. + assertPlan( + logicalQueryPlan, + anyTree( + tableFunctionProcessor( + outerMost, + group( + orderBy("tag2", "tag1", "tag3", "time"), + 3, + tableFunctionProcessor( + middle, + group( + orderBy("tag3", "tag2", "tag1", "time"), + 3, + tableFunctionProcessor( + inner, + group( + orderBy("tag1", "tag2", "tag3", "time"), + 3, + scanAllColumns())))))))); + } + @Test public void testLeafFunction() { PlanTester planTester = new PlanTester(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/MergeSortMatcher.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/MergeSortMatcher.java new file mode 100644 index 0000000000000..c739b265db519 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/MergeSortMatcher.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.assertions; + +import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.MergeSortNode; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata; +import org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.Ordering; + +import java.util.List; + +import static com.google.common.base.MoreObjects.toStringHelper; +import static com.google.common.base.Preconditions.checkState; +import static java.util.Objects.requireNonNull; +import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.MatchResult.NO_MATCH; +import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.MatchResult.match; +import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.Util.orderingSchemeMatches; + +class MergeSortMatcher implements Matcher { + private final List orderBy; + + public MergeSortMatcher(List orderBy) { + this.orderBy = requireNonNull(orderBy, "orderBy is null"); + } + + @Override + public boolean shapeMatches(PlanNode node) { + return node instanceof MergeSortNode; + } + + @Override + public MatchResult detailMatches( + PlanNode node, SessionInfo sessionInfo, Metadata metadata, SymbolAliases symbolAliases) { + checkState( + shapeMatches(node), + "Plan testing framework error: shapeMatches returned false in detailMatches in %s", + this.getClass().getName()); + MergeSortNode mergeSortNode = (MergeSortNode) node; + if (!orderingSchemeMatches(orderBy, mergeSortNode.getOrderingScheme(), symbolAliases)) { + return NO_MATCH; + } + return match(); + } + + @Override + public String toString() { + return toStringHelper(this).add("orderBy", orderBy).toString(); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java index 6d5102d408a19..9ffd9b9bdecc5 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java @@ -868,6 +868,10 @@ public static PlanMatchPattern mergeSort(PlanMatchPattern... sources) { return node(MergeSortNode.class, sources); } + public static PlanMatchPattern mergeSort(List orderBy, PlanMatchPattern... sources) { + return node(MergeSortNode.class, sources).with(new MergeSortMatcher(orderBy)); + } + public static PlanMatchPattern collect(PlanMatchPattern... sources) { return node(CollectNode.class, sources); }