This is an automated email from the ASF dual-hosted git repository. shuwenwei pushed a commit to branch fix/copy-to-tsfile-unordered-results in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e93753b51c28fbda9ef3995f3b1a54acdf3d280e Author: shuwenwei <[email protected]> AuthorDate: Tue Oct 20 11:41:57 2026 +0800 Support unordered query results in COPY TO TsFile Sort COPY TO input by target tags and time before TsFile writing, preserving the inner query's LIMIT and OFFSET semantics. Skip sorting when timestamps are generated. Add planner unit tests for streaming sort, distributed merge, TopK and offset, aliases, and generated timestamps. Tests and compilation were not run as requested. --- .../relational/planner/TableLogicalPlanner.java | 35 ++++- .../OrderingPropertyPropagationTest.java | 147 +++++++++++++++++++++ 2 files changed, 181 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableLogicalPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableLogicalPlanner.java index 8f9928b7ad2..5c2951cb1f3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableLogicalPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableLogicalPlanner.java @@ -29,12 +29,15 @@ import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchem import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.TableSchema; 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.SortOrder; import org.apache.iotdb.commons.queryengine.plan.relational.planner.Symbol; 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.OffsetNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.OutputNode; 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.Query; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Table; @@ -48,6 +51,7 @@ import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.common.header.DatasetHeader; +import org.apache.iotdb.db.queryengine.execution.operator.process.copyto.tsfile.CopyToTsFileOptions; import org.apache.iotdb.db.queryengine.execution.warnings.WarningCollector; import org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet; import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher; @@ -98,7 +102,10 @@ import org.apache.tsfile.read.common.type.StringType; import org.apache.tsfile.read.common.type.TypeFactory; import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.Optional; @@ -656,10 +663,36 @@ public class TableLogicalPlanner { .infer(analysis, innerQueryRelationPlan, innerQueryRespDatasetHeader.getColumnHeaders()); statement.getOptions().check(innerQueryRespDatasetHeader.getColumnHeaders()); + PlanNode copyToSource = innerQueryRelationPlan.getRoot(); + if (statement.getOptions() instanceof CopyToTsFileOptions tsFileOptions + && !tsFileOptions.isGenerateNewTimeColumn()) { + List<ColumnHeader> columnHeaders = innerQueryRespDatasetHeader.getColumnHeaders(); + List<Symbol> outputSymbols = outputNode.getOutputSymbols(); + Map<String, Symbol> columnSymbols = new HashMap<>(); + for (int i = 0; i < columnHeaders.size(); i++) { + columnSymbols.put(columnHeaders.get(i).getColumnName(), outputSymbols.get(i)); + } + Map<Symbol, SortOrder> orderings = new LinkedHashMap<>(); + for (String tagColumn : tsFileOptions.getTargetTagColumns()) { + orderings.put(columnSymbols.get(tagColumn), SortOrder.ASC_NULLS_FIRST); + } + orderings.put( + columnSymbols.get(tsFileOptions.getTargetTimeColumn()), SortOrder.ASC_NULLS_FIRST); + // Sort the complete inner query result, after its LIMIT and OFFSET. + copyToSource = + new SortNode( + queryContext.getQueryId().genPlanNodeId(), + copyToSource, + new OrderingScheme(new ArrayList<>(orderings.keySet()), orderings), + false, + false); + analysis.setSortNode(true); + } + PlanNode newRoot = new CopyToNode( queryContext.getQueryId().genPlanNodeId(), - innerQueryRelationPlan.getRoot(), + copyToSource, statement.getTargetFileName(), statement.getOptions(), // recording permittedOutputs of CopyToNode's child diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/OrderingPropertyPropagationTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/OrderingPropertyPropagationTest.java index cf0d52bfc66..4661a40cc57 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/OrderingPropertyPropagationTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/OrderingPropertyPropagationTest.java @@ -20,20 +20,28 @@ package org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; +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.FillNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.MergeSortNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.OffsetNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.PatternRecognitionNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.RowNumberNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.SortNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.StreamSortNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.TableFunctionProcessorNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.TopKNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.TopKRankingNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.ValueFillNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.WindowNode; +import org.apache.iotdb.commons.schema.column.ColumnHeader; +import org.apache.iotdb.db.queryengine.execution.operator.process.copyto.tsfile.CopyToTsFileOptions; import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CopyToNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode; +import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.junit.Test; @@ -42,6 +50,7 @@ import java.util.Arrays; import java.util.List; import java.util.stream.Collectors; +import static org.apache.iotdb.commons.queryengine.plan.relational.planner.SortOrder.ASC_NULLS_FIRST; import static org.apache.iotdb.commons.queryengine.plan.relational.planner.SortOrder.DESC_NULLS_LAST; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -437,6 +446,144 @@ public class OrderingPropertyPropagationTest { assertTrue(findNodes(diff, StreamSortNode.class).isEmpty()); } + @Test + public void testCopyToSortUsesStreamingScanAndDistributedMerge() { + for (String source : + List.of( + "table1", + "table1(time, tag1, tag2, tag3, s1)", + "(SELECT time, tag1, tag2, tag3, s1 FROM table1)")) { + PlanTester tester = new PlanTester(); + PlanNode logical = tester.createPlan("COPY " + source + " TO 'copy.tsfile'").getRootNode(); + assertTrue(tester.getAnalysis().isQuery()); + assertTrue(tester.getAnalysis().hasSortNode()); + CopyToNode copyTo = onlyCopyTo(findNodes(logical, CopyToNode.class)); + List<SortNode> sorts = findNodes(copyTo.getChild(), SortNode.class); + assertEquals(1, sorts.size()); + assertTrue(sorts.get(0) instanceof StreamSortNode); + assertCopyToOrdering( + copyTo, sorts.get(0).getOrderingScheme(), "tag1", "tag2", "tag3", "time"); + + CopyToNode distributedCopyTo = onlyCopyTo(findNodes(tester, CopyToNode.class)); + List<MergeSortNode> merges = findNodes(tester, MergeSortNode.class); + assertFalse(merges.isEmpty()); + for (MergeSortNode merge : merges) { + assertCopyToOrdering( + distributedCopyTo, merge.getOrderingScheme(), "tag1", "tag2", "tag3", "time"); + } + assertTrue(findNodes(tester, SortNode.class).isEmpty()); + List<DeviceTableScanNode> scans = findNodes(tester, DeviceTableScanNode.class); + assertFalse(scans.isEmpty()); + scans.forEach(scan -> assertEquals(Ordering.ASC, scan.getScanOrder())); + } + } + + @Test + public void testCopyToSortPreservesInnerTopKAndOffset() { + for (String orderColumn : List.of("time", "s1")) { + PlanTester tester = new PlanTester(); + PlanNode logical = + tester + .createPlan( + "COPY (SELECT time, s1 FROM table1 ORDER BY " + + orderColumn + + " DESC OFFSET 2 LIMIT 3) TO 'copy_limit.tsfile'") + .getRootNode(); + CopyToNode copyTo = onlyCopyTo(findNodes(logical, CopyToNode.class)); + List<SortNode> sorts = findNodes(copyTo.getChild(), SortNode.class); + assertEquals(1, sorts.size()); + assertCopyToOrdering(copyTo, sorts.get(0).getOrderingScheme(), "time"); + List<OffsetNode> offsets = findNodes(sorts.get(0), OffsetNode.class); + assertEquals(1, offsets.size()); + assertEquals(2, offsets.get(0).getCount()); + List<TopKNode> topKs = findNodes(offsets.get(0), TopKNode.class); + assertEquals(1, topKs.size()); + Symbol orderSymbol = + copyTo.getInnerQueryOutputSymbols().get(orderColumn.equals("time") ? 0 : 1); + assertEquals(5, topKs.get(0).getCount()); + assertEquals(List.of(orderSymbol), topKs.get(0).getOrderingScheme().getOrderBy()); + assertEquals(DESC_NULLS_LAST, topKs.get(0).getOrderingScheme().getOrdering(orderSymbol)); + + CopyToNode distributedCopyTo = onlyCopyTo(findNodes(tester, CopyToNode.class)); + List<SortNode> outerSorts = sortsAbove(tester, OffsetNode.class); + assertEquals(1, outerSorts.size()); + assertCopyToOrdering(distributedCopyTo, outerSorts.get(0).getOrderingScheme(), "time"); + offsets = findNodes(tester, OffsetNode.class); + assertEquals(1, offsets.size()); + assertEquals(2, offsets.get(0).getCount()); + topKs = findNodes(tester, TopKNode.class); + assertFalse(topKs.isEmpty()); + for (TopKNode topK : topKs) { + assertEquals(5, topK.getCount()); + assertEquals(List.of(orderSymbol), topK.getOrderingScheme().getOrderBy()); + assertEquals(DESC_NULLS_LAST, topK.getOrderingScheme().getOrdering(orderSymbol)); + } + } + } + + @Test + public void testCopyToSortResolvesAliasesAndTargetTagOrder() { + PlanTester tester = new PlanTester(); + PlanNode logical = + tester + .createPlan( + "COPY (SELECT time AS t, tag1 AS k1, tag2 AS k2, tag3 AS k3, s1 AS v" + + " FROM table1) TO 'copy_alias.tsfile' WITH (TIME t, TAGS(k2, k1, k3))") + .getRootNode(); + CopyToNode copyTo = onlyCopyTo(findNodes(logical, CopyToNode.class)); + List<SortNode> sorts = findNodes(copyTo.getChild(), SortNode.class); + assertEquals(1, sorts.size()); + assertTrue(sorts.get(0) instanceof StreamSortNode); + assertCopyToOrdering(copyTo, sorts.get(0).getOrderingScheme(), "k2", "k1", "k3", "t"); + + CopyToNode distributedCopyTo = onlyCopyTo(findNodes(tester, CopyToNode.class)); + List<MergeSortNode> merges = findNodes(tester, MergeSortNode.class); + assertFalse(merges.isEmpty()); + for (MergeSortNode merge : merges) { + assertCopyToOrdering( + distributedCopyTo, merge.getOrderingScheme(), "k2", "k1", "k3", "t"); + } + assertTrue(findNodes(tester, SortNode.class).isEmpty()); + } + + @Test + public void testCopyToGeneratedTimeNeedsNoSort() { + PlanTester tester = new PlanTester(); + PlanNode logical = + tester + .createPlan("COPY (SELECT tag1, tag2, tag3, s1 FROM table1) TO 'copy_generated.tsfile'") + .getRootNode(); + CopyToNode copyTo = onlyCopyTo(findNodes(logical, CopyToNode.class)); + assertTrue(((CopyToTsFileOptions) copyTo.getCopyToOptions()).isGenerateNewTimeColumn()); + assertFalse(tester.getAnalysis().hasSortNode()); + assertTrue(findNodes(logical, SortNode.class).isEmpty()); + onlyCopyTo(findNodes(tester, CopyToNode.class)); + assertTrue(findNodes(tester, SortNode.class).isEmpty()); + assertTrue(findNodes(tester, MergeSortNode.class).isEmpty()); + } + + private static CopyToNode onlyCopyTo(List<CopyToNode> nodes) { + assertEquals(1, nodes.size()); + return nodes.get(0); + } + + private static void assertCopyToOrdering( + CopyToNode copyTo, OrderingScheme ordering, String... columnNames) { + List<String> outputNames = + copyTo.getInnerQueryDatasetHeader().getColumnHeaders().stream() + .map(ColumnHeader::getColumnName) + .collect(Collectors.toList()); + List<Symbol> expected = new ArrayList<>(); + for (String columnName : columnNames) { + int index = outputNames.indexOf(columnName); + assertTrue(columnName, index >= 0); + expected.add(copyTo.getInnerQueryOutputSymbols().get(index)); + } + assertEquals(expected, ordering.getOrderBy()); + expected.forEach(symbol -> assertEquals(ASC_NULLS_FIRST, ordering.getOrdering(symbol))); + assertTrue(copyTo.getChild().getOutputSymbols().containsAll(expected)); + } + private static PlanTester plan(String sql) { PlanTester tester = new PlanTester(); tester.createPlan(sql);
