This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new e7fd2894912 [Fix] Support unordered query results in COPY TO TsFile
(#18928)
e7fd2894912 is described below
commit e7fd289491233a66ab5895c4866f92f7842f8801
Author: shuwenwei <[email protected]>
AuthorDate: Tue Oct 20 19:30:55 2026 +0800
[Fix] Support unordered query results in COPY TO TsFile (#18928)
---
.../it/insertquery/IoTDBInsertQueryIT.java | 4 +-
.../it/query/recent/IoTDBComplexQueryIT.java | 9 +
.../query/recent/copyto/IoTDBCopyToTsFileIT.java | 185 ++++++++++++++
.../iotdb/db/i18n/DataNodeQueryMessages.java | 11 +
.../iotdb/db/i18n/DataNodeQueryMessages.java | 11 +
.../process/copyto/TableCopyToOperator.java | 7 +-
.../process/copyto/tsfile/CopyToTsFileOptions.java | 7 +
.../planner/plan/node/PlanGraphJsonPrinter.java | 11 +
.../plan/planner/plan/node/PlanGraphPrinter.java | 36 +++
.../relational/analyzer/StatementAnalyzer.java | 8 +-
.../relational/planner/TableLogicalPlanner.java | 59 ++++-
.../TableModelTypeProviderExtractor.java | 3 +-
.../plan/relational/planner/node/CopyToNode.java | 13 +-
.../optimizations/LogicalOptimizeFactory.java | 2 +
.../RemoveRedundantSortForCopyTo.java | 88 +++++++
.../optimizations/TransformSortToStreamSort.java | 4 +
.../optimizations/UnaliasSymbolReferences.java | 3 +-
.../plan/relational/sql/parser/AstBuilder.java | 4 +
.../OrderingPropertyPropagationTest.java | 281 +++++++++++++++++++++
.../db/relational/grammar/sql/RelationalSql.g4 | 4 +-
20 files changed, 728 insertions(+), 22 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 75a7725c414..d6aa44a18f1 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 class IoTDBInsertQueryIT {
e.getMessage(),
e.getMessage()
.contains(
- "700: line 1:9: mismatched input 'INSERT'. Expecting: '(',
'ANALYZE', 'EXECUTE', <query>"));
+ "700: line 1:9: mismatched input 'INSERT'. Expecting: '(',
'ANALYZE', 'COPY', 'EXECUTE', <query>"));
}
try {
@@ -473,7 +473,7 @@ public class IoTDBInsertQueryIT {
e.getMessage(),
e.getMessage()
.contains(
- "700: line 1:17: mismatched input 'INSERT'. Expecting: '(',
'EXECUTE', 'VERBOSE', <query>"));
+ "700: line 1:17: mismatched input 'INSERT'. Expecting: '(',
'COPY', 'EXECUTE', 'VERBOSE', <query>"));
}
}
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 55b42142249..e3ca746ecb7 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 class IoTDBComplexQueryIT {
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/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 388d833f700..f0093a16bbc 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.IDeviceID;
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 org.slf4j.LoggerFactory;
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);
}
@@ -97,6 +105,66 @@ public class IoTDBCopyToTsFileIT {
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 {
@@ -743,6 +811,123 @@ public class IoTDBCopyToTsFileIT {
}
}
+ @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<String> 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<String> columns = new ArrayList<>(tagColumns);
+ columns.add("s1");
+ columns.add("s2");
+ List<String> actualRows = new ArrayList<>();
+ Map<List<String>, 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<String> 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<String> expected = new ArrayList<>(Arrays.asList(expectedRows));
+ Collections.sort(expected);
+ Collections.sort(actualRows);
+ Assert.assertEquals(expected, actualRows);
+ }
+
@Test
public void testFindTimeColumn()
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 d6fcfc64eb1..db0fa44dbc5 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
@@ -297,6 +297,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 =
@@ -4054,6 +4057,14 @@ public final class DataNodeQueryMessages {
public static final String
EXCEPTION_LOAD_CONSENSUS_INVALID_PIECE_REF_F3498507 =
"Invalid LOAD consensus piece ref: path %s, offset %d, size %d";
+ 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";
+
public static final String
EXCEPTION_CHUNKDATA_CONTAINS_NO_PHYSICAL_CHUNK_607BCFC8 =
"ChunkData contains no physical chunk";
public static final String EXCEPTION_DEVICE_CANNOT_BE_NULL_F1EB20B6 =
"device cannot be null";
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 e06c0ded22c..f500838c1ca 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
@@ -285,6 +285,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 =
@@ -4810,6 +4813,14 @@ public final class DataNodeQueryMessages {
public static final String
EXCEPTION_LOAD_CONSENSUS_INVALID_PIECE_REF_F3498507 =
"无效的 LOAD 共识分片引用：路径 %s，偏移 %d，大小 %d";
+ 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";
+
public static final String
EXCEPTION_CHUNKDATA_CONTAINS_NO_PHYSICAL_CHUNK_607BCFC8 =
"ChunkData 中不包含任何物理 chunk";
public static final String EXCEPTION_DEVICE_CANNOT_BE_NULL_F1EB20B6 =
"device 不能为 null";
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 f43fbd2b730..14ead195e8d 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 class TableCopyToOperator implements ProcessOperator {
@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/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 22b2e25a6df..ffdba2995b1 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 class CopyToTsFileOptions implements CopyToOptions {
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);
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 ecd9d2a58a5..a5f6e061453 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.OutputN
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 @@ public class PlanGraphJsonPrinter {
} 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 3316299ea17..557bd714d92 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.ValueFi
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.CrossSeriesAg
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 class PlanGraphPrinter implements
PlanVisitor<List<String>, PlanGraphPrin
return render(node, boxValue, context);
}
+ @Override
+ public List<String> visitCopyTo(CopyToNode node, GraphContext context) {
+ List<String> 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<String> visitOutput(OutputNode node, GraphContext context) {
List<String> 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 e1c9db8d0f2..5214b30be99 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 class StatementAnalyzer {
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 class StatementAnalyzer {
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 8f9928b7ad2..c68ebb7e254 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;
@@ -247,7 +254,7 @@ public class TableLogicalPlanner {
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));
@@ -346,14 +353,14 @@ public class TableLogicalPlanner {
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 {
@@ -621,8 +628,7 @@ public class TableLogicalPlanner {
}
private RelationPlan planExplainAnalyze(final ExplainAnalyze statement,
final Analysis analysis) {
- RelationPlan originalQueryPlan =
- createRelationPlan(analysis, (Query) (statement.getStatement()));
+ RelationPlan originalQueryPlan = planStatementWithoutOutput(analysis,
statement.getStatement());
Symbol symbol =
symbolAllocator.newSymbol(ColumnHeaderConstant.EXPLAIN_ANALYZE,
StringType.getInstance());
@@ -635,7 +641,9 @@ public class TableLogicalPlanner {
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,
@@ -656,17 +664,50 @@ 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);
+ }
+
+ List<Symbol> 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(),
- innerQueryRelationPlan.getRoot(),
+ copyToSource,
statement.getTargetFileName(),
statement.getOptions(),
// recording permittedOutputs of CopyToNode's child
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 aaf3725ec2a..356efb97406 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 @@ public class TableModelTypeProviderExtractor {
@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 2e251bd7a19..ff6899444c4 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<Symbol> childPermittedOutputs;
private final List<Symbol> innerQueryOutputSymbols;
private final DatasetHeader innerQueryDatasetHeader;
+ private final List<Symbol> outputSymbols;
public CopyToNode(
PlanNodeId id,
@@ -49,7 +50,8 @@ public class CopyToNode extends SingleChildProcessNode {
CopyToOptions copyToOptions,
List<Symbol> childPermittedOutputs,
DatasetHeader innerQueryDatasetHeader,
- List<Symbol> innerQueryOutputSymbols) {
+ List<Symbol> innerQueryOutputSymbols,
+ List<Symbol> outputSymbols) {
super(id);
this.child = child;
this.targetFilePath = targetFilePath;
@@ -57,6 +59,7 @@ public class CopyToNode extends SingleChildProcessNode {
this.childPermittedOutputs = childPermittedOutputs;
this.innerQueryDatasetHeader = innerQueryDatasetHeader;
this.innerQueryOutputSymbols = innerQueryOutputSymbols;
+ this.outputSymbols = outputSymbols;
}
public DatasetHeader getInnerQueryDatasetHeader() {
@@ -88,7 +91,8 @@ public class CopyToNode extends SingleChildProcessNode {
copyToOptions,
childPermittedOutputs,
innerQueryDatasetHeader,
- innerQueryOutputSymbols);
+ innerQueryOutputSymbols,
+ outputSymbols);
}
@Override
@@ -98,7 +102,7 @@ public class CopyToNode extends SingleChildProcessNode {
@Override
public List<Symbol> getOutputSymbols() {
- return copyToOptions.getOutputSymbols();
+ return outputSymbols;
}
@Override
@@ -115,7 +119,8 @@ public class CopyToNode extends SingleChildProcessNode {
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/LogicalOptimizeFactory.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java
index 78bb5f984a1..ddf09e7f98d 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 class LogicalOptimizeFactory {
ImmutableList.Builder<PlanOptimizer> 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 00000000000..d4295cce17e
--- /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<PlanNode, Void> {
+
+ @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/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 a89b2d31694..90c2c91931d 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 class TransformSortToStreamSort implements
PlanOptimizer {
context.setCanTransform(false);
DeviceTableScanNode deviceTableScanNode = context.getTableScanNode();
+ if (deviceTableScanNode == null) {
+ node.setChild(child);
+ return node;
+ }
Map<Symbol, ColumnSchema> tableColumnSchema;
if (deviceTableScanNode instanceof ExternalTsFileScanNode) {
tableColumnSchema =
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 131ba8af383..89e377bcffe 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 class UnaliasSymbolReferences implements
PlanOptimizer {
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 1d2417ea644..71686b52cf7 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
@@ -1874,6 +1874,8 @@ public class AstBuilder extends
RelationalSqlBaseVisitor<Node> {
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 {
@@ -1905,6 +1907,8 @@ public class AstBuilder extends
RelationalSqlBaseVisitor<Node> {
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/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..4d972991c39 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,31 +20,42 @@
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;
+import org.mockito.Mockito;
import java.util.ArrayList;
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;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
public class OrderingPropertyPropagationTest {
@@ -437,6 +448,276 @@ public class OrderingPropertyPropagationTest {
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<SortNode> 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<MergeSortNode> 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 ")) {
+ 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<SortNode> 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 :
+ 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 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());
+ 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 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<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");
+ onlyCopyTo(findNodes(tester, CopyToNode.class));
+ assertTrue(findNodes(tester, SortNode.class).isEmpty());
+ }
+ }
+
+ @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 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<SortNode> sorts = findNodes(copyTo.getChild(), SortNode.class);
+ assertEquals(2, 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<SortNode> 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();
+ 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 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<SortNode> 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();
+ 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);
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 f4d80cfc244..706b464f9ff 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
@@ -968,8 +968,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
