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/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/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..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 @@ -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); } @@ -97,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 { @@ -743,6 +811,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 { 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 c89034e8831df..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 = @@ -4037,6 +4040,14 @@ private 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 a3307bd07cb42..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 = @@ -4793,6 +4796,14 @@ private 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 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/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); 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 e1c9db8d0f2bf..5214b30be992b 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 8f9928b7ad2d9..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 @@ -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; @@ -247,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)); @@ -346,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 { @@ -621,8 +628,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())); + RelationPlan originalQueryPlan = planStatementWithoutOutput(analysis, statement.getStatement()); Symbol symbol = symbolAllocator.newSymbol(ColumnHeaderConstant.EXPLAIN_ANALYZE, StringType.getInstance()); @@ -635,7 +641,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, @@ -656,17 +664,50 @@ 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); + } + + 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(), - 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 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/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/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/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 1d2417ea6442f..71686b52cf7ac 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 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 { @@ -1905,6 +1907,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/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..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 @@ -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 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 ")) { + 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 : + 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 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 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")) { + 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 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(); + 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 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(); + 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); 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 f4d80cfc24440..706b464f9ffc1 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