From e93753b51c28fbda9ef3995f3b1a54acdf3d280e Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 20 Oct 2026 11:41:57 +0800 Subject: [PATCH 1/8] 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. --- .../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 8f9928b7ad2d9..5c2951cb1f335 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.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.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.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 @@ private RelationPlan createRelationPlan(final CopyTo statement, final Analysis a .infer(analysis, innerQueryRelationPlan, innerQueryRespDatasetHeader.getColumnHeaders()); statement.getOptions().check(innerQueryRespDatasetHeader.getColumnHeaders()); + PlanNode copyToSource = innerQueryRelationPlan.getRoot(); + if (statement.getOptions() instanceof CopyToTsFileOptions tsFileOptions + && !tsFileOptions.isGenerateNewTimeColumn()) { + List columnHeaders = innerQueryRespDatasetHeader.getColumnHeaders(); + List outputSymbols = outputNode.getOutputSymbols(); + Map columnSymbols = new HashMap<>(); + for (int i = 0; i < columnHeaders.size(); i++) { + columnSymbols.put(columnHeaders.get(i).getColumnName(), outputSymbols.get(i)); + } + Map 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 cf0d52bfc667d..4661a40cc5715 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.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 void testMismatchedWindowPartitionAndDiffKeepFullSort() { 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 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 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 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 sorts = findNodes(copyTo.getChild(), SortNode.class); + assertEquals(1, sorts.size()); + assertCopyToOrdering(copyTo, sorts.get(0).getOrderingScheme(), "time"); + List offsets = findNodes(sorts.get(0), OffsetNode.class); + assertEquals(1, offsets.size()); + assertEquals(2, offsets.get(0).getCount()); + List 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 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 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 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 nodes) { + assertEquals(1, nodes.size()); + return nodes.get(0); + } + + private static void assertCopyToOrdering( + CopyToNode copyTo, OrderingScheme ordering, String... columnNames) { + List outputNames = + copyTo.getInnerQueryDatasetHeader().getColumnHeaders().stream() + .map(ColumnHeader::getColumnName) + .collect(Collectors.toList()); + List 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); From 19346d7dbea339c21b3aaaefc865b9455a18fe4f Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 20 Oct 2026 11:59:05 +0800 Subject: [PATCH 2/8] spotless --- .../planner/optimizations/OrderingPropertyPropagationTest.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 4661a40cc5715..16bf18791f4df 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 @@ -540,8 +540,7 @@ public void testCopyToSortResolvesAliasesAndTargetTagOrder() { List merges = findNodes(tester, MergeSortNode.class); assertFalse(merges.isEmpty()); for (MergeSortNode merge : merges) { - assertCopyToOrdering( - distributedCopyTo, merge.getOrderingScheme(), "k2", "k1", "k3", "t"); + assertCopyToOrdering(distributedCopyTo, merge.getOrderingScheme(), "k2", "k1", "k3", "t"); } assertTrue(findNodes(tester, SortNode.class).isEmpty()); } From 30517b1cf9b0543e24892675929153c5108ec62d Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 20 Oct 2026 12:14:33 +0800 Subject: [PATCH 3/8] Add integration coverage for unordered COPY TO TsFile results Cover descending time, field ordering, LIMIT and OFFSET, aliased timestamps and reordered tags, and exports without tags. Read back all rows and use small TsBlocks to exercise multiple write batches. Set MEMORY_THRESHOLD to 1000000 for the new cases. All five integration tests passed according to user validation; tests and compilation were not rerun by the agent. --- .../recent/copyto/IoTDBCopyToTsFileIT.java | 125 ++++++++++++++++++ 1 file changed, 125 insertions(+) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/copyto/IoTDBCopyToTsFileIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/copyto/IoTDBCopyToTsFileIT.java index 388d833f700fe..7092bc265c61c 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/copyto/IoTDBCopyToTsFileIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/copyto/IoTDBCopyToTsFileIT.java @@ -32,6 +32,9 @@ import org.apache.tsfile.file.metadata.StringArrayDeviceID; import org.apache.tsfile.file.metadata.TimeseriesMetadata; import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.query.dataset.ResultSet; +import org.apache.tsfile.read.v4.ITsFileReader; +import org.apache.tsfile.read.v4.TsFileReaderBuilder; import org.junit.After; import org.junit.AfterClass; import org.junit.Assert; @@ -45,6 +48,10 @@ import java.io.File; import java.io.IOException; import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; @@ -76,6 +83,7 @@ public class IoTDBCopyToTsFileIT { @BeforeClass public static void setUp() throws Exception { + EnvFactory.getEnv().getConfig().getCommonConfig().setMaxTsBlockLineNumber(2); EnvFactory.getEnv().initClusterEnvironment(); prepareTableData(createSqls); } @@ -743,6 +751,123 @@ public void testSpecifiedDatabaseTable() } } + @Test + public void testCopyWithDescendingTime() throws Exception { + assertCopyToRows( + "COPY (SELECT time, tag1, tag2, s1, s2 FROM table1 ORDER BY time DESC, tag1)" + + " TO 'copy_descending_time.tsfile' WITH (MEMORY_THRESHOLD 1000000)", + List.of("tag1", "tag2"), + "1,t1_1,t2,1,1", + "2,t1_1,t2,2,2", + "3,t1_1,t2,3,3", + "1,t1_2,t2,1,1", + "2,t1_2,t2,2,2", + "3,t1_2,t2,3,3"); + } + + @Test + public void testCopyWithUnorderedTimeByField() throws Exception { + assertCopyToRows( + "COPY (SELECT time, tag1, tag2, s1, s2 FROM table1 ORDER BY s1 % 2, time DESC, tag1)" + + " TO 'copy_field_order.tsfile' WITH (MEMORY_THRESHOLD 1000000)", + List.of("tag1", "tag2"), + "1,t1_1,t2,1,1", + "2,t1_1,t2,2,2", + "3,t1_1,t2,3,3", + "1,t1_2,t2,1,1", + "2,t1_2,t2,2,2", + "3,t1_2,t2,3,3"); + } + + @Test + public void testCopyWithDescendingTimeAndLimitOffset() throws Exception { + assertCopyToRows( + "COPY (SELECT time, tag1, tag2, s1, s2 FROM table1" + + " ORDER BY time DESC, tag1 OFFSET 1 LIMIT 3) TO 'copy_descending_limit.tsfile'" + + " WITH (MEMORY_THRESHOLD 1000000)", + List.of("tag1", "tag2"), + "2,t1_1,t2,2,2", + "2,t1_2,t2,2,2", + "3,t1_2,t2,3,3"); + } + + @Test + public void testCopyWithDescendingAliasedTimeAndReorderedTags() throws Exception { + assertCopyToRows( + "COPY (SELECT time AS t, tag1 AS renamed_tag1, tag2 AS renamed_tag2, s1, s2" + + " FROM table1 ORDER BY t DESC, renamed_tag1)" + + " TO 'copy_descending_alias.tsfile'" + + " WITH (TIME t, TAGS(renamed_tag2, renamed_tag1), MEMORY_THRESHOLD 1000000)", + List.of("renamed_tag2", "renamed_tag1"), + "1,t2,t1_1,1,1", + "2,t2,t1_1,2,2", + "3,t2,t1_1,3,3", + "1,t2,t1_2,1,1", + "2,t2,t1_2,2,2", + "3,t2,t1_2,3,3"); + } + + @Test + public void testCopyWithDescendingTimeWithoutTags() throws Exception { + assertCopyToRows( + "COPY (SELECT time, s1, s2 FROM table1 WHERE tag1 = 't1_1' ORDER BY time DESC)" + + " TO 'copy_descending_no_tags.tsfile' WITH (MEMORY_THRESHOLD 1000000)", + Collections.emptyList(), + "1,1,1", + "2,2,2", + "3,3,3"); + } + + private void assertCopyToRows(String sql, List tagColumns, String... expectedRows) + throws Exception { + long deviceCount; + try (ITableSession session = + EnvFactory.getEnv().getTableSessionConnectionWithDB(DATABASE_NAME); + SessionDataSet dataSet = session.executeQueryStatement(sql)) { + SessionDataSet.DataIterator iterator = dataSet.iterator(); + Assert.assertTrue(iterator.next()); + targetFilePath = iterator.getString(1); + Assert.assertTrue(new File(targetFilePath).isFile()); + Assert.assertEquals(expectedRows.length, iterator.getLong(2)); + deviceCount = iterator.getLong(3); + Assert.assertTrue(iterator.getLong(4) > 0); + Assert.assertEquals("table1", iterator.getString(5)); + Assert.assertEquals(tagColumns.toString(), iterator.getString(7)); + Assert.assertFalse(iterator.next()); + } + + List columns = new ArrayList<>(tagColumns); + columns.add("s1"); + columns.add("s2"); + List actualRows = new ArrayList<>(); + Map, Long> lastTimeByDevice = new HashMap<>(); + try (ITsFileReader reader = new TsFileReaderBuilder().file(new File(targetFilePath)).build(); + ResultSet result = reader.query("table1", columns, Long.MIN_VALUE, Long.MAX_VALUE)) { + while (result.next()) { + long time = result.getLong(1); + StringBuilder row = new StringBuilder(Long.toString(time)); + List tags = new ArrayList<>(); + for (int i = 0; i < tagColumns.size(); i++) { + String tag = result.isNull(i + 2) ? null : result.getString(i + 2); + tags.add(tag); + row.append(',').append(tag); + } + Long lastTime = lastTimeByDevice.put(tags, time); + if (lastTime != null) { + Assert.assertTrue(time > lastTime); + } + row.append(',').append(result.getInt(tagColumns.size() + 2)); + row.append(',').append(result.getInt(tagColumns.size() + 3)); + actualRows.add(row.toString()); + } + } + Assert.assertEquals(deviceCount, lastTimeByDevice.size()); + List expected = new ArrayList<>(Arrays.asList(expectedRows)); + Collections.sort(expected); + Collections.sort(actualRows); + Assert.assertEquals(expected, actualRows); + } + @Test public void testFindTimeColumn() throws IoTDBConnectionException, StatementExecutionException, IOException { From 92336b94061861bfa1adf6cb1a09ac2b9dfa27f5 Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 20 Oct 2026 15:07:20 +0800 Subject: [PATCH 4/8] Support EXPLAIN and EXPLAIN ANALYZE for COPY TO Add integration coverage for plan output and normalize writer initialization failures to COPY_TO_WRITE_ERROR. --- .../recent/copyto/IoTDBCopyToTsFileIT.java | 60 +++++++++++++++++++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 7 +++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 7 +++ .../process/copyto/TableCopyToOperator.java | 7 ++- .../plan/node/PlanGraphJsonPrinter.java | 11 ++++ .../planner/plan/node/PlanGraphPrinter.java | 36 +++++++++++ .../analyzer/StatementAnalyzer.java | 8 ++- .../planner/TableLogicalPlanner.java | 23 ++++--- .../TableModelTypeProviderExtractor.java | 3 +- .../relational/planner/node/CopyToNode.java | 13 ++-- .../UnaliasSymbolReferences.java | 3 +- .../relational/sql/parser/AstBuilder.java | 4 ++ .../relational/grammar/sql/RelationalSql.g4 | 4 +- 13 files changed, 168 insertions(+), 18 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/copyto/IoTDBCopyToTsFileIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/copyto/IoTDBCopyToTsFileIT.java index 7092bc265c61c..f0093a16bbc0c 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/copyto/IoTDBCopyToTsFileIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/copyto/IoTDBCopyToTsFileIT.java @@ -105,6 +105,66 @@ public static void tearDown() throws Exception { EnvFactory.getEnv().cleanClusterEnvironment(); } + @Test + public void testExplainCopyTable() throws Exception { + assertExplainCopyToContainsNodes("explain", "table1"); + } + + @Test + public void testExplainCopyQueryJson() throws Exception { + assertExplainCopyToContainsNodes( + "explain (format json)", "(select * from table1 order by time desc limit 3)"); + } + + @Test + public void testExplainAnalyzeCopyTable() throws Exception { + assertExplainCopyToContainsNodes("explain analyze", "table1(time, tag1, tag2, s1, s2)"); + } + + @Test + public void testExplainAnalyzeCopyQueryJson() throws Exception { + assertExplainCopyToContainsNodes( + "explain analyze verbose (format json)", + "(select * from table1 order by time desc limit 3)"); + } + + private void assertExplainCopyToContainsNodes(String explainPrefix, String source) + throws Exception { + File targetDirectory = Files.createTempDirectory("iotdb-explain-copy-to").toFile(); + File targetFile = new File(targetDirectory, "result.tsfile"); + String targetPath = targetFile.getAbsolutePath().replace("\\", "\\\\").replace("'", "''"); + String exportDirectoryPath = + targetDirectory.getAbsolutePath().replace("\\", "\\\\").replace("'", "''"); + + try (ITableSession session = + EnvFactory.getEnv().getTableSessionConnectionWithDB(DATABASE_NAME)) { + session.executeNonQueryStatement( + "set configuration \"copy_to_allowed_export_dirs\"='" + exportDirectoryPath + "'"); + try (SessionDataSet dataSet = + session.executeQueryStatement( + explainPrefix + + " copy " + + source + + " to '" + + targetPath + + "' (memory_threshold 1000000)")) { + SessionDataSet.DataIterator iterator = dataSet.iterator(); + StringBuilder output = new StringBuilder(); + while (iterator.next()) { + output.append(iterator.getString(1)).append('\n'); + } + String result = output.toString(); + Assert.assertTrue(result, result.contains("TableScan")); + Assert.assertTrue(result, result.contains("CopyTo")); + } finally { + session.executeNonQueryStatement("set configuration \"copy_to_allowed_export_dirs\"=''"); + } + } finally { + Files.deleteIfExists(targetFile.toPath()); + Files.deleteIfExists(targetDirectory.toPath()); + } + } + @Test public void testCopyTable() throws IoTDBConnectionException, StatementExecutionException, IOException { diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index 1672c9024f573..182a0dedde092 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -3938,4 +3938,11 @@ private DataNodeQueryMessages() {} public static final String EXCEPTION_ONLY_INMEMORYDEVICEENTRYDATASET_SUPPORTS_GET_INLINE_DEVICE_ENTRIES_07A52CAB = "Only InMemoryDeviceEntryDataSet supports get inline device entries"; + public static final String MESSAGE_COPYTO_ARG_38343199 = "CopyTo-%s"; + public static final String MESSAGE_TARGETFILE_ARG_EC484B24 = "TargetFile: %s"; + public static final String MESSAGE_FORMAT_ARG_CE5DC4EE = "Format: %s"; + public static final String MESSAGE_TARGETTABLE_ARG_00BE4714 = "TargetTable: %s"; + public static final String MESSAGE_TIMECOLUMN_ARG_54F6A2A6 = "TimeColumn: %s"; + public static final String MESSAGE_TAGCOLUMNS_ARG_6520A0C2 = "TagColumns: %s"; + public static final String MESSAGE_MEMORYTHRESHOLD_ARG_3676AB46 = "MemoryThreshold: %s"; } diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index 114fc1240ffda..d38536fd02c3a 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -4694,4 +4694,11 @@ private DataNodeQueryMessages() {} public static final String EXCEPTION_ONLY_INMEMORYDEVICEENTRYDATASET_SUPPORTS_GET_INLINE_DEVICE_ENTRIES_07A52CAB = "只有 InMemoryDeviceEntryDataSet 支持获取内存中的设备条目"; + public static final String MESSAGE_COPYTO_ARG_38343199 = "CopyTo-%s"; + public static final String MESSAGE_TARGETFILE_ARG_EC484B24 = "目标文件：%s"; + public static final String MESSAGE_FORMAT_ARG_CE5DC4EE = "格式：%s"; + public static final String MESSAGE_TARGETTABLE_ARG_00BE4714 = "目标表：%s"; + public static final String MESSAGE_TIMECOLUMN_ARG_54F6A2A6 = "时间列：%s"; + public static final String MESSAGE_TAGCOLUMNS_ARG_6520A0C2 = "TAG 列：%s"; + public static final String MESSAGE_MEMORYTHRESHOLD_ARG_3676AB46 = "内存阈值：%s"; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/copyto/TableCopyToOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/copyto/TableCopyToOperator.java index f43fbd2b730e8..14ead195e8d3d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/copyto/TableCopyToOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/copyto/TableCopyToOperator.java @@ -83,7 +83,12 @@ public OperatorContext getOperatorContext() { @Override public TsBlock next() throws Exception { - IFormatCopyToWriter formatWriter = getWriter(); + IFormatCopyToWriter formatWriter; + try { + formatWriter = getWriter(); + } catch (IOException e) { + throw newCopyToWriteError(e); + } if (!childOperator.hasNext()) { try { formatWriter.seal(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphJsonPrinter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphJsonPrinter.java index ecd9d2a58a576..a5f6e06145306 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphJsonPrinter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphJsonPrinter.java @@ -31,6 +31,8 @@ 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.planner.node.UnionNode; +import org.apache.iotdb.db.queryengine.execution.operator.process.copyto.tsfile.CopyToTsFileOptions; +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.relational.planner.node.ExchangeNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExplainAnalyzeNode; @@ -96,6 +98,15 @@ private static JsonObject buildProperties(PlanNode node) { } else if (node instanceof ExplainAnalyzeNode) { ExplainAnalyzeNode n = (ExplainAnalyzeNode) node; properties.add("ChildPermittedOutputs", toJsonArray(n.getChildPermittedOutputs())); + } else if (node instanceof CopyToNode n) { + properties.addProperty("TargetFile", n.getTargetFilePath()); + properties.addProperty("Format", n.getCopyToOptions().getFormat().name()); + if (n.getCopyToOptions() instanceof CopyToTsFileOptions options) { + properties.addProperty("TargetTable", options.getTargetTableName()); + properties.addProperty("TimeColumn", options.getTargetTimeColumn()); + properties.add("TagColumns", toJsonArray(options.getTargetTagColumns())); + properties.addProperty("MemoryThreshold", options.getTargetMemoryThreshold()); + } } else if (node instanceof TableScanNode) { buildTableScanProperties(properties, (TableScanNode) node); } else if (node instanceof ExchangeNode) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java index 3316299ea17d6..557bd714d920f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java @@ -51,6 +51,7 @@ import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.ValuesNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.WindowNode; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; +import org.apache.iotdb.db.queryengine.execution.operator.process.copyto.tsfile.CopyToTsFileOptions; import org.apache.iotdb.db.queryengine.plan.analyze.TemplatedInfo; import org.apache.iotdb.db.queryengine.plan.expression.Expression; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.AggregationMergeSortNode; @@ -95,6 +96,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.DeviceViewIntoPathDescriptor; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.IntoPathDescriptor; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTreeDeviceViewScanNode; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CopyToNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CteScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExchangeNode; @@ -866,6 +868,40 @@ public List visitInto( return render(node, boxValue, context); } + @Override + public List visitCopyTo(CopyToNode node, GraphContext context) { + List boxValue = new ArrayList<>(); + boxValue.add( + String.format( + DataNodeQueryMessages.MESSAGE_COPYTO_ARG_38343199, node.getPlanNodeId().getId())); + boxValue.add( + String.format( + DataNodeQueryMessages.MESSAGE_TARGETFILE_ARG_EC484B24, node.getTargetFilePath())); + boxValue.add( + String.format( + DataNodeQueryMessages.MESSAGE_FORMAT_ARG_CE5DC4EE, + node.getCopyToOptions().getFormat())); + if (node.getCopyToOptions() instanceof CopyToTsFileOptions options) { + boxValue.add( + String.format( + DataNodeQueryMessages.MESSAGE_TARGETTABLE_ARG_00BE4714, + options.getTargetTableName())); + boxValue.add( + String.format( + DataNodeQueryMessages.MESSAGE_TIMECOLUMN_ARG_54F6A2A6, + options.getTargetTimeColumn())); + boxValue.add( + String.format( + DataNodeQueryMessages.MESSAGE_TAGCOLUMNS_ARG_6520A0C2, + options.getTargetTagColumns())); + boxValue.add( + String.format( + DataNodeQueryMessages.MESSAGE_MEMORYTHRESHOLD_ARG_3676AB46, + options.getTargetMemoryThreshold())); + } + return render(node, boxValue, context); + } + @Override public List visitOutput(OutputNode node, GraphContext context) { List boxValue = new ArrayList<>(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java index 88f605f6dfd65..afb58c645c00a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java @@ -1410,7 +1410,9 @@ public Scope visitExplain(Explain node, Optional context) { queryContext.setExplainType(ExplainType.EXPLAIN); queryContext.setExplainOutputFormat(node.getOutputFormat()); analysis.setFinishQueryAfterAnalyze(); - return visitQuery((Query) node.getStatement(), context); + Scope scope = process(node.getStatement(), context); + analysis.setScope(node, scope); + return scope; } @Override @@ -1437,7 +1439,9 @@ public Scope visitExplainAnalyze(ExplainAnalyze node, Optional context) { queryContext.setExplainType(ExplainType.EXPLAIN_ANALYZE); queryContext.setVerbose(node.isVerbose()); queryContext.setExplainOutputFormat(node.getOutputFormat()); - return visitQuery((Query) node.getStatement(), context); + Scope scope = process(node.getStatement(), context); + analysis.setScope(node, scope); + return scope; } @Override 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 5c2951cb1f335..ff71e384280a6 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 @@ -254,7 +254,7 @@ private RelationPlan planStatementWithoutOutput(Analysis analysis, Statement sta return createRelationPlan(analysis, (Query) statement); } if (statement instanceof Explain) { - return createRelationPlan(analysis, (Query) ((Explain) statement).getStatement()); + return planStatementWithoutOutput(analysis, ((Explain) statement).getStatement()); } if (statement instanceof WrappedStatement) { return createRelationPlan(analysis, ((WrappedStatement) statement)); @@ -353,14 +353,14 @@ private PlanNode createOutputPlan(RelationPlan plan, Analysis analysis) { int columnNumber = 0; // TODO perfect the logic of outputDescriptor if (plan.getRoot() instanceof CopyToNode) { + outputs.addAll(plan.getRoot().getOutputSymbols()); for (ColumnHeader columnHeader : ((CopyToNode) plan.getRoot()).getCopyToOptions().getRespColumnHeaders()) { - outputs.add(new Symbol(columnHeader.getColumnName())); names.add(columnHeader.getColumnName()); columnHeaders.add(columnHeader); } - } else if (queryContext.isExplainAnalyze() && !queryContext.isInnerTriggeredQuery()) { - outputs.add(new Symbol(ColumnHeaderConstant.EXPLAIN_ANALYZE)); + } else if (plan.getRoot() instanceof ExplainAnalyzeNode) { + outputs.addAll(plan.getRoot().getOutputSymbols()); names.add(ColumnHeaderConstant.EXPLAIN_ANALYZE); columnHeaders.add(new ColumnHeader(ColumnHeaderConstant.EXPLAIN_ANALYZE, TSDataType.TEXT)); } else { @@ -629,7 +629,7 @@ private void planTraverseDevice(final AbstractTraverseDevice statement, final An private RelationPlan planExplainAnalyze(final ExplainAnalyze statement, final Analysis analysis) { RelationPlan originalQueryPlan = - createRelationPlan(analysis, (Query) (statement.getStatement())); + planStatementWithoutOutput(analysis, statement.getStatement()); Symbol symbol = symbolAllocator.newSymbol(ColumnHeaderConstant.EXPLAIN_ANALYZE, StringType.getInstance()); @@ -642,7 +642,9 @@ private RelationPlan planExplainAnalyze(final ExplainAnalyze statement, final An queryContext.getTimeOut(), symbol, // recording permittedOutputs of ExplainAnalyzeNode's child - getChildPermittedOutputs(analysis, statement.getStatement(), originalQueryPlan), + statement.getStatement() instanceof CopyTo + ? originalQueryPlan.getRoot().getOutputSymbols() + : getChildPermittedOutputs(analysis, statement.getStatement(), originalQueryPlan), statement.getOutputFormat()); return new RelationPlan( newRoot, @@ -689,6 +691,12 @@ private RelationPlan createRelationPlan(final CopyTo statement, final Analysis a analysis.setSortNode(true); } + List outputSymbols = new ArrayList<>(); + for (ColumnHeader columnHeader : statement.getOptions().getRespColumnHeaders()) { + outputSymbols.add( + symbolAllocator.newSymbol( + columnHeader.getColumnName(), TypeFactory.getType(columnHeader.getColumnType()))); + } PlanNode newRoot = new CopyToNode( queryContext.getQueryId().genPlanNodeId(), @@ -699,7 +707,8 @@ private RelationPlan createRelationPlan(final CopyTo statement, final Analysis a getChildPermittedOutputs( analysis, statement.getQueryStatement(), innerQueryRelationPlan), innerQueryRespDatasetHeader, - outputNode.getOutputSymbols()); + outputNode.getOutputSymbols(), + outputSymbols); return new RelationPlan( newRoot, innerQueryRelationPlan.getScope(), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableModelTypeProviderExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableModelTypeProviderExtractor.java index aaf3725ec2a97..356efb9740613 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableModelTypeProviderExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableModelTypeProviderExtractor.java @@ -107,7 +107,8 @@ private void addOutputSymbolsToTypeProvider(PlanNode node) { @Override public Void visitCopyTo(CopyToNode node, Void context) { - visitPlan(node.getChild(), context); + node.getChild().accept(this, context); + addOutputSymbolsToTypeProvider(node); return null; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/CopyToNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/CopyToNode.java index 2e251bd7a1964..ff6899444c45f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/CopyToNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/CopyToNode.java @@ -41,6 +41,7 @@ public class CopyToNode extends SingleChildProcessNode { private final List childPermittedOutputs; private final List innerQueryOutputSymbols; private final DatasetHeader innerQueryDatasetHeader; + private final List outputSymbols; public CopyToNode( PlanNodeId id, @@ -49,7 +50,8 @@ public CopyToNode( CopyToOptions copyToOptions, List childPermittedOutputs, DatasetHeader innerQueryDatasetHeader, - List innerQueryOutputSymbols) { + List innerQueryOutputSymbols, + List outputSymbols) { super(id); this.child = child; this.targetFilePath = targetFilePath; @@ -57,6 +59,7 @@ public CopyToNode( this.childPermittedOutputs = childPermittedOutputs; this.innerQueryDatasetHeader = innerQueryDatasetHeader; this.innerQueryOutputSymbols = innerQueryOutputSymbols; + this.outputSymbols = outputSymbols; } public DatasetHeader getInnerQueryDatasetHeader() { @@ -88,7 +91,8 @@ public PlanNode clone() { copyToOptions, childPermittedOutputs, innerQueryDatasetHeader, - innerQueryOutputSymbols); + innerQueryOutputSymbols, + outputSymbols); } @Override @@ -98,7 +102,7 @@ public R accept(IPlanVisitor visitor, C context) { @Override public List getOutputSymbols() { - return copyToOptions.getOutputSymbols(); + return outputSymbols; } @Override @@ -115,7 +119,8 @@ public PlanNode replaceChildren(List newChildren) { copyToOptions, childPermittedOutputs, innerQueryDatasetHeader, - innerQueryOutputSymbols); + innerQueryOutputSymbols, + outputSymbols); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java index 131ba8af383c2..89e377bcffe05 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java @@ -484,7 +484,8 @@ public PlanAndMappings visitCopyTo(CopyToNode node, UnaliasContext context) { node.getCopyToOptions(), newChildPermittedOutputs, node.getInnerQueryDatasetHeader(), - newInnerQueryOutputSymbols), + newInnerQueryOutputSymbols, + node.getOutputSymbols()), mapping); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java index 7b8102960d09d..fab87b3c18418 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java @@ -1855,6 +1855,8 @@ public Node visitExplain(RelationalSqlParser.ExplainContext ctx) { Statement innerStatement; if (ctx.query() != null) { innerStatement = (Statement) visit(ctx.query()); + } else if (ctx.copyToStatement() != null) { + innerStatement = (Statement) visit(ctx.copyToStatement()); } else if (ctx.executeStatement() != null) { innerStatement = (Statement) visit(ctx.executeStatement()); } else { @@ -1886,6 +1888,8 @@ public Node visitExplainAnalyze(RelationalSqlParser.ExplainAnalyzeContext ctx) { Statement innerStatement; if (ctx.query() != null) { innerStatement = (Statement) visit(ctx.query()); + } else if (ctx.copyToStatement() != null) { + innerStatement = (Statement) visit(ctx.copyToStatement()); } else if (ctx.executeStatement() != null) { innerStatement = (Statement) visit(ctx.executeStatement()); } else { diff --git a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 index 629f4d5f4d4b0..44c681c7168dc 100644 --- a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 +++ b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 @@ -963,8 +963,8 @@ copyToStatementOption // ------------------------------------------- Query Statement --------------------------------------------------------- queryStatement : query #statementDefault - | EXPLAIN ('(' FORMAT identifier ')')? (query | executeStatement | executeImmediateStatement) #explain - | EXPLAIN ANALYZE VERBOSE? ('(' FORMAT identifier ')')? (query | executeStatement | executeImmediateStatement) #explainAnalyze + | EXPLAIN ('(' FORMAT identifier ')')? (query | copyToStatement | executeStatement | executeImmediateStatement) #explain + | EXPLAIN ANALYZE VERBOSE? ('(' FORMAT identifier ')')? (query | copyToStatement | executeStatement | executeImmediateStatement) #explainAnalyze ; query From f1ec16b8d95bf084c1731d26de7b5c86ffbf1d99 Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 20 Oct 2026 15:21:16 +0800 Subject: [PATCH 5/8] spotless --- .../plan/relational/planner/TableLogicalPlanner.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 ff71e384280a6..c68ebb7e254fe 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 @@ -628,8 +628,7 @@ private void planTraverseDevice(final AbstractTraverseDevice statement, final An } private RelationPlan planExplainAnalyze(final ExplainAnalyze statement, final Analysis analysis) { - RelationPlan originalQueryPlan = - planStatementWithoutOutput(analysis, statement.getStatement()); + RelationPlan originalQueryPlan = planStatementWithoutOutput(analysis, statement.getStatement()); Symbol symbol = symbolAllocator.newSymbol(ColumnHeaderConstant.EXPLAIN_ANALYZE, StringType.getInstance()); From def06d17fd0baa8747d020b621211ac5bf180157 Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 20 Oct 2026 16:42:58 +0800 Subject: [PATCH 6/8] Validate COPY TO time columns and update EXPLAIN error expectations --- .../relational/it/insertquery/IoTDBInsertQueryIT.java | 4 ++-- .../en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java | 3 +++ .../zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java | 3 +++ .../process/copyto/tsfile/CopyToTsFileOptions.java | 7 +++++++ 4 files changed, 15 insertions(+), 2 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java index 75a7725c41498..d6aa44a18f147 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java @@ -461,7 +461,7 @@ public void testExplain() throws SQLException { e.getMessage(), e.getMessage() .contains( - "700: line 1:9: mismatched input 'INSERT'. Expecting: '(', 'ANALYZE', 'EXECUTE', ")); + "700: line 1:9: mismatched input 'INSERT'. Expecting: '(', 'ANALYZE', 'COPY', 'EXECUTE', ")); } try { @@ -473,7 +473,7 @@ public void testExplain() throws SQLException { e.getMessage(), e.getMessage() .contains( - "700: line 1:17: mismatched input 'INSERT'. Expecting: '(', 'EXECUTE', 'VERBOSE', ")); + "700: line 1:17: mismatched input 'INSERT'. Expecting: '(', 'COPY', 'EXECUTE', 'VERBOSE', ")); } } diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index 7b0b189f53841..8bb740269cea1 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -280,6 +280,9 @@ public final class DataNodeQueryMessages { + " (configure copy_to_allowed_export_dirs to permit it): "; public static final String DATA_TYPE_OF_TARGET_TIME_COLUMN_IS_NOT = "Data type of target time column is not TIMESTAMP"; + public static final String + EXCEPTION_SPECIFIED_TIME_COLUMN_ARG_DOES_NOT_EXIST_IN_THE_QUERY_RESULT_34791B26 = + "Specified TIME column %s does not exist in the query result"; public static final String DUPLICATE_COLUMN_NAMES_IN_QUERY_DATASET = "Duplicate column names in query dataset."; public static final String SOME_SPECIFIED_TAG_COLUMNS_ARE_NOT_EXIST_IN = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index e4c928bff6742..68f686925c473 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -268,6 +268,9 @@ public final class DataNodeQueryMessages { "COPY TO 目标路径不在允许的导出目录内（可通过 copy_to_allowed_export_dirs 配置）："; public static final String DATA_TYPE_OF_TARGET_TIME_COLUMN_IS_NOT = "目标时间列的数据类型不是 TIMESTAMP"; + public static final String + EXCEPTION_SPECIFIED_TIME_COLUMN_ARG_DOES_NOT_EXIST_IN_THE_QUERY_RESULT_34791B26 = + "指定的 TIME 列 %s 在查询结果中不存在"; public static final String DUPLICATE_COLUMN_NAMES_IN_QUERY_DATASET = "查询数据集中存在重复的列名。"; public static final String SOME_SPECIFIED_TAG_COLUMNS_ARE_NOT_EXIST_IN = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/copyto/tsfile/CopyToTsFileOptions.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/copyto/tsfile/CopyToTsFileOptions.java index 22b2e25a6df36..ffdba2995b1af 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/copyto/tsfile/CopyToTsFileOptions.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/copyto/tsfile/CopyToTsFileOptions.java @@ -201,6 +201,13 @@ public void check(List columnHeaders) { if (columns.size() != columnHeaders.size()) { throw new SemanticException(DataNodeQueryMessages.DUPLICATE_COLUMN_NAMES_IN_QUERY_DATASET); } + if (!generateNewTimeColumn && !columns.contains(targetTimeColumn)) { + throw new SemanticException( + String.format( + DataNodeQueryMessages + .EXCEPTION_SPECIFIED_TIME_COLUMN_ARG_DOES_NOT_EXIST_IN_THE_QUERY_RESULT_34791B26, + targetTimeColumn)); + } if (foundTagColumns != targetTagColumns.size()) { throw new SemanticException( DataNodeQueryMessages.SOME_SPECIFIED_TAG_COLUMNS_ARE_NOT_EXIST_IN); From 600ba96f537cebdfb1f371c021588b8a5eecc84a Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 20 Oct 2026 16:54:04 +0800 Subject: [PATCH 7/8] Handle sort optimization for queries without table scans --- .../it/query/recent/IoTDBComplexQueryIT.java | 9 ++++++ .../TransformSortToStreamSort.java | 4 +++ .../OrderingPropertyPropagationTest.java | 28 +++++++++++++++++++ 3 files changed, 41 insertions(+) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBComplexQueryIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBComplexQueryIT.java index 55b421422497f..e3ca746ecb7ea 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBComplexQueryIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBComplexQueryIT.java @@ -82,6 +82,15 @@ public void queryTest1() { DATABASE_NAME); } + @Test + public void testTableLessQueryWithOrderBy() { + tableResultSetEqualTest( + "SELECT CAST(1 AS TIMESTAMP) AS time, 1 AS v ORDER BY v, time", + new String[] {"time", "v"}, + new String[] {"1970-01-01T00:00:00.001Z,1,"}, + DATABASE_NAME); + } + @Test public void testTableLessQuery() { String[] expectedHeader; 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 a89b2d3169469..90c2c91931d13 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 @@ -97,6 +97,10 @@ public PlanNode visitSort(SortNode node, Context context) { context.setCanTransform(false); DeviceTableScanNode deviceTableScanNode = context.getTableScanNode(); + if (deviceTableScanNode == null) { + node.setChild(child); + return node; + } Map tableColumnSchema; if (deviceTableScanNode instanceof ExternalTsFileScanNode) { tableColumnSchema = 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 16bf18791f4df..1db1846e2610a 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 @@ -446,6 +446,34 @@ public void testMismatchedWindowPartitionAndDiffKeepFullSort() { assertTrue(findNodes(diff, StreamSortNode.class).isEmpty()); } + @Test + public void testCopyToWithoutFromKeepsRegularSort() { + for (String prefix : List.of("", "EXPLAIN ", "EXPLAIN ANALYZE ")) { + PlanTester tester = new PlanTester(); + PlanNode logical = + tester + .createPlan( + prefix + + "COPY (SELECT CAST(1 AS TIMESTAMP) AS time, 1 AS v)" + + " TO 'constant.tsfile' (MEMORY_THRESHOLD 1000000)") + .getRootNode(); + CopyToNode copyTo = onlyCopyTo(findNodes(logical, CopyToNode.class)); + assertFalse(((CopyToTsFileOptions) copyTo.getCopyToOptions()).isGenerateNewTimeColumn()); + List sorts = findNodes(copyTo.getChild(), SortNode.class); + assertEquals(1, sorts.size()); + assertCopyToOrdering(copyTo, sorts.get(0).getOrderingScheme(), "time"); + assertTrue(findNodes(logical, StreamSortNode.class).isEmpty()); + assertTrue(findNodes(logical, DeviceTableScanNode.class).isEmpty()); + + CopyToNode distributedCopyTo = onlyCopyTo(findNodes(tester, CopyToNode.class)); + sorts = findNodes(tester, SortNode.class); + assertEquals(1, sorts.size()); + assertCopyToOrdering(distributedCopyTo, sorts.get(0).getOrderingScheme(), "time"); + assertTrue(findNodes(tester, StreamSortNode.class).isEmpty()); + assertTrue(findNodes(tester, DeviceTableScanNode.class).isEmpty()); + } + } + @Test public void testCopyToSortUsesStreamingScanAndDistributedMerge() { for (String source : From 7cafbb3a382494898dbb873868c5063df88980db Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 20 Oct 2026 17:33:41 +0800 Subject: [PATCH 8/8] Optimize redundant inner sorts for COPY TO exports --- .../optimizations/LogicalOptimizeFactory.java | 2 + .../RemoveRedundantSortForCopyTo.java | 88 ++++++++++++++ .../OrderingPropertyPropagationTest.java | 109 +++++++++++++++++- 3 files changed, 198 insertions(+), 1 deletion(-) create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/RemoveRedundantSortForCopyTo.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java index 78bb5f984a15c..ddf09e7f98d4d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java @@ -211,6 +211,8 @@ public LogicalOptimizeFactory(PlannerContext plannerContext) { ImmutableList.Builder optimizerBuilder = ImmutableList.builder(); optimizerBuilder.add( + // Remove superseded COPY TO ordering before scan ordering and limit pushdown. + new RemoveRedundantSortForCopyTo(), new IterativeOptimizer( plannerContext, ruleStats, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/RemoveRedundantSortForCopyTo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/RemoveRedundantSortForCopyTo.java new file mode 100644 index 0000000000000..d4295cce17ea7 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/RemoveRedundantSortForCopyTo.java @@ -0,0 +1,88 @@ +/* + * 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.optimizations; + +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; +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.db.queryengine.execution.operator.process.copyto.tsfile.CopyToTsFileOptions; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CopyToNode; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExplainAnalyzeNode; + +import com.google.common.collect.ImmutableList; + +/** Removes an inner sort superseded by the TAG/TIME sort required for COPY TO TsFile. */ +public class RemoveRedundantSortForCopyTo implements PlanOptimizer { + + @Override + public PlanNode optimize(PlanNode plan, Context context) { + if (!context.getAnalysis().hasSortNode()) { + return plan; + } + return plan.accept(new Rewriter(), null); + } + + private static class Rewriter implements PlanVisitor { + + @Override + public PlanNode visitPlan(PlanNode node, Void context) { + // COPY TO is a statement, so only its output and EXPLAIN ANALYZE wrappers can precede it. + if (node instanceof OutputNode || node instanceof ExplainAnalyzeNode) { + PlanNode child = node.getChildren().get(0); + PlanNode rewrittenChild = child.accept(this, context); + return child == rewrittenChild + ? node + : node.replaceChildren(ImmutableList.of(rewrittenChild)); + } + return node; + } + + @Override + public PlanNode visitCopyTo(CopyToNode node, Void context) { + if (node.getCopyToOptions() instanceof CopyToTsFileOptions options + && !options.isGenerateNewTimeColumn() + && node.getChild() instanceof SortNode sortNode) { + PlanNode source = removeRedundantSort(sortNode.getChild()); + if (source == sortNode.getChild()) { + return node; + } + return node.replaceChildren( + ImmutableList.of(sortNode.replaceChildren(ImmutableList.of(source)))); + } + return node; + } + + private PlanNode removeRedundantSort(PlanNode node) { + if (node instanceof SortNode sortNode) { + return sortNode.getChild(); + } + // Stop at LIMIT, OFFSET and order-dependent computations; only identity projections are safe. + if (node instanceof ProjectNode projectNode && projectNode.isIdentity()) { + PlanNode child = removeRedundantSort(projectNode.getChild()); + return child == projectNode.getChild() + ? node + : projectNode.replaceChildren(ImmutableList.of(child)); + } + return node; + } + } +} 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 1db1846e2610a..4d972991c39a1 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 @@ -44,6 +44,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.junit.Test; +import org.mockito.Mockito; import java.util.ArrayList; import java.util.Arrays; @@ -54,6 +55,7 @@ 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; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; public class OrderingPropertyPropagationTest { @@ -446,6 +448,35 @@ public void testMismatchedWindowPartitionAndDiffKeepFullSort() { assertTrue(findNodes(diff, StreamSortNode.class).isEmpty()); } + @Test + public void testPlainQueryPreservesOrderBy() { + PlanTester tester = new PlanTester(); + PlanNode logical = + tester + .createPlan("SELECT time, tag1, tag2, tag3, s1 FROM table1 ORDER BY time DESC") + .getRootNode(); + List sorts = findNodes(logical, SortNode.class); + assertEquals(1, sorts.size()); + Symbol time = sorts.get(0).getOrderingScheme().getOrderBy().get(0); + assertEquals("time", time.getName()); + assertEquals(DESC_NULLS_LAST, sorts.get(0).getOrderingScheme().getOrdering(time)); + + SortNode queryRoot = Mockito.spy(sorts.get(0)); + PlanOptimizer.Context context = Mockito.mock(PlanOptimizer.Context.class); + Mockito.when(context.getAnalysis()).thenReturn(tester.getAnalysis()); + assertSame(queryRoot, new RemoveRedundantSortForCopyTo().optimize(queryRoot, context)); + Mockito.verify(queryRoot, Mockito.never()).getChild(); + Mockito.verify(queryRoot, Mockito.never()).getChildren(); + assertSame(logical, new RemoveRedundantSortForCopyTo().optimize(logical, context)); + + List merges = findNodes(tester, MergeSortNode.class); + assertFalse(merges.isEmpty()); + for (MergeSortNode merge : merges) { + assertEquals(List.of(time), merge.getOrderingScheme().getOrderBy()); + assertEquals(DESC_NULLS_LAST, merge.getOrderingScheme().getOrdering(time)); + } + } + @Test public void testCopyToWithoutFromKeepsRegularSort() { for (String prefix : List.of("", "EXPLAIN ", "EXPLAIN ANALYZE ")) { @@ -480,7 +511,9 @@ public void testCopyToSortUsesStreamingScanAndDistributedMerge() { List.of( "table1", "table1(time, tag1, tag2, tag3, s1)", - "(SELECT time, tag1, tag2, tag3, s1 FROM table1)")) { + "(SELECT time, tag1, tag2, tag3, s1 FROM table1)", + "(SELECT time, tag1, tag2, tag3, s1 FROM table1 ORDER BY time DESC)", + "(SELECT time, tag1, tag2, tag3, s1 FROM table1 ORDER BY s1 DESC)")) { PlanTester tester = new PlanTester(); PlanNode logical = tester.createPlan("COPY " + source + " TO 'copy.tsfile'").getRootNode(); assertTrue(tester.getAnalysis().isQuery()); @@ -506,6 +539,28 @@ public void testCopyToSortUsesStreamingScanAndDistributedMerge() { } } + @Test + public void testExplainCopyToRemovesRedundantSort() { + for (String prefix : List.of("EXPLAIN ", "EXPLAIN ANALYZE ")) { + PlanTester tester = new PlanTester(); + PlanNode logical = + tester + .createPlan( + prefix + + "COPY (SELECT time, tag1, tag2, tag3, s1 FROM table1 ORDER BY time DESC)" + + " TO 'copy_explain.tsfile'") + .getRootNode(); + CopyToNode copyTo = onlyCopyTo(findNodes(logical, CopyToNode.class)); + List 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"); + onlyCopyTo(findNodes(tester, CopyToNode.class)); + assertTrue(findNodes(tester, SortNode.class).isEmpty()); + } + } + @Test public void testCopyToSortPreservesInnerTopKAndOffset() { for (String orderColumn : List.of("time", "s1")) { @@ -549,6 +604,34 @@ public void testCopyToSortPreservesInnerTopKAndOffset() { } } + @Test + public void testCopyToSortPreservesOffsetWithoutLimit() { + PlanTester tester = new PlanTester(); + PlanNode logical = + tester + .createPlan( + "COPY (SELECT time, s1 FROM table1 ORDER BY time DESC OFFSET 2)" + + " TO 'copy_offset.tsfile'") + .getRootNode(); + CopyToNode copyTo = onlyCopyTo(findNodes(logical, CopyToNode.class)); + List sorts = findNodes(copyTo.getChild(), SortNode.class); + assertEquals(2, sorts.size()); + assertCopyToOrdering(copyTo, sorts.get(0).getOrderingScheme(), "time"); + List offsets = findNodes(sorts.get(0), OffsetNode.class); + assertEquals(1, offsets.size()); + assertEquals(2, offsets.get(0).getCount()); + List innerSorts = findNodes(offsets.get(0), SortNode.class); + assertEquals(1, innerSorts.size()); + Symbol time = copyTo.getInnerQueryOutputSymbols().get(0); + assertEquals(DESC_NULLS_LAST, innerSorts.get(0).getOrderingScheme().getOrdering(time)); + + onlyCopyTo(findNodes(tester, CopyToNode.class)); + assertEquals(1, sortsAbove(tester, OffsetNode.class).size()); + offsets = findNodes(tester, OffsetNode.class); + assertEquals(1, offsets.size()); + assertEquals(2, offsets.get(0).getCount()); + } + @Test public void testCopyToSortResolvesAliasesAndTargetTagOrder() { PlanTester tester = new PlanTester(); @@ -573,6 +656,30 @@ public void testCopyToSortResolvesAliasesAndTargetTagOrder() { assertTrue(findNodes(tester, SortNode.class).isEmpty()); } + @Test + public void testCopyToGeneratedTimePreservesInnerOrderBy() { + PlanTester tester = new PlanTester(); + PlanNode logical = + tester + .createPlan("COPY (SELECT s1 FROM table1 ORDER BY s1 DESC) TO 'copy_generated.tsfile'") + .getRootNode(); + CopyToNode copyTo = onlyCopyTo(findNodes(logical, CopyToNode.class)); + assertTrue(((CopyToTsFileOptions) copyTo.getCopyToOptions()).isGenerateNewTimeColumn()); + List sorts = findNodes(copyTo.getChild(), SortNode.class); + assertEquals(1, sorts.size()); + Symbol value = copyTo.getInnerQueryOutputSymbols().get(0); + assertEquals(List.of(value), sorts.get(0).getOrderingScheme().getOrderBy()); + assertEquals(DESC_NULLS_LAST, sorts.get(0).getOrderingScheme().getOrdering(value)); + + onlyCopyTo(findNodes(tester, CopyToNode.class)); + sorts = findNodes(tester, SortNode.class); + assertFalse(sorts.isEmpty()); + for (SortNode sort : sorts) { + assertEquals(List.of(value), sort.getOrderingScheme().getOrderBy()); + assertEquals(DESC_NULLS_LAST, sort.getOrderingScheme().getOrdering(value)); + } + } + @Test public void testCopyToGeneratedTimeNeedsNoSort() { PlanTester tester = new PlanTester();