[Fix] Support unordered query results in COPY TO TsFile

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', <query>"));
+                  "700: line 1:9: mismatched input 'INSERT'. Expecting: '(', 'ANALYZE', 'COPY', 'EXECUTE', <query>"));
     }
 
     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', <query>"));
+                  "700: line 1:17: mismatched input 'INSERT'. Expecting: '(', 'COPY', 'EXECUTE', 'VERBOSE', <query>"));
     }
   }
 

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;

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<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 {

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";

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";

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();

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<ColumnHeader> 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);

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) {