You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by cr...@apache.org on 2026/10/22 07:29:46 UTC
(iotdb) 01/01: Add measurement files, time ranges, progress and logs to export-data
This is an automated email from the ASF dual-hosted git repository.
CritasWang pushed a commit to branch wx/plm-doc-861
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit edf4484904e034ce2d70a10fcfbbd318897c0adf
Author: CritasWang <cr...@outlook.com>
AuthorDate: Thu Oct 22 15:29:17 2026 +0800
Add measurement files, time ranges, progress and logs to export-data
- -mf/--measurement_file exports the measurements matched by the IoTDB path
patterns in a file, one output per measurement named by its full path.
The whole file is validated before any query, and each unmatched pattern
is reported with its line.
- -start_time/-end_time form an inclusive interval for measurement-file and
automatic Table exports. Queries run as written and report the ignored
bounds as a WARNING.
- Progress is reported every two seconds: one redrawn line in a terminal,
one line per update otherwise, with an ETA in measurement-file mode.
- Every run appends a JSON Lines log, <target>/log_export_data_<time>.log
by default or the path given with -l/--export_data_log, ending with one
FINISH event. The console ends with the summary, after the log path.
- CSV and SQL files are split by data rows; the INSERT statements of one
row stay in one file.
- Ctrl+C and SIGTERM stop at the next row, close the query and the session
and exit with 130 or 143; failures exit with 1.
- A Table query exported as SQL or TsFile still writes to the table it
reads, as in #18828; -table is ignored with a WARNING, and every query is
checked before the first one runs.
---
.../iotdb/tools/it/ExportDataMeasurementIT.java | 505 ++++++++++++++
.../en/org/apache/iotdb/cli/i18n/CliMessages.java | 141 ++++
.../zh/org/apache/iotdb/cli/i18n/CliMessages.java | 141 ++++
.../org/apache/iotdb/tool/common/Constants.java | 5 +
.../org/apache/iotdb/tool/common/OptionsUtil.java | 98 ++-
.../apache/iotdb/tool/data/AbstractExportData.java | 187 ++++--
.../org/apache/iotdb/tool/data/ExportConfig.java | 363 +++++++++++
.../org/apache/iotdb/tool/data/ExportConsole.java | 137 ++++
.../org/apache/iotdb/tool/data/ExportData.java | 526 +++++++--------
.../apache/iotdb/tool/data/ExportDataTable.java | 635 +++++++++---------
.../org/apache/iotdb/tool/data/ExportDataTree.java | 709 ++++++++++----------
.../apache/iotdb/tool/data/ExportFileNames.java | 114 ++++
.../iotdb/tool/data/ExportProgressReporter.java | 638 ++++++++++++++++++
.../org/apache/iotdb/tool/data/ExportTask.java | 31 +
.../apache/iotdb/tool/data/ExportTextWriter.java | 126 ++++
.../apache/iotdb/tool/data/ExportTimeRange.java | 84 +++
.../apache/iotdb/tool/data/MeasurementFile.java | 145 ++++
.../iotdb/tool/data/DataToolRegressionTest.java | 293 ++++++---
.../iotdb/tool/data/ExportDataFeatureTest.java | 726 +++++++++++++++++++++
.../iotdb/tool/data/ExportDataRegressionTest.java | 300 +++++++++
scripts/tools/README.md | 130 ++++
scripts/tools/export-data.sh | 18 +-
scripts/tools/windows/export-data.bat | 6 +-
23 files changed, 4901 insertions(+), 1157 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/tools/it/ExportDataMeasurementIT.java b/integration-test/src/test/java/org/apache/iotdb/tools/it/ExportDataMeasurementIT.java
new file mode 100644
index 00000000000..37b869901a5
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/tools/it/ExportDataMeasurementIT.java
@@ -0,0 +1,505 @@
+/*
+ * 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.tools.it;
+
+import org.apache.iotdb.isession.ISession;
+import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionConfig;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.AfterClass;
+import org.junit.Assume;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/** Runs the actual export command and verifies files, exit status and round trips. */
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class ExportDataMeasurementIT {
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().initClusterEnvironment();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @Test
+ public void csvPatternsSplitsPrefixTimeAndEmptyMeasurements() throws Exception {
+ try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+ session.executeNonQueryStatement(
+ "CREATE TIMESERIES root.export_csv.d.pre1 WITH DATATYPE=INT32");
+ session.executeNonQueryStatement(
+ "CREATE TIMESERIES root.export_csv.d.pre2 WITH DATATYPE=TEXT");
+ session.executeNonQueryStatement(
+ "CREATE TIMESERIES root.export_csv.d.empty WITH DATATYPE=INT32");
+ for (int t = 100; t <= 300; t += 100) {
+ session.executeNonQueryStatement(
+ "INSERT INTO root.export_csv.d(time,pre1,pre2) VALUES(" + t + "," + t + ",'温度,计')");
+ }
+ }
+ Path input =
+ file(
+ "\uFEFF# patterns\nroot.export_csv.d.pre*\nroot.export_csv.*.*\nroot.export_csv.**\nroot.missing.**\nroot.export_csv.d.pre*\n");
+ Path target = folder.newFolder("csv output").toPath();
+ Result result =
+ export(
+ target,
+ "-ft",
+ "csv",
+ "-mf",
+ input.toString(),
+ "-pfn",
+ "backup",
+ "-lpf",
+ "1",
+ "-tf",
+ "timestamp",
+ "-start_time",
+ "100",
+ "-end_time",
+ "200");
+ assertSuccess(result);
+ assertEquals(
+ "Time,root.export_csv.d.pre1\r\n100,100\r\n",
+ Files.readString(target.resolve("backup_root.export_csv.d.pre1_0.csv")));
+ assertEquals(
+ "Time,root.export_csv.d.pre1\r\n200,200\r\n",
+ Files.readString(target.resolve("backup_root.export_csv.d.pre1_1.csv")));
+ assertEquals(4, outputs(target, ".csv").size());
+ String log = log(target);
+ assertTrue(log, log.contains(member("total_measurement_num", 3)));
+ assertTrue(log, log.contains(member("exported_measurement_num", 3)));
+ assertTrue(log, log.contains(member("zero_row_tasks", 1)));
+ assertTrue(log, log.contains(member("exported_percent", "100.00%")));
+ assertTrue(log, log.contains(member("event", "WARNING")));
+ assertTrue(log, log.contains(member("line", 5)));
+ assertTrue(log, log.contains(member("exported_data_rows", 4)));
+ // The console is for people: the warning and the summary, never the JSON events.
+ assertTrue(
+ result.text,
+ result
+ .text
+ .lines()
+ .anyMatch(
+ "WARNING: No measurements matched root.missing.** (measurement file line 5)."
+ ::equals));
+ assertTrue(
+ result.text,
+ result.text.contains("Export completely! Measurements: 3/3, rows: 4, files: 4, elapsed: "));
+ assertFalse(result.text, result.text.contains("{\""));
+ }
+
+ @Test
+ public void sqlAndTsFileRoundTripAlignedTypesAndQuotedPaths() throws Exception {
+ for (String format : Arrays.asList("sql", "tsfile")) {
+ String database = "root.export_" + format;
+ String d = database + ".d";
+ try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+ session.executeNonQueryStatement(
+ "CREATE ALIGNED TIMESERIES "
+ + d
+ + "(s INT32,txt TEXT,stamp TIMESTAMP,day DATE,blob BLOB,`a.b` STRING)");
+ session.executeNonQueryStatement(
+ "CREATE TIMESERIES " + database + ".plain.n WITH DATATYPE=DOUBLE");
+ session.executeNonQueryStatement(
+ "INSERT INTO "
+ + d
+ + "(time,s,txt,stamp,day,blob,`a.b`) ALIGNED VALUES(100,10,'温度',123,'2026-10-14',X'CAFE','quoted'),(200,20,'温度2',456,'2026-10-15',X'BEEF','quoted2'),(300,30,'外部',789,'2026-10-16',X'ABCD','outside')");
+ session.executeNonQueryStatement(
+ "INSERT INTO " + database + ".plain(time,n) VALUES(100,1.5),(200,2.5),(300,3.5)");
+ }
+ Path target = folder.newFolder(format).toPath();
+ Path input = file(database + ".**\n");
+ assertSuccess(
+ export(
+ target,
+ "-ft",
+ format,
+ "-mf",
+ input.toString(),
+ "-start_time",
+ "100",
+ "-end_time",
+ "200"));
+ List<Path> files = outputs(target, "." + format);
+ assertEquals(7, files.size());
+ try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+ session.executeNonQueryStatement(
+ format.equals("sql")
+ ? "DELETE FROM " + database + ".** WHERE time >= 0"
+ : "DELETE DATABASE " + database);
+ for (Path path : files) {
+ if (format.equals("sql")) {
+ for (String sql : Files.readAllLines(path)) {
+ session.executeNonQueryStatement(sql);
+ }
+ } else {
+ session.executeNonQueryStatement(
+ "LOAD '" + path.toAbsolutePath() + "' WITH ('verify'='true')");
+ }
+ }
+ try (SessionDataSet data =
+ session.executeQueryStatement("SELECT s,txt,stamp,day,blob,`a.b` FROM " + d)) {
+ SessionDataSet.DataIterator it = data.iterator();
+ assertTrue(it.next());
+ assertEquals(100, it.getLong(1));
+ assertEquals(10, it.getInt(2));
+ assertEquals("温度", it.getString(3));
+ assertEquals(123, it.getLong(4));
+ assertEquals("2026-10-14", it.getString(5));
+ assertEquals("0xcafe", it.getString(6).toLowerCase());
+ assertEquals("quoted", it.getString(7));
+ assertTrue(it.next());
+ assertEquals(200, it.getLong(1));
+ assertEquals(20, it.getInt(2));
+ assertFalse(it.next());
+ }
+ try (SessionDataSet data =
+ session.executeQueryStatement("SELECT n FROM " + database + ".plain")) {
+ assertTrue(data.hasNext());
+ assertEquals(1.5, data.next().getField(0).getDoubleV(), 0.0);
+ assertTrue(data.hasNext());
+ assertEquals(2.5, data.next().getField(0).getDoubleV(), 0.0);
+ assertFalse(data.hasNext());
+ }
+ }
+ }
+ }
+
+ @Test
+ public void originalQueryAndTableModesAndFailures() throws Exception {
+ try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+ session.executeNonQueryStatement(
+ "INSERT INTO root.export_query.d(time,s) VALUES(100,10),(200,20)");
+ }
+ Path query = folder.newFolder("query").toPath();
+ assertSuccess(
+ export(
+ query,
+ "-ft",
+ "csv",
+ "-q",
+ "SELECT s FROM root.export_query.d",
+ "-tf",
+ "timestamp",
+ "-start_time",
+ "200"));
+ assertTrue(Files.readString(query.resolve("dump0_0.csv")).contains("100,10"));
+ String queryLog = log(query);
+ assertFalse(queryLog.contains("exported_measurement_num"));
+ // The query runs as written, so the ignored time range is a warning rather than a START field.
+ assertTrue(queryLog, queryLog.contains(member("event", "WARNING")));
+ assertTrue(
+ queryLog,
+ queryLog
+ .lines()
+ .filter(line -> line.contains(member("event", "START")))
+ .noneMatch(line -> line.contains("\"start_time\":")));
+ Path failure = folder.newFolder("failed").toPath();
+ Result badQuery = export(failure, "-ft", "csv", "-q", "invalid sql");
+ assertNotEquals(0, badQuery.code);
+ assertTrue(badQuery.text, badQuery.text.contains("Export failed after "));
+ assertFalse(badQuery.text, badQuery.text.contains("Export completely!"));
+ assertTrue(log(failure).contains(member("status", "FAILED")));
+ assertFalse(log(failure).contains(member("status", "SUCCESS")));
+ Path empty = folder.newFolder("unmatched").toPath();
+ assertNotEquals(
+ 0, export(empty, "-ft", "csv", "-mf", file("root.no_measurements.**\n").toString()).code);
+ assertNotEquals(
+ 0, export(folder.newFolder("alias").toPath(), "-ft", "csv", "-f", "backup").code);
+ assertNotEquals(
+ 0,
+ export(
+ folder.newFolder("badfile").toPath(),
+ "-ft",
+ "csv",
+ "-mf",
+ file("root.export_query.d.s\ninvalid..path\n").toString())
+ .code);
+ try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) {
+ session.executeNonQueryStatement("CREATE DATABASE export_table");
+ session.executeNonQueryStatement("USE export_table");
+ session.executeNonQueryStatement(
+ "CREATE TABLE metrics(id STRING TAG, v INT32 FIELD, stamp TIMESTAMP FIELD, day DATE FIELD, blob BLOB FIELD)");
+ session.executeNonQueryStatement(
+ "INSERT INTO metrics(time,id,v,stamp,day,blob) VALUES(100,'d1',10,123,'2026-10-14',X'CAFE'),(200,'d1',20,456,'2026-10-15',X'BEEF'),(300,'d1',30,789,'2026-10-16',X'ABCD')");
+ }
+ for (String format : Arrays.asList("csv", "sql", "tsfile")) {
+ Path target = folder.newFolder("table_" + format).toPath();
+ assertSuccess(
+ export(
+ target,
+ "-sql_dialect",
+ "table",
+ "-db",
+ "export_table",
+ "-ft",
+ format,
+ "-start_time",
+ "100",
+ "-end_time",
+ "200"));
+ assertTrue(log(target).contains(member("exported_data_rows", 2)));
+ assertFalse(log(target).contains("exported_measurement_num"));
+ assertEquals(1, outputs(target, "." + format).size());
+ // SQL and TsFile output take the table from the query.
+ Path custom = folder.newFolder("table_query_" + format).toPath();
+ assertSuccess(
+ export(
+ custom,
+ "-sql_dialect",
+ "table",
+ "-db",
+ "export_table",
+ "-ft",
+ format,
+ "-q",
+ "SELECT * FROM metrics WHERE time=100"));
+ assertTrue(log(custom).contains(member("exported_data_rows", 1)));
+ }
+ assertNotEquals(
+ 0,
+ export(
+ folder.newFolder("table_mf").toPath(),
+ "-sql_dialect",
+ "table",
+ "-db",
+ "export_table",
+ "-ft",
+ "csv",
+ "-mf",
+ file("root.**\n").toString())
+ .code);
+ }
+
+ @Test
+ public void cancellationHasOneTerminalEvent() throws Exception {
+ org.junit.Assume.assumeFalse(System.getProperty("os.name").toLowerCase().startsWith("windows"));
+ StringBuilder patterns = new StringBuilder();
+ for (int i = 0; i < 10000; i++) {
+ patterns.append("root.cancel.d").append(i).append(".s\n");
+ }
+ Path input = file(patterns.toString());
+ Path target = folder.newFolder("cancelled").toPath();
+ Path console = folder.newFile().toPath();
+ List<String> command = command(target, "-ft", "csv", "-mf", input.toString());
+ Process process =
+ new ProcessBuilder(command)
+ .redirectErrorStream(true)
+ .redirectOutput(console.toFile())
+ .start();
+ try {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(15);
+ while (!Files.readString(console).contains("Export started")
+ && System.nanoTime() < deadline
+ && process.isAlive()) {
+ Thread.sleep(25);
+ }
+ assertTrue(Files.readString(console), process.isAlive());
+ process.destroy();
+ assertTrue(process.waitFor(30, TimeUnit.SECONDS));
+ String output = Files.readString(console);
+ assertEquals(output, 1, output.split("Export cancelled after ", -1).length - 1);
+ assertFalse(output, output.contains("Export completely!"));
+ String log = log(target);
+ assertEquals(log, 1, log.split(member("event", "FINISH"), -1).length - 1);
+ assertTrue(log, log.contains(member("status", "CANCELLED")));
+ // destroy() sends SIGTERM; the process exits with its status, which FINISH reports.
+ assertEquals(output, 143, process.exitValue());
+ assertTrue(log, log.contains(member("exit_code", 143)));
+ // The worker closes its own session, so no transport is closed under a request.
+ assertFalse(output, output.contains("SocketException"));
+ } finally {
+ if (process.isAlive()) {
+ process.destroyForcibly();
+ }
+ }
+ }
+
+ @Test
+ public void benchmarkTwentyThousandMeasurements() throws Exception {
+ Assume.assumeTrue(Boolean.getBoolean("export.measurements.benchmark"));
+ List<IMeasurementSchema> schemas = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ schemas.add(new MeasurementSchema("s" + i, TSDataType.INT32));
+ }
+ StringBuilder patterns = new StringBuilder();
+ try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+ for (int d = 0; d < 200; d++) {
+ String device = "root.export_bench.d" + d;
+ Tablet tablet = new Tablet(device, schemas, 1440);
+ for (int row = 0; row < 1440; row++) {
+ tablet.addTimestamp(row, row * 60000L);
+ for (int s = 0; s < 100; s++) {
+ tablet.addValue("s" + s, row, row);
+ }
+ }
+ session.insertTablet(tablet);
+ for (int s = 0; s < 100; s++) {
+ patterns.append(device).append(".s").append(s).append('\n');
+ }
+ }
+ }
+ Path input = file(patterns.toString());
+ Path target = folder.newFolder("benchmark").toPath();
+ long started = System.nanoTime();
+ assertSuccess(export(target, "-ft", "csv", "-mf", input.toString(), "-tf", "timestamp"));
+ long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started);
+ List<Path> outputs = outputs(target, ".csv");
+ assertEquals(20000, outputs.size());
+ long rows = 0;
+ for (Path output : outputs) {
+ try (Stream<String> lines = Files.lines(output)) {
+ rows += lines.count() - 1;
+ }
+ }
+ assertEquals(28800000, rows);
+ String log = log(target);
+ assertTrue(log.contains(member("exported_measurement_num", 20000)));
+ assertTrue(log.contains(member("exported_data_rows", 28800000)));
+ System.out.println(
+ "export_measurement_benchmark devices=200 measurements=20000 points="
+ + rows
+ + " elapsed_ms="
+ + elapsedMillis);
+ Files.writeString(
+ Path.of("target/export-measurement-benchmark.txt"),
+ "devices=200 measurements=20000 points=" + rows + " elapsed_ms=" + elapsedMillis + "\n");
+ }
+
+ private Result export(Path target, String... args) throws Exception {
+ List<String> command = command(target, args);
+ Path console = folder.newFile().toPath();
+ Process process =
+ new ProcessBuilder(command)
+ .redirectErrorStream(true)
+ .redirectOutput(console.toFile())
+ .start();
+ try {
+ if (!process.waitFor(60, TimeUnit.SECONDS)) {
+ // A benchmark may legitimately take longer; regular tests retain the bounded wait.
+ if (Boolean.getBoolean("export.measurements.benchmark")) {
+ while (!process.waitFor(60, TimeUnit.SECONDS)) {
+ System.out.println("Export benchmark is running.");
+ }
+ } else {
+ fail("Export command timed out: " + Files.readString(console));
+ }
+ }
+ return new Result(process.exitValue(), Files.readString(console));
+ } finally {
+ if (process.isAlive()) {
+ process.destroyForcibly();
+ }
+ }
+ }
+
+ private List<String> command(Path target, String... args) {
+ List<String> command = new ArrayList<>();
+ command.add(Path.of(System.getProperty("java.home"), "bin", "java").toString());
+ command.add("-cp");
+ command.add(EnvFactory.getEnv().getLibPath());
+ command.add("org.apache.iotdb.tool.data.ExportData");
+ command.addAll(
+ Arrays.asList(
+ "-h",
+ EnvFactory.getEnv().getIP(),
+ "-p",
+ EnvFactory.getEnv().getPort(),
+ "-u",
+ SessionConfig.DEFAULT_USER,
+ "-pw",
+ SessionConfig.DEFAULT_PASSWORD,
+ "-t",
+ target.toString()));
+ command.addAll(Arrays.asList(args));
+ return command;
+ }
+
+ private void assertSuccess(Result result) {
+ assertEquals(result.text, 0, result.code);
+ assertEquals(result.text, 1, result.text.split("Export completely!", -1).length - 1);
+ assertFalse(result.text, result.text.contains("Export failed"));
+ }
+
+ /** A member as the JSON Lines log writes it, such as "status":"FAILED" or "exit_code":130. */
+ private static String member(String key, Object value) {
+ return '"' + key + "\":" + (value instanceof Number ? value : "\"" + value + '"');
+ }
+
+ private Path file(String content) throws Exception {
+ return Files.writeString(folder.newFile().toPath(), content, StandardCharsets.UTF_8);
+ }
+
+ private List<Path> outputs(Path directory, String suffix) throws Exception {
+ try (Stream<Path> files = Files.list(directory)) {
+ return files
+ .filter(file -> file.toString().endsWith(suffix))
+ .sorted()
+ .collect(Collectors.toList());
+ }
+ }
+
+ private String log(Path directory) throws Exception {
+ return Files.readString(outputs(directory, ".log").get(0));
+ }
+
+ private static class Result {
+ final int code;
+ final String text;
+
+ Result(int code, String text) {
+ this.code = code;
+ this.text = text;
+ }
+ }
+}
diff --git a/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java b/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java
index ec00a7dc559..7b0dde1ea72 100644
--- a/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java
+++ b/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java
@@ -116,4 +116,145 @@ public final class CliMessages {
public static final String LOG_INPUT_TIME_FORMAT_ARG_NOT_SUPPORTED_00172A7B = "Input time format {} is not supported, ";
public static final String LOG_PLEASE_INPUT_LIKE_YYYY_MM_DD_HH_MM_SS_SSS_9318BFC7 = "please input like yyyy-MM-dd\\ HH:mm:ss.SSS or yyyy-MM-dd'T'HH:mm:ss.SSS%n";
+ // Export data
+ public static final String MESSAGE_EXPORT_LOG_FILE_DEFAULT_TARGET_LOG_EXPORT_DATA_YYYYMMDDHHMMSS_LOG_B29FC35F =
+ "Export log file (default: <target>/log_export_data_yyyyMMddHHmmss.log).";
+ public static final String MESSAGE_INCLUSIVE_START_TIME_FOR_MEASUREMENT_FILE_OR_AUTOMATIC_TABLE_EXPORT_69FE75CF =
+ "Inclusive start time for measurement-file or automatic Table export.";
+ public static final String MESSAGE_INCLUSIVE_END_TIME_FOR_MEASUREMENT_FILE_OR_AUTOMATIC_TABLE_EXPORT_D9C24D95 =
+ "Inclusive end time for measurement-file or automatic Table export.";
+ public static final String MESSAGE_UTF_8_IOTDB_PATH_PATTERN_FILE_FOR_TREE_EXPORT_MUTUALLY_EXCLUSIVE_WITH_Q_DD1F2CDA =
+ "UTF-8 IoTDB path-pattern file for Tree export; mutually exclusive with -q.";
+ public static final String MESSAGE_TABLE_TO_EXPORT_OPTIONAL_IGNORED_WHEN_Q_IS_GIVEN_4F37C0B5 =
+ "Table to export (optional). Ignored when -q is given.";
+ public static final String
+ MESSAGE_DATABASE_TO_EXPORT_REQUIRED_EXPORTS_ALL_OF_ITS_TABLES_UNLESS_TABLE_OR_Q_IS_GIVEN_F125EEA1 =
+ "Database to export (required). Exports all of its tables unless -table or -q is given.";
+ public static final String MESSAGE_MAXIMUM_DATA_ROWS_PER_CSV_OR_SQL_FILE_DEFAULT_10000_000CEFB9 =
+ "Maximum data rows per CSV or SQL file (default 10000).";
+ public static final String EXCEPTION_FILE_PREFIX_MUST_NOT_CONTAIN_PATH_SEPARATORS_ARG_C89823EE =
+ "File prefix must not contain path separators: %s";
+ public static final String EXCEPTION_INVALID_OUTPUT_TIME_FORMAT_ARG_9C690675 =
+ "Invalid output time format: %s";
+ public static final String EXCEPTION_EXPORTING_INFORMATION_SCHEMA_IS_NOT_SUPPORTED_6FD0FB85 =
+ "Exporting information_schema is not supported.";
+ public static final String EXCEPTION_THE_EXPORT_LOG_MUST_DIFFER_FROM_THE_MEASUREMENT_INPUT_FILE_088A542C =
+ "The export log must differ from the measurement input file.";
+ public static final String LOG_PLEASE_INPUT_YOUR_PASSWORD_AEE49DC5 =
+ "Please input your password:";
+ public static final String LOG_PLEASE_INPUT_YOUR_TRUST_STORE_4C50D824 =
+ "Please input your trust_store:";
+ public static final String LOG_PLEASE_INPUT_YOUR_TRUST_STORE_PASSWORD_86ED5CB6 =
+ "Please input your trust_store_password:";
+ public static final String LOG_PLEASE_INPUT_YOUR_KEY_STORE_AC5DB91F =
+ "Please input your key_store:";
+ public static final String LOG_PLEASE_INPUT_YOUR_KEY_STORE_PASSWORD_19ED16D4 =
+ "Please input your key_store_password:";
+ public static final String EXCEPTION_UNSUPPORTED_SQL_DIALECT_ARG_C4D0D97C =
+ "Unsupported sql_dialect: %s";
+ public static final String EXCEPTION_SPECIFY_AN_EXPORT_FORMAT_CSV_SQL_OR_TSFILE_RECEIVED_ARG_5B5211B5 =
+ "Specify an export format: csv, sql or tsfile (received: %s).";
+ public static final String EXCEPTION_MEASUREMENT_FILE_IS_SUPPORTED_ONLY_IN_THE_TREE_MODEL_1ED604B5 =
+ "-measurement_file is supported only in the Tree model.";
+ public static final String EXCEPTION_MEASUREMENT_FILE_AND_Q_ARE_MUTUALLY_EXCLUSIVE_7A7457D6 =
+ "-measurement_file and -q are mutually exclusive.";
+ public static final String EXCEPTION_UNEXPECTED_POSITIONAL_ARGUMENTS_ARG_027F44CB =
+ "Unexpected positional arguments: %s";
+ public static final String EXCEPTION_INVALID_POSITIVE_INTEGER_FOR_OPTION_ARG_ARG_30FB488D =
+ "Invalid positive integer for option %s: %s";
+ public static final String EXCEPTION_OPTION_ARG_REQUIRES_TRUE_OR_FALSE_ARG_0DC9BE3D =
+ "Option %s requires true or false: %s";
+ public static final String EXCEPTION_START_TIME_MUST_BE_LESS_THAN_OR_EQUAL_TO_END_TIME_3551199C =
+ "start_time must be less than or equal to end_time.";
+ public static final String EXCEPTION_CANNOT_PARSE_TIME_ARG_ARG_40257DF7 =
+ "Cannot parse time %s: %s";
+ public static final String EXCEPTION_MEASUREMENT_INPUT_MUST_BE_A_READABLE_REGULAR_FILE_ARG_43D4A272 =
+ "Measurement input must be a readable regular file: %s";
+ public static final String EXCEPTION_A_COMPLETE_IOTDB_PATH_PATTERN_STARTING_WITH_ROOT_IS_REQUIRED_7D000A79 =
+ "A complete IoTDB path pattern starting with root is required.";
+ public static final String EXCEPTION_INVALID_MEASUREMENT_INPUT_ARG_AT_LINE_ARG_CONTENT_ARG_ARG_27A0F948 =
+ "Invalid measurement input %s at line %d, content=%s: %s";
+ public static final String EXCEPTION_CANNOT_READ_UTF_8_MEASUREMENT_FILE_ARG_AT_LINE_ARG_ARG_A5739D3F =
+ "Cannot read UTF-8 measurement file %s at line %d: %s";
+ public static final String EXCEPTION_MEASUREMENT_INPUT_CONTAINS_NO_PATH_PATTERNS_ARG_96760070 =
+ "Measurement input contains no path patterns: %s";
+ public static final String EXCEPTION_MEASUREMENT_OUTPUT_FILENAME_EXCEEDS_THE_SUPPORTED_LENGTH_ARG_FF01275D =
+ "Measurement output filename exceeds the supported length: %s";
+ public static final String EXCEPTION_MEASUREMENTS_ARG_AND_ARG_MAP_TO_THE_SAME_OUTPUT_FILENAME_ARG_58E1FD4D =
+ "Measurements %s and %s map to the same output filename: %s";
+ public static final String EXCEPTION_EXPORT_OUTPUT_CONFLICTS_WITH_THE_LOG_FILE_ARG_F01BE79D =
+ "Export output conflicts with the log file: %s";
+ public static final String LOG_EXPORT_STARTED_45465087 =
+ "Export started.";
+ public static final String LOG_EXPORT_COMPLETELY_34200D19 =
+ "Export completely!";
+ public static final String EXCEPTION_EXPORT_CANCELLED_85D44F54 =
+ "Export cancelled.";
+ public static final String LOG_EXPORT_FAILED_77EE72CD =
+ "Export failed.";
+ public static final String LOG_CANNOT_WRITE_OR_CLOSE_THE_EXPORT_LOG_32579AC6 =
+ "Cannot write or close the export log.";
+ public static final String LOG_EXPORT_STARTED_ARG_ARG_TARGET_ARG_LOG_ARG_BD58AECB =
+ "Export started (%s, %s): target %s, log %s";
+ public static final String LOG_MEASUREMENTS_TO_EXPORT_ARG_PATH_PATTERNS_ARG_A0AC5F31 =
+ "Measurements to export: %d (path patterns: %d).";
+ public static final String LOG_EXPORTING_ARG_ARG_ARG_ARG_ROWS_ARG_ETA_ARG_C41CA790 =
+ "Exporting %s %d/%d %s rows %s ETA %s";
+ public static final String LOG_ARG_ARG_ARG_MEASUREMENTS_ARG_ROWS_ELAPSED_ARG_ETA_ARG_CURRENT_ARG_B6DEA0BD =
+ "[%s] %d/%d measurements, %s rows, elapsed %s, ETA %s, current %s";
+ public static final String LOG_EXPORTING_ARG_ROWS_ARG_ARG_ROWS_S_56BD441C =
+ "Exporting %s rows %s %s rows/s";
+ public static final String LOG_ARG_ROWS_ELAPSED_ARG_ARG_ROWS_S_AE3CF660 =
+ "%s rows, elapsed %s, %s rows/s";
+ public static final String LOG_WARNING_ARG_05BE32E9 =
+ "WARNING: %s";
+ public static final String LOG_EXPORT_COMPLETELY_MEASUREMENTS_ARG_ARG_ROWS_ARG_FILES_ARG_ELAPSED_ARG_36D9F8E5 =
+ "Export completely! Measurements: %d/%d, rows: %s, files: %s, elapsed: %s.";
+ public static final String LOG_EXPORT_COMPLETELY_ROWS_ARG_FILES_ARG_ELAPSED_ARG_A94BBB8B =
+ "Export completely! Rows: %s, files: %s, elapsed: %s.";
+ public static final String LOG_EXPORT_FAILED_AFTER_ARG_ARG_C123F0EF =
+ "Export failed after %s: %s";
+ public static final String LOG_EXPORT_FAILED_AFTER_ARG_DA6D56F9 =
+ "Export failed after %s.";
+ public static final String LOG_EXPORT_CANCELLED_AFTER_ARG_ROWS_EXPORTED_ARG_0999F99F =
+ "Export cancelled after %s. Rows exported: %s.";
+ public static final String LOG_LOG_ARG_FBF61F63 =
+ "Log: %s";
+ public static final String LOG_ARG_82AC62F7 =
+ "%.2f%%";
+ public static final String EXCEPTION_AGGREGATE_QUERIES_CANNOT_BE_EXPORTED_AS_SQL_OR_TSFILE_48BC1932 =
+ "Aggregate queries cannot be exported as SQL or TsFile.";
+ public static final String LOG_NO_MEASUREMENTS_MATCHED_ARG_MEASUREMENT_FILE_LINE_ARG_94C4EF10 =
+ "No measurements matched %s (measurement file line %d).";
+ public static final String LOG_START_TIME_AND_END_TIME_DO_NOT_APPLY_TO_QUERIES_AND_ARE_IGNORED_02DC37CC =
+ "-start_time and -end_time do not apply to queries and are ignored.";
+ public static final String LOG_TABLE_DOES_NOT_APPLY_TO_QUERIES_AND_IS_IGNORED_DBA31209 =
+ "-table does not apply to queries and is ignored.";
+ public static final String EXCEPTION_NO_MEASUREMENTS_REMAIN_AFTER_PATTERN_EXPANSION_AND_DEDUPLICATION_40D930B7 =
+ "No measurements remain after pattern expansion and deduplication.";
+ public static final String EXCEPTION_THE_TARGET_DATABASE_ARG_DOES_NOT_EXIST_80E096EB =
+ "The target database %s does not exist.";
+ public static final String EXCEPTION_THERE_ARE_NO_TABLES_OR_THE_TARGET_TABLE_ARG_DOES_NOT_EXIST_E5F88A98 =
+ "There are no tables or the target table %s does not exist.";
+ public static final String LOG_PLEASE_INPUT_QUERY_EEAD5433 =
+ "Please input query: ";
+ public static final String EXCEPTION_NO_EXPORT_QUERY_WAS_PROVIDED_09E24A92 =
+ "No export query was provided.";
+ public static final String LOG_EXPORT_FAILED_ARG_57F9420F =
+ "Export failed: %s";
+ public static final String EXCEPTION_CANNOT_OBTAIN_SCHEMA_FOR_MEASUREMENT_ARG_D3F1DAF3 =
+ "Cannot obtain schema for measurement %s.";
+ public static final String EXCEPTION_SQL_TSFILE_OUTPUT_REQUIRES_THE_TIME_COLUMN_AND_UNALIASED_COLUMNS_FROM_TABLE_ARG_C5E77CD3 =
+ "SQL/TsFile output requires the time column and unaliased columns from table %s.";
+ public static final String EXCEPTION_SQL_OUTPUT_REQUIRES_ORIGINAL_MEASUREMENT_PATHS_OR_AN_ALIGN_BY_DEVICE_QUERY_ADA72EA8 =
+ "SQL output requires original measurement paths or an ALIGN BY DEVICE query.";
+
+
+
+
+ public static final String EXCEPTION_EXPORT_OUTPUT_CONFLICTS_WITH_THE_MEASUREMENT_INPUT_FILE_ARG_1CBEBB6C =
+ "Export output conflicts with the measurement input file: %s";
+ public static final String EXCEPTION_TREE_TSFILE_OUTPUT_REQUIRES_A_TIME_COLUMN_AND_ORIGINAL_MEASUREMENT_PATHS_1DCEE8EA =
+ "Tree TsFile output requires a time column and original measurement paths.";
+
}
diff --git a/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java b/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java
index 8321a28d08e..c4a2d1b78ab 100644
--- a/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java
+++ b/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java
@@ -112,4 +112,145 @@ public final class CliMessages {
public static final String LOG_INPUT_TIME_FORMAT_ARG_NOT_SUPPORTED_00172A7B = "不支持输入时间格式 {}，";
public static final String LOG_PLEASE_INPUT_LIKE_YYYY_MM_DD_HH_MM_SS_SSS_9318BFC7 = "请输入类似 yyyy-MM-dd\\ HH:mm:ss.SSS 或 yyyy-MM-dd'T'HH:mm:ss.SSS 的格式%n";
+ // Export data
+ public static final String MESSAGE_EXPORT_LOG_FILE_DEFAULT_TARGET_LOG_EXPORT_DATA_YYYYMMDDHHMMSS_LOG_B29FC35F =
+ "导出日志文件（默认：<target>/log_export_data_yyyyMMddHHmmss.log）。";
+ public static final String MESSAGE_INCLUSIVE_START_TIME_FOR_MEASUREMENT_FILE_OR_AUTOMATIC_TABLE_EXPORT_69FE75CF =
+ "测点文件或 Table 自动导出的开始时间（含边界）。";
+ public static final String MESSAGE_INCLUSIVE_END_TIME_FOR_MEASUREMENT_FILE_OR_AUTOMATIC_TABLE_EXPORT_D9C24D95 =
+ "测点文件或 Table 自动导出的结束时间（含边界）。";
+ public static final String MESSAGE_UTF_8_IOTDB_PATH_PATTERN_FILE_FOR_TREE_EXPORT_MUTUALLY_EXCLUSIVE_WITH_Q_DD1F2CDA =
+ "Tree 导出的 UTF-8 IoTDB 路径 pattern 文件，与 -q 互斥。";
+ public static final String MESSAGE_TABLE_TO_EXPORT_OPTIONAL_IGNORED_WHEN_Q_IS_GIVEN_4F37C0B5 =
+ "要导出的表（可选）。指定 -q 时忽略。";
+ public static final String
+ MESSAGE_DATABASE_TO_EXPORT_REQUIRED_EXPORTS_ALL_OF_ITS_TABLES_UNLESS_TABLE_OR_Q_IS_GIVEN_F125EEA1 =
+ "要导出的数据库（必填）。不带 -table 或 -q 时导出其中所有表。";
+ public static final String MESSAGE_MAXIMUM_DATA_ROWS_PER_CSV_OR_SQL_FILE_DEFAULT_10000_000CEFB9 =
+ "每个 CSV 或 SQL 文件的最大数据行数（默认 10000）。";
+ public static final String EXCEPTION_FILE_PREFIX_MUST_NOT_CONTAIN_PATH_SEPARATORS_ARG_C89823EE =
+ "文件名前缀不能包含路径分隔符：%s";
+ public static final String EXCEPTION_INVALID_OUTPUT_TIME_FORMAT_ARG_9C690675 =
+ "输出时间格式无效：%s";
+ public static final String EXCEPTION_EXPORTING_INFORMATION_SCHEMA_IS_NOT_SUPPORTED_6FD0FB85 =
+ "不支持导出 information_schema。";
+ public static final String EXCEPTION_THE_EXPORT_LOG_MUST_DIFFER_FROM_THE_MEASUREMENT_INPUT_FILE_088A542C =
+ "导出日志不能与测点输入文件相同。";
+ public static final String LOG_PLEASE_INPUT_YOUR_PASSWORD_AEE49DC5 =
+ "请输入密码：";
+ public static final String LOG_PLEASE_INPUT_YOUR_TRUST_STORE_4C50D824 =
+ "请输入 trust_store：";
+ public static final String LOG_PLEASE_INPUT_YOUR_TRUST_STORE_PASSWORD_86ED5CB6 =
+ "请输入 trust_store_password：";
+ public static final String LOG_PLEASE_INPUT_YOUR_KEY_STORE_AC5DB91F =
+ "请输入 key_store：";
+ public static final String LOG_PLEASE_INPUT_YOUR_KEY_STORE_PASSWORD_19ED16D4 =
+ "请输入 key_store_password：";
+ public static final String EXCEPTION_UNSUPPORTED_SQL_DIALECT_ARG_C4D0D97C =
+ "不支持的 sql_dialect：%s";
+ public static final String EXCEPTION_SPECIFY_AN_EXPORT_FORMAT_CSV_SQL_OR_TSFILE_RECEIVED_ARG_5B5211B5 =
+ "请指定导出格式 csv、sql 或 tsfile（收到：%s）。";
+ public static final String EXCEPTION_MEASUREMENT_FILE_IS_SUPPORTED_ONLY_IN_THE_TREE_MODEL_1ED604B5 =
+ "-measurement_file 仅支持 Tree 模型。";
+ public static final String EXCEPTION_MEASUREMENT_FILE_AND_Q_ARE_MUTUALLY_EXCLUSIVE_7A7457D6 =
+ "-measurement_file 和 -q 互斥。";
+ public static final String EXCEPTION_UNEXPECTED_POSITIONAL_ARGUMENTS_ARG_027F44CB =
+ "不支持的位置参数：%s";
+ public static final String EXCEPTION_INVALID_POSITIVE_INTEGER_FOR_OPTION_ARG_ARG_30FB488D =
+ "选项 %s 的正整数值无效：%s";
+ public static final String EXCEPTION_OPTION_ARG_REQUIRES_TRUE_OR_FALSE_ARG_0DC9BE3D =
+ "选项 %s 必须为 true 或 false：%s";
+ public static final String EXCEPTION_START_TIME_MUST_BE_LESS_THAN_OR_EQUAL_TO_END_TIME_3551199C =
+ "start_time 必须小于或等于 end_time。";
+ public static final String EXCEPTION_CANNOT_PARSE_TIME_ARG_ARG_40257DF7 =
+ "无法解析时间 %s：%s";
+ public static final String EXCEPTION_MEASUREMENT_INPUT_MUST_BE_A_READABLE_REGULAR_FILE_ARG_43D4A272 =
+ "测点输入必须是可读的普通文件：%s";
+ public static final String EXCEPTION_A_COMPLETE_IOTDB_PATH_PATTERN_STARTING_WITH_ROOT_IS_REQUIRED_7D000A79 =
+ "必须提供以 root 开头的完整 IoTDB 路径 pattern。";
+ public static final String EXCEPTION_INVALID_MEASUREMENT_INPUT_ARG_AT_LINE_ARG_CONTENT_ARG_ARG_27A0F948 =
+ "测点输入 %s 的第 %d 行无效，原文=%s：%s";
+ public static final String EXCEPTION_CANNOT_READ_UTF_8_MEASUREMENT_FILE_ARG_AT_LINE_ARG_ARG_A5739D3F =
+ "无法读取 UTF-8 测点文件 %s 的第 %d 行：%s";
+ public static final String EXCEPTION_MEASUREMENT_INPUT_CONTAINS_NO_PATH_PATTERNS_ARG_96760070 =
+ "测点输入没有有效路径 pattern：%s";
+ public static final String EXCEPTION_MEASUREMENT_OUTPUT_FILENAME_EXCEEDS_THE_SUPPORTED_LENGTH_ARG_FF01275D =
+ "测点输出文件名超过支持的长度：%s";
+ public static final String EXCEPTION_MEASUREMENTS_ARG_AND_ARG_MAP_TO_THE_SAME_OUTPUT_FILENAME_ARG_58E1FD4D =
+ "测点 %s 和 %s 对应同一个输出文件名：%s";
+ public static final String EXCEPTION_EXPORT_OUTPUT_CONFLICTS_WITH_THE_LOG_FILE_ARG_F01BE79D =
+ "导出输出与日志文件冲突：%s";
+ public static final String LOG_EXPORT_STARTED_45465087 =
+ "导出已开始。";
+ public static final String LOG_EXPORT_COMPLETELY_34200D19 =
+ "导出完成！";
+ public static final String EXCEPTION_EXPORT_CANCELLED_85D44F54 =
+ "导出已取消。";
+ public static final String LOG_EXPORT_FAILED_77EE72CD =
+ "导出失败。";
+ public static final String LOG_CANNOT_WRITE_OR_CLOSE_THE_EXPORT_LOG_32579AC6 =
+ "无法写入或关闭导出日志。";
+ public static final String LOG_EXPORT_STARTED_ARG_ARG_TARGET_ARG_LOG_ARG_BD58AECB =
+ "导出已开始（%s，%s）：输出目录 %s，日志 %s";
+ public static final String LOG_MEASUREMENTS_TO_EXPORT_ARG_PATH_PATTERNS_ARG_A0AC5F31 =
+ "待导出测点：%d 个（路径 pattern：%d 个）。";
+ public static final String LOG_EXPORTING_ARG_ARG_ARG_ARG_ROWS_ARG_ETA_ARG_C41CA790 =
+ "导出中 %s %d/%d %s 行 %s 剩余 %s";
+ public static final String LOG_ARG_ARG_ARG_MEASUREMENTS_ARG_ROWS_ELAPSED_ARG_ETA_ARG_CURRENT_ARG_B6DEA0BD =
+ "[%s] %d/%d 个测点，%s 行，已用 %s，剩余 %s，当前 %s";
+ public static final String LOG_EXPORTING_ARG_ROWS_ARG_ARG_ROWS_S_56BD441C =
+ "导出中 %s 行 %s %s 行/秒";
+ public static final String LOG_ARG_ROWS_ELAPSED_ARG_ARG_ROWS_S_AE3CF660 =
+ "%s 行，已用 %s，%s 行/秒";
+ public static final String LOG_WARNING_ARG_05BE32E9 =
+ "警告：%s";
+ public static final String LOG_EXPORT_COMPLETELY_MEASUREMENTS_ARG_ARG_ROWS_ARG_FILES_ARG_ELAPSED_ARG_36D9F8E5 =
+ "导出完成！测点：%d/%d，行数：%s，文件：%s 个，耗时：%s。";
+ public static final String LOG_EXPORT_COMPLETELY_ROWS_ARG_FILES_ARG_ELAPSED_ARG_A94BBB8B =
+ "导出完成！行数：%s，文件：%s 个，耗时：%s。";
+ public static final String LOG_EXPORT_FAILED_AFTER_ARG_ARG_C123F0EF =
+ "导出失败（已用 %s）：%s";
+ public static final String LOG_EXPORT_FAILED_AFTER_ARG_DA6D56F9 =
+ "导出失败（已用 %s）。";
+ public static final String LOG_EXPORT_CANCELLED_AFTER_ARG_ROWS_EXPORTED_ARG_0999F99F =
+ "导出已取消（已用 %s），已导出 %s 行。";
+ public static final String LOG_LOG_ARG_FBF61F63 =
+ "日志：%s";
+ public static final String LOG_ARG_82AC62F7 =
+ "%.2f%%";
+ public static final String EXCEPTION_AGGREGATE_QUERIES_CANNOT_BE_EXPORTED_AS_SQL_OR_TSFILE_48BC1932 =
+ "聚合查询不能导出为 SQL 或 TsFile。";
+ public static final String LOG_NO_MEASUREMENTS_MATCHED_ARG_MEASUREMENT_FILE_LINE_ARG_94C4EF10 =
+ "%s 未匹配到任何测点（测点文件第 %d 行）。";
+ public static final String LOG_START_TIME_AND_END_TIME_DO_NOT_APPLY_TO_QUERIES_AND_ARE_IGNORED_02DC37CC =
+ "-start_time 和 -end_time 对查询不生效，已忽略。";
+ public static final String LOG_TABLE_DOES_NOT_APPLY_TO_QUERIES_AND_IS_IGNORED_DBA31209 =
+ "-table 对查询不生效，已忽略。";
+ public static final String EXCEPTION_NO_MEASUREMENTS_REMAIN_AFTER_PATTERN_EXPANSION_AND_DEDUPLICATION_40D930B7 =
+ "pattern 展开并去重后没有实际测点。";
+ public static final String EXCEPTION_THE_TARGET_DATABASE_ARG_DOES_NOT_EXIST_80E096EB =
+ "目标数据库 %s 不存在。";
+ public static final String EXCEPTION_THERE_ARE_NO_TABLES_OR_THE_TARGET_TABLE_ARG_DOES_NOT_EXIST_E5F88A98 =
+ "没有可导出的表，或目标表 %s 不存在。";
+ public static final String LOG_PLEASE_INPUT_QUERY_EEAD5433 =
+ "请输入查询：";
+ public static final String EXCEPTION_NO_EXPORT_QUERY_WAS_PROVIDED_09E24A92 =
+ "没有提供导出查询。";
+ public static final String LOG_EXPORT_FAILED_ARG_57F9420F =
+ "导出失败：%s";
+ public static final String EXCEPTION_CANNOT_OBTAIN_SCHEMA_FOR_MEASUREMENT_ARG_D3F1DAF3 =
+ "无法获取测点 %s 的 schema。";
+ public static final String EXCEPTION_SQL_TSFILE_OUTPUT_REQUIRES_THE_TIME_COLUMN_AND_UNALIASED_COLUMNS_FROM_TABLE_ARG_C5E77CD3 =
+ "SQL/TsFile 输出需要 time 列及表 %s 的原始列（不能使用别名）。";
+ public static final String EXCEPTION_SQL_OUTPUT_REQUIRES_ORIGINAL_MEASUREMENT_PATHS_OR_AN_ALIGN_BY_DEVICE_QUERY_ADA72EA8 =
+ "SQL 输出需要原始测点路径或 ALIGN BY DEVICE 查询。";
+
+
+
+
+ public static final String EXCEPTION_EXPORT_OUTPUT_CONFLICTS_WITH_THE_MEASUREMENT_INPUT_FILE_ARG_1CBEBB6C =
+ "导出输出与测点输入文件冲突：%s";
+ public static final String EXCEPTION_TREE_TSFILE_OUTPUT_REQUIRES_A_TIME_COLUMN_AND_ORIGINAL_MEASUREMENT_PATHS_1DCEE8EA =
+ "Tree TsFile 输出需要 time 列及原始测点路径。";
+
}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/Constants.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/Constants.java
index b6999dea427..86c8419ab1b 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/Constants.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/Constants.java
@@ -231,6 +231,11 @@ public class Constants {
public static final String TARGET_FILE_NAME = "prefix_file_name";
public static final String TARGET_FILE_DESC = "Export file name .(optional)";
+ public static final String MEASUREMENT_FILE_ARGS = "mf";
+ public static final String MEASUREMENT_FILE_NAME = "measurement_file";
+ public static final String EXPORT_DATA_LOG_ARGS = "l";
+ public static final String EXPORT_DATA_LOG_NAME = "export_data_log";
+
public static final String RPC_MAX_FRAME_SIZE_ARGS = "mfs";
public static final String RPC_MAX_FRAME_SIZE_NAME = "rpc_max_frame_size";
public static final String RPC_MAX_FRAME_SIZE_DESC =
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/OptionsUtil.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/OptionsUtil.java
index e7ca2317be6..59d07a7bfdc 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/OptionsUtil.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/OptionsUtil.java
@@ -19,17 +19,13 @@
package org.apache.iotdb.tool.common;
+import org.apache.iotdb.cli.i18n.CliMessages;
+
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
public class OptionsUtil extends Constants {
- private static boolean isImport = true;
-
- public static void setIsImport(boolean isImport) {
- OptionsUtil.isImport = isImport;
- }
-
public static Options createHelpOptions() {
final Options options = new Options();
Option opHelp = Option.builder(HELP_ARGS).longOpt(HELP_ARGS).hasArg().desc(HELP_DESC).build();
@@ -167,6 +163,10 @@ public class OptionsUtil extends Constants {
}
public static Options createImportCommonOptions() {
+ return createDataCommonOptions(true);
+ }
+
+ private static Options createDataCommonOptions(boolean importing) {
Options options = new Options();
Option opFileType =
@@ -175,7 +175,7 @@ public class OptionsUtil extends Constants {
.argName(FILE_TYPE_ARGS_NAME)
.required()
.hasArg()
- .desc(isImport ? FILE_TYPE_DESC_IMPORT : FILE_TYPE_DESC_EXPORT)
+ .desc(importing ? FILE_TYPE_DESC_IMPORT : FILE_TYPE_DESC_EXPORT)
.build();
options.addOption(opFileType);
@@ -191,7 +191,7 @@ public class OptionsUtil extends Constants {
}
public static Options createExportCommonOptions() {
- Options options = createImportCommonOptions();
+ Options options = createDataCommonOptions(false);
Option opFile =
Option.builder(TARGET_DIR_ARGS)
@@ -239,11 +239,53 @@ public class OptionsUtil extends Constants {
.build();
options.addOption(opRpcMaxFrameSize);
+ options.addOption(
+ Option.builder(EXPORT_DATA_LOG_ARGS)
+ .longOpt(EXPORT_DATA_LOG_NAME)
+ .hasArg()
+ .argName(EXPORT_DATA_LOG_NAME)
+ .desc(
+ CliMessages
+ .MESSAGE_EXPORT_LOG_FILE_DEFAULT_TARGET_LOG_EXPORT_DATA_YYYYMMDDHHMMSS_LOG_B29FC35F)
+ .build());
+ options.addOption(
+ Option.builder(START_TIME_ARGS)
+ .longOpt(START_TIME_ARGS)
+ .hasArg()
+ .desc(
+ CliMessages
+ .MESSAGE_INCLUSIVE_START_TIME_FOR_MEASUREMENT_FILE_OR_AUTOMATIC_TABLE_EXPORT_69FE75CF)
+ .build());
+ options.addOption(
+ Option.builder(END_TIME_ARGS)
+ .longOpt(END_TIME_ARGS)
+ .hasArg()
+ .desc(
+ CliMessages
+ .MESSAGE_INCLUSIVE_END_TIME_FOR_MEASUREMENT_FILE_OR_AUTOMATIC_TABLE_EXPORT_D9C24D95)
+ .build());
+ options.addOption(
+ Option.builder(TIME_ZONE_ARGS)
+ .longOpt(TIME_ZONE_NAME)
+ .hasArg()
+ .desc(TIME_ZONE_DESC)
+ .build());
+
return options;
}
public static Options createTreeExportCommonOptions() {
- return createExportCommonOptions();
+ Options options = createExportCommonOptions();
+ options.addOption(
+ Option.builder(MEASUREMENT_FILE_ARGS)
+ .longOpt(MEASUREMENT_FILE_NAME)
+ .hasArg()
+ .argName(MEASUREMENT_FILE_NAME)
+ .desc(
+ CliMessages
+ .MESSAGE_UTF_8_IOTDB_PATH_PATTERN_FILE_FOR_TREE_EXPORT_MUTUALLY_EXCLUSIVE_WITH_Q_DD1F2CDA)
+ .build());
+ return options;
}
public static Options createTableExportCommonOptions() {
@@ -255,7 +297,9 @@ public class OptionsUtil extends Constants {
.argName(DB_ARGS)
.hasArg()
.required()
- .desc(DB_DESC)
+ .desc(
+ CliMessages
+ .MESSAGE_DATABASE_TO_EXPORT_REQUIRED_EXPORTS_ALL_OF_ITS_TABLES_UNLESS_TABLE_OR_Q_IS_GIVEN_F125EEA1)
.build();
options.addOption(opDatabase);
@@ -264,28 +308,10 @@ public class OptionsUtil extends Constants {
.longOpt(TABLE_ARGS)
.argName(TABLE_ARGS)
.hasArg()
- .desc(TABLE_DESC_EXPORT)
+ .desc(CliMessages.MESSAGE_TABLE_TO_EXPORT_OPTIONAL_IGNORED_WHEN_Q_IS_GIVEN_4F37C0B5)
.build();
options.addOption(opTable);
- Option opStartTime =
- Option.builder(START_TIME_ARGS)
- .longOpt(START_TIME_ARGS)
- .argName(START_TIME_ARGS)
- .hasArg()
- .desc(START_TIME_DESC)
- .build();
- options.addOption(opStartTime);
-
- Option opEndTime =
- Option.builder(END_TIME_ARGS)
- .longOpt(END_TIME_ARGS)
- .argName(END_TIME_ARGS)
- .hasArg()
- .desc(END_TIME_DESC)
- .build();
- options.addOption(opEndTime);
-
return options;
}
@@ -297,7 +323,7 @@ public class OptionsUtil extends Constants {
.longOpt(TABLE_ARGS)
.argName(TABLE_ARGS)
.hasArg()
- .desc(TABLE_DESC_EXPORT)
+ .desc(CliMessages.MESSAGE_TABLE_TO_EXPORT_OPTIONAL_IGNORED_WHEN_Q_IS_GIVEN_4F37C0B5)
.build();
options.addOption(opTable);
@@ -315,7 +341,7 @@ public class OptionsUtil extends Constants {
.longOpt(LINES_PER_FILE_NAME)
.argName(LINES_PER_FILE_NAME)
.hasArg()
- .desc(LINES_PER_FILE_DESC)
+ .desc(CliMessages.MESSAGE_MAXIMUM_DATA_ROWS_PER_CSV_OR_SQL_FILE_DEFAULT_10000_000CEFB9)
.build();
options.addOption(opLinesPerFile);
@@ -348,7 +374,7 @@ public class OptionsUtil extends Constants {
.longOpt(TABLE_ARGS)
.argName(TABLE_ARGS)
.hasArg()
- .desc(TABLE_DESC_EXPORT)
+ .desc(CliMessages.MESSAGE_TABLE_TO_EXPORT_OPTIONAL_IGNORED_WHEN_Q_IS_GIVEN_4F37C0B5)
.build();
options.addOption(opTable);
@@ -390,7 +416,7 @@ public class OptionsUtil extends Constants {
.longOpt(TABLE_ARGS)
.argName(TABLE_ARGS)
.hasArg()
- .desc(TABLE_DESC_EXPORT)
+ .desc(CliMessages.MESSAGE_TABLE_TO_EXPORT_OPTIONAL_IGNORED_WHEN_Q_IS_GIVEN_4F37C0B5)
.build();
options.addOption(opTable);
@@ -408,7 +434,7 @@ public class OptionsUtil extends Constants {
.longOpt(LINES_PER_FILE_NAME)
.argName(LINES_PER_FILE_NAME)
.hasArg()
- .desc(LINES_PER_FILE_DESC)
+ .desc(CliMessages.MESSAGE_MAXIMUM_DATA_ROWS_PER_CSV_OR_SQL_FILE_DEFAULT_10000_000CEFB9)
.build();
options.addOption(opLinesPerFile);
@@ -450,7 +476,7 @@ public class OptionsUtil extends Constants {
.longOpt(LINES_PER_FILE_NAME)
.argName(LINES_PER_FILE_NAME)
.hasArg()
- .desc(LINES_PER_FILE_DESC)
+ .desc(CliMessages.MESSAGE_MAXIMUM_DATA_ROWS_PER_CSV_OR_SQL_FILE_DEFAULT_10000_000CEFB9)
.build();
options.addOption(opLinesPerFile);
@@ -492,7 +518,7 @@ public class OptionsUtil extends Constants {
.longOpt(LINES_PER_FILE_NAME)
.argName(LINES_PER_FILE_NAME)
.hasArg()
- .desc(LINES_PER_FILE_DESC)
+ .desc(CliMessages.MESSAGE_MAXIMUM_DATA_ROWS_PER_CSV_OR_SQL_FILE_DEFAULT_10000_000CEFB9)
.build();
options.addOption(opLinesPerFile);
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractExportData.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractExportData.java
index e3c33950d42..ad1815d42f8 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractExportData.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractExportData.java
@@ -19,79 +19,174 @@
package org.apache.iotdb.tool.data;
-import org.apache.iotdb.cli.utils.IoTPrinter;
+import org.apache.iotdb.cli.i18n.CliMessages;
+import org.apache.iotdb.exception.ArgsErrorException;
import org.apache.iotdb.isession.SessionDataSet;
-import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.RpcUtils;
-import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.tool.common.Constants;
+import org.apache.iotdb.session.Session;
-import org.apache.thrift.TException;
-
-import java.time.Instant;
-import java.time.ZonedDateTime;
+import java.io.IOException;
+import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
-import java.util.regex.Matcher;
import java.util.regex.Pattern;
-public abstract class AbstractExportData extends AbstractDataTool {
+/** Model-specific exporters share configuration and explicit resource ownership. */
+public abstract class AbstractExportData implements AutoCloseable {
+ private static final Pattern AGGREGATE =
+ Pattern.compile(
+ "\\b(count|sum|avg|extreme|max_value|min_value|first_value|last_value|max_time|min_time|stddev|stddev_pop|stddev_samp|variance|var_pop|var_samp|max_by|min_by)\\b\\s*\\(",
+ Pattern.CASE_INSENSITIVE);
+ protected final ExportConfig config;
+ protected final ExportProgressReporter progress;
+ protected ZoneId zoneId;
+ protected String timestampPrecision;
+ private final List<Resource<?>> resources = new ArrayList<>();
+ private boolean closed;
- private static final IoTPrinter ioTPrinter = new IoTPrinter(System.out);
+ AbstractExportData(ExportConfig config, ExportProgressReporter progress) {
+ this.config = config;
+ this.progress = progress;
+ }
- public abstract void init()
- throws IoTDBConnectionException, StatementExecutionException, TException;
+ public abstract void init() throws Exception;
- public abstract boolean exportBySql(String sql, int index);
+ abstract void export(ExportTask task) throws Exception;
- protected static String formatSqlValue(String value, String type) {
- if (value == null) {
- return "NULL";
+ protected void readTimeSettings(Session session) throws Exception {
+ session.open(false);
+ timestampPrecision = session.getTimestampPrecision();
+ if (config.timeZone != null) {
+ session.setTimeZone(config.timeZone);
}
- if ("TEXT".equalsIgnoreCase(type) || "STRING".equalsIgnoreCase(type)) {
- return "'" + value.replace("'", "''") + "'";
+ zoneId = ZoneId.of(session.getTimeZone());
+ }
+
+ ExportTimeRange timeRange() throws ArgsErrorException {
+ return ExportTimeRange.parse(config.startTime, config.endTime, zoneId, timestampPrecision);
+ }
+
+ protected static void legalCheck(String sql) throws ArgsErrorException {
+ if (AGGREGATE.matcher(sql).find()) {
+ throw new ArgsErrorException(
+ CliMessages.EXCEPTION_AGGREGATE_QUERIES_CANNOT_BE_EXPORTED_AS_SQL_OR_TSFILE_48BC1932);
}
- return value;
}
- protected static List<String> getCsvHeaders(SessionDataSet dataSet, boolean treeModel) {
- List<String> headers = new ArrayList<>(dataSet.getColumnNames());
- if (Boolean.TRUE.equals(needDataTypePrinted)) {
- List<String> types = dataSet.getColumnTypes();
+ /** With -dt, value columns are written as name(TYPE); time and the Tree Device column are not. */
+ protected List<String> csvHeaders(List<String> names, List<String> types, boolean tree) {
+ List<String> headers = new ArrayList<>(names);
+ if (config.printDataType) {
for (int i = 0; i < headers.size(); i++) {
String name = headers.get(i);
- if (!"Time".equalsIgnoreCase(name) && !(treeModel && "Device".equals(name))) {
- headers.set(i, String.format("%s(%s)", name, types.get(i)));
+ if (!"Time".equalsIgnoreCase(name) && !(tree && "Device".equals(name))) {
+ headers.set(i, name + "(" + types.get(i) + ")");
}
}
}
return headers;
}
- protected static void legalCheck(String sql) {
- String aggregatePattern =
- "\\b(count|sum|avg|extreme|max_value|min_value|first_value|last_value|max_time|min_time|stddev|stddev_pop|stddev_samp|variance|var_pop|var_samp|max_by|min_by)\\b\\s*\\(";
- Pattern pattern = Pattern.compile(aggregatePattern, Pattern.CASE_INSENSITIVE);
- Matcher matcher = pattern.matcher(sql);
- if (matcher.find()) {
- ioTPrinter.println("The sql you entered is invalid, please don't use aggregate query.");
- System.exit(Constants.CODE_ERROR);
+ /** One value of an INSERT statement, in a form both SQL dialects read back. */
+ protected static String sqlValue(SessionDataSet.DataIterator iterator, String type, int column)
+ throws Exception {
+ if (iterator.isNull(column)) {
+ return "NULL";
+ }
+ switch (type) {
+ case "TEXT":
+ case "STRING":
+ case "DATE":
+ return "'" + iterator.getString(column).replace("'", "''") + "'";
+ case "TIMESTAMP":
+ return Long.toString(iterator.getLong(column));
+ case "BLOB":
+ return "X'" + iterator.getString(column).substring(2) + "'";
+ default:
+ return iterator.getString(column);
}
}
- protected static String timeTrans(Long time) {
- switch (timeFormat) {
- case "default":
- return RpcUtils.parseLongToDateWithPrecision(
- DateTimeFormatter.ISO_OFFSET_DATE_TIME, time, zoneId, timestampPrecision);
- case "timestamp":
- case "long":
- case "number":
- return String.valueOf(time);
- default:
- return ZonedDateTime.ofInstant(Instant.ofEpochMilli(time), zoneId)
- .format(DateTimeFormatter.ofPattern(timeFormat));
+ protected String timeTrans(long time) {
+ if ("timestamp".equals(config.timeFormat)
+ || "number".equals(config.timeFormat)
+ || "long".equals(config.timeFormat)) {
+ return Long.toString(time);
+ }
+ DateTimeFormatter formatter =
+ "default".equals(config.timeFormat)
+ ? DateTimeFormatter.ISO_OFFSET_DATE_TIME
+ : DateTimeFormatter.ofPattern(config.timeFormat);
+ return RpcUtils.parseLongToDateWithPrecision(formatter, time, zoneId, timestampPrecision);
+ }
+
+ protected <T extends AutoCloseable> Resource<T> own(T resource) throws Exception {
+ synchronized (resources) {
+ if (closed) {
+ resource.close();
+ throw new IOException(CliMessages.EXCEPTION_EXPORT_CANCELLED_85D44F54);
+ }
+ Resource<T> owned = new Resource<>(resource);
+ resources.add(owned);
+ return owned;
+ }
+ }
+
+ /** Closing from the shutdown hook and normal execution releases each handle only once. */
+ protected final class Resource<T extends AutoCloseable> implements AutoCloseable {
+ private final T value;
+ private boolean released;
+
+ private Resource(T value) {
+ this.value = value;
+ }
+
+ T get() {
+ return value;
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ if (!released) {
+ released = true;
+ try {
+ value.close();
+ } finally {
+ synchronized (resources) {
+ resources.remove(this);
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ List<Resource<?>> remaining;
+ synchronized (resources) {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ remaining = new ArrayList<>(resources);
+ }
+ Collections.reverse(remaining);
+ Exception failure = null;
+ for (Resource<?> resource : remaining) {
+ try {
+ resource.close();
+ } catch (Exception e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
}
}
}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportConfig.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportConfig.java
new file mode 100644
index 00000000000..2e2d60b568d
--- /dev/null
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportConfig.java
@@ -0,0 +1,363 @@
+/*
+ * 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.tool.data;
+
+import org.apache.iotdb.cli.i18n.CliMessages;
+import org.apache.iotdb.cli.type.ExitType;
+import org.apache.iotdb.cli.utils.CliContext;
+import org.apache.iotdb.cli.utils.JlineUtils;
+import org.apache.iotdb.exception.ArgsErrorException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.TableSessionBuilder;
+import org.apache.iotdb.tool.common.Constants;
+import org.apache.iotdb.tool.common.OptionsUtil;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.jline.reader.LineReader;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Locale;
+
+/** Immutable options for one invocation, independent of the import tool's static configuration. */
+final class ExportConfig {
+ final boolean tree;
+ final String format;
+ final String host;
+ final int port;
+ final String username;
+ final String password;
+ final boolean useSsl;
+ final String trustStore;
+ final String trustStorePassword;
+ final String keyStore;
+ final String keyStorePassword;
+ final String sslProtocol;
+ final Path targetDirectory;
+ final Path measurementFile;
+ final Path logFile;
+ final String prefix;
+ final String query;
+ final String database;
+ final String table;
+ final String startTime;
+ final String endTime;
+ final String timeZone;
+ final String timeFormat;
+ final long timeout;
+ final int rpcMaxFrameSize;
+ final int linesPerFile;
+ final boolean printDataType;
+ final boolean aligned;
+
+ private ExportConfig(CommandLine options, ZonedDateTime startedAt)
+ throws ArgsErrorException, IOException {
+ tree =
+ Constants.SQL_DIALECT_VALUE_TREE.equalsIgnoreCase(
+ options.getOptionValue(Constants.SQL_DIALECT_ARGS, Constants.SQL_DIALECT_VALUE_TREE));
+ format = options.getOptionValue(Constants.FILE_TYPE_ARGS).toLowerCase(Locale.ROOT);
+ host = options.getOptionValue(Constants.HOST_ARGS, Constants.HOST_DEFAULT_VALUE);
+ port = (int) positiveNumber(options, Constants.PORT_ARGS, 6667, 65535);
+ username = options.getOptionValue(Constants.USERNAME_ARGS, Constants.USERNAME_DEFAULT_VALUE);
+ timeout = positiveNumber(options, Constants.TIMEOUT_ARGS, Long.MAX_VALUE, Long.MAX_VALUE);
+ rpcMaxFrameSize =
+ (int)
+ positiveNumber(
+ options, Constants.RPC_MAX_FRAME_SIZE_ARGS, 536870912, Integer.MAX_VALUE);
+ linesPerFile =
+ (int) positiveNumber(options, Constants.LINES_PER_FILE_ARGS, 10000, Integer.MAX_VALUE);
+ printDataType = booleanOption(options, Constants.DATA_TYPE_ARGS, false);
+ aligned = booleanOption(options, Constants.ALIGNED_ARGS, false);
+ useSsl = booleanOption(options, Constants.USE_SSL_ARGS, false);
+ targetDirectory =
+ Path.of(options.getOptionValue(Constants.TARGET_DIR_ARGS)).toAbsolutePath().normalize();
+ String input = options.getOptionValue(Constants.MEASUREMENT_FILE_ARGS);
+ measurementFile = input == null ? null : Path.of(input).toAbsolutePath().normalize();
+ query = options.getOptionValue(Constants.QUERY_COMMAND_ARGS);
+ prefix =
+ options.getOptionValue(
+ Constants.TARGET_FILE_ARGS,
+ measurementFile == null ? Constants.DUMP_FILE_NAME_DEFAULT : "");
+ if (prefix.contains("/") || prefix.contains("\\") || prefix.indexOf('\0') >= 0) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages.EXCEPTION_FILE_PREFIX_MUST_NOT_CONTAIN_PATH_SEPARATORS_ARG_C89823EE,
+ prefix));
+ }
+ String log = options.getOptionValue(Constants.EXPORT_DATA_LOG_ARGS);
+ logFile =
+ log == null
+ ? targetDirectory.resolve(
+ "log_export_data_"
+ + startedAt.format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss"))
+ + ".log")
+ : Path.of(log).toAbsolutePath().normalize();
+ database = lower(options.getOptionValue(Constants.DB_ARGS));
+ table = lower(options.getOptionValue(Constants.TABLE_ARGS));
+ startTime = options.getOptionValue(Constants.START_TIME_ARGS);
+ endTime = options.getOptionValue(Constants.END_TIME_ARGS);
+ timeZone = options.getOptionValue(Constants.TIME_ZONE_ARGS);
+ timeFormat = options.getOptionValue(Constants.TIME_FORMAT_ARGS, "default");
+ if (!"default".equals(timeFormat)
+ && !"timestamp".equals(timeFormat)
+ && !"number".equals(timeFormat)
+ && !"long".equals(timeFormat)) {
+ try {
+ DateTimeFormatter.ofPattern(timeFormat);
+ } catch (IllegalArgumentException e) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages.EXCEPTION_INVALID_OUTPUT_TIME_FORMAT_ARG_9C690675, timeFormat));
+ }
+ }
+ if (!tree && "information_schema".equals(database)) {
+ throw new ArgsErrorException(
+ CliMessages.EXCEPTION_EXPORTING_INFORMATION_SCHEMA_IS_NOT_SUPPORTED_6FD0FB85);
+ }
+ if (measurementFile != null && logFile.equals(measurementFile)) {
+ throw new ArgsErrorException(
+ CliMessages
+ .EXCEPTION_THE_EXPORT_LOG_MUST_DIFFER_FROM_THE_MEASUREMENT_INPUT_FILE_088A542C);
+ }
+ // Only construct a terminal reader if an option actually needs a prompt.
+ LineReader reader = null;
+ if ((options.hasOption(Constants.PW_ARGS) && options.getOptionValue(Constants.PW_ARGS) == null)
+ || (useSsl
+ && (options.getOptionValue(Constants.TRUST_STORE_ARGS) == null
+ || options.getOptionValue(Constants.TRUST_STORE_PWD_ARGS) == null
+ || ((options.hasOption(Constants.KEY_STORE_ARGS)
+ || options.hasOption(Constants.KEY_STORE_PWD_ARGS))
+ && (options.getOptionValue(Constants.KEY_STORE_ARGS) == null
+ || options.getOptionValue(Constants.KEY_STORE_PWD_ARGS) == null))))) {
+ reader =
+ JlineUtils.getLineReader(
+ new CliContext(System.in, System.out, System.err, ExitType.EXCEPTION),
+ username,
+ host,
+ Integer.toString(port));
+ }
+ password =
+ options.hasOption(Constants.PW_ARGS)
+ ? valueOrPrompt(
+ options,
+ Constants.PW_ARGS,
+ reader,
+ CliMessages.LOG_PLEASE_INPUT_YOUR_PASSWORD_AEE49DC5)
+ : Constants.PW_DEFAULT_VALUE;
+ trustStore =
+ useSsl
+ ? valueOrPrompt(
+ options,
+ Constants.TRUST_STORE_ARGS,
+ reader,
+ CliMessages.LOG_PLEASE_INPUT_YOUR_TRUST_STORE_4C50D824)
+ : null;
+ trustStorePassword =
+ useSsl
+ ? valueOrPrompt(
+ options,
+ Constants.TRUST_STORE_PWD_ARGS,
+ reader,
+ CliMessages.LOG_PLEASE_INPUT_YOUR_TRUST_STORE_PASSWORD_86ED5CB6)
+ : null;
+ keyStore =
+ useSsl
+ && (options.hasOption(Constants.KEY_STORE_ARGS)
+ || options.hasOption(Constants.KEY_STORE_PWD_ARGS))
+ ? valueOrPrompt(
+ options,
+ Constants.KEY_STORE_ARGS,
+ reader,
+ CliMessages.LOG_PLEASE_INPUT_YOUR_KEY_STORE_AC5DB91F)
+ : null;
+ keyStorePassword =
+ keyStore == null
+ ? null
+ : valueOrPrompt(
+ options,
+ Constants.KEY_STORE_PWD_ARGS,
+ reader,
+ CliMessages.LOG_PLEASE_INPUT_YOUR_KEY_STORE_PASSWORD_19ED16D4);
+ sslProtocol = options.getOptionValue(Constants.SSL_PROTOCOL_ARGS);
+ }
+
+ static CommandLine inspect(String[] args) throws ParseException {
+ // Parse a union without required flags to select the dialect/format, without silently
+ // accepting unknown arguments. The selected option set is subsequently parsed strictly.
+ Options all = new Options();
+ Options[] sets = {
+ OptionsUtil.createExportCsvOptions(),
+ OptionsUtil.createExportSqlOptions(),
+ OptionsUtil.createExportTsFileOptions(),
+ OptionsUtil.createTableExportCsvOptions(),
+ OptionsUtil.createTableExportSqlOptions(),
+ OptionsUtil.createTableExportTsFileOptions()
+ };
+ for (Options set : sets) {
+ for (Option option : set.getOptions()) {
+ Option copy = (Option) option.clone();
+ copy.setRequired(false);
+ all.addOption(copy);
+ }
+ }
+ all.addOption(Option.builder(Constants.HELP_ARGS).hasArg().optionalArg(true).build());
+ return new DefaultParser(false).parse(all, args);
+ }
+
+ static ExportConfig parse(String[] args, ZonedDateTime startedAt)
+ throws ParseException, ArgsErrorException, IOException {
+ CommandLine inspected = inspect(args);
+ String dialect = inspected.getOptionValue(Constants.SQL_DIALECT_ARGS, "tree");
+ if (!"tree".equalsIgnoreCase(dialect) && !"table".equalsIgnoreCase(dialect)) {
+ throw new ArgsErrorException(
+ String.format(CliMessages.EXCEPTION_UNSUPPORTED_SQL_DIALECT_ARG_C4D0D97C, dialect));
+ }
+ String format = inspected.getOptionValue(Constants.FILE_TYPE_ARGS);
+ if (format == null
+ || (!"csv".equalsIgnoreCase(format)
+ && !"sql".equalsIgnoreCase(format)
+ && !"tsfile".equalsIgnoreCase(format))) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages
+ .EXCEPTION_SPECIFY_AN_EXPORT_FORMAT_CSV_SQL_OR_TSFILE_RECEIVED_ARG_5B5211B5,
+ format));
+ }
+ if (inspected.hasOption(Constants.MEASUREMENT_FILE_ARGS)) {
+ if (!"tree".equalsIgnoreCase(dialect)) {
+ throw new ArgsErrorException(
+ CliMessages.EXCEPTION_MEASUREMENT_FILE_IS_SUPPORTED_ONLY_IN_THE_TREE_MODEL_1ED604B5);
+ }
+ if (inspected.hasOption(Constants.QUERY_COMMAND_ARGS)) {
+ throw new ArgsErrorException(
+ CliMessages.EXCEPTION_MEASUREMENT_FILE_AND_Q_ARE_MUTUALLY_EXCLUSIVE_7A7457D6);
+ }
+ }
+ CommandLine selected =
+ new DefaultParser(false).parse(options("tree".equalsIgnoreCase(dialect), format), args);
+ if (!selected.getArgList().isEmpty()) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages.EXCEPTION_UNEXPECTED_POSITIONAL_ARGUMENTS_ARG_027F44CB,
+ selected.getArgList()));
+ }
+ return new ExportConfig(selected, startedAt);
+ }
+
+ static Options options(boolean tree, String format) {
+ if ("csv".equalsIgnoreCase(format)) {
+ return tree
+ ? OptionsUtil.createExportCsvOptions()
+ : OptionsUtil.createTableExportCsvOptions();
+ }
+ if ("sql".equalsIgnoreCase(format)) {
+ return tree
+ ? OptionsUtil.createExportSqlOptions()
+ : OptionsUtil.createTableExportSqlOptions();
+ }
+ return tree
+ ? OptionsUtil.createExportTsFileOptions()
+ : OptionsUtil.createTableExportTsFileOptions();
+ }
+
+ private static long positiveNumber(CommandLine options, String key, long fallback, long maximum)
+ throws ArgsErrorException {
+ String value = options.getOptionValue(key);
+ if (value == null) {
+ return fallback;
+ }
+ try {
+ long parsed = Long.parseLong(value);
+ if (parsed > 0 && parsed <= maximum) {
+ return parsed;
+ }
+ } catch (NumberFormatException ignored) {
+ // Report the option as well as the original value.
+ }
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages.EXCEPTION_INVALID_POSITIVE_INTEGER_FOR_OPTION_ARG_ARG_30FB488D,
+ key,
+ value));
+ }
+
+ private static boolean booleanOption(CommandLine options, String key, boolean fallback)
+ throws ArgsErrorException {
+ String value = options.getOptionValue(key);
+ if (value == null) {
+ return fallback;
+ }
+ if (!"true".equalsIgnoreCase(value) && !"false".equalsIgnoreCase(value)) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages.EXCEPTION_OPTION_ARG_REQUIRES_TRUE_OR_FALSE_ARG_0DC9BE3D, key, value));
+ }
+ return Boolean.parseBoolean(value);
+ }
+
+ private static String lower(String value) {
+ return value == null ? null : value.toLowerCase(Locale.ROOT);
+ }
+
+ private static String valueOrPrompt(
+ CommandLine options, String key, LineReader reader, String prompt) {
+ String value = options.getOptionValue(key);
+ return value == null ? reader.readLine(prompt, '\0') : value;
+ }
+
+ Session newTreeSession() {
+ Session.Builder builder =
+ new Session.Builder()
+ .host(host)
+ .port(port)
+ .username(username)
+ .password(password)
+ .zoneId(null)
+ .thriftMaxFrameSize(rpcMaxFrameSize);
+ if (useSsl) {
+ builder.useSSL(true).trustStore(trustStore).trustStorePwd(trustStorePassword);
+ if (keyStore != null) {
+ builder.keyStore(keyStore).keyStorePwd(keyStorePassword);
+ }
+ if (sslProtocol != null) {
+ builder.sslProtocol(sslProtocol);
+ }
+ }
+ return builder.build();
+ }
+
+ void configureSsl(TableSessionBuilder builder) {
+ if (useSsl) {
+ builder.useSSL(true).trustStore(trustStore).trustStorePwd(trustStorePassword);
+ if (keyStore != null) {
+ builder.keyStore(keyStore).keyStorePwd(keyStorePassword);
+ }
+ if (sslProtocol != null) {
+ builder.sslProtocol(sslProtocol);
+ }
+ }
+ }
+}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportConsole.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportConsole.java
new file mode 100644
index 00000000000..8601165bd7d
--- /dev/null
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportConsole.java
@@ -0,0 +1,137 @@
+/*
+ * 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.tool.data;
+
+import org.jline.utils.WCWidth;
+
+import java.io.Console;
+import java.io.PrintStream;
+
+/** What a person reads while exporting; the export log keeps the machine-readable events. */
+final class ExportConsole {
+ // Querying the terminal for its size would replace the signal handlers that report the exit
+ // status, so the width comes from COLUMNS, which the shell launcher sets, or is taken as 80.
+ private static final int DEFAULT_WIDTH = 80;
+ private static final String ELLIPSIS = "...";
+
+ private final PrintStream out;
+ private final boolean terminal;
+ private final int width;
+ private int drawn;
+
+ ExportConsole(PrintStream out, boolean terminal) {
+ this(out, terminal, columns());
+ }
+
+ ExportConsole(PrintStream out, boolean terminal, int width) {
+ this.out = out;
+ this.terminal = terminal;
+ this.width = width;
+ }
+
+ static boolean isTerminal() {
+ Console console = System.console();
+ if (console == null) {
+ return false;
+ }
+ try {
+ // From JDK 22 redirected streams also get a console; isTerminal tells them apart.
+ return (Boolean) Console.class.getMethod("isTerminal").invoke(console);
+ } catch (ReflectiveOperationException e) {
+ return true;
+ }
+ }
+
+ private static int columns() {
+ try {
+ int columns = Integer.parseInt(System.getenv("COLUMNS"));
+ return columns >= 20 ? columns : DEFAULT_WIDTH;
+ } catch (NumberFormatException e) {
+ return DEFAULT_WIDTH;
+ }
+ }
+
+ /** Prints a lasting line, first erasing a progress line that would otherwise share its row. */
+ void line(String text) {
+ erase();
+ out.println(text);
+ out.flush();
+ }
+
+ /**
+ * A terminal redraws one line in place, shortening {@code tail} from the left to fit. Redirected
+ * output gets {@code plain} as a line of its own, so files and container logs hold no redraws; so
+ * does a terminal too narrow for {@code text}, since a carriage return reaches only the last row
+ * of a wrapped line and would leave the rest of every redraw behind.
+ */
+ void progress(String text, String tail, String plain) {
+ if (!terminal || width(text) > width - 1) {
+ erase();
+ out.println(plain);
+ out.flush();
+ return;
+ }
+ // The last column stays free: filling it makes some terminals wrap before the next redraw.
+ int room = width - 1 - width(text) - 2;
+ String shown =
+ tail.isEmpty() || room <= ELLIPSIS.length() ? text : text + " " + left(tail, room);
+ int shownWidth = width(shown);
+ out.print('\r' + shown + " ".repeat(Math.max(0, drawn - shownWidth)));
+ out.flush();
+ drawn = shownWidth;
+ }
+
+ private void erase() {
+ if (drawn > 0) {
+ out.print('\r' + " ".repeat(drawn) + '\r');
+ drawn = 0;
+ }
+ }
+
+ /** The end of {@code text} that fits in {@code columns}, marked as shortened when cut. */
+ static String left(String text, int columns) {
+ if (width(text) <= columns) {
+ return text;
+ }
+ int budget = columns - ELLIPSIS.length();
+ int start = text.length();
+ int used = 0;
+ while (start > 0) {
+ int codePoint = text.codePointBefore(start);
+ int codePointWidth = Math.max(0, WCWidth.wcwidth(codePoint));
+ if (used + codePointWidth > budget) {
+ break;
+ }
+ used += codePointWidth;
+ start -= Character.charCount(codePoint);
+ }
+ // Cut between path nodes, so the shown part is whole nodes such as line19.s0.
+ int node = text.indexOf('.', start);
+ if (start > 0 && text.charAt(start - 1) != '.' && node >= 0 && node + 1 < text.length()) {
+ start = node + 1;
+ }
+ return ELLIPSIS + text.substring(start);
+ }
+
+ /** Display columns: a CJK character takes two. */
+ static int width(String text) {
+ return text.codePoints().map(c -> Math.max(0, WCWidth.wcwidth(c))).sum();
+ }
+}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportData.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportData.java
index 63a2fdf45c7..a199d8ba182 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportData.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportData.java
@@ -19,313 +19,291 @@
package org.apache.iotdb.tool.data;
+import org.apache.iotdb.cli.i18n.CliMessages;
import org.apache.iotdb.cli.type.ExitType;
import org.apache.iotdb.cli.utils.CliContext;
-import org.apache.iotdb.cli.utils.IoTPrinter;
import org.apache.iotdb.cli.utils.JlineUtils;
import org.apache.iotdb.exception.ArgsErrorException;
-import org.apache.iotdb.rpc.IoTDBConnectionException;
-import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.tool.common.Constants;
-import org.apache.iotdb.tool.common.OptionsUtil;
import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.CommandLineParser;
-import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.HelpFormatter;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
-import org.apache.thrift.TException;
-import org.apache.tsfile.external.commons.lang3.ObjectUtils;
-import org.apache.tsfile.external.commons.lang3.StringUtils;
-import org.jline.reader.LineReader;
+import org.jline.reader.EndOfFileException;
+import org.jline.reader.UserInterruptException;
+import org.jline.utils.Signals;
-import java.io.File;
import java.io.IOException;
-import java.util.Arrays;
+import java.nio.file.Files;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * Export CSV file.
- *
- * @version 1.0.0 20170719
- */
-public class ExportData extends AbstractDataTool {
+/** Entry point and lifecycle for one export invocation. */
+public final class ExportData {
+ /** How long cancellation waits for the worker to close its own query and session. */
+ private static final long CANCEL_GRACE_SECONDS = 3;
+
+ /** POSIX numbers of the signals that shut the JVM down; it then exits with 128 + number. */
+ private static final Map<String, Integer> TERMINATING_SIGNALS =
+ Map.of("HUP", 1, "INT", 2, "TERM", 15);
+
+ /** Exit status chosen by a terminating signal, or 0 before one arrives. */
+ private static final AtomicInteger SIGNAL_EXIT_CODE = new AtomicInteger();
+
+ /** Reads the Tree query when neither a query nor a measurement file is supplied. */
+ @FunctionalInterface
+ interface QueryPrompt {
+ String read(ExportConfig config) throws IOException;
+ }
- private static final IoTPrinter ioTPrinter = new IoTPrinter(System.out);
+ private ExportData() {}
- @SuppressWarnings({
- "squid:S3776",
- "squid:S2093"
- }) // Suppress high Cognitive Complexity warning, ignore try-with-resources
- /* main function of export csv tool. */
public static void main(String[] args) {
- OptionsUtil.setIsImport(false);
- Options helpOptions = OptionsUtil.createHelpOptions();
- Options tsFileOptions = OptionsUtil.createExportTsFileOptions();
- Options csvOptions = OptionsUtil.createExportCsvOptions();
- Options sqlOptions = OptionsUtil.createExportSqlOptions();
- HelpFormatter hf = new HelpFormatter();
- CommandLine commandLine = null;
- CommandLineParser parser = new DefaultParser();
- hf.setOptionComparator(null); // avoid reordering
- hf.setWidth(Constants.MAX_HELP_CONSOLE_WIDTH);
+ System.exit(run(args));
+ }
- if (args == null || args.length == 0) {
- printHelpOptions(
- Constants.EXPORT_CLI_HEAD,
- Constants.EXPORT_CLI_PREFIX,
- hf,
- tsFileOptions,
- csvOptions,
- sqlOptions,
- true);
- System.exit(Constants.CODE_ERROR);
- }
+ static int run(String[] args) {
+ return run(args, ExportData::promptQuery);
+ }
+
+ // Tests replace the prompt: JLine reads the process terminal rather than a replaced System.in.
+ static int run(String[] args, QueryPrompt queryPrompt) {
+ ZonedDateTime startedAt = ZonedDateTime.now();
+ ExportProgressReporter progress = null;
+ AbstractExportData exporter = null;
+ Thread hook = null;
+ AtomicBoolean cancelled = new AtomicBoolean();
+ CountDownLatch finished = new CountDownLatch(1);
+ Throwable failure = null;
try {
- commandLine = parser.parse(helpOptions, args, true);
- } catch (ParseException e) {
- printHelpOptions(
- Constants.EXPORT_CLI_HEAD,
- Constants.EXPORT_CLI_PREFIX,
- hf,
- tsFileOptions,
- csvOptions,
- sqlOptions,
- true);
- System.exit(Constants.CODE_ERROR);
- }
- final List<String> argList = Arrays.asList(args);
- int helpIndex = argList.indexOf(Constants.MINUS + Constants.HELP_ARGS);
- int sql_dialect = argList.indexOf(Constants.MINUS + Constants.SQL_DIALECT_ARGS); // -sql_dialect
- if (sql_dialect >= 0
- && !Constants.SQL_DIALECT_VALUE_TREE.equalsIgnoreCase(argList.get(sql_dialect + 1))) {
- final String sqlDialectValue = argList.get(sql_dialect + 1);
- if (Constants.SQL_DIALECT_VALUE_TABLE.equalsIgnoreCase(sqlDialectValue)) {
- sqlDialectTree = false;
- csvOptions = OptionsUtil.createTableExportCsvOptions();
- tsFileOptions = OptionsUtil.createTableExportTsFileOptions();
- sqlOptions = OptionsUtil.createTableExportSqlOptions();
- } else {
- ioTPrinter.println(String.format("sql_dialect %s is not support", sqlDialectValue));
- printHelpOptions(
- Constants.IMPORT_CLI_HEAD,
- Constants.IMPORT_CLI_PREFIX,
- hf,
- tsFileOptions,
- csvOptions,
- sqlOptions,
- true);
- System.exit(Constants.CODE_ERROR);
+ if (args == null || args.length == 0) {
+ help(true, null);
+ return Constants.CODE_ERROR;
}
- }
- int ftIndex = argList.indexOf(Constants.MINUS + Constants.FILE_TYPE_ARGS);
- if (ftIndex < 0) {
- ftIndex = argList.indexOf(Constants.MINUS + Constants.FILE_TYPE_NAME);
- }
- if (helpIndex >= 0) {
- fileType = argList.get(helpIndex + 1);
- if (StringUtils.isNotBlank(fileType)) {
- if (Constants.TSFILE_SUFFIXS.equalsIgnoreCase(fileType)) {
- printHelpOptions(null, Constants.EXPORT_CLI_PREFIX, hf, tsFileOptions, null, null, false);
- } else if (Constants.CSV_SUFFIXS.equalsIgnoreCase(fileType)) {
- printHelpOptions(null, Constants.EXPORT_CLI_PREFIX, hf, null, csvOptions, null, false);
- } else if (Constants.SQL_SUFFIXS.equalsIgnoreCase(fileType)) {
- printHelpOptions(null, Constants.EXPORT_CLI_PREFIX, hf, null, null, sqlOptions, false);
- } else {
- ioTPrinter.println(String.format("File type %s is not support", fileType));
- printHelpOptions(
- Constants.EXPORT_CLI_HEAD,
- Constants.EXPORT_CLI_PREFIX,
- hf,
- tsFileOptions,
- csvOptions,
- sqlOptions,
- true);
- }
- } else {
- printHelpOptions(
- Constants.EXPORT_CLI_HEAD,
- Constants.EXPORT_CLI_PREFIX,
- hf,
- tsFileOptions,
- csvOptions,
- sqlOptions,
- true);
+ CommandLine inspected = ExportConfig.inspect(args);
+ if (inspected.hasOption(Constants.HELP_ARGS)) {
+ help(
+ !"table".equalsIgnoreCase(inspected.getOptionValue(Constants.SQL_DIALECT_ARGS)),
+ inspected.getOptionValue(Constants.HELP_ARGS));
+ return Constants.CODE_OK;
}
- System.exit(Constants.CODE_ERROR);
- } else if (ftIndex >= 0) {
- fileType = argList.get(ftIndex + 1);
- if (StringUtils.isNotBlank(fileType)) {
- if (Constants.TSFILE_SUFFIXS.equalsIgnoreCase(fileType)) {
- try {
- commandLine = parser.parse(tsFileOptions, args, true);
- } catch (ParseException e) {
- ioTPrinter.println("Parse error: " + e.getMessage());
- printHelpOptions(
- null, Constants.EXPORT_CLI_PREFIX, hf, tsFileOptions, null, null, false);
- System.exit(Constants.CODE_ERROR);
- }
- } else if (Constants.CSV_SUFFIXS.equalsIgnoreCase(fileType)) {
- try {
- commandLine = parser.parse(csvOptions, args, true);
- } catch (ParseException e) {
- ioTPrinter.println("Parse error: " + e.getMessage());
- printHelpOptions(null, Constants.EXPORT_CLI_PREFIX, hf, null, csvOptions, null, false);
- System.exit(Constants.CODE_ERROR);
- }
- } else if (Constants.SQL_SUFFIXS.equalsIgnoreCase(fileType)) {
- try {
- commandLine = parser.parse(sqlOptions, args, true);
- } catch (ParseException e) {
- ioTPrinter.println("Parse error: " + e.getMessage());
- printHelpOptions(null, Constants.EXPORT_CLI_PREFIX, hf, null, null, sqlOptions, false);
- System.exit(Constants.CODE_ERROR);
- }
- } else {
- ioTPrinter.println(String.format("File type %s is not support", fileType));
- printHelpOptions(
- Constants.EXPORT_CLI_HEAD,
- Constants.EXPORT_CLI_PREFIX,
- hf,
- tsFileOptions,
- csvOptions,
- sqlOptions,
- true);
- System.exit(Constants.CODE_ERROR);
- }
- } else {
- printHelpOptions(
- Constants.EXPORT_CLI_HEAD,
- Constants.EXPORT_CLI_PREFIX,
- hf,
- tsFileOptions,
- csvOptions,
- sqlOptions,
- true);
- System.exit(Constants.CODE_ERROR);
+ ExportConfig config = ExportConfig.parse(args, startedAt);
+ if (config.measurementFile != null
+ && Files.exists(config.measurementFile)
+ && Files.exists(config.logFile)
+ && Files.isSameFile(config.measurementFile, config.logFile)) {
+ throw new ArgsErrorException(
+ CliMessages
+ .EXCEPTION_THE_EXPORT_LOG_MUST_DIFFER_FROM_THE_MEASUREMENT_INPUT_FILE_088A542C);
}
- } else {
- ioTPrinter.println(
- String.format(Constants.REQUIRED_ARGS_ERROR_MSG, Constants.FILE_TYPE_NAME));
- System.exit(Constants.CODE_ERROR);
- }
- int exitCode = Constants.CODE_OK;
- try {
- parseBasicParams(commandLine);
- parseSpecialParams(commandLine);
- if (!checkTimeFormat()) {
- System.exit(Constants.CODE_ERROR);
+ String query = config.query;
+ if (query == null && config.tree && config.measurementFile == null) {
+ // Prompt before the progress timer starts, so periodic events cannot interleave with input.
+ query = queryPrompt.read(config);
}
- AbstractExportData exportData;
- exportData = new ExportDataTree();
- exportData.init();
- if (!sqlDialectTree) {
- exportData = new ExportDataTable();
- exportData.init();
+ Files.createDirectories(config.targetDirectory);
+ progress = new ExportProgressReporter(config, System.nanoTime(), System.out);
+ exporter =
+ config.tree
+ ? new ExportDataTree(config, progress)
+ : new ExportDataTable(config, progress);
+ ExportProgressReporter reporter = progress;
+ AbstractExportData ownedExporter = exporter;
+ Thread worker = Thread.currentThread();
+ hook =
+ new Thread(
+ () -> {
+ int signalExitCode = SIGNAL_EXIT_CODE.get();
+ cancelled.set(true);
+ reporter.cancel(signalExitCode == 0 ? Constants.CODE_ERROR : signalExitCode);
+ worker.interrupt();
+ try {
+ // The worker closes its query and session once an in-flight call returns.
+ // Closing them underneath that call makes Thrift log a socket error.
+ if (finished.await(CANCEL_GRACE_SECONDS, TimeUnit.SECONDS)) {
+ return;
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ // The worker is blocked in a call that ignores interrupts; closing ends it.
+ Throwable closeFailure = null;
+ try {
+ ownedExporter.close();
+ } catch (Exception e) {
+ closeFailure = e;
+ }
+ try {
+ reporter.finish(ExportProgressReporter.Status.CANCELLED, closeFailure);
+ } catch (Exception ignored) {
+ // finish already reports a log failure to the console.
+ }
+ },
+ "export-data-shutdown");
+ recordTerminatingSignals();
+ Runtime.getRuntime().addShutdownHook(hook);
+ progress.start(
+ config,
+ query != null,
+ () -> {
+ worker.interrupt();
+ // Unblock IO cleanup without making the progress timer access the session.
+ Thread cleanup =
+ new Thread(
+ () -> {
+ try {
+ ownedExporter.close();
+ } catch (Exception ignored) {
+ // The reporter retains the original log failure.
+ }
+ },
+ "export-data-log-failure");
+ cleanup.setDaemon(true);
+ cleanup.start();
+ });
+ MeasurementFile input =
+ config.measurementFile == null ? null : MeasurementFile.read(config.measurementFile);
+ List<ExportTask> tasks = query == null ? null : queryTasks(config, query);
+ progress.phase(ExportProgressReporter.Phase.CONNECTING);
+ exporter.init();
+ if (input != null) {
+ tasks = ((ExportDataTree) exporter).planMeasurements(input, exporter.timeRange());
+ } else if (tasks == null) {
+ tasks = ((ExportDataTable) exporter).planTables(exporter.timeRange());
+ } else if (!config.tree) {
+ tasks = ((ExportDataTable) exporter).planQueries(tasks);
}
- if (queryCommand == null) {
- if (sqlDialectTree) {
- LineReader lineReader =
- JlineUtils.getLineReader(
- new CliContext(System.in, System.out, System.err, ExitType.EXCEPTION),
- username,
- host,
- port);
- String sql = lineReader.readLine(Constants.EXPORT_CLI_PREFIX + "> please input query: ");
- ioTPrinter.println(sql);
- String[] values = sql.trim().split(";");
- for (int i = 0; i < values.length; i++) {
- if (!exportData.exportBySql(values[i], i)) {
- exitCode = Constants.CODE_ERROR;
- }
- }
- } else {
- if (!exportData.exportBySql(null, 0)) {
- exitCode = Constants.CODE_ERROR;
+ for (ExportTask task : tasks) {
+ progress.checkActive();
+ exporter.export(task);
+ }
+ progress.checkActive();
+ progress.phase(ExportProgressReporter.Phase.CLOSING);
+ } catch (Exception e) {
+ failure = e;
+ if (e instanceof UserInterruptException || e instanceof EndOfFileException) {
+ cancelled.set(true);
+ }
+ } finally {
+ Exception closeFailure = null;
+ if (exporter != null) {
+ try {
+ exporter.close();
+ } catch (Exception e) {
+ closeFailure = e;
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
}
}
- } else {
- String[] values = queryCommand.trim().split(";");
- for (int i = 0; i < values.length; i++) {
- if (!exportData.exportBySql(values[i], i)) {
- exitCode = Constants.CODE_ERROR;
+ }
+ if (progress != null) {
+ try {
+ // A cancelled export reports cleanup failures, not the interruption it caused.
+ progress.finish(
+ cancelled.get()
+ ? ExportProgressReporter.Status.CANCELLED
+ : failure == null
+ ? ExportProgressReporter.Status.SUCCESS
+ : ExportProgressReporter.Status.FAILED,
+ cancelled.get() ? closeFailure : failure);
+ } catch (Exception e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
}
}
+ } else if (cancelled.get()) {
+ // Input was abandoned at a prompt before the export started.
+ System.out.println(CliMessages.EXCEPTION_EXPORT_CANCELLED_85D44F54);
+ } else if (failure != null) {
+ System.out.println(
+ String.format(CliMessages.LOG_EXPORT_FAILED_ARG_57F9420F, failure.getMessage()));
}
- } catch (IOException e) {
- ioTPrinter.println("Failed to operate on file, because " + e.getMessage());
- exitCode = Constants.CODE_ERROR;
- } catch (ArgsErrorException e) {
- ioTPrinter.println("Invalid args: " + e.getMessage());
- exitCode = Constants.CODE_ERROR;
- } catch (IoTDBConnectionException | StatementExecutionException e) {
- ioTPrinter.println("Connect failed because " + e.getMessage());
- exitCode = Constants.CODE_ERROR;
- } catch (TException e) {
- ioTPrinter.println(
- "Can not get the timestamp precision from server because " + e.getMessage());
- exitCode = Constants.CODE_ERROR;
+ if (hook != null) {
+ try {
+ Runtime.getRuntime().removeShutdownHook(hook);
+ } catch (IllegalStateException ignored) {
+ // JVM shutdown owns the cancellation hook once it has started.
+ }
+ }
+ // A running cancellation hook waits for this before the JVM exits.
+ finished.countDown();
}
- System.exit(exitCode);
+ return failure == null
+ && !cancelled.get()
+ && (progress == null || progress.status() == ExportProgressReporter.Status.SUCCESS)
+ ? Constants.CODE_OK
+ : Constants.CODE_ERROR;
}
- private static void parseSpecialParams(CommandLine commandLine) throws ArgsErrorException {
- targetDirectory =
- checkRequiredArg(Constants.TARGET_DIR_ARGS, Constants.TARGET_DIR_NAME, commandLine, null);
- targetFile = commandLine.getOptionValue(Constants.TARGET_FILE_ARGS);
- needDataTypePrinted = Boolean.valueOf(commandLine.getOptionValue(Constants.DATA_TYPE_ARGS));
- queryCommand = commandLine.getOptionValue(Constants.QUERY_COMMAND_ARGS);
- exportType = commandLine.getOptionValue(Constants.FILE_TYPE_ARGS);
- String timeoutString = commandLine.getOptionValue(Constants.TIMEOUT_ARGS);
- if (timeoutString != null) {
- timeout = Long.parseLong(timeoutString);
- }
- String rpcMaxFrameSizeString = commandLine.getOptionValue(Constants.RPC_MAX_FRAME_SIZE_ARGS);
- if (rpcMaxFrameSizeString != null) {
- rpcMaxFrameSize = Integer.parseInt(rpcMaxFrameSizeString);
- }
- if (needDataTypePrinted == null) {
- needDataTypePrinted = true;
- }
- if (targetFile == null) {
- targetFile = Constants.DUMP_FILE_NAME_DEFAULT;
- }
- timeFormat = commandLine.getOptionValue(Constants.TIME_FORMAT_ARGS);
- if (timeFormat == null) {
- timeFormat = "default";
- }
- timeZoneID = commandLine.getOptionValue(Constants.TIME_ZONE_ARGS);
- if (!targetDirectory.endsWith("/") && !targetDirectory.endsWith("\\")) {
- targetDirectory += File.separator;
- }
- final File file = new File(targetDirectory);
- if (!file.isDirectory() && !file.mkdirs()) {
- ioTPrinter.println(String.format("Failed to create directories %s", targetDirectory));
- System.exit(Constants.CODE_ERROR);
- }
- if (commandLine.getOptionValue(Constants.LINES_PER_FILE_ARGS) != null) {
- linesPerFile = Integer.parseInt(commandLine.getOptionValue(Constants.LINES_PER_FILE_ARGS));
- }
- if (commandLine.getOptionValue(Constants.ALIGNED_ARGS) != null) {
- aligned = Boolean.valueOf(commandLine.getOptionValue(Constants.ALIGNED_ARGS));
+ // Shutdown hooks cannot see the signal, so record it to report the status the JVM exits with.
+ private static void recordTerminatingSignals() {
+ TERMINATING_SIGNALS.forEach(
+ (name, number) ->
+ // Without the signal on this platform, the JVM default handler stays in place.
+ Signals.register(
+ name,
+ () -> {
+ SIGNAL_EXIT_CODE.set(128 + number);
+ System.exit(128 + number);
+ }));
+ }
+
+ private static String promptQuery(ExportConfig config) throws IOException {
+ return JlineUtils.getLineReader(
+ new CliContext(System.in, System.out, System.err, ExitType.EXCEPTION),
+ config.username,
+ config.host,
+ Integer.toString(config.port))
+ .readLine(CliMessages.LOG_PLEASE_INPUT_QUERY_EEAD5433);
+ }
+
+ static List<ExportTask> queryTasks(ExportConfig config, String query) throws ArgsErrorException {
+ if (query == null || query.trim().isEmpty()) {
+ throw new ArgsErrorException(CliMessages.EXCEPTION_NO_EXPORT_QUERY_WAS_PROVIDED_09E24A92);
}
- if (commandLine.getOptionValue(Constants.DB_ARGS) != null) {
- database = commandLine.getOptionValue(Constants.DB_ARGS).toLowerCase();
- if (ObjectUtils.isNotEmpty(database) && "information_schema".equalsIgnoreCase(database)) {
- ioTPrinter.println(
- String.format("Does not support exporting system databases %s", database));
- System.exit(Constants.CODE_ERROR);
+ List<ExportTask> tasks = new ArrayList<>();
+ for (String sql : query.trim().split(";")) {
+ if (sql.trim().isEmpty()) {
+ continue;
}
+ tasks.add(
+ new ExportTask(
+ sql.trim(),
+ config.targetDirectory.resolve(config.prefix + tasks.size()),
+ null,
+ null));
}
- if (commandLine.getOptionValue(Constants.TABLE_ARGS) != null) {
- table = commandLine.getOptionValue(Constants.TABLE_ARGS).toLowerCase();
- }
- if (commandLine.getOptionValue(Constants.START_TIME_ARGS) != null) {
- startTime = commandLine.getOptionValue(Constants.START_TIME_ARGS);
- }
- if (commandLine.getOptionValue(Constants.END_TIME_ARGS) != null) {
- endTime = commandLine.getOptionValue(Constants.END_TIME_ARGS);
+ if (tasks.isEmpty()) {
+ throw new ArgsErrorException(CliMessages.EXCEPTION_NO_EXPORT_QUERY_WAS_PROVIDED_09E24A92);
}
+ return tasks;
+ }
+
+ private static void help(boolean tree, String format) {
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.setOptionComparator(null);
+ formatter.setWidth(Constants.MAX_HELP_CONSOLE_WIDTH);
+ boolean all =
+ format == null
+ || (!"csv".equalsIgnoreCase(format)
+ && !"sql".equalsIgnoreCase(format)
+ && !"tsfile".equalsIgnoreCase(format));
+ AbstractDataTool.printHelpOptions(
+ all ? Constants.EXPORT_CLI_HEAD : null,
+ Constants.EXPORT_CLI_PREFIX,
+ formatter,
+ all || "tsfile".equalsIgnoreCase(format) ? ExportConfig.options(tree, "tsfile") : null,
+ all || "csv".equalsIgnoreCase(format) ? ExportConfig.options(tree, "csv") : null,
+ all || "sql".equalsIgnoreCase(format) ? ExportConfig.options(tree, "sql") : null,
+ all);
}
}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTable.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTable.java
index d60b5d559b6..55e5742b1ab 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTable.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTable.java
@@ -20,16 +20,15 @@
package org.apache.iotdb.tool.data;
import org.apache.iotdb.cli.i18n.CliMessages;
-import org.apache.iotdb.cli.utils.IoTPrinter;
import org.apache.iotdb.commons.queryengine.plan.relational.sql.parser.CaseInsensitiveStream;
import org.apache.iotdb.db.relational.grammar.sql.RelationalSqlBaseVisitor;
import org.apache.iotdb.db.relational.grammar.sql.RelationalSqlLexer;
import org.apache.iotdb.db.relational.grammar.sql.RelationalSqlParser;
+import org.apache.iotdb.exception.ArgsErrorException;
import org.apache.iotdb.isession.ITableSession;
import org.apache.iotdb.isession.SessionDataSet;
-import org.apache.iotdb.rpc.IoTDBConnectionException;
-import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.rpc.UrlUtils;
+import org.apache.iotdb.session.Session;
import org.apache.iotdb.session.TableSessionBuilder;
import org.apache.iotdb.tool.common.Constants;
@@ -40,162 +39,114 @@ import org.antlr.v4.runtime.CommonTokenStream;
import org.antlr.v4.runtime.RecognitionException;
import org.antlr.v4.runtime.Recognizer;
import org.antlr.v4.runtime.misc.ParseCancellationException;
+import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.exception.write.WriteProcessException;
-import org.apache.tsfile.external.commons.collections4.CollectionUtils;
-import org.apache.tsfile.external.commons.lang3.ObjectUtils;
-import org.apache.tsfile.external.commons.lang3.StringUtils;
import org.apache.tsfile.file.metadata.ColumnSchema;
import org.apache.tsfile.file.metadata.ColumnSchemaBuilder;
-import org.apache.tsfile.fileSystem.FSFactoryProducer;
-import org.apache.tsfile.read.common.Field;
import org.apache.tsfile.read.common.RowRecord;
import org.apache.tsfile.write.record.Tablet;
-import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.v4.ITsFileWriter;
import org.apache.tsfile.write.v4.TsFileWriterBuilder;
-import java.io.File;
-import java.io.FileWriter;
-import java.io.IOException;
import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
-import java.util.stream.Collectors;
public class ExportDataTable extends AbstractExportData {
+ private ITableSession session;
- private static final IoTPrinter ioTPrinter = new IoTPrinter(System.out);
- private static ITableSession tableSession;
- private static List<String> tables = new ArrayList<>();
- private static List<String> allTables = new ArrayList<>();
- private static long processedRows;
- private static long lastPrintTime;
+ ExportDataTable(ExportConfig config, ExportProgressReporter progress) {
+ super(config, progress);
+ }
+
+ ExportDataTable(ExportConfig config, ExportProgressReporter progress, ITableSession session)
+ throws Exception {
+ this(config, progress);
+ this.session = own(session).get();
+ zoneId = ZoneId.of("UTC");
+ timestampPrecision = "ms";
+ }
@Override
- public void init() throws IoTDBConnectionException, StatementExecutionException {
- TableSessionBuilder tableSessionBuilder =
- new TableSessionBuilder()
- .nodeUrls(Collections.singletonList(UrlUtils.formatTEndPointIpv4AndIpv6Url(host, port)))
- .username(username)
- .password(password)
- .database(database)
- .thriftMaxFrameSize(rpcMaxFrameSize);
- if (useSsl) {
- tableSessionBuilder = configureSsl(tableSessionBuilder);
+ public void init() throws Exception {
+ // TableSession does not expose precision or timezone; retain the existing metadata handshake.
+ try (Resource<Session> metadata = own(config.newTreeSession())) {
+ readTimeSettings(metadata.get());
}
- tableSession = tableSessionBuilder.build();
- SessionDataSet sessionDataSet = tableSession.executeQueryStatement("show databases", timeout);
- List<String> databases = new ArrayList<>();
- while (sessionDataSet.hasNext()) {
- databases.add(sessionDataSet.next().getField(0).getStringValue());
- }
- if (CollectionUtils.isEmpty(databases) || !databases.contains(database)) {
- ioTPrinter.println(String.format(Constants.TARGET_DATABASE_NOT_EXIST_MSG, database));
- System.exit(1);
- }
- sessionDataSet = tableSession.executeQueryStatement("show tables", timeout);
- allTables.clear();
- while (sessionDataSet.hasNext()) {
- allTables.add(sessionDataSet.next().getField(0).getStringValue());
+ TableSessionBuilder builder =
+ new TableSessionBuilder()
+ .nodeUrls(
+ Collections.singletonList(
+ UrlUtils.formatTEndPointIpv4AndIpv6Url(config.host, config.port)))
+ .username(config.username)
+ .password(config.password)
+ .database(config.database)
+ .zoneId(zoneId)
+ .thriftMaxFrameSize(config.rpcMaxFrameSize);
+ config.configureSsl(builder);
+ session = own(builder.build()).get();
+ if (!names("show databases").contains(config.database)) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages.EXCEPTION_THE_TARGET_DATABASE_ARG_DOES_NOT_EXIST_80E096EB,
+ config.database));
}
- boolean filterByTable = StringUtils.isBlank(queryCommand) && ObjectUtils.isNotEmpty(table);
- if (CollectionUtils.isEmpty(allTables) || (filterByTable && !allTables.contains(table))) {
- ioTPrinter.println(String.format(Constants.TARGET_TABLE_NOT_EXIST_MSG, table));
- System.exit(1);
+ }
+
+ List<ExportTask> planTables(ExportTimeRange range) throws Exception {
+ List<String> tables = names("show tables");
+ if (tables.isEmpty() || (config.table != null && !tables.contains(config.table))) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages
+ .EXCEPTION_THERE_ARE_NO_TABLES_OR_THE_TARGET_TABLE_ARG_DOES_NOT_EXIST_E5F88A98,
+ config.table));
}
- tables.clear();
- if (filterByTable) {
- tables.add(table);
- } else {
- tables.addAll(allTables);
+ if (config.table != null) {
+ tables = Collections.singletonList(config.table);
}
- if (ObjectUtils.isNotEmpty(sessionDataSet)) {
- sessionDataSet.close();
+ List<ExportTask> tasks = new ArrayList<>();
+ for (String table : tables) {
+ tasks.add(
+ new ExportTask(
+ "SELECT * FROM " + identifier(table) + range.whereClause(),
+ config.targetDirectory.resolve(config.prefix + tasks.size()),
+ null,
+ table));
}
+ return tasks;
}
- @Override
- public boolean exportBySql(String sql, int index) {
- List<String> exportSql = new ArrayList<>();
- boolean customQuery = StringUtils.isNotBlank(sql);
- boolean success = true;
- if (customQuery) {
- if (Constants.SQL_SUFFIXS.equalsIgnoreCase(exportType)
- || Constants.TSFILE_SUFFIXS.equalsIgnoreCase(exportType)) {
- legalCheck(sql);
- }
- exportSql.add(sql);
- } else {
- StringBuilder sqlBuilder;
- for (String table : tables) {
- sqlBuilder = new StringBuilder("select * from ").append(quoteIdentifier(table));
- if (StringUtils.isNotBlank(startTime) || StringUtils.isNotBlank(endTime)) {
- sqlBuilder.append(" where ");
- if (StringUtils.isNotBlank(startTime)) {
- sqlBuilder.append("time >= ").append(startTime);
- }
- if (StringUtils.isNotBlank(startTime) && StringUtils.isNotBlank(endTime)) {
- sqlBuilder.append(" and ");
- }
- if (StringUtils.isNotBlank(endTime)) {
- sqlBuilder.append("time <= ").append(endTime);
- }
- }
- exportSql.add(sqlBuilder.toString());
- }
+ /** SQL and TsFile output are written to the table each query reads; -table does not apply. */
+ List<ExportTask> planQueries(List<ExportTask> tasks) throws Exception {
+ if ("csv".equals(config.format)) {
+ return tasks;
}
- for (int i = 0; i < exportSql.size(); i++) {
- String path = targetDirectory + targetFile + (customQuery ? index : i);
- try {
- String queryTable =
- customQuery
- ? (Constants.SQL_SUFFIXS.equalsIgnoreCase(exportType)
- || Constants.TSFILE_SUFFIXS.equalsIgnoreCase(exportType)
- ? getQueryTable(sql)
- : null)
- : tables.get(i);
- try (SessionDataSet sessionDataSet =
- tableSession.executeQueryStatement(exportSql.get(i), timeout)) {
- if (Constants.SQL_SUFFIXS.equalsIgnoreCase(exportType)) {
- exportToSqlFile(sessionDataSet, queryTable, path);
- } else if (Constants.TSFILE_SUFFIXS.equalsIgnoreCase(exportType)) {
- long start = System.currentTimeMillis();
- boolean isComplete = exportToTsFile(sessionDataSet, path + ".tsfile", queryTable);
- if (isComplete) {
- long end = System.currentTimeMillis();
- ioTPrinter.println("Export completely!cost: " + (end - start) + " ms.");
- }
- } else {
- exportToCsvFile(sessionDataSet, path);
- }
- sessionDataSet.closeOperationHandle();
- ioTPrinter.println(Constants.EXPORT_COMPLETELY);
- }
- } catch (StatementExecutionException
- | IoTDBConnectionException
- | IOException
- | WriteProcessException e) {
- ioTPrinter.println("Cannot dump result because: " + e.getMessage());
- success = false;
- }
+ // Every query is checked before the first one runs, so a rejected query writes no files.
+ List<String> tables = names("show tables");
+ List<ExportTask> planned = new ArrayList<>(tasks.size());
+ for (ExportTask task : tasks) {
+ planned.add(
+ new ExportTask(task.sql(), task.outputBase(), null, queryTable(task.sql(), tables)));
}
- return success;
+ return planned;
}
- private static String getQueryTable(String sql) throws StatementExecutionException {
- String message =
- String.format(
- CliMessages
- .EXCEPTION_SQL_TSFILE_EXPORT_REQUIRES_A_QUERY_FROM_A_SINGLE_TABLE_IN_DATABASE_ARG_C9EB3768,
- database);
+ private String queryTable(String sql, List<String> tables) throws ArgsErrorException {
+ ArgsErrorException invalid =
+ new ArgsErrorException(
+ String.format(
+ CliMessages
+ .EXCEPTION_SQL_TSFILE_EXPORT_REQUIRES_A_QUERY_FROM_A_SINGLE_TABLE_IN_DATABASE_ARG_C9EB3768,
+ config.database));
RelationalSqlLexer lexer =
new RelationalSqlLexer(new CaseInsensitiveStream(CharStreams.fromString(sql)));
- RelationalSqlParser parser = new RelationalSqlParser(new CommonTokenStream(lexer));
lexer.removeErrorListeners();
lexer.addErrorListener(
new BaseErrorListener() {
@@ -205,11 +156,12 @@ public class ExportDataTable extends AbstractExportData {
Object offendingSymbol,
int line,
int charPositionInLine,
- String error,
+ String message,
RecognitionException cause) {
throw new ParseCancellationException(message);
}
});
+ RelationalSqlParser parser = new RelationalSqlParser(new CommonTokenStream(lexer));
parser.removeErrorListeners();
parser.setErrorHandler(new BailErrorStrategy());
List<RelationalSqlParser.QualifiedNameContext> sources = new ArrayList<>();
@@ -222,22 +174,22 @@ public class ExportDataTable extends AbstractExportData {
}
}.visit(parser.singleStatement());
} catch (ParseCancellationException e) {
- throw new StatementExecutionException(message);
+ throw invalid;
}
- // Multiple source relations do not have an unambiguous destination table.
+ // A query reading several relations has no single destination table.
if (sources.size() != 1) {
- throw new StatementExecutionException(message);
+ throw invalid;
}
- List<RelationalSqlParser.IdentifierContext> names = sources.get(0).identifier();
- if (names.size() > 2
- || (names.size() == 2 && !identifierValue(names.get(0)).equalsIgnoreCase(database))) {
- throw new StatementExecutionException(message);
+ List<RelationalSqlParser.IdentifierContext> parts = sources.get(0).identifier();
+ if (parts.size() > 2
+ || (parts.size() == 2 && !identifierValue(parts.get(0)).equals(config.database))) {
+ throw invalid;
}
- String name = identifierValue(names.get(names.size() - 1));
- if (!allTables.contains(name)) {
- throw new StatementExecutionException(message);
+ String table = identifierValue(parts.get(parts.size() - 1));
+ if (!tables.contains(table)) {
+ throw invalid;
}
- return name;
+ return table;
}
private static String identifierValue(RelationalSqlParser.IdentifierContext context) {
@@ -250,239 +202,236 @@ public class ExportDataTable extends AbstractExportData {
return value.toLowerCase(Locale.ROOT);
}
- private static String quoteIdentifier(String name) {
- return "\"" + name.replace("\"", "\"\"") + "\"";
- }
-
- private void exportToSqlFile(SessionDataSet sessionDataSet, String table, String filePath)
- throws IOException, IoTDBConnectionException, StatementExecutionException {
- processedRows = 0;
- lastPrintTime = 0;
- StringBuilder sqlBuilder;
- List<String> headers = sessionDataSet.getColumnNames();
- String prevSql =
- "insert into "
- + quoteIdentifier(table)
- + "("
- + headers.stream()
- .map(ExportDataTable::quoteIdentifier)
- .collect(Collectors.joining(","))
- + ") values(";
- SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
- List<String> columnTypeList = iterator.getColumnTypeList();
- int totalColumns = columnTypeList.size();
- int fileIndex = 0;
- boolean fromOuterLoop = false;
- while (iterator.next()) {
- final String finalFilePath = filePath + "_" + fileIndex + ".sql";
- int countLine = 0;
- fromOuterLoop = true;
- try (FileWriter writer = new FileWriter(finalFilePath)) {
- while (countLine++ < linesPerFile && (fromOuterLoop || iterator.next())) {
- fromOuterLoop = false;
- sqlBuilder = new StringBuilder();
- sqlBuilder.append(prevSql);
- for (int curColumnIndex = 0; curColumnIndex < totalColumns; curColumnIndex++) {
- if (curColumnIndex > 0) {
- sqlBuilder.append(",");
- }
- String columnType = columnTypeList.get(curColumnIndex);
- String columnValue = iterator.getString(curColumnIndex + 1);
- sqlBuilder.append(formatSqlValue(columnValue, columnType));
- }
- sqlBuilder.append(");\n");
- writer.write(sqlBuilder.toString());
- processedRows += 1;
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
- }
- }
- writer.flush();
- fileIndex++;
+ private List<String> names(String sql) throws Exception {
+ List<String> names = new ArrayList<>();
+ try (Resource<SessionDataSet> result =
+ own(session.executeQueryStatement(sql, config.timeout))) {
+ while (result.get().hasNext()) {
+ progress.checkActive();
+ names.add(result.get().next().getField(0).getStringValue());
}
}
- ioTPrinter.print("\n");
+ return names;
}
- private Boolean exportToTsFile(SessionDataSet sessionDataSet, String filePath, String table)
- throws IOException,
- IoTDBConnectionException,
- StatementExecutionException,
- WriteProcessException {
- processedRows = 0;
- lastPrintTime = 0;
- List<String> columnNamesRaw = sessionDataSet.getColumnNames();
- List<TSDataType> columnTypesRaw =
- sessionDataSet.getColumnTypes().stream().map(t -> getType(t)).collect(Collectors.toList());
- File f = FSFactoryProducer.getFSFactory().getFile(filePath);
- if (f.exists()) {
- Files.delete(f.toPath());
+ @Override
+ void export(ExportTask task) throws Exception {
+ if (!"csv".equals(config.format)) {
+ legalCheck(task.sql());
}
- boolean isEmpty = false;
- Map<String, Integer> deviceColumnIndices = new HashMap<>();
- List<ColumnSchema> columnSchemas = collectSchemas(columnNamesRaw, deviceColumnIndices, table);
- try (ITsFileWriter tsFileWriter =
- new TsFileWriterBuilder()
- .file(f)
- .tableSchema(new org.apache.tsfile.file.metadata.TableSchema(table, columnSchemas))
- .memoryThreshold(Constants.memoryThreshold)
- .build()) {
- List<String> columnNames = new ArrayList<>(columnNamesRaw);
- List<TSDataType> columnTypes = new ArrayList<>(columnTypesRaw);
- int timeIndex = columnNamesRaw.indexOf(timeColumn.toLowerCase());
- if (timeIndex >= 0) {
- columnNames.remove(timeIndex);
- columnTypes.remove(timeIndex);
+ progress.taskStarted(task);
+ try (Resource<SessionDataSet> result =
+ own(session.executeQueryStatement(task.sql(), config.timeout))) {
+ switch (config.format) {
+ case "sql":
+ exportSql(result.get(), task);
+ break;
+ case "tsfile":
+ exportTsFile(result.get(), task);
+ break;
+ default:
+ exportCsv(result.get(), task);
}
- Tablet tablet = new Tablet(columnNames, columnTypes);
- if (ObjectUtils.isNotEmpty(tablet)) {
- writeWithTablets(sessionDataSet, tablet, deviceColumnIndices, tsFileWriter);
- } else {
- isEmpty = true;
- }
- }
- if (isEmpty) {
- ioTPrinter.println("!!!Warning:Tablet is empty,no data can be exported.");
- return false;
}
- return true;
+ progress.taskCompleted();
}
- private void exportToCsvFile(SessionDataSet sessionDataSet, String filePath)
- throws IOException, IoTDBConnectionException, StatementExecutionException {
- processedRows = 0;
- lastPrintTime = 0;
- List<String> headers = getCsvHeaders(sessionDataSet, false);
- int fileIndex = 0;
- SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
- List<String> columnTypeList = iterator.getColumnTypeList();
- int totalColumns = columnTypeList.size();
- boolean fromOuterloop = false;
- while (iterator.next()) {
- final String finalFilePath = filePath + "_" + fileIndex + ".csv";
- final CSVPrinterWrapper csvPrinterWrapper = new CSVPrinterWrapper(finalFilePath);
- try {
- csvPrinterWrapper.printRecord(headers);
- fromOuterloop = true;
- int countLine = 0;
- while (countLine++ < linesPerFile && (fromOuterloop || iterator.next())) {
- fromOuterloop = false;
- for (int curColumnIndex = 0; curColumnIndex < totalColumns; curColumnIndex++) {
- String curType = columnTypeList.get(curColumnIndex);
- if (curType.equalsIgnoreCase("TIMESTAMP")) {
- csvPrinterWrapper.print(timeTrans(iterator.getLong(curColumnIndex + 1)));
- } else {
- if (iterator.isNull(curColumnIndex + 1)) {
- csvPrinterWrapper.print("");
- } else {
- String columnValue = iterator.getString(curColumnIndex + 1);
- if (curType.equalsIgnoreCase("TEXT") || curType.equalsIgnoreCase("STRING")) {
- csvPrinterWrapper.print("\"" + columnValue + "\"");
- } else {
- csvPrinterWrapper.print(columnValue);
- }
- }
+ private void exportCsv(SessionDataSet data, ExportTask task) throws Exception {
+ List<String> names = data.getColumnNames();
+ SessionDataSet.DataIterator iterator = data.iterator();
+ List<String> types = iterator.getColumnTypeList();
+ try (Resource<ExportTextWriter> output =
+ own(new ExportTextWriter(config, task, progress, csvHeaders(names, types, false)))) {
+ while (iterator.next()) {
+ progress.checkActive();
+ List<Object> values = new ArrayList<>(names.size());
+ for (int i = 0; i < names.size(); i++) {
+ if (iterator.isNull(i + 1)) {
+ values.add("");
+ } else if ("TIMESTAMP".equals(types.get(i))) {
+ values.add(timeTrans(iterator.getLong(i + 1)));
+ } else {
+ String value = iterator.getString(i + 1);
+ if ("TEXT".equals(types.get(i)) || "STRING".equals(types.get(i))) {
+ value = '"' + value + '"';
}
- }
- csvPrinterWrapper.println();
- processedRows += 1;
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
+ values.add(value);
}
}
- fileIndex++;
- csvPrinterWrapper.flush();
- } finally {
- csvPrinterWrapper.close();
+ output.get().csvRow(values);
}
+ output.get().complete();
}
- ioTPrinter.print("\n");
}
- private static void writeWithTablets(
- SessionDataSet sessionDataSet,
- Tablet tablet,
- Map<String, Integer> deviceColumnIndices,
- ITsFileWriter tsFileWriter)
- throws IoTDBConnectionException,
- StatementExecutionException,
- IOException,
- WriteProcessException {
- while (sessionDataSet.hasNext()) {
- RowRecord rowRecord = sessionDataSet.next();
- List<Field> fields = rowRecord.getFields();
- int rowIndex = tablet.getRowSize();
- tablet.addTimestamp(
- rowIndex, fields.get(deviceColumnIndices.get(timeColumn.toLowerCase())).getLongV());
- List<IMeasurementSchema> schemas = tablet.getSchemas();
- for (int i = 0; i < schemas.size(); i++) {
- IMeasurementSchema measurementSchema = schemas.get(i);
- // -1 for time not in fields
- final String measurementName = measurementSchema.getMeasurementName();
- if (timeColumn.equalsIgnoreCase(measurementName)) {
- continue;
+ private void exportSql(SessionDataSet data, ExportTask task) throws Exception {
+ validateColumns(data, task.table());
+ List<String> headers = data.getColumnNames();
+ SessionDataSet.DataIterator iterator = data.iterator();
+ List<String> types = iterator.getColumnTypeList();
+ List<String> quotedHeaders = new ArrayList<>();
+ for (String header : headers) {
+ quotedHeaders.add(identifier(header));
+ }
+ String prefix =
+ "insert into "
+ + identifier(task.table())
+ + "("
+ + String.join(",", quotedHeaders)
+ + ") values(";
+ try (Resource<ExportTextWriter> output =
+ own(new ExportTextWriter(config, task, progress, null))) {
+ while (iterator.next()) {
+ progress.checkActive();
+ StringBuilder sql = new StringBuilder(prefix);
+ for (int i = 0; i < headers.size(); i++) {
+ if (i > 0) {
+ sql.append(',');
+ }
+ sql.append(sqlValue(iterator, types.get(i), i + 1));
}
- Object value =
- fields
- .get(deviceColumnIndices.get(measurementName))
- .getObjectValue(measurementSchema.getType());
- tablet.addValue(measurementName, rowIndex, value);
+ output.get().sqlRow(sql.append(");\n").toString());
}
+ output.get().complete();
+ }
+ }
- if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
- writeToTsFile(tsFileWriter, tablet);
- processedRows += tablet.getRowSize();
- tablet.initBitMaps();
- tablet.reset();
- }
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
+ private Map<String, ColumnSchema> validateColumns(SessionDataSet data, String table)
+ throws Exception {
+ Map<String, ColumnSchema> schemas = new LinkedHashMap<>();
+ try (Resource<SessionDataSet> result =
+ own(session.executeQueryStatement("DESCRIBE " + identifier(table), config.timeout))) {
+ while (result.get().hasNext()) {
+ progress.checkActive();
+ RowRecord row = result.get().next();
+ String name = row.getField(0).getStringValue();
+ if (!"time".equals(name)) {
+ schemas.put(
+ name,
+ new ColumnSchemaBuilder()
+ .name(name)
+ .dataType(TSDataType.valueOf(row.getField(1).getStringValue()))
+ .category(ColumnCategory.valueOf(row.getField(2).getStringValue()))
+ .build());
+ }
}
}
- if (tablet.getRowSize() != 0) {
- writeToTsFile(tsFileWriter, tablet);
- processedRows += tablet.getRowSize();
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
+ List<String> columns = data.getColumnNames();
+ List<String> types = data.getColumnTypes();
+ if (!columns.contains("time")) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages
+ .EXCEPTION_SQL_TSFILE_OUTPUT_REQUIRES_THE_TIME_COLUMN_AND_UNALIASED_COLUMNS_FROM_TABLE_ARG_C5E77CD3,
+ table));
+ }
+ for (int i = 0; i < columns.size(); i++) {
+ String column = columns.get(i);
+ if ("time".equals(column)) {
+ continue;
+ }
+ ColumnSchema schema = schemas.get(column);
+ if (schema == null || schema.getDataType() != TSDataType.valueOf(types.get(i))) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages
+ .EXCEPTION_SQL_TSFILE_OUTPUT_REQUIRES_THE_TIME_COLUMN_AND_UNALIASED_COLUMNS_FROM_TABLE_ARG_C5E77CD3,
+ table));
}
}
- ioTPrinter.print("\n");
- }
-
- private static void writeToTsFile(ITsFileWriter tsFileWriter, Tablet tablet)
- throws IOException, WriteProcessException {
- tsFileWriter.write(tablet);
+ return schemas;
}
- private List<ColumnSchema> collectSchemas(
- List<String> columnNames, Map<String, Integer> deviceColumnIndices, String table)
- throws IoTDBConnectionException, StatementExecutionException {
- List<ColumnSchema> columnSchemas = new ArrayList<>();
- SessionDataSet sessionDataSet =
- tableSession.executeQueryStatement("describe " + quoteIdentifier(table));
- while (sessionDataSet.hasNext()) {
- RowRecord rowRecord = sessionDataSet.next();
- final String columnName = rowRecord.getField(0).getStringValue();
- if (timeColumn.equalsIgnoreCase(columnName)) {
+ private void exportTsFile(SessionDataSet data, ExportTask task) throws Exception {
+ Map<String, ColumnSchema> available = validateColumns(data, task.table());
+ List<String> columns = data.getColumnNames();
+ List<ColumnSchema> schemas = new ArrayList<>();
+ List<String> names = new ArrayList<>();
+ List<TSDataType> types = new ArrayList<>();
+ List<ColumnCategory> categories = new ArrayList<>();
+ List<Integer> indices = new ArrayList<>();
+ for (int i = 0; i < columns.size(); i++) {
+ String column = columns.get(i);
+ if ("time".equals(column)) {
continue;
}
- columnSchemas.add(
- new ColumnSchemaBuilder()
- .name(columnName)
- .dataType(getType(rowRecord.getField(1).getStringValue()))
- .category(getColumnCategory(rowRecord.getField(2).getStringValue()))
- .build());
+ ColumnSchema schema = available.get(column);
+ schemas.add(schema);
+ names.add(column);
+ types.add(schema.getDataType());
+ categories.add(schema.getColumnCategory());
+ indices.add(i + 1);
}
- for (int i = 0; i < columnNames.size(); i++) {
- deviceColumnIndices.put(columnNames.get(i), i);
+ // All tags are needed to preserve device identity in a TsFile.
+ for (ColumnSchema schema : available.values()) {
+ if (schema.getColumnCategory() == ColumnCategory.TAG
+ && !names.contains(schema.getColumnName())) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages
+ .EXCEPTION_SQL_TSFILE_OUTPUT_REQUIRES_THE_TIME_COLUMN_AND_UNALIASED_COLUMNS_FROM_TABLE_ARG_C5E77CD3,
+ task.table()));
+ }
}
- if (ObjectUtils.isNotEmpty(sessionDataSet)) {
- sessionDataSet.close();
+ Path file = Path.of(task.outputBase() + ".tsfile");
+ ExportFileNames.checkOutputCollision(file, config);
+ progress.fileStarted(file);
+ Files.deleteIfExists(file);
+ long startRows = progress.rows();
+ try (Resource<ITsFileWriter> output =
+ own(
+ new TsFileWriterBuilder()
+ .file(file.toFile())
+ .tableSchema(new org.apache.tsfile.file.metadata.TableSchema(task.table(), schemas))
+ .memoryThreshold(Constants.memoryThreshold)
+ .build())) {
+ Tablet tablet = new Tablet(task.table(), names, types, categories);
+ tablet.initBitMaps();
+ SessionDataSet.DataIterator iterator = data.iterator();
+ int time = columns.indexOf("time") + 1;
+ while (iterator.next()) {
+ progress.checkActive();
+ int row = tablet.getRowSize();
+ tablet.addTimestamp(row, iterator.getLong(time));
+ for (int i = 0; i < names.size(); i++) {
+ int column = indices.get(i);
+ Object value = null;
+ if (!iterator.isNull(column)) {
+ switch (types.get(i)) {
+ case TIMESTAMP:
+ value = iterator.getLong(column);
+ break;
+ case DATE:
+ value = iterator.getDate(column);
+ break;
+ case BLOB:
+ value = iterator.getBlob(column);
+ break;
+ default:
+ value = iterator.getObject(column);
+ }
+ }
+ tablet.addValue(names.get(i), row, value);
+ }
+ if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
+ writeTablet(output.get(), tablet);
+ }
+ }
+ if (tablet.getRowSize() > 0) {
+ writeTablet(output.get(), tablet);
+ }
}
- return columnSchemas;
+ progress.fileCompleted(file, progress.rows() - startRows);
+ }
+
+ private void writeTablet(ITsFileWriter writer, Tablet tablet) throws Exception {
+ writer.write(tablet);
+ progress.rowsWritten(tablet.getRowSize());
+ tablet.reset();
+ tablet.initBitMaps();
+ }
+
+ private static String identifier(String name) {
+ return "\"" + name.replace("\"", "\"\"") + "\"";
}
}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java
index bab4fa88b92..65c5d13dfdd 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java
@@ -19,21 +19,15 @@
package org.apache.iotdb.tool.data;
-import org.apache.iotdb.cli.utils.IoTPrinter;
-import org.apache.iotdb.isession.SessionConfig;
+import org.apache.iotdb.cli.i18n.CliMessages;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.exception.ArgsErrorException;
import org.apache.iotdb.isession.SessionDataSet;
-import org.apache.iotdb.rpc.IoTDBConnectionException;
-import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.Session;
-import org.apache.iotdb.tool.common.Constants;
-import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.exception.write.WriteProcessException;
-import org.apache.tsfile.external.commons.collections4.CollectionUtils;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.fileSystem.FSFactoryProducer;
import org.apache.tsfile.read.common.Field;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
@@ -42,16 +36,13 @@ import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
-import java.io.File;
-import java.io.FileWriter;
-import java.io.IOException;
import java.nio.file.Files;
-import java.time.ZoneId;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
@@ -59,420 +50,398 @@ import static org.apache.iotdb.commons.schema.SchemaConstant.AUDIT_DATABASE;
import static org.apache.iotdb.commons.schema.SchemaConstant.SYSTEM_DATABASE;
public class ExportDataTree extends AbstractExportData {
+ private Session session;
+ private final Map<String, Boolean> alignment = new HashMap<>();
+ private final Map<String, ExportTask.Measurement> measurements = new LinkedHashMap<>();
- private static final IoTPrinter ioTPrinter = new IoTPrinter(System.out);
- private static Session session;
- private static long processedRows;
- private static long lastPrintTime;
+ ExportDataTree(ExportConfig config, ExportProgressReporter progress) {
+ super(config, progress);
+ }
- @Override
- public void init() throws IoTDBConnectionException, StatementExecutionException, TException {
- Session.Builder sessionBuilder =
- new Session.Builder()
- .host(host)
- .port(Integer.parseInt(port))
- .username(username)
- .password(password)
- .fetchSize(SessionConfig.DEFAULT_FETCH_SIZE)
- .zoneId(null)
- .thriftDefaultBufferSize(SessionConfig.DEFAULT_INITIAL_BUFFER_CAPACITY)
- .thriftMaxFrameSize(rpcMaxFrameSize)
- .enableRedirection(SessionConfig.DEFAULT_REDIRECTION_MODE)
- .version(SessionConfig.DEFAULT_VERSION);
- if (useSsl) {
- sessionBuilder = configureSsl(sessionBuilder);
- }
- session = sessionBuilder.build();
- session.open(false);
- timestampPrecision = session.getTimestampPrecision();
- if (timeZoneID != null) {
- session.setTimeZone(timeZoneID);
- }
- zoneId = ZoneId.of(session.getTimeZone());
+ // A connected Session can be supplied for output regression tests.
+ ExportDataTree(ExportConfig config, ExportProgressReporter progress, Session session)
+ throws Exception {
+ this(config, progress);
+ this.session = own(session).get();
+ zoneId = java.time.ZoneId.of("UTC");
+ timestampPrecision = "ms";
}
@Override
- public boolean exportBySql(String sql, int index) {
- if (Constants.SQL_SUFFIXS.equalsIgnoreCase(exportType)
- || Constants.TSFILE_SUFFIXS.equalsIgnoreCase(exportType)) {
- legalCheck(sql);
- }
- final String path = targetDirectory + targetFile + index;
- try (SessionDataSet sessionDataSet = session.executeQueryStatement(sql, timeout)) {
- if (Constants.SQL_SUFFIXS.equalsIgnoreCase(exportType)) {
- exportToSqlFile(sessionDataSet, path);
- } else if (Constants.TSFILE_SUFFIXS.equalsIgnoreCase(exportType)) {
- long start = System.currentTimeMillis();
- boolean isComplete = exportToTsFile(sessionDataSet, path + ".tsfile");
- if (isComplete) {
- long end = System.currentTimeMillis();
- ioTPrinter.println("Export completely!cost: " + (end - start) + " ms.");
+ public void init() throws Exception {
+ session = own(config.newTreeSession()).get();
+ readTimeSettings(session);
+ }
+
+ List<ExportTask> planMeasurements(MeasurementFile input, ExportTimeRange range) throws Exception {
+ progress.phase(ExportProgressReporter.Phase.DISCOVERING);
+ long matches = 0;
+ int unmatched = 0;
+ for (MeasurementFile.PatternLine pattern : input.patterns()) {
+ progress.checkActive();
+ long patternMatches = 0;
+ try (Resource<SessionDataSet> result =
+ own(
+ session.executeQueryStatement(
+ "show timeseries " + pattern.path().getFullPath(), config.timeout))) {
+ SessionDataSet data = result.get();
+ List<String> columns = data.getColumnNames();
+ while (data.hasNext()) {
+ progress.checkActive();
+ ExportTask.Measurement measurement = measurement(data.next(), columns);
+ measurements.putIfAbsent(measurement.path().getFullPath(), measurement);
+ patternMatches++;
}
- } else {
- exportToCsvFile(sessionDataSet, path);
}
- sessionDataSet.closeOperationHandle();
- ioTPrinter.println(Constants.EXPORT_COMPLETELY);
- return true;
- } catch (StatementExecutionException
- | IoTDBConnectionException
- | IOException
- | WriteProcessException e) {
- ioTPrinter.println("Cannot dump result because: " + e.getMessage());
- return false;
+ matches += patternMatches;
+ if (patternMatches == 0) {
+ unmatched++;
+ // The message names the pattern; the log also records which file it came from.
+ progress.warning(
+ String.format(
+ Locale.ROOT,
+ CliMessages.LOG_NO_MEASUREMENTS_MATCHED_ARG_MEASUREMENT_FILE_LINE_ARG_94C4EF10,
+ pattern.original(),
+ pattern.line()),
+ "source",
+ input.source(),
+ "line",
+ pattern.line(),
+ "pattern",
+ pattern.original());
+ }
+ }
+ progress.discovered(input, matches, measurements.size(), unmatched);
+ if (measurements.isEmpty()) {
+ throw new ArgsErrorException(
+ CliMessages
+ .EXCEPTION_NO_MEASUREMENTS_REMAIN_AFTER_PATTERN_EXPANSION_AND_DEDUPLICATION_40D930B7);
+ }
+ ExportFileNames names = new ExportFileNames();
+ List<ExportTask> tasks = new ArrayList<>(measurements.size());
+ for (ExportTask.Measurement measurement : measurements.values()) {
+ PartialPath path = measurement.path();
+ java.nio.file.Path output = names.measurement(config, path.getFullPath());
+ // Validate collisions before the first data query. Split paths are also checked when opened.
+ ExportFileNames.checkOutputCollision(
+ java.nio.file.Path.of(
+ output + ("tsfile".equals(config.format) ? ".tsfile" : "_0." + config.format)),
+ config);
+ String sql =
+ "SELECT "
+ + path.getMeasurement()
+ + " FROM "
+ + devicePath(path).getFullPath()
+ + range.whereClause();
+ tasks.add(new ExportTask(sql, output, measurement, null));
}
+ return tasks;
}
- private void exportToSqlFileWithAlignDevice(SessionDataSet sessionDataSet, String filePath)
- throws IOException, IoTDBConnectionException, StatementExecutionException {
- processedRows = 0;
- lastPrintTime = 0;
-
- List<String> measurementNames = new ArrayList<>(sessionDataSet.getColumnNames());
- if (CollectionUtils.isEmpty(measurementNames) || measurementNames.size() <= 1) {
- return;
- } else {
- measurementNames.remove("Time");
- measurementNames.remove("Device");
- }
- String sqlPrefix =
- String.format(
- "INSERT INTO %s(TIMESTAMP,%s) ALIGNED VALUES (%s);\n",
- "%s", String.join(",", measurementNames), "%d,%s");
+ private static PartialPath devicePath(PartialPath measurement) {
+ String[] nodes = measurement.getNodes();
+ return new PartialPath(java.util.Arrays.copyOf(nodes, nodes.length - 1));
+ }
- SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
- List<String> columnTypeList = iterator.getColumnTypeList();
- int totalColumns = columnTypeList.size();
- String deviceName = null;
- int fileIndex = 0;
- int currentLines = 0;
- String filePathTemplate = filePath + "_%d" + ".sql";
- FileWriter writer = null;
- while (iterator.next()) {
- if (writer == null) {
- writer = new FileWriter(String.format(filePathTemplate, fileIndex));
- }
- deviceName = iterator.getString(2);
- if (deviceName.startsWith(SYSTEM_DATABASE + ".")
- || deviceName.startsWith(AUDIT_DATABASE + ".")) {
- continue;
- }
- List<String> values = new ArrayList<>();
- for (int index = 2; index < totalColumns; index++) {
- values.add(formatSqlValue(iterator.getString(index + 1), columnTypeList.get(index)));
- }
- long timestamp = iterator.getLong(1);
- writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values)));
- processedRows += 1;
- currentLines += 1;
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
- }
- if (currentLines >= linesPerFile) {
- writer.flush();
- writer.close();
- fileIndex += 1;
- writer = null;
- currentLines = 0;
- }
+ private ExportTask.Measurement measurement(RowRecord record, List<String> columns)
+ throws Exception {
+ List<Field> fields = record.getFields();
+ PartialPath path = new PartialPath(fields.get(columns.indexOf("Timeseries")).getStringValue());
+ TSDataType type =
+ AbstractDataTool.getType(fields.get(columns.indexOf("DataType")).getStringValue());
+ MeasurementSchema schema = new MeasurementSchema(path.getMeasurement(), type);
+ String encoding = fields.get(columns.indexOf("Encoding")).getStringValue();
+ String compression = fields.get(columns.indexOf("Compression")).getStringValue();
+ // Logical views may omit physical encoding and compression metadata.
+ if (encoding != null && !"null".equals(encoding)) {
+ schema.setEncoding(TSEncoding.valueOf(encoding));
}
- if (writer != null) {
- writer.flush();
- writer.close();
+ if (compression != null && !"null".equals(compression)) {
+ schema.setCompressionType(CompressionType.valueOf(compression));
}
- ioTPrinter.print("\n");
+ return new ExportTask.Measurement(path, schema);
}
- private void exportToSqlFileWithoutAlign(SessionDataSet sessionDataSet, String filePath)
- throws IoTDBConnectionException, StatementExecutionException, IOException {
- processedRows = 0;
- lastPrintTime = 0;
- List<String> headers = sessionDataSet.getColumnNames();
- Map<String, List<Integer>> deviceColumnIndices = new LinkedHashMap<>();
- Map<Integer, String> measurements = new HashMap<>();
- for (int i = 1; i < headers.size(); i++) {
- Path path = new Path(headers.get(i), true);
- String device = path.getDeviceString();
- if (device.startsWith(SYSTEM_DATABASE + ".") || device.startsWith(AUDIT_DATABASE + ".")) {
- continue;
+ @Override
+ void export(ExportTask task) throws Exception {
+ if (!"csv".equals(config.format) && task.measurement() == null) {
+ legalCheck(task.sql());
+ }
+ progress.taskStarted(task);
+ try (Resource<SessionDataSet> result =
+ own(session.executeQueryStatement(task.sql(), config.timeout))) {
+ SessionDataSet data = result.get();
+ switch (config.format) {
+ case "sql":
+ exportSql(data, task);
+ break;
+ case "tsfile":
+ exportTsFile(data, task);
+ break;
+ default:
+ exportCsv(data, task);
}
- deviceColumnIndices.computeIfAbsent(device, key -> new ArrayList<>()).add(i);
- measurements.put(i, path.getMeasurement());
}
+ progress.taskCompleted();
+ }
- SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
- List<String> columnTypes = iterator.getColumnTypeList();
- int fileIndex = 0;
- int currentLines = 0;
- FileWriter writer = null;
- try {
+ private void exportCsv(SessionDataSet data, ExportTask task) throws Exception {
+ List<String> names = data.getColumnNames();
+ SessionDataSet.DataIterator iterator = data.iterator();
+ List<String> types = iterator.getColumnTypeList();
+ try (Resource<ExportTextWriter> output =
+ own(new ExportTextWriter(config, task, progress, csvHeaders(names, types, true)))) {
while (iterator.next()) {
- for (Map.Entry<String, List<Integer>> entry : deviceColumnIndices.entrySet()) {
- List<String> names = new ArrayList<>();
- List<String> values = new ArrayList<>();
- for (int index : entry.getValue()) {
- if (!iterator.isNull(index + 1)) {
- names.add(measurements.get(index));
- values.add(formatSqlValue(iterator.getString(index + 1), columnTypes.get(index)));
- }
- }
- // A timestamp present in another device must not create an empty record here.
- if (values.isEmpty()) {
+ progress.checkActive();
+ List<Object> values = new ArrayList<>(types.size());
+ for (int i = 0; i < types.size(); i++) {
+ if (iterator.isNull(i + 1)) {
+ values.add("");
continue;
}
- if (writer == null) {
- writer = new FileWriter(filePath + "_" + fileIndex + ".sql");
+ if (i == 0 && "Time".equals(names.get(i))) {
+ values.add(timeTrans(iterator.getLong(i + 1)));
+ continue;
}
- writer.write(
- String.format(
- "INSERT INTO %s(TIMESTAMP,%s) VALUES (%d,%s);\n",
- entry.getKey(),
- String.join(",", names),
- iterator.getLong(1),
- String.join(",", values)));
- if (++currentLines >= linesPerFile) {
- writer.close();
- writer = null;
- fileIndex++;
- currentLines = 0;
+ String value = iterator.getString(i + 1);
+ if (("TEXT".equalsIgnoreCase(types.get(i)) || "STRING".equalsIgnoreCase(types.get(i)))
+ && !value.startsWith("root.")) {
+ value = '"' + value + '"';
}
+ values.add(value);
}
- processedRows++;
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
- }
- }
- } finally {
- if (writer != null) {
- writer.close();
+ output.get().csvRow(values);
}
+ output.get().complete();
}
- ioTPrinter.print("\n");
}
- private void exportToSqlFile(SessionDataSet sessionDataSet, String filePath)
- throws IOException, IoTDBConnectionException, StatementExecutionException {
- List<String> headers = sessionDataSet.getColumnNames();
- boolean alignByDeviceQuery = headers.contains("Device");
- if (alignByDeviceQuery) {
- exportToSqlFileWithAlignDevice(sessionDataSet, filePath);
+ private void exportSql(SessionDataSet data, ExportTask task) throws Exception {
+ List<String> headers = data.getColumnNames();
+ SessionDataSet.DataIterator iterator = data.iterator();
+ List<String> types = iterator.getColumnTypeList();
+ boolean byDevice = headers.contains("Device");
+ // Column lists are built once: parsing the column paths for every row slowed SQL export.
+ SqlColumns deviceRow = new SqlColumns();
+ Map<String, SqlColumns> deviceColumns = new LinkedHashMap<>();
+ if (byDevice) {
+ for (int i = 2; i < headers.size(); i++) {
+ deviceRow.add(i, headers.get(i));
+ }
} else {
- exportToSqlFileWithoutAlign(sessionDataSet, filePath);
- }
- }
-
- private static Boolean exportToTsFile(SessionDataSet sessionDataSet, String filePath)
- throws IOException,
- IoTDBConnectionException,
- StatementExecutionException,
- WriteProcessException {
- List<String> columnNames = sessionDataSet.getColumnNames();
- List<String> columnTypes = sessionDataSet.getColumnTypes();
- File f = FSFactoryProducer.getFSFactory().getFile(filePath);
- if (f.exists()) {
- Files.delete(f.toPath());
- }
- boolean isEmpty = false;
- try (TsFileWriter tsFileWriter = new TsFileWriter(f)) {
- // device -> column indices in columnNames
- Map<String, List<Integer>> deviceColumnIndices = new HashMap<>();
- Set<String> alignedDevices = new HashSet<>();
- Map<String, List<IMeasurementSchema>> deviceSchemaMap = new LinkedHashMap<>();
-
- collectSchemas(
- columnNames, columnTypes, deviceSchemaMap, alignedDevices, deviceColumnIndices);
-
- List<Tablet> tabletList = constructTablets(deviceSchemaMap, alignedDevices, tsFileWriter);
-
- if (!tabletList.isEmpty()) {
- writeWithTablets(
- sessionDataSet, tabletList, alignedDevices, tsFileWriter, deviceColumnIndices);
- tsFileWriter.flush();
- } else {
- isEmpty = true;
+ for (int i = 1; i < headers.size(); i++) {
+ PartialPath path = new PartialPath(headers.get(i));
+ if (!"root".equals(path.getFirstNode()) || path.hasWildcard()) {
+ throw new ArgsErrorException(
+ CliMessages
+ .EXCEPTION_SQL_OUTPUT_REQUIRES_ORIGINAL_MEASUREMENT_PATHS_OR_AN_ALIGN_BY_DEVICE_QUERY_ADA72EA8);
+ }
+ deviceColumns
+ .computeIfAbsent(devicePath(path).getFullPath(), key -> new SqlColumns())
+ .add(i, path.getMeasurement());
}
}
- if (isEmpty) {
- ioTPrinter.println("!!!Warning:Tablet is empty,no data can be exported.");
- return false;
- }
- return true;
- }
-
- private void exportToCsvFile(SessionDataSet sessionDataSet, String filePath)
- throws IOException, IoTDBConnectionException, StatementExecutionException {
- processedRows = 0;
- lastPrintTime = 0;
- List<String> headers = getCsvHeaders(sessionDataSet, true);
- int fileIndex = 0;
- SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
- List<String> columnTypeList = iterator.getColumnTypeList();
- int totalColumns = columnTypeList.size();
- boolean fromOuterloop = false;
- while (iterator.next()) {
- final String finalFilePath = filePath + "_" + fileIndex + ".csv";
- CSVPrinterWrapper csvPrinterWrapper = new CSVPrinterWrapper(finalFilePath);
- try {
- csvPrinterWrapper.printRecord(headers);
- fromOuterloop = true;
- int i = 0;
- while (i++ < linesPerFile && (fromOuterloop || iterator.next())) {
- fromOuterloop = false;
- csvPrinterWrapper.print(timeTrans(iterator.getLong(1)));
- for (int curColumnIndex = 1; curColumnIndex < totalColumns; curColumnIndex++) {
- if (iterator.isNull(curColumnIndex + 1)) {
- csvPrinterWrapper.print("");
- } else {
- String columnValue = iterator.getString(curColumnIndex + 1);
- String curType = columnTypeList.get(curColumnIndex);
- if ((curType.equalsIgnoreCase("TEXT") || curType.equalsIgnoreCase("STRING"))
- && !columnValue.startsWith("root.")) {
- csvPrinterWrapper.print("\"" + columnValue + "\"");
- } else {
- csvPrinterWrapper.print(columnValue);
- }
- }
- }
- csvPrinterWrapper.println();
- processedRows += 1;
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
+ try (Resource<ExportTextWriter> output =
+ own(new ExportTextWriter(config, task, progress, null))) {
+ while (iterator.next()) {
+ progress.checkActive();
+ // One result row is one data row: its statements for every device stay in one file.
+ StringBuilder row = new StringBuilder();
+ if (byDevice) {
+ // Every row of a device has the same columns, so a missing value is written as NULL.
+ appendInsert(row, iterator, types, iterator.getString(2), deviceRow, true, false);
+ } else {
+ for (Map.Entry<String, SqlColumns> device : deviceColumns.entrySet()) {
+ boolean aligned =
+ config.aligned || (task.measurement() != null && isAligned(device.getKey()));
+ // A row also carries the times of other devices: null columns are left out.
+ appendInsert(row, iterator, types, device.getKey(), device.getValue(), aligned, true);
}
}
- fileIndex++;
- csvPrinterWrapper.flush();
- } finally {
- csvPrinterWrapper.close();
+ if (row.length() > 0) {
+ output.get().sqlRow(row.toString());
+ }
}
+ output.get().complete();
}
- ioTPrinter.print("\n");
}
- private static void writeWithTablets(
- SessionDataSet sessionDataSet,
- List<Tablet> tabletList,
- Set<String> alignedDevices,
- TsFileWriter tsFileWriter,
- Map<String, List<Integer>> deviceColumnIndices)
- throws IoTDBConnectionException,
- StatementExecutionException,
- IOException,
- WriteProcessException {
- processedRows = 0;
- lastPrintTime = 0;
- while (sessionDataSet.hasNext()) {
- RowRecord rowRecord = sessionDataSet.next();
- List<Field> fields = rowRecord.getFields();
-
- for (Tablet tablet : tabletList) {
- String deviceId = tablet.getDeviceId();
- List<Integer> columnIndices = deviceColumnIndices.get(deviceId);
- int rowIndex = tablet.getRowSize();
- tablet.addTimestamp(rowIndex, rowRecord.getTimestamp());
- List<IMeasurementSchema> schemas = tablet.getSchemas();
+ /** The result columns of one device and the measurement list its INSERT statements name. */
+ private static final class SqlColumns {
+ private final List<Integer> columns = new ArrayList<>();
+ private final List<String> measurements = new ArrayList<>();
+ private final StringBuilder names = new StringBuilder();
- for (int i = 0, columnIndicesSize = columnIndices.size(); i < columnIndicesSize; i++) {
- Integer columnIndex = columnIndices.get(i);
- IMeasurementSchema measurementSchema = schemas.get(i);
- Object value = fields.get(columnIndex - 1).getObjectValue(measurementSchema.getType());
- tablet.addValue(measurementSchema.getMeasurementName(), rowIndex, value);
- }
+ private void add(int column, String measurement) {
+ columns.add(column);
+ measurements.add(measurement);
+ names.append(',').append(measurement);
+ }
+ }
- if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
- writeToTsFile(alignedDevices, tsFileWriter, tablet);
- processedRows += tablet.getRowSize();
- tablet.reset();
+ private static void appendInsert(
+ StringBuilder row,
+ SessionDataSet.DataIterator iterator,
+ List<String> types,
+ String device,
+ SqlColumns columns,
+ boolean aligned,
+ boolean skipNulls)
+ throws Exception {
+ if (device.startsWith(SYSTEM_DATABASE + ".") || device.startsWith(AUDIT_DATABASE + ".")) {
+ return;
+ }
+ StringBuilder values = new StringBuilder();
+ // Stays null while every column has a value, so the prepared list is used.
+ StringBuilder names = null;
+ for (int i = 0; i < columns.columns.size(); i++) {
+ int column = columns.columns.get(i) + 1;
+ if (skipNulls && iterator.isNull(column)) {
+ if (names == null) {
+ names = new StringBuilder();
+ for (String measurement : columns.measurements.subList(0, i)) {
+ names.append(',').append(measurement);
+ }
}
+ continue;
}
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
+ if (names != null) {
+ names.append(',').append(columns.measurements.get(i));
}
+ values.append(',').append(sqlValue(iterator, types.get(column - 1), column));
}
-
- for (Tablet tablet : tabletList) {
- if (tablet.getRowSize() != 0) {
- writeToTsFile(alignedDevices, tsFileWriter, tablet);
- processedRows += tablet.getRowSize();
- }
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
- }
+ if (values.length() == 0) {
+ return;
}
- ioTPrinter.print("\n");
- }
-
- private static void writeToTsFile(
- Set<String> deviceFilterSet, TsFileWriter tsFileWriter, Tablet tablet)
- throws IOException, WriteProcessException {
- if (deviceFilterSet.contains(tablet.getDeviceId())) {
- tsFileWriter.writeAligned(tablet);
- } else {
- tsFileWriter.writeTree(tablet);
+ row.append("INSERT INTO ")
+ .append(device)
+ .append("(TIMESTAMP")
+ .append(names == null ? columns.names : names)
+ .append(')');
+ if (aligned) {
+ row.append(" ALIGNED");
}
+ row.append(" VALUES (").append(iterator.getLong(1)).append(values).append(");\n");
}
- private static List<Tablet> constructTablets(
- Map<String, List<IMeasurementSchema>> deviceSchemaMap,
- Set<String> alignedDevices,
- TsFileWriter tsFileWriter)
- throws WriteProcessException {
- List<Tablet> tabletList = new ArrayList<>(deviceSchemaMap.size());
- for (Map.Entry<String, List<IMeasurementSchema>> stringListEntry : deviceSchemaMap.entrySet()) {
- String deviceId = stringListEntry.getKey();
- List<IMeasurementSchema> schemaList = stringListEntry.getValue();
- Tablet tablet = new Tablet(deviceId, schemaList);
- tablet.initBitMaps();
- Path path = new Path(tablet.getDeviceId());
- if (alignedDevices.contains(tablet.getDeviceId())) {
- tsFileWriter.registerAlignedTimeseries(path, schemaList);
- } else {
- tsFileWriter.registerTimeseries(path, schemaList);
+ private boolean isAligned(String device) throws Exception {
+ Boolean known = alignment.get(device);
+ if (known != null) {
+ return known;
+ }
+ boolean aligned = false;
+ try (Resource<SessionDataSet> result =
+ own(session.executeQueryStatement("show devices " + device, config.timeout))) {
+ SessionDataSet data = result.get();
+ if (data.hasNext()) {
+ List<Field> fields = data.next().getFields();
+ aligned = fields.size() > 1 && "true".equalsIgnoreCase(fields.get(1).getStringValue());
}
- tabletList.add(tablet);
}
- return tabletList;
+ alignment.put(device, aligned);
+ return aligned;
}
- private static void collectSchemas(
- List<String> columnNames,
- List<String> columnTypes,
- Map<String, List<IMeasurementSchema>> deviceSchemaMap,
- Set<String> alignedDevices,
- Map<String, List<Integer>> deviceColumnIndices)
- throws IoTDBConnectionException, StatementExecutionException {
+ private void exportTsFile(SessionDataSet data, ExportTask task) throws Exception {
+ java.nio.file.Path file = java.nio.file.Path.of(task.outputBase() + ".tsfile");
+ ExportFileNames.checkOutputCollision(file, config);
+ List<String> columnNames = data.getColumnNames();
+ Map<String, List<IMeasurementSchema>> schemas = new LinkedHashMap<>();
+ Map<String, List<Integer>> indices = new LinkedHashMap<>();
+ Set<String> alignedDevices = new HashSet<>();
for (int i = 0; i < columnNames.size(); i++) {
- String column = columnNames.get(i);
- if (!column.startsWith("root.")) {
+ String name = columnNames.get(i);
+ if (i == 0 && "Time".equals(name)) {
continue;
}
- TSDataType tsDataType = getType(columnTypes.get(i));
- Path path = new Path(column, true);
- String deviceId = path.getDeviceString();
- // query whether the device is aligned or not
- try (SessionDataSet deviceDataSet =
- session.executeQueryStatement("show devices " + deviceId, timeout)) {
- List<Field> deviceList = deviceDataSet.next().getFields();
- if (deviceList.size() > 1 && "true".equals(deviceList.get(1).getStringValue())) {
- alignedDevices.add(deviceId);
+ if (i == 0 || !name.startsWith("root.")) {
+ throw new ArgsErrorException(
+ CliMessages
+ .EXCEPTION_TREE_TSFILE_OUTPUT_REQUIRES_A_TIME_COLUMN_AND_ORIGINAL_MEASUREMENT_PATHS_1DCEE8EA);
+ }
+ ExportTask.Measurement measurement = measurements.get(name);
+ if (measurement == null
+ && task.measurement() != null
+ && task.measurement().path().getFullPath().equals(name)) {
+ measurement = task.measurement();
+ }
+ if (measurement == null) {
+ try (Resource<SessionDataSet> result =
+ own(session.executeQueryStatement("show timeseries " + name, config.timeout))) {
+ if (!result.get().hasNext()) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages.EXCEPTION_CANNOT_OBTAIN_SCHEMA_FOR_MEASUREMENT_ARG_D3F1DAF3, name));
+ }
+ measurement = measurement(result.get().next(), result.get().getColumnNames());
}
+ measurements.put(name, measurement);
}
+ String device = devicePath(measurement.path()).getFullPath();
+ if (isAligned(device)) {
+ alignedDevices.add(device);
+ }
+ schemas.computeIfAbsent(device, key -> new ArrayList<>()).add(measurement.schema());
+ indices.computeIfAbsent(device, key -> new ArrayList<>()).add(i);
+ }
+ progress.fileStarted(file);
+ Files.deleteIfExists(file);
+ long startRows = progress.rows();
+ try (Resource<TsFileWriter> output = own(new TsFileWriter(file.toFile()))) {
+ TsFileWriter writer = output.get();
+ List<Tablet> tablets = new ArrayList<>();
+ for (Map.Entry<String, List<IMeasurementSchema>> schema : schemas.entrySet()) {
+ Tablet tablet = new Tablet(schema.getKey(), schema.getValue());
+ tablet.initBitMaps();
+ if (alignedDevices.contains(schema.getKey())) {
+ writer.registerAlignedTimeseries(new Path(schema.getKey()), schema.getValue());
+ } else {
+ writer.registerTimeseries(new Path(schema.getKey()), schema.getValue());
+ }
+ tablets.add(tablet);
+ }
+ long pendingRows = 0;
+ while (!tablets.isEmpty() && data.hasNext()) {
+ progress.checkActive();
+ RowRecord row = data.next();
+ for (Tablet tablet : tablets) {
+ int rowIndex = tablet.getRowSize();
+ tablet.addTimestamp(rowIndex, row.getTimestamp());
+ List<Integer> columns = indices.get(tablet.getDeviceId());
+ for (int i = 0; i < columns.size(); i++) {
+ IMeasurementSchema schema = tablet.getSchemas().get(i);
+ tablet.addValue(
+ schema.getMeasurementName(),
+ rowIndex,
+ row.getFields().get(columns.get(i) - 1).getObjectValue(schema.getType()));
+ }
+ }
+ pendingRows++;
+ if (tablets.get(0).getRowSize() == tablets.get(0).getMaxRowNumber()) {
+ writeTablets(tablets, alignedDevices, writer);
+ progress.rowsWritten(pendingRows);
+ pendingRows = 0;
+ }
+ }
+ if (pendingRows > 0) {
+ writeTablets(tablets, alignedDevices, writer);
+ progress.rowsWritten(pendingRows);
+ }
+ writer.flush();
+ }
+ progress.fileCompleted(file, progress.rows() - startRows);
+ }
- // query timeseries metadata
- MeasurementSchema measurementSchema =
- new MeasurementSchema(path.getMeasurement(), tsDataType);
- List<Field> seriesList =
- session.executeQueryStatement("show timeseries " + column, timeout).next().getFields();
- measurementSchema.setEncoding(TSEncoding.valueOf(seriesList.get(4).getStringValue()));
- measurementSchema.setCompressionType(
- CompressionType.valueOf(seriesList.get(5).getStringValue()));
-
- deviceSchemaMap.computeIfAbsent(deviceId, key -> new ArrayList<>()).add(measurementSchema);
- deviceColumnIndices.computeIfAbsent(deviceId, key -> new ArrayList<>()).add(i);
+ private static void writeTablets(List<Tablet> tablets, Set<String> aligned, TsFileWriter writer)
+ throws Exception {
+ for (Tablet tablet : tablets) {
+ if (aligned.contains(tablet.getDeviceId())) {
+ writer.writeAligned(tablet);
+ } else {
+ writer.writeTree(tablet);
+ }
+ tablet.reset();
+ tablet.initBitMaps();
}
}
}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportFileNames.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportFileNames.java
new file mode 100644
index 00000000000..871d8d13ef5
--- /dev/null
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportFileNames.java
@@ -0,0 +1,114 @@
+/*
+ * 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.tool.data;
+
+import org.apache.iotdb.cli.i18n.CliMessages;
+import org.apache.iotdb.exception.ArgsErrorException;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.text.Normalizer;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+
+/** Retains readable full paths, escaping characters that cannot be used on Windows or Unix. */
+final class ExportFileNames {
+ private final Map<String, String> names = new HashMap<>();
+
+ Path measurement(ExportConfig config, String measurement) throws ArgsErrorException {
+ String name = encode((config.prefix.isEmpty() ? "" : config.prefix + "_") + measurement);
+ // Leave room for the split index and extension on filesystems with 255-byte components.
+ if (name.getBytes(StandardCharsets.UTF_8).length > 220) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages
+ .EXCEPTION_MEASUREMENT_OUTPUT_FILENAME_EXCEEDS_THE_SUPPORTED_LENGTH_ARG_FF01275D,
+ measurement));
+ }
+ String key = Normalizer.normalize(name, Normalizer.Form.NFC).toLowerCase(Locale.ROOT);
+ String previous = names.putIfAbsent(key, measurement);
+ if (previous != null && !previous.equals(measurement)) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages
+ .EXCEPTION_MEASUREMENTS_ARG_AND_ARG_MAP_TO_THE_SAME_OUTPUT_FILENAME_ARG_58E1FD4D,
+ previous,
+ measurement,
+ name));
+ }
+ return config.targetDirectory.resolve(name);
+ }
+
+ private static String encode(String value) {
+ StringBuilder result = new StringBuilder();
+ for (int i = 0; i < value.length(); i++) {
+ char character = value.charAt(i);
+ if (character < 32
+ || "\\/:*?\"<>|%".indexOf(character) >= 0
+ || (i == value.length() - 1 && (character == '.' || character == ' '))) {
+ for (byte b : String.valueOf(character).getBytes(StandardCharsets.UTF_8)) {
+ result.append('%');
+ result.append(Character.toUpperCase(Character.forDigit((b & 0xff) >>> 4, 16)));
+ result.append(Character.toUpperCase(Character.forDigit(b & 15, 16)));
+ }
+ } else {
+ result.append(character);
+ }
+ }
+ return result.toString();
+ }
+
+ static void checkOutputCollision(Path output, ExportConfig config) throws ArgsErrorException {
+ checkLogCollision(output, config.logFile);
+ Path input = config.measurementFile;
+ if (input != null
+ && (output.equals(input)
+ || (java.nio.file.Files.exists(output)
+ && java.nio.file.Files.exists(input)
+ && sameFile(output, input)))) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages
+ .EXCEPTION_EXPORT_OUTPUT_CONFLICTS_WITH_THE_MEASUREMENT_INPUT_FILE_ARG_1CBEBB6C,
+ output));
+ }
+ }
+
+ static void checkLogCollision(Path output, Path log) throws ArgsErrorException {
+ if (output.equals(log)
+ || (java.nio.file.Files.exists(output)
+ && java.nio.file.Files.exists(log)
+ && sameFile(output, log))) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages.EXCEPTION_EXPORT_OUTPUT_CONFLICTS_WITH_THE_LOG_FILE_ARG_F01BE79D,
+ output));
+ }
+ }
+
+ private static boolean sameFile(Path left, Path right) throws ArgsErrorException {
+ try {
+ return java.nio.file.Files.isSameFile(left, right);
+ } catch (java.io.IOException e) {
+ throw new ArgsErrorException(e.getMessage());
+ }
+ }
+}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportProgressReporter.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportProgressReporter.java
new file mode 100644
index 00000000000..fc3a75fdace
--- /dev/null
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportProgressReporter.java
@@ -0,0 +1,638 @@
+/*
+ * 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.tool.data;
+
+import org.apache.iotdb.cli.i18n.CliMessages;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+import org.apache.iotdb.tool.common.Constants;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.io.PrintStream;
+import java.io.Writer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.text.NumberFormat;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * One task's counters, its JSON Lines log and its console view. The timer never touches a Session
+ * or result set.
+ */
+final class ExportProgressReporter implements AutoCloseable {
+ enum Status {
+ RUNNING,
+ SUCCESS,
+ FAILED,
+ CANCELLED
+ }
+
+ enum Phase {
+ VALIDATING,
+ CONNECTING,
+ DISCOVERING,
+ EXPORTING,
+ CLOSING
+ }
+
+ private final Writer log;
+ private final ExportConsole console;
+ private final String model;
+ private final long startNanos;
+ private final AtomicLong rows = new AtomicLong();
+ private final ScheduledExecutorService timer;
+ private volatile IOException logFailure;
+ private volatile boolean stopping;
+ private volatile int cancelExitCode = Constants.CODE_ERROR;
+ private volatile Status status = Status.RUNNING;
+ private volatile Phase phase = Phase.VALIDATING;
+ private long totalMeasurements = -1;
+ private long completedMeasurements;
+ private long completedTasks;
+ private long zeroRowTasks;
+ private long files;
+ private long taskStartRows;
+ private long exportStartNanos;
+ private String currentMeasurement = "";
+ private String output = "";
+ private Path logFile;
+ private boolean closed;
+ private Runnable failureHandler = () -> {};
+
+ ExportProgressReporter(ExportConfig config, long startNanos, PrintStream console)
+ throws IOException {
+ this(
+ Files.newBufferedWriter(
+ config.logFile,
+ StandardCharsets.UTF_8,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.APPEND),
+ config.tree ? "tree" : "table",
+ startNanos,
+ new ExportConsole(console, ExportConsole.isTerminal()));
+ }
+
+ // Writer injection exercises real write/flush failures without relying on filesystem permissions.
+ ExportProgressReporter(Writer log, String model, long startNanos, PrintStream console) {
+ this(log, model, startNanos, new ExportConsole(console, false));
+ }
+
+ ExportProgressReporter(Writer log, String model, long startNanos, ExportConsole console) {
+ this.log = log;
+ this.console = console;
+ this.model = model;
+ this.startNanos = startNanos;
+ timer =
+ Executors.newSingleThreadScheduledExecutor(
+ runnable -> {
+ Thread thread = new Thread(runnable, "export-data-progress");
+ thread.setDaemon(true);
+ return thread;
+ });
+ }
+
+ synchronized void start(ExportConfig config, boolean queryMode, Runnable onFailure)
+ throws IOException {
+ failureHandler = onFailure;
+ logFile = config.logFile;
+ if (emit(
+ "INFO",
+ "START",
+ Status.RUNNING,
+ fields(
+ "format",
+ config.format,
+ "input",
+ config.measurementFile,
+ "mode",
+ config.measurementFile != null ? "measurements" : queryMode ? "query" : "automatic",
+ // A query runs as written, so a time range here would misstate what it exports.
+ "start_time",
+ queryMode ? null : config.startTime,
+ "end_time",
+ queryMode ? null : config.endTime,
+ "target",
+ config.targetDirectory,
+ "log",
+ config.logFile,
+ "lines_per_file",
+ config.linesPerFile,
+ "message",
+ CliMessages.LOG_EXPORT_STARTED_45465087))) {
+ console.line(
+ String.format(
+ Locale.ROOT,
+ CliMessages.LOG_EXPORT_STARTED_ARG_ARG_TARGET_ARG_LOG_ARG_BD58AECB,
+ model,
+ config.format,
+ config.targetDirectory,
+ config.logFile));
+ }
+ if (queryMode && (config.startTime != null || config.endTime != null)) {
+ warning(
+ CliMessages.LOG_START_TIME_AND_END_TIME_DO_NOT_APPLY_TO_QUERIES_AND_ARE_IGNORED_02DC37CC,
+ "start_time",
+ config.startTime,
+ "end_time",
+ config.endTime);
+ }
+ if (queryMode && config.table != null) {
+ warning(
+ CliMessages.LOG_TABLE_DOES_NOT_APPLY_TO_QUERIES_AND_IS_IGNORED_DBA31209,
+ "table",
+ config.table);
+ }
+ ScheduledExecutorUtil.safelyScheduleAtFixedRate(timer, this::tick, 2, 2, TimeUnit.SECONDS);
+ }
+
+ private void tick() {
+ try {
+ progress();
+ } catch (Exception e) {
+ logFailure =
+ e instanceof IOException
+ ? (IOException) e
+ : new IOException(CliMessages.LOG_CANNOT_WRITE_OR_CLOSE_THE_EXPORT_LOG_32579AC6, e);
+ stopping = true;
+ failureHandler.run();
+ }
+ }
+
+ synchronized void progress() throws IOException {
+ if (!stopping
+ && status == Status.RUNNING
+ && !closed
+ && emit("INFO", "PROGRESS", Status.RUNNING, fields())) {
+ showProgress();
+ }
+ }
+
+ private void showProgress() {
+ String exported = count(rows.get());
+ String elapsed = duration(elapsedSeconds());
+ if (totalMeasurements < 0) {
+ // Without a measurement file there is no total, so no percentage or remaining time either.
+ String rate = count(rowsPerSecond());
+ console.progress(
+ String.format(
+ Locale.ROOT,
+ CliMessages.LOG_EXPORTING_ARG_ROWS_ARG_ARG_ROWS_S_56BD441C,
+ exported,
+ elapsed,
+ rate),
+ "",
+ String.format(
+ Locale.ROOT,
+ CliMessages.LOG_ARG_ROWS_ELAPSED_ARG_ARG_ROWS_S_AE3CF660,
+ exported,
+ elapsed,
+ rate));
+ return;
+ }
+ // Whole points, so the console never shows 100% before the last measurement completes.
+ String percent =
+ (totalMeasurements == 0 ? 0 : completedMeasurements * 100 / totalMeasurements) + "%";
+ long remaining = remainingSeconds();
+ String eta = remaining < 0 ? "--" : duration(remaining);
+ console.progress(
+ String.format(
+ Locale.ROOT,
+ CliMessages.LOG_EXPORTING_ARG_ARG_ARG_ARG_ROWS_ARG_ETA_ARG_C41CA790,
+ percent,
+ completedMeasurements,
+ totalMeasurements,
+ exported,
+ elapsed,
+ eta),
+ currentMeasurement,
+ String.format(
+ Locale.ROOT,
+ CliMessages
+ .LOG_ARG_ARG_ARG_MEASUREMENTS_ARG_ROWS_ELAPSED_ARG_ETA_ARG_CURRENT_ARG_B6DEA0BD,
+ percent,
+ completedMeasurements,
+ totalMeasurements,
+ exported,
+ elapsed,
+ eta,
+ currentMeasurement.isEmpty() ? "-" : currentMeasurement));
+ }
+
+ void phase(Phase phase) throws IOException {
+ checkActive();
+ this.phase = phase;
+ }
+
+ synchronized void discovered(MeasurementFile input, long matches, int measurements, int unmatched)
+ throws IOException {
+ checkActive();
+ totalMeasurements = measurements;
+ exportStartNanos = System.nanoTime();
+ if (emit(
+ "INFO",
+ "DISCOVERY",
+ Status.RUNNING,
+ fields(
+ "source",
+ input.source(),
+ "input_lines",
+ input.lines(),
+ "patterns",
+ input.patterns().size(),
+ "duplicate_patterns",
+ input.duplicates(),
+ "matches",
+ matches,
+ "duplicate_matches",
+ matches - measurements,
+ "unmatched_patterns",
+ unmatched))) {
+ console.line(
+ String.format(
+ Locale.ROOT,
+ CliMessages.LOG_MEASUREMENTS_TO_EXPORT_ARG_PATH_PATTERNS_ARG_A0AC5F31,
+ measurements,
+ input.patterns().size()));
+ }
+ }
+
+ /** {@code message} says on its own what the warning is about; its details go to the log only. */
+ synchronized void warning(String message, Object... details) throws IOException {
+ Map<String, Object> values = fields(details);
+ values.put("message", message);
+ if (emit("WARN", "WARNING", Status.RUNNING, values)) {
+ console.line(String.format(Locale.ROOT, CliMessages.LOG_WARNING_ARG_05BE32E9, message));
+ }
+ }
+
+ synchronized void taskStarted(ExportTask task) throws IOException {
+ checkActive();
+ phase = Phase.EXPORTING;
+ currentMeasurement = task.measurement() == null ? "" : task.measurement().path().getFullPath();
+ // FILE_START names the output once the task opens a file; a task without rows opens none.
+ output = "";
+ taskStartRows = rows.get();
+ emit("INFO", "TASK_START", Status.RUNNING, fields("sql", task.sql(), "table", task.table()));
+ }
+
+ synchronized void taskCompleted() throws IOException {
+ checkActive();
+ long taskRows = rows.get() - taskStartRows;
+ completedTasks++;
+ if (totalMeasurements >= 0) {
+ completedMeasurements++;
+ }
+ if (taskRows == 0) {
+ zeroRowTasks++;
+ }
+ emit("INFO", "TASK_COMPLETE", Status.RUNNING, fields("task_rows", taskRows));
+ }
+
+ synchronized void fileStarted(Path file) throws IOException {
+ checkActive();
+ output = file.toString();
+ emit("INFO", "FILE_START", Status.RUNNING, fields());
+ }
+
+ synchronized void fileCompleted(Path file, long fileRows) throws IOException {
+ checkActive();
+ files++;
+ emit("INFO", "FILE_COMPLETE", Status.RUNNING, fields("output", file, "file_rows", fileRows));
+ }
+
+ void rowsWritten(long count) throws IOException {
+ checkActive();
+ rows.addAndGet(count);
+ }
+
+ long rows() {
+ return rows.get();
+ }
+
+ void checkActive() throws IOException {
+ if (logFailure != null) {
+ throw logFailure;
+ }
+ if (stopping || Thread.currentThread().isInterrupted()) {
+ throw new InterruptedIOException(CliMessages.EXCEPTION_EXPORT_CANCELLED_85D44F54);
+ }
+ }
+
+ /** Stops at the next check; a cancelled finish reports the status the process exits with. */
+ void cancel(int exitCode) {
+ cancelExitCode = exitCode;
+ stopping = true;
+ }
+
+ Status status() {
+ return status;
+ }
+
+ synchronized void finish(Status terminal, Throwable error) throws IOException {
+ if (status != Status.RUNNING || closed) {
+ return;
+ }
+ stopping = true;
+ timer.shutdownNow();
+ if (terminal == Status.SUCCESS && logFailure != null) {
+ terminal = Status.FAILED;
+ error = logFailure;
+ }
+ String reason = reason(error);
+ String line =
+ event(
+ terminal == Status.SUCCESS ? "INFO" : "ERROR",
+ "FINISH",
+ terminal,
+ fields(
+ "exit_code",
+ exitCode(terminal),
+ "completed_tasks",
+ completedTasks,
+ "zero_row_tasks",
+ zeroRowTasks,
+ "message",
+ terminal == Status.SUCCESS
+ ? CliMessages.LOG_EXPORT_COMPLETELY_34200D19
+ : terminal == Status.CANCELLED
+ ? CliMessages.EXCEPTION_EXPORT_CANCELLED_85D44F54
+ : CliMessages.LOG_EXPORT_FAILED_77EE72CD,
+ "error",
+ reason));
+ IOException writeError = null;
+ try {
+ log.write(line + System.lineSeparator());
+ log.flush();
+ } catch (IOException e) {
+ writeError = e;
+ }
+ try {
+ log.close();
+ } catch (IOException e) {
+ if (writeError == null) {
+ writeError = e;
+ } else {
+ writeError.addSuppressed(e);
+ }
+ }
+ closed = true;
+ if (writeError != null) {
+ logFailure = writeError;
+ // The log cannot record its own failure, so only the console reports it.
+ reason =
+ CliMessages.LOG_CANNOT_WRITE_OR_CLOSE_THE_EXPORT_LOG_32579AC6 + ' ' + reason(writeError);
+ if (terminal == Status.CANCELLED) {
+ console.line(String.format(Locale.ROOT, CliMessages.LOG_WARNING_ARG_05BE32E9, reason));
+ } else {
+ terminal = Status.FAILED;
+ }
+ }
+ status = terminal;
+ // Scripts read the outcome from the last console line, so the log path is printed first.
+ if (logFile != null) {
+ console.line(String.format(Locale.ROOT, CliMessages.LOG_LOG_ARG_FBF61F63, logFile));
+ }
+ console.line(summary(terminal, reason));
+ if (writeError != null) {
+ throw writeError;
+ }
+ }
+
+ private String summary(Status terminal, String reason) {
+ String exported = count(rows.get());
+ String elapsed = duration(elapsedSeconds());
+ if (terminal == Status.SUCCESS) {
+ return totalMeasurements >= 0
+ ? String.format(
+ Locale.ROOT,
+ CliMessages
+ .LOG_EXPORT_COMPLETELY_MEASUREMENTS_ARG_ARG_ROWS_ARG_FILES_ARG_ELAPSED_ARG_36D9F8E5,
+ completedMeasurements,
+ totalMeasurements,
+ exported,
+ count(files),
+ elapsed)
+ : String.format(
+ Locale.ROOT,
+ CliMessages.LOG_EXPORT_COMPLETELY_ROWS_ARG_FILES_ARG_ELAPSED_ARG_A94BBB8B,
+ exported,
+ count(files),
+ elapsed);
+ }
+ if (terminal == Status.CANCELLED) {
+ return String.format(
+ Locale.ROOT,
+ CliMessages.LOG_EXPORT_CANCELLED_AFTER_ARG_ROWS_EXPORTED_ARG_0999F99F,
+ elapsed,
+ exported);
+ }
+ return reason == null
+ ? String.format(Locale.ROOT, CliMessages.LOG_EXPORT_FAILED_AFTER_ARG_DA6D56F9, elapsed)
+ : String.format(
+ Locale.ROOT, CliMessages.LOG_EXPORT_FAILED_AFTER_ARG_ARG_C123F0EF, elapsed, reason);
+ }
+
+ private static String reason(Throwable error) {
+ if (error == null) {
+ return null;
+ }
+ return error.getMessage() == null ? error.toString() : error.getMessage();
+ }
+
+ private int exitCode(Status terminal) {
+ return terminal == Status.SUCCESS
+ ? Constants.CODE_OK
+ : terminal == Status.CANCELLED ? cancelExitCode : Constants.CODE_ERROR;
+ }
+
+ /** Appends one event to the log; false once the log has closed and records nothing more. */
+ private boolean emit(String level, String event, Status eventStatus, Map<String, Object> details)
+ throws IOException {
+ if (closed || status != Status.RUNNING) {
+ return false;
+ }
+ String line = event(level, event, eventStatus, details);
+ try {
+ log.write(line + System.lineSeparator());
+ log.flush();
+ } catch (IOException e) {
+ logFailure = e;
+ throw e;
+ }
+ return true;
+ }
+
+ private String event(
+ String level, String event, Status eventStatus, Map<String, Object> details) {
+ // The export requirement's progress fields come first, in its order; event details follow.
+ // These keys are the stable, locale-independent log schema, not translated prose.
+ Map<String, Object> values =
+ fields(
+ "progress_id",
+ ProcessHandle.current().pid(),
+ "timestamp",
+ ZonedDateTime.now()
+ .truncatedTo(ChronoUnit.SECONDS)
+ .format(DateTimeFormatter.ISO_OFFSET_DATE_TIME),
+ "model",
+ model,
+ "status",
+ eventStatus,
+ "exported_data_rows",
+ rows.get());
+ if (totalMeasurements >= 0) {
+ values.put("exported_measurement_num", completedMeasurements);
+ values.put("total_measurement_num", totalMeasurements);
+ values.put(
+ "exported_percent",
+ String.format(
+ Locale.ROOT,
+ CliMessages.LOG_ARG_82AC62F7,
+ totalMeasurements == 0 ? 0.0 : completedMeasurements * 100.0 / totalMeasurements));
+ long remaining = remainingSeconds();
+ values.put("estimated_remaining", remaining < 0 ? "UNKNOWN" : duration(remaining));
+ }
+ values.put("elapsed_time", duration(elapsedSeconds()));
+ values.put("output", output.isEmpty() ? null : output);
+ values.put("event", event);
+ values.put("level", level);
+ values.put("phase", phase);
+ values.put("rows_per_second", rowsPerSecond());
+ values.put("files", files);
+ values.put("current_measurement", currentMeasurement.isEmpty() ? null : currentMeasurement);
+ values.putAll(details);
+ return json(values);
+ }
+
+ /** Seconds left at the average measurement duration, or -1 while too few have completed. */
+ private long remainingSeconds() {
+ // Measurements differ widely in size, so the first few say little about the rest.
+ long sample = Math.min(totalMeasurements, Math.max(3, (totalMeasurements + 19) / 20));
+ if (completedMeasurements == 0 || completedMeasurements < sample) {
+ return -1;
+ }
+ // Rounded up, so work still left never shows as 00:00:00.
+ return (long)
+ Math.ceil(
+ (System.nanoTime() - exportStartNanos)
+ / 1_000_000_000.0
+ * (totalMeasurements - completedMeasurements)
+ / completedMeasurements);
+ }
+
+ private long elapsedSeconds() {
+ return Math.max(0, System.nanoTime() - startNanos) / 1_000_000_000;
+ }
+
+ private long rowsPerSecond() {
+ return Math.round(
+ rows.get() / Math.max(1.0, Math.max(0, System.nanoTime() - startNanos) / 1_000_000_000.0));
+ }
+
+ static Map<String, Object> fields(Object... values) {
+ Map<String, Object> fields = new LinkedHashMap<>();
+ for (int i = 0; i < values.length; i += 2) {
+ fields.put((String) values[i], values[i + 1]);
+ }
+ return fields;
+ }
+
+ /** One flat JSON object without null members; integers stay numbers, the rest are strings. */
+ static String json(Map<String, Object> values) {
+ StringBuilder json = new StringBuilder("{");
+ values.forEach(
+ (key, value) -> {
+ if (value == null) {
+ return;
+ }
+ if (json.length() > 1) {
+ json.append(',');
+ }
+ string(json, key).append(':');
+ if (value instanceof Integer || value instanceof Long) {
+ json.append(value);
+ } else {
+ string(json, String.valueOf(value));
+ }
+ });
+ return json.append('}').toString();
+ }
+
+ private static StringBuilder string(StringBuilder json, String value) {
+ json.append('"');
+ for (int i = 0; i < value.length(); i++) {
+ char c = value.charAt(i);
+ if (c == '"' || c == '\\') {
+ json.append('\\').append(c);
+ } else if (c == '\n') {
+ json.append("\\n");
+ } else if (c == '\r') {
+ json.append("\\r");
+ } else if (c == '\t') {
+ json.append("\\t");
+ } else if (c < 0x20) {
+ json.append("\\u00")
+ .append(Character.forDigit(c >> 4, 16))
+ .append(Character.forDigit(c & 0xF, 16));
+ } else {
+ json.append(c);
+ }
+ }
+ return json.append('"');
+ }
+
+ private static String count(long value) {
+ return NumberFormat.getIntegerInstance(Locale.ROOT).format(value);
+ }
+
+ private static String duration(long seconds) {
+ long hours = seconds / 3600;
+ long minutes = seconds / 60 % 60;
+ long remainder = seconds % 60;
+ return (hours < 10 ? "0" : "")
+ + hours
+ + ":"
+ + (minutes < 10 ? "0" : "")
+ + minutes
+ + ":"
+ + (remainder < 10 ? "0" : "")
+ + remainder;
+ }
+
+ @Override
+ public synchronized void close() throws IOException {
+ timer.shutdownNow();
+ if (!closed) {
+ closed = true;
+ log.close();
+ }
+ }
+}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportTask.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportTask.java
new file mode 100644
index 00000000000..5d647eb67e5
--- /dev/null
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportTask.java
@@ -0,0 +1,31 @@
+/*
+ * 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.tool.data;
+
+import org.apache.iotdb.commons.path.PartialPath;
+
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import java.nio.file.Path;
+
+/** Query and output metadata travel together, rather than in separately indexed lists. */
+record ExportTask(String sql, Path outputBase, Measurement measurement, String table) {
+ record Measurement(PartialPath path, MeasurementSchema schema) {}
+}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportTextWriter.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportTextWriter.java
new file mode 100644
index 00000000000..be0295a5ad8
--- /dev/null
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportTextWriter.java
@@ -0,0 +1,126 @@
+/*
+ * 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.tool.data;
+
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVPrinter;
+import org.apache.commons.csv.QuoteMode;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+
+/**
+ * Shared CSV/SQL splitting and reliable IO. Partial files are closed without a completion event.
+ */
+final class ExportTextWriter implements AutoCloseable {
+ private final ExportConfig config;
+ private final ExportTask task;
+ private final ExportProgressReporter progress;
+ private final List<String> headers;
+ private BufferedWriter writer;
+ private CSVPrinter csv;
+ private Path file;
+ private int index;
+ private long fileRows;
+
+ ExportTextWriter(
+ ExportConfig config, ExportTask task, ExportProgressReporter progress, List<String> headers) {
+ this.config = config;
+ this.task = task;
+ this.progress = progress;
+ this.headers = headers;
+ }
+
+ private void open() throws Exception {
+ if (writer != null) {
+ return;
+ }
+ progress.checkActive();
+ file = Path.of(task.outputBase() + "_" + index + (headers == null ? ".sql" : ".csv"));
+ ExportFileNames.checkOutputCollision(file, config);
+ progress.fileStarted(file);
+ writer = Files.newBufferedWriter(file, StandardCharsets.UTF_8);
+ fileRows = 0;
+ if (headers != null) {
+ csv =
+ CSVFormat.Builder.create(CSVFormat.DEFAULT)
+ .setHeader()
+ .setSkipHeaderRecord(true)
+ .setEscape('\\')
+ .setQuoteMode(QuoteMode.NONE)
+ .build()
+ .print(writer);
+ csv.printRecord(headers);
+ }
+ }
+
+ synchronized void csvRow(List<Object> values) throws Exception {
+ open();
+ csv.printRecord(values);
+ rowWritten();
+ }
+
+ synchronized void sqlRow(String sql) throws Exception {
+ open();
+ writer.write(sql);
+ rowWritten();
+ }
+
+ private void rowWritten() throws IOException {
+ fileRows++;
+ progress.rowsWritten(1);
+ if (fileRows >= config.linesPerFile) {
+ completeFile();
+ }
+ }
+
+ synchronized void complete() throws IOException {
+ completeFile();
+ }
+
+ private void completeFile() throws IOException {
+ if (writer == null) {
+ return;
+ }
+ close();
+ progress.fileCompleted(file, fileRows);
+ index++;
+ }
+
+ @Override
+ public synchronized void close() throws IOException {
+ if (writer != null) {
+ try {
+ if (csv != null) {
+ csv.close();
+ } else {
+ writer.close();
+ }
+ } finally {
+ writer = null;
+ csv = null;
+ }
+ }
+ }
+}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportTimeRange.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportTimeRange.java
new file mode 100644
index 00000000000..9205c9cd67e
--- /dev/null
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportTimeRange.java
@@ -0,0 +1,84 @@
+/*
+ * 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.tool.data;
+
+import org.apache.iotdb.cli.i18n.CliMessages;
+import org.apache.iotdb.commons.queryengine.utils.DateTimeUtils;
+import org.apache.iotdb.exception.ArgsErrorException;
+
+import java.time.ZoneId;
+
+/** A closed interval, resolved using the connected server's precision and session time zone. */
+final class ExportTimeRange {
+ private final Long start;
+ private final Long end;
+
+ private ExportTimeRange(Long start, Long end) {
+ this.start = start;
+ this.end = end;
+ }
+
+ static ExportTimeRange parse(String start, String end, ZoneId zone, String precision)
+ throws ArgsErrorException {
+ Long startValue = parseBound(start, zone, precision);
+ Long endValue = parseBound(end, zone, precision);
+ if (startValue != null && endValue != null && startValue > endValue) {
+ throw new ArgsErrorException(
+ CliMessages.EXCEPTION_START_TIME_MUST_BE_LESS_THAN_OR_EQUAL_TO_END_TIME_3551199C);
+ }
+ return new ExportTimeRange(startValue, endValue);
+ }
+
+ private static Long parseBound(String value, ZoneId zone, String precision)
+ throws ArgsErrorException {
+ if (value == null || value.trim().isEmpty()) {
+ return null;
+ }
+ String trimmed = value.trim();
+ try {
+ return Long.parseLong(trimmed);
+ } catch (NumberFormatException ignored) {
+ try {
+ return DateTimeUtils.convertDatetimeStrToLong(trimmed, zone, precision);
+ } catch (RuntimeException e) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages.EXCEPTION_CANNOT_PARSE_TIME_ARG_ARG_40257DF7, value, e.getMessage()));
+ }
+ }
+ }
+
+ String whereClause() {
+ if (start == null && end == null) {
+ return "";
+ }
+ StringBuilder sql = new StringBuilder(" WHERE ");
+ if (start != null) {
+ sql.append("time >= ").append(start);
+ }
+ if (start != null && end != null) {
+ sql.append(" AND ");
+ }
+ if (end != null) {
+ sql.append("time <= ").append(end);
+ }
+ return sql.toString();
+ }
+}
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/MeasurementFile.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/MeasurementFile.java
new file mode 100644
index 00000000000..d5f56c5f895
--- /dev/null
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/MeasurementFile.java
@@ -0,0 +1,145 @@
+/*
+ * 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.tool.data;
+
+import org.apache.iotdb.cli.i18n.CliMessages;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.exception.ArgsErrorException;
+
+import java.io.BufferedInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.CodingErrorAction;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Validates the entire UTF-8 input before any metadata or data query is issued. */
+record MeasurementFile(Path source, List<PatternLine> patterns, int lines, int duplicates) {
+ record PatternLine(int line, String original, PartialPath path) {}
+
+ static MeasurementFile read(Path source) throws IOException, ArgsErrorException {
+ if (!Files.isRegularFile(source) || !Files.isReadable(source)) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages.EXCEPTION_MEASUREMENT_INPUT_MUST_BE_A_READABLE_REGULAR_FILE_ARG_43D4A272,
+ source));
+ }
+ Map<String, PatternLine> distinct = new LinkedHashMap<>();
+ int lineNumber = 0;
+ int duplicates = 0;
+ String original = "";
+ try (BufferedInputStream reader = new BufferedInputStream(Files.newInputStream(source))) {
+ byte[] bytes;
+ while ((bytes = readLine(reader)) != null) {
+ try {
+ original =
+ StandardCharsets.UTF_8
+ .newDecoder()
+ .onMalformedInput(CodingErrorAction.REPORT)
+ .onUnmappableCharacter(CodingErrorAction.REPORT)
+ .decode(ByteBuffer.wrap(bytes))
+ .toString();
+ } catch (IOException e) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages
+ .EXCEPTION_INVALID_MEASUREMENT_INPUT_ARG_AT_LINE_ARG_CONTENT_ARG_ARG_27A0F948,
+ source,
+ lineNumber + 1,
+ new String(bytes, StandardCharsets.UTF_8),
+ e.getMessage()));
+ }
+ lineNumber++;
+ String pattern = original;
+ if (lineNumber == 1 && pattern.startsWith("\uFEFF")) {
+ pattern = pattern.substring(1);
+ }
+ pattern = pattern.trim();
+ if (pattern.isEmpty() || pattern.startsWith("#")) {
+ continue;
+ }
+ try {
+ PartialPath path = new PartialPath(pattern);
+ if (!"root".equals(path.getFirstNode()) || path.getNodeLength() < 2) {
+ throw new ArgsErrorException(
+ CliMessages
+ .EXCEPTION_A_COMPLETE_IOTDB_PATH_PATTERN_STARTING_WITH_ROOT_IS_REQUIRED_7D000A79);
+ }
+ PatternLine previous =
+ distinct.putIfAbsent(path.getFullPath(), new PatternLine(lineNumber, original, path));
+ if (previous != null) {
+ duplicates++;
+ }
+ } catch (IllegalPathException | ArgsErrorException e) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages
+ .EXCEPTION_INVALID_MEASUREMENT_INPUT_ARG_AT_LINE_ARG_CONTENT_ARG_ARG_27A0F948,
+ source,
+ lineNumber,
+ original,
+ e.getMessage()));
+ }
+ }
+ } catch (IOException e) {
+ throw new IOException(
+ String.format(
+ CliMessages.EXCEPTION_CANNOT_READ_UTF_8_MEASUREMENT_FILE_ARG_AT_LINE_ARG_ARG_A5739D3F,
+ source,
+ lineNumber + 1,
+ e.getMessage()),
+ e);
+ }
+ if (distinct.isEmpty()) {
+ throw new ArgsErrorException(
+ String.format(
+ CliMessages.EXCEPTION_MEASUREMENT_INPUT_CONTAINS_NO_PATH_PATTERNS_ARG_96760070,
+ source));
+ }
+ return new MeasurementFile(
+ source, List.copyOf(new ArrayList<>(distinct.values())), lineNumber, duplicates);
+ }
+
+ private static byte[] readLine(BufferedInputStream input) throws IOException {
+ ByteArrayOutputStream line = new ByteArrayOutputStream();
+ int next;
+ while ((next = input.read()) != -1) {
+ if (next == '\n') {
+ return line.toByteArray();
+ }
+ if (next == '\r') {
+ input.mark(1);
+ if (input.read() != '\n') {
+ input.reset();
+ }
+ return line.toByteArray();
+ }
+ line.write(next);
+ }
+ return line.size() == 0 ? null : line.toByteArray();
+ }
+}
diff --git a/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/DataToolRegressionTest.java b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/DataToolRegressionTest.java
index ff4a59ada91..43f90cca6a5 100644
--- a/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/DataToolRegressionTest.java
+++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/DataToolRegressionTest.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.tool.data;
+import org.apache.iotdb.cli.i18n.CliMessages;
+import org.apache.iotdb.exception.ArgsErrorException;
import org.apache.iotdb.isession.ITableSession;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.isession.pool.ITableSessionPool;
@@ -33,6 +35,8 @@ import org.apache.iotdb.tool.tsfile.ImportTsFileScanTool;
import org.apache.commons.csv.CSVParser;
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.RowRecord;
+import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.write.record.Tablet;
import org.junit.After;
import org.junit.Before;
@@ -40,15 +44,19 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
-import java.io.File;
import java.io.IOException;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.io.StringWriter;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
+import java.nio.charset.StandardCharsets;
import java.nio.file.FileAlreadyExistsException;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -61,12 +69,15 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.mockito.AdditionalMatchers.not;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
@@ -87,8 +98,6 @@ public class DataToolRegressionTest extends AbstractDataTool {
for (Class<?> owner :
Arrays.asList(
AbstractDataTool.class,
- ExportDataTree.class,
- ExportDataTable.class,
ImportDataTree.class,
ImportDataTable.class,
ImportTsFileBase.class,
@@ -101,20 +110,14 @@ public class DataToolRegressionTest extends AbstractDataTool {
}
}
directory = temporaryFolder.getRoot().toPath();
- targetDirectory = directory + File.separator;
- targetFile = "dump";
- exportType = "sql";
fileType = "csv";
- timeFormat = "timestamp";
timeColumn = "Time";
timestampPrecision = "ms";
database = "db";
table = "t";
aligned = false;
- linesPerFile = 10000;
batchPointSize = 100;
threadNum = 1;
- needDataTypePrinted = false;
}
@After
@@ -126,31 +129,37 @@ public class DataToolRegressionTest extends AbstractDataTool {
@Test
public void testMultiDeviceSqlPreservesColumnMappingAndSplitsFiles() throws Exception {
- linesPerFile = 1;
- mockTree(
- dataSet(
- List.of("Time", "root.a.d1.s1", "root.a.d2.s2", "root.a.d1.s3"),
- List.of("INT64", "INT32", "INT32", "INT32"),
- new Object[] {1L, 11, 22, 33},
- new Object[] {2L, null, 44, null}));
- assertTrue(new ExportDataTree().exportBySql("select * from root.a.*", 0));
- assertEquals("INSERT INTO root.a.d1(TIMESTAMP,s1,s3) VALUES (1,11,33);\n", read("dump0_0.sql"));
- assertEquals("INSERT INTO root.a.d2(TIMESTAMP,s2) VALUES (1,22);\n", read("dump0_1.sql"));
- assertEquals("INSERT INTO root.a.d2(TIMESTAMP,s2) VALUES (2,44);\n", read("dump0_2.sql"));
- assertFalse(Files.exists(directory.resolve("dump0_3.sql")));
+ Session session =
+ mockTree(
+ dataSet(
+ List.of("Time", "root.a.d1.s1", "root.a.d2.s2", "root.a.d1.s3"),
+ List.of("INT64", "INT32", "INT32", "INT32"),
+ new Object[] {1L, 11, 22, 33},
+ new Object[] {2L, null, 44, null}));
+ exportTree(session, "select * from root.a.*", 0, "-ft", "sql", "-lpf", "1");
+ // -lpf counts data rows, so the statements of one row share a file.
+ assertEquals(
+ "INSERT INTO root.a.d1(TIMESTAMP,s1,s3) VALUES (1,11,33);\n"
+ + "INSERT INTO root.a.d2(TIMESTAMP,s2) VALUES (1,22);\n",
+ read("dump0_0.sql"));
+ assertEquals("INSERT INTO root.a.d2(TIMESTAMP,s2) VALUES (2,44);\n", read("dump0_1.sql"));
+ assertFalse(Files.exists(directory.resolve("dump0_2.sql")));
}
@Test
public void testCustomTableQueriesUseDistinctFilesAndActualTables() throws Exception {
ITableSession session = mockTable("table_a", "table_b");
+ describe(session, "table_a", "v", "INT32", "FIELD");
+ describe(session, "table_b", "v", "INT32", "FIELD");
+ String qualified = "SELECT * FROM \"db\".\"table_b\" WHERE tag = 'from table_a'";
SessionDataSet first =
dataSet(List.of("time", "v"), List.of("TIMESTAMP", "INT32"), new Object[] {1L, 11});
+ when(session.executeQueryStatement(eq("select * from table_a"), anyLong())).thenReturn(first);
SessionDataSet second =
dataSet(List.of("time", "v"), List.of("TIMESTAMP", "INT32"), new Object[] {2L, 22});
- when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(first, second);
- ExportDataTable tool = new ExportDataTable();
- assertTrue(tool.exportBySql("select * from table_a", 0));
- assertTrue(tool.exportBySql("SELECT * FROM \"db\".\"table_b\" WHERE tag = 'from table_a'", 1));
+ when(session.executeQueryStatement(eq(qualified), anyLong())).thenReturn(second);
+ exportTable(session, "select * from table_a", 0, "-ft", "sql");
+ exportTable(session, qualified, 1, "-ft", "sql");
assertEquals("insert into \"table_a\"(\"time\",\"v\") values(1,11);\n", read("dump0_0.sql"));
assertEquals("insert into \"table_b\"(\"time\",\"v\") values(2,22);\n", read("dump1_0.sql"));
}
@@ -158,27 +167,26 @@ public class DataToolRegressionTest extends AbstractDataTool {
@Test
public void testCustomTableQueryOverridesTableFilter() throws Exception {
ITableSession session = mockTable("table_a", "table_b");
- table = "table_a";
- set(ExportDataTable.class, "tables", new ArrayList<>(List.of("table_a")));
+ describe(session, "table_b", "v", "INT32", "FIELD");
SessionDataSet result =
dataSet(List.of("time", "v"), List.of("TIMESTAMP", "INT32"), new Object[] {1L, 11});
- when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(result);
+ when(session.executeQueryStatement(eq("select * from table_b"), anyLong())).thenReturn(result);
- assertTrue(new ExportDataTable().exportBySql("select * from table_b", 0));
+ exportTable(session, "select * from table_b", 0, "-ft", "sql", "-table", "table_a");
assertEquals("insert into \"table_b\"(\"time\",\"v\") values(1,11);\n", read("dump0_0.sql"));
}
@Test
public void testDefaultTableExportStillUsesTableFilter() throws Exception {
ITableSession session = mockTable("table_a", "table_b");
- table = "table_a";
- set(ExportDataTable.class, "tables", new ArrayList<>(List.of("table_a")));
+ describe(session, "table_a", "v", "INT32", "FIELD");
SessionDataSet result =
dataSet(List.of("time", "v"), List.of("TIMESTAMP", "INT32"), new Object[] {1L, 11});
- when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(result);
+ when(session.executeQueryStatement(eq("SELECT * FROM \"table_a\""), anyLong()))
+ .thenReturn(result);
- assertTrue(new ExportDataTable().exportBySql(null, 0));
- verify(session).executeQueryStatement("select * from \"table_a\"", timeout);
+ exportTable(session, null, 0, "-ft", "sql", "-table", "table_a");
+ verify(session).executeQueryStatement(eq("SELECT * FROM \"table_a\""), anyLong());
assertEquals("insert into \"table_a\"(\"time\",\"v\") values(1,11);\n", read("dump0_0.sql"));
assertFalse(Files.exists(directory.resolve("dump1_0.sql")));
}
@@ -186,10 +194,12 @@ public class DataToolRegressionTest extends AbstractDataTool {
@Test
public void testQuotedTableAndColumnNames() throws Exception {
ITableSession session = mockTable("odd\"name");
+ describe(session, "odd\"name", "select", "INT32", "FIELD");
SessionDataSet result =
dataSet(List.of("time", "select"), List.of("TIMESTAMP", "INT32"), new Object[] {1L, 11});
- when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(result);
- assertTrue(new ExportDataTable().exportBySql("select * from \"ODD\"\"NAME\"", 0));
+ when(session.executeQueryStatement(eq("select * from \"ODD\"\"NAME\""), anyLong()))
+ .thenReturn(result);
+ exportTable(session, "select * from \"ODD\"\"NAME\"", 0, "-ft", "sql");
assertEquals(
"insert into \"odd\"\"name\"(\"time\",\"select\") values(1,11);\n", read("dump0_0.sql"));
}
@@ -197,26 +207,41 @@ public class DataToolRegressionTest extends AbstractDataTool {
@Test
public void testAmbiguousAndCrossDatabaseTableQueriesFailBeforeWriting() throws Exception {
ITableSession session = mockTable("table_a", "table_b");
- ExportDataTable tool = new ExportDataTable();
- assertFalse(
- tool.exportBySql("select * from table_a join table_b on table_a.time = table_b.time", 0));
- assertFalse(tool.exportBySql("select * from other_db.table_b", 1));
- assertFalse(tool.exportBySql("select * from", 2));
- assertFalse(tool.exportBySql("select * from table_b #invalid_character", 3));
- verify(session, never()).executeQueryStatement(anyString(), anyLong());
+ String message =
+ String.format(
+ CliMessages
+ .EXCEPTION_SQL_TSFILE_EXPORT_REQUIRES_A_QUERY_FROM_A_SINGLE_TABLE_IN_DATABASE_ARG_C9EB3768,
+ "db");
+ for (String query :
+ List.of(
+ "select * from table_a join table_b on table_a.time = table_b.time",
+ "select * from other_db.table_b",
+ "select * from",
+ "select * from table_b #invalid_character",
+ // All queries are checked before the first one runs.
+ "select * from table_a; select * from table_a, table_b")) {
+ ArgsErrorException e =
+ assertThrows(
+ query,
+ ArgsErrorException.class,
+ () -> exportTable(session, null, 0, "-ft", "sql", "-q", query));
+ assertEquals(query, message, e.getMessage());
+ }
+ verify(session, never()).executeQueryStatement(not(eq("show tables")), anyLong());
assertFalse(Files.exists(directory.resolve("dump0_0.sql")));
}
@Test
public void testTableSqlEscapesStringsAndPreservesNull() throws Exception {
ITableSession session = mockTable("t");
+ describe(session, "t", "tag", "STRING", "TAG", "value", "TEXT", "FIELD");
SessionDataSet result =
dataSet(
List.of("time", "tag", "value"),
List.of("TIMESTAMP", "STRING", "TEXT"),
new Object[] {1L, "O'Brien\\path", null});
- when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(result);
- assertTrue(new ExportDataTable().exportBySql("select * from t", 0));
+ when(session.executeQueryStatement(eq("select * from t"), anyLong())).thenReturn(result);
+ exportTable(session, "select * from t", 0, "-ft", "sql");
assertEquals(
"insert into \"t\"(\"time\",\"tag\",\"value\") values(1,'O''Brien\\path',NULL);\n",
read("dump0_0.sql"));
@@ -224,12 +249,13 @@ public class DataToolRegressionTest extends AbstractDataTool {
@Test
public void testAlignedTreeSqlEscapesStringsAndPreservesNull() throws Exception {
- mockTree(
- dataSet(
- List.of("Time", "Device", "s1", "s2"),
- List.of("INT64", "TEXT", "TEXT", "STRING"),
- new Object[] {1L, "root.a.d1", "a'b\"\\c", null}));
- assertTrue(new ExportDataTree().exportBySql("select * from root.a.* align by device", 0));
+ Session session =
+ mockTree(
+ dataSet(
+ List.of("Time", "Device", "s1", "s2"),
+ List.of("INT64", "TEXT", "TEXT", "STRING"),
+ new Object[] {1L, "root.a.d1", "a'b\"\\c", null}));
+ exportTree(session, "select * from root.a.* align by device", 0, "-ft", "sql");
assertEquals(
"INSERT INTO root.a.d1(TIMESTAMP,s1,s2) ALIGNED VALUES (1,'a''b\"\\c',NULL);\n",
read("dump0_0.sql"));
@@ -237,43 +263,49 @@ public class DataToolRegressionTest extends AbstractDataTool {
@Test
public void testTreeCsvTypeHeadersInEverySplitFile() throws Exception {
- exportType = "csv";
- linesPerFile = 1;
- needDataTypePrinted = true;
- mockTree(
- dataSet(
- List.of("Time", "Device", "s1"),
- List.of("INT64", "TEXT", "DOUBLE"),
- new Object[] {1L, "root.a.d1", 1.5},
- new Object[] {2L, "root.a.d1", 2.5}));
- assertTrue(new ExportDataTree().exportBySql("select * from root.a.* align by device", 0));
+ Session typed =
+ mockTree(
+ dataSet(
+ List.of("Time", "Device", "s1"),
+ List.of("INT64", "TEXT", "DOUBLE"),
+ new Object[] {1L, "root.a.d1", 1.5},
+ new Object[] {2L, "root.a.d1", 2.5}));
+ exportTree(
+ typed,
+ "select * from root.a.* align by device",
+ 0,
+ "-ft",
+ "csv",
+ "-lpf",
+ "1",
+ "-dt",
+ "true");
assertEquals("Time,Device,s1(DOUBLE)", firstLine("dump0_0.csv"));
assertEquals("Time,Device,s1(DOUBLE)", firstLine("dump0_1.csv"));
- needDataTypePrinted = false;
- mockTree(
- dataSet(
- List.of("Time", "root.a.d1.s1"), List.of("INT64", "DOUBLE"), new Object[] {1L, 1.5}));
- assertTrue(new ExportDataTree().exportBySql("select * from root.a.d1", 1));
+ Session plain =
+ mockTree(
+ dataSet(
+ List.of("Time", "root.a.d1.s1"),
+ List.of("INT64", "DOUBLE"),
+ new Object[] {1L, 1.5}));
+ exportTree(plain, "select * from root.a.d1", 1, "-ft", "csv");
assertEquals("Time,root.a.d1.s1", firstLine("dump1_0.csv"));
}
@Test
public void testTableCsvTypeHeadersAndMultipleQueryFiles() throws Exception {
- exportType = "csv";
- needDataTypePrinted = true;
ITableSession session = mockTable("t");
SessionDataSet first =
dataSet(
List.of("time", "device", "v"),
List.of("TIMESTAMP", "STRING", "DOUBLE"),
new Object[] {1L, "sensor", 1.5});
+ when(session.executeQueryStatement(eq("select * from t"), anyLong())).thenReturn(first);
SessionDataSet second =
dataSet(List.of("time", "v"), List.of("TIMESTAMP", "DOUBLE"), new Object[] {2L, 2.5});
- when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(first, second);
- ExportDataTable tool = new ExportDataTable();
- assertTrue(tool.exportBySql("select * from t", 0));
- needDataTypePrinted = false;
- assertTrue(tool.exportBySql("select time,v from t", 1));
+ when(session.executeQueryStatement(eq("select time,v from t"), anyLong())).thenReturn(second);
+ exportTable(session, "select * from t", 0, "-ft", "csv", "-dt", "true");
+ exportTable(session, "select time,v from t", 1, "-ft", "csv");
assertEquals("time,device(STRING),v(DOUBLE)", firstLine("dump0_0.csv"));
assertEquals("time,v", firstLine("dump1_0.csv"));
}
@@ -283,8 +315,12 @@ public class DataToolRegressionTest extends AbstractDataTool {
Session session = mock(Session.class);
when(session.executeQueryStatement(anyString(), anyLong()))
.thenThrow(new IoTDBConnectionException("disconnect"));
- set(ExportDataTree.class, "session", session);
- assertFalse(new ExportDataTree().exportBySql("select * from root.a.d1", 0));
+ // ExportData.run reports the failure: FINISH is FAILED and the exit code is 1.
+ IoTDBConnectionException e =
+ assertThrows(
+ IoTDBConnectionException.class,
+ () -> exportTree(session, "select * from root.a.d1", 0, "-ft", "sql"));
+ assertEquals("disconnect", e.getMessage());
}
@Test
@@ -709,8 +745,6 @@ public class DataToolRegressionTest extends AbstractDataTool {
})
.when(session)
.insert(any(Tablet.class));
- exportType = "csv";
- needDataTypePrinted = typed;
for (int i = 0; i < 2; i++) {
List<String> columns =
i == 0
@@ -719,7 +753,7 @@ public class DataToolRegressionTest extends AbstractDataTool {
Object[] row = i == 0 ? new Object[] {1L, 1.5, 2.5} : new Object[] {1L, 2.5, 1.5};
SessionDataSet result = dataSet(columns, List.of("TIMESTAMP", "DOUBLE", "DOUBLE"), row);
when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(result);
- assertTrue(new ExportDataTable().exportBySql("select * from t", i));
+ exportTable(session, "select * from t", i, "-ft", "csv", "-dt", String.valueOf(typed));
long failedFiles = loadFileFailedNum.sum();
long successfulFiles = loadFileSuccessfulNum.sum();
ImportDataScanTool.putToQueue(directory.resolve("dump" + i + "_0.csv").toString());
@@ -750,14 +784,12 @@ public class DataToolRegressionTest extends AbstractDataTool {
})
.when(session)
.insert(any(Tablet.class));
- exportType = "csv";
for (boolean typed : List.of(false, true)) {
- needDataTypePrinted = typed;
SessionDataSet result =
dataSet(List.of("time", column), List.of("TIMESTAMP", "DOUBLE"), new Object[] {1L, 1.5});
when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(result);
int index = typed ? 1 : 0;
- assertTrue(new ExportDataTable().exportBySql("select * from t", index));
+ exportTable(session, "select * from t", index, "-ft", "csv", "-dt", String.valueOf(typed));
String csv = "dump" + index + "_0.csv";
assertEquals("time," + column + (typed ? "(DOUBLE)" : ""), firstLine(csv));
try (CSVParser records = readCsvFile(directory.resolve(csv).toString())) {
@@ -1001,20 +1033,109 @@ public class DataToolRegressionTest extends AbstractDataTool {
}
}
- private static void mockTree(SessionDataSet dataSet) throws Exception {
+ /** Exports one query to dump<index> files, as ExportData.run does for one -q statement. */
+ private void exportTree(Session session, String sql, int index, String... options)
+ throws Exception {
+ ExportConfig config = exportConfig(false, options);
+ try (ExportProgressReporter progress = progress();
+ ExportDataTree exporter = new ExportDataTree(config, progress, session)) {
+ exporter.export(
+ new ExportTask(sql, config.targetDirectory.resolve(config.prefix + index), null, null));
+ }
+ }
+
+ /**
+ * Plans and exports like ExportData.run: the given query is written to dump<index> files, a null
+ * query exports the -q option or, without one, the tables of the database.
+ */
+ private void exportTable(ITableSession session, String sql, int index, String... options)
+ throws Exception {
+ ExportConfig config = exportConfig(true, options);
+ try (ExportProgressReporter progress = progress();
+ ExportDataTable exporter = new ExportDataTable(config, progress, session)) {
+ List<ExportTask> tasks;
+ if (sql != null) {
+ tasks =
+ exporter.planQueries(
+ List.of(
+ new ExportTask(
+ sql, config.targetDirectory.resolve(config.prefix + index), null, null)));
+ } else if (config.query != null) {
+ tasks = exporter.planQueries(ExportData.queryTasks(config, config.query));
+ } else {
+ tasks = exporter.planTables(exporter.timeRange());
+ }
+ for (ExportTask task : tasks) {
+ exporter.export(task);
+ }
+ }
+ }
+
+ private ExportConfig exportConfig(boolean table, String... options) throws Exception {
+ List<String> args = new ArrayList<>(List.of("-t", directory.toString(), "-tf", "timestamp"));
+ if (table) {
+ args.addAll(List.of("-sql_dialect", "table", "-db", "db"));
+ }
+ args.addAll(Arrays.asList(options));
+ return ExportConfig.parse(args.toArray(new String[0]), ZonedDateTime.now());
+ }
+
+ private static ExportProgressReporter progress() {
+ return new ExportProgressReporter(
+ new StringWriter(),
+ "tree",
+ System.nanoTime(),
+ new PrintStream(OutputStream.nullOutputStream()));
+ }
+
+ private static Session mockTree(SessionDataSet dataSet) throws Exception {
Session session = mock(Session.class);
when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(dataSet);
- set(ExportDataTree.class, "session", session);
+ return session;
}
+ /** A table session whose SHOW TABLES lists the given tables. */
private static ITableSession mockTable(String... names) throws Exception {
ITableSession session = mock(ITableSession.class);
- set(ExportDataTable.class, "tableSession", session);
- set(ExportDataTable.class, "tables", new ArrayList<>(Arrays.asList(names)));
- set(ExportDataTable.class, "allTables", new ArrayList<>(Arrays.asList(names)));
+ String[][] rows = new String[names.length][];
+ for (int i = 0; i < names.length; i++) {
+ rows[i] = new String[] {names[i]};
+ }
+ when(session.executeQueryStatement(eq("show tables"), anyLong()))
+ .thenAnswer(invocation -> rows(rows));
return session;
}
+ /** Answers DESCRIBE for a table: its time column, then name, type and category per column. */
+ private static void describe(ITableSession session, String table, String... columns)
+ throws Exception {
+ List<String[]> rows = new ArrayList<>();
+ rows.add(new String[] {"time", "TIMESTAMP", "TIME"});
+ for (int i = 0; i < columns.length; i += 3) {
+ rows.add(Arrays.copyOfRange(columns, i, i + 3));
+ }
+ when(session.executeQueryStatement(
+ eq("DESCRIBE \"" + table.replace("\"", "\"\"") + "\""), anyLong()))
+ .thenAnswer(invocation -> rows(rows.toArray(new String[0][])));
+ }
+
+ /** A metadata result whose rows hold the given strings. */
+ private static SessionDataSet rows(String[]... rows) throws Exception {
+ SessionDataSet dataSet = mock(SessionDataSet.class);
+ AtomicInteger row = new AtomicInteger();
+ when(dataSet.hasNext()).thenAnswer(invocation -> row.get() < rows.length);
+ when(dataSet.next())
+ .thenAnswer(
+ invocation -> {
+ RowRecord record = new RowRecord(0);
+ for (String value : rows[row.getAndIncrement()]) {
+ record.addField(new Binary(value, StandardCharsets.UTF_8), TSDataType.TEXT);
+ }
+ return record;
+ });
+ return dataSet;
+ }
+
private static SessionDataSet dataSet(List<String> names, List<String> types, Object[]... rows)
throws Exception {
SessionDataSet dataSet = mock(SessionDataSet.class);
diff --git a/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ExportDataFeatureTest.java b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ExportDataFeatureTest.java
new file mode 100644
index 00000000000..1dd2ee68b7c
--- /dev/null
+++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ExportDataFeatureTest.java
@@ -0,0 +1,726 @@
+/*
+ * 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.tool.data;
+
+import org.apache.iotdb.cli.i18n.CliMessages;
+import org.apache.iotdb.exception.ArgsErrorException;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tool.common.Constants;
+
+import org.apache.commons.cli.ParseException;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.Field;
+import org.apache.tsfile.read.common.RowRecord;
+import org.apache.tsfile.utils.Binary;
+import org.jline.reader.EndOfFileException;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.io.StringWriter;
+import java.io.Writer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Input, planning and lifecycle contracts exercised without a server. */
+public class ExportDataFeatureTest {
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+ private static final ZonedDateTime START = ZonedDateTime.parse("2026-10-14T12:34:56+08:00");
+
+ @Test
+ public void preservesPrefixOptionsAndRejectsFAlias() throws Exception {
+ ExportConfig defaults = config("-mf", "measurements.txt");
+ assertEquals("", defaults.prefix);
+ assertEquals("log_export_data_20261014123456.log", defaults.logFile.getFileName().toString());
+ assertEquals("backup", config("-mf", "measurements.txt", "-pfn", "backup").prefix);
+ assertEquals(
+ "backup", config("-mf", "measurements.txt", "--prefix_file_name", "backup").prefix);
+ assertEquals("dump", config("-q", "select * from root.**").prefix);
+ assertThrows(ParseException.class, () -> config("-f", "backup"));
+ assertThrows(ArgsErrorException.class, () -> config("-progress"));
+ assertThrows(
+ ArgsErrorException.class, () -> config("-mf", "file", "-q", "select s from root.sg.d"));
+ assertThrows(
+ ArgsErrorException.class,
+ () -> config("-sql_dialect", "table", "-db", "db", "-mf", "file"));
+ assertThrows(ArgsErrorException.class, () -> config("-lpf", "0"));
+ assertThrows(ArgsErrorException.class, () -> config("-dt", "maybe"));
+ assertThrows(ParseException.class, () -> config("-p"));
+ }
+
+ @Test
+ public void acceptsBomCommentsQuotedPathsAndIoTDBPatterns() throws Exception {
+ Path input =
+ input(
+ "\uFEFF# comment\r\n\n root.sg.d.pre* \nroot.sg.**\rroot.sg.d.`a.b`\nroot.sg.d.pre*\n");
+ MeasurementFile file = MeasurementFile.read(input);
+ assertEquals(6, file.lines());
+ assertEquals(3, file.patterns().size());
+ assertEquals(1, file.duplicates());
+ assertEquals("root.sg.d.pre*", file.patterns().get(0).path().getFullPath());
+ assertEquals("`a.b`", file.patterns().get(2).path().getMeasurement());
+ }
+
+ @Test
+ public void reportsInvalidLineAndUtf8AtActualLine() throws Exception {
+ Path input = input("root.sg.d.s\n# valid\nroot.sg..s\n");
+ ArgsErrorException invalid =
+ assertThrows(ArgsErrorException.class, () -> MeasurementFile.read(input));
+ assertTrue(invalid.getMessage().contains("3"));
+ assertTrue(invalid.getMessage().contains("root.sg..s"));
+ byte[] bytes = "root.sg.d.s\n# comment\n".getBytes(StandardCharsets.UTF_8);
+ byte[] bad = Arrays.copyOf(bytes, bytes.length + 2);
+ bad[bytes.length] = (byte) 0xc3;
+ bad[bytes.length + 1] = 0x28;
+ Files.write(input, bad);
+ ArgsErrorException utf8 =
+ assertThrows(ArgsErrorException.class, () -> MeasurementFile.read(input));
+ assertTrue(utf8.getMessage().contains("3"));
+ assertThrows(ArgsErrorException.class, () -> MeasurementFile.read(input("# empty\n \n")));
+ assertThrows(ArgsErrorException.class, () -> MeasurementFile.read(input("other.sg.s\n")));
+ }
+
+ @Test
+ public void plansEachDistinctPatternAndDeduplicatesActualMeasurements() throws Exception {
+ Path input = input("root.sg.d.pre*\nroot.sg.**\nroot.sg.d.pre*\nroot.empty.**\n");
+ ExportConfig config = config("-mf", input.toString(), "-pfn", "backup");
+ StringWriter log = new StringWriter();
+ Session session = mock(Session.class);
+ SessionDataSet first = metadata("root.sg.d.pre1");
+ SessionDataSet second = metadata("root.sg.d.pre1", "root.sg.d.pre2");
+ SessionDataSet empty = metadata();
+ when(session.executeQueryStatement("show timeseries root.sg.d.pre*", config.timeout))
+ .thenReturn(first);
+ when(session.executeQueryStatement("show timeseries root.sg.**", config.timeout))
+ .thenReturn(second);
+ when(session.executeQueryStatement("show timeseries root.empty.**", config.timeout))
+ .thenReturn(empty);
+ ByteArrayOutputStream console = new ByteArrayOutputStream();
+ try (ExportProgressReporter reporter =
+ new ExportProgressReporter(
+ log,
+ "tree",
+ System.nanoTime(),
+ new PrintStream(console, true, StandardCharsets.UTF_8));
+ ExportDataTree exporter = new ExportDataTree(config, reporter, session)) {
+ List<ExportTask> tasks =
+ exporter.planMeasurements(
+ MeasurementFile.read(input),
+ ExportTimeRange.parse("100", "200", ZoneId.of("UTC"), "ms"));
+ assertEquals(2, tasks.size());
+ assertEquals(
+ "SELECT pre1 FROM root.sg.d WHERE time >= 100 AND time <= 200", tasks.get(0).sql());
+ assertEquals("backup_root.sg.d.pre1", tasks.get(0).outputBase().getFileName().toString());
+ assertEquals("root.sg.d.pre2", tasks.get(1).measurement().path().getFullPath());
+ reporter.finish(ExportProgressReporter.Status.SUCCESS, null);
+ }
+ verify(session, times(3)).executeQueryStatement(anyString(), anyLong());
+ verify(first).close();
+ verify(second).close();
+ verify(empty).close();
+ verify(session).close();
+ String content = log.toString();
+ String warning = event(log, "WARNING");
+ assertTrue(warning, warning.contains(member("line", 4)));
+ assertTrue(warning, warning.contains(member("pattern", "root.empty.**")));
+ // JSON escapes the backslashes of a Windows path.
+ assertTrue(warning, warning.contains(member("source", input.toString().replace("\\", "\\\\"))));
+ // The console names the unmatched pattern itself, without the long source path.
+ String expected =
+ String.format(
+ CliMessages.LOG_WARNING_ARG_05BE32E9,
+ String.format(
+ CliMessages.LOG_NO_MEASUREMENTS_MATCHED_ARG_MEASUREMENT_FILE_LINE_ARG_94C4EF10,
+ "root.empty.**",
+ 4));
+ String shown = console.toString(StandardCharsets.UTF_8);
+ assertTrue(shown, shown.lines().anyMatch(expected::equals));
+ // Measurement fields keep the requirement's order between the rows and the elapsed time.
+ assertTrue(
+ content,
+ content.contains(
+ String.join(
+ ",",
+ member("status", "RUNNING"),
+ member("exported_data_rows", 0),
+ member("exported_measurement_num", 0),
+ member("total_measurement_num", 2),
+ member("exported_percent", "0.00%"),
+ member("estimated_remaining", "UNKNOWN"))
+ + ",\"elapsed_time\":"));
+ assertTrue(content, content.contains(member("duplicate_matches", 1)));
+ }
+
+ @Test
+ public void failsOnlyWhenNoPatternMatches() throws Exception {
+ Path input = input("root.empty.**\n");
+ ExportConfig config = config("-mf", input.toString());
+ Session session = mock(Session.class);
+ SessionDataSet empty = metadata();
+ when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(empty);
+ try (ExportProgressReporter reporter = reporter(new StringWriter());
+ ExportDataTree exporter = new ExportDataTree(config, reporter, session)) {
+ assertThrows(
+ ArgsErrorException.class,
+ () ->
+ exporter.planMeasurements(
+ MeasurementFile.read(input),
+ ExportTimeRange.parse(null, null, ZoneId.of("UTC"), "ms")));
+ }
+ }
+
+ @Test
+ public void timeRangesUsePrecisionTimezoneAndInclusiveBounds() throws Exception {
+ ZoneId zone = ZoneId.of("Asia/Shanghai");
+ assertEquals(
+ " WHERE time >= 100 AND time <= 100",
+ ExportTimeRange.parse("100", "100", zone, "ns").whereClause());
+ assertEquals(" WHERE time <= 99", ExportTimeRange.parse(null, "99", zone, "ms").whereClause());
+ assertEquals(
+ " WHERE time >= 1000",
+ ExportTimeRange.parse("1970-01-01T08:00:00.001", null, zone, "us").whereClause());
+ assertEquals(
+ " WHERE time >= 1000000",
+ ExportTimeRange.parse("1970-01-01T00:00:00.001Z", null, zone, "ns").whereClause());
+ assertThrows(ArgsErrorException.class, () -> ExportTimeRange.parse("200", "100", zone, "ms"));
+ assertThrows(ArgsErrorException.class, () -> ExportTimeRange.parse("oops", null, zone, "ms"));
+ ExportConfig config = config("-q", "select s from root.sg.d", "-start_time", "oops");
+ assertEquals(
+ "select s from root.sg.d", ExportData.queryTasks(config, config.query).get(0).sql());
+ }
+
+ @Test
+ public void fileNamesEscapeAndRejectCollisionsAndLogAliases() throws Exception {
+ ExportConfig config = config("-mf", "file");
+ ExportFileNames names = new ExportFileNames();
+ assertEquals(
+ "root.sg.d.`a%3Ab`", names.measurement(config, "root.sg.d.`a:b`").getFileName().toString());
+ names.measurement(config, "root.sg.d.A");
+ assertThrows(ArgsErrorException.class, () -> names.measurement(config, "root.sg.d.a"));
+ assertThrows(
+ ArgsErrorException.class, () -> names.measurement(config, "root." + "x".repeat(230)));
+ Path file = folder.newFile().toPath();
+ Path alias = folder.getRoot().toPath().resolve("alias");
+ Files.createLink(alias, file);
+ assertThrows(ArgsErrorException.class, () -> ExportFileNames.checkLogCollision(alias, file));
+ }
+
+ @Test
+ public void independentTimerReportsWithoutRowsAndStopsAfterOneTerminal() throws Exception {
+ CountDownLatch tick = new CountDownLatch(1);
+ StringWriter log =
+ new StringWriter() {
+ @Override
+ public void write(String value) {
+ super.write(value);
+ if (value.contains(member("event", "PROGRESS"))) {
+ tick.countDown();
+ }
+ }
+ };
+ ByteArrayOutputStream console = new ByteArrayOutputStream();
+ try (ExportProgressReporter reporter =
+ new ExportProgressReporter(
+ log,
+ "tree",
+ System.nanoTime(),
+ new PrintStream(console, true, StandardCharsets.UTF_8))) {
+ reporter.start(
+ config("-q", "select * from root.**"), true, () -> fail("unexpected log error"));
+ assertTrue(tick.await(5, TimeUnit.SECONDS));
+ reporter.finish(ExportProgressReporter.Status.SUCCESS, null);
+ reporter.progress();
+ reporter.finish(ExportProgressReporter.Status.FAILED, new IOException("later"));
+ }
+ String content = log.toString();
+ assertTrue(
+ content,
+ content.startsWith(
+ "{" + member("progress_id", ProcessHandle.current().pid()) + ",\"timestamp\":"));
+ assertTrue(content, content.contains(member("mode", "query")));
+ // The requirement's fields lead each event, in its order; event details follow them.
+ assertTrue(
+ content,
+ Pattern.compile(
+ Pattern.quote(
+ String.join(
+ ",",
+ member("model", "tree"),
+ member("status", "RUNNING"),
+ member("exported_data_rows", 0)))
+ + ",\"elapsed_time\":\"00:00:0\\d\","
+ + Pattern.quote(
+ String.join(
+ ",",
+ member("event", "PROGRESS"),
+ member("level", "INFO"),
+ member("phase", "VALIDATING"))))
+ .matcher(content)
+ .find());
+ assertFalse(content.contains("exported_measurement_num"));
+ assertEquals(1, content.split(member("event", "FINISH"), -1).length - 1);
+ // Every event is one JSON object on its own line.
+ content.lines().forEach(line -> assertTrue(line, line.startsWith("{") && line.endsWith("}")));
+
+ // A query has no total, so the console shows rows, elapsed time and rate, then one summary.
+ String shown = console.toString(StandardCharsets.UTF_8);
+ assertTrue(
+ shown, printed(shown, CliMessages.LOG_EXPORT_STARTED_ARG_ARG_TARGET_ARG_LOG_ARG_BD58AECB));
+ assertTrue(shown, printed(shown, CliMessages.LOG_ARG_ROWS_ELAPSED_ARG_ARG_ROWS_S_AE3CF660));
+ assertTrue(
+ shown,
+ printed(shown, CliMessages.LOG_EXPORT_COMPLETELY_ROWS_ARG_FILES_ARG_ELAPSED_ARG_A94BBB8B));
+ assertFalse(shown, printed(shown, CliMessages.LOG_EXPORT_FAILED_AFTER_ARG_ARG_C123F0EF));
+ assertFalse(shown, shown.contains("{"));
+ // Scripts check the last line for the outcome, so the log path comes before the summary.
+ List<String> lines = shown.lines().collect(Collectors.toList());
+ assertTrue(
+ shown,
+ printed(
+ lines.get(lines.size() - 1),
+ CliMessages.LOG_EXPORT_COMPLETELY_ROWS_ARG_FILES_ARG_ELAPSED_ARG_A94BBB8B));
+ assertTrue(shown, printed(lines.get(lines.size() - 2), CliMessages.LOG_LOG_ARG_FBF61F63));
+ }
+
+ @Test
+ public void remainingTimeWaitsForSeveralMeasurements() throws Exception {
+ StringWriter log = new StringWriter();
+ ByteArrayOutputStream console = new ByteArrayOutputStream();
+ ExportConfig config = config();
+ try (ExportProgressReporter reporter =
+ new ExportProgressReporter(
+ log,
+ "tree",
+ System.nanoTime(),
+ new PrintStream(console, true, StandardCharsets.UTF_8))) {
+ reporter.discovered(MeasurementFile.read(input("root.sg.**\n")), 40, 40, 0);
+ for (int i = 0; i < 3; i++) {
+ reporter.taskStarted(
+ new ExportTask(
+ "select s from root.sg.d" + i,
+ config.targetDirectory.resolve("m" + i),
+ null,
+ null));
+ reporter.taskCompleted();
+ reporter.progress();
+ }
+ reporter.finish(ExportProgressReporter.Status.SUCCESS, null);
+ }
+ List<String> progress =
+ log.toString()
+ .lines()
+ .filter(line -> line.contains(member("event", "PROGRESS")))
+ .collect(Collectors.toList());
+ // 5% of 40 is two measurements, yet at least three must complete before estimating.
+ assertTrue(progress.get(1), progress.get(1).contains(member("estimated_remaining", "UNKNOWN")));
+ assertFalse(
+ progress.get(2), progress.get(2).contains(member("estimated_remaining", "UNKNOWN")));
+ List<String> shown =
+ console
+ .toString(StandardCharsets.UTF_8)
+ .lines()
+ .filter(
+ line ->
+ printed(
+ line,
+ CliMessages
+ .LOG_ARG_ARG_ARG_MEASUREMENTS_ARG_ROWS_ELAPSED_ARG_ETA_ARG_CURRENT_ARG_B6DEA0BD))
+ .collect(Collectors.toList());
+ assertEquals(3, shown.size());
+ assertTrue(shown.get(0), shown.get(0).startsWith("[2%] 1/40 "));
+ assertTrue(shown.get(1), shown.get(1).contains(" --"));
+ assertFalse(shown.get(2), shown.get(2).contains(" --"));
+ }
+
+ @Test
+ public void terminalRedrawsOneFittedLineAndRedirectedOutputGetsLines() {
+ ByteArrayOutputStream terminalBytes = new ByteArrayOutputStream();
+ ExportConsole terminal =
+ new ExportConsole(new PrintStream(terminalBytes, true, StandardCharsets.UTF_8), true, 30);
+ terminal.progress("Exporting 10%", "root.factory.line19.s0", "unused");
+ terminal.progress("Exporting 20%", "", "unused");
+ terminal.line("done");
+ // The tail keeps whole path nodes from the right so the line, with its last column free, fits
+ // 30 columns. A shorter redraw blanks what the previous one left behind, and a lasting line
+ // first erases the progress line instead of continuing it.
+ assertEquals(
+ "\rExporting 10% ...line19.s0"
+ + "\rExporting 20%"
+ + " ".repeat(14)
+ + "\r"
+ + " ".repeat(13)
+ + "\rdone"
+ + System.lineSeparator(),
+ terminalBytes.toString(StandardCharsets.UTF_8));
+
+ ByteArrayOutputStream plainBytes = new ByteArrayOutputStream();
+ ExportConsole plain =
+ new ExportConsole(new PrintStream(plainBytes, true, StandardCharsets.UTF_8), false, 30);
+ plain.progress("Exporting 10%", "root.factory.line19.s0", "[10%] plain");
+ plain.line("done");
+ assertEquals(
+ "[10%] plain" + System.lineSeparator() + "done" + System.lineSeparator(),
+ plainBytes.toString(StandardCharsets.UTF_8));
+
+ // CJK characters take two columns, so cutting counts columns rather than characters.
+ assertEquals(10, ExportConsole.width("导出中 59%"));
+ assertEquals("...乙丙", ExportConsole.left("测点甲乙丙", 7));
+ assertEquals("root.sg", ExportConsole.left("root.sg", 7));
+ }
+
+ @Test
+ public void terminalTooNarrowForTheLineGetsLinesInsteadOfRedraws() {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ ExportConsole console =
+ new ExportConsole(new PrintStream(bytes, true, StandardCharsets.UTF_8), true, 14);
+ console.progress("Exporting 10%", "root.factory.line19.s0", "[10%] first");
+ console.progress("Exporting 100%", "root.factory.line19.s0", "[100%] second");
+ console.line("done");
+ // 13 columns fit beside the free last column and are redrawn in place. A carriage return
+ // reaches only the last row of a wrapped line, so the 14-column line erases the redrawn one
+ // and becomes a line of its own.
+ assertEquals(
+ "\rExporting 10%"
+ + "\r"
+ + " ".repeat(13)
+ + "\r"
+ + "[100%] second"
+ + System.lineSeparator()
+ + "done"
+ + System.lineSeparator(),
+ bytes.toString(StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void jsonEscapesStringsAndKeepsIntegersAsNumbers() {
+ assertEquals(
+ "{\"text\":\"a\\\"b\\\\c\\nd\\te\\u0001中\",\"rows\":3,\"code\":130,\"path\":\"x y\"}",
+ ExportProgressReporter.json(
+ ExportProgressReporter.fields(
+ "text",
+ "a\"b\\c\nd\te" + (char) 1 + "中",
+ "rows",
+ 3L,
+ "code",
+ 130,
+ "skipped",
+ null,
+ "path",
+ Path.of("x y"))));
+ }
+
+ @Test
+ public void interactiveQueryIsReadBeforeProgressAndLogStart() throws Exception {
+ Path target = folder.getRoot().toPath();
+ PrintStream originalOut = System.out;
+ ByteArrayOutputStream console = new ByteArrayOutputStream();
+ AtomicBoolean prompted = new AtomicBoolean();
+ int code;
+ try {
+ System.setOut(new PrintStream(console, true, StandardCharsets.UTF_8));
+ code =
+ ExportData.run(
+ new String[] {"-ft", "csv", "-t", target.toString()},
+ config -> {
+ prompted.set(true);
+ // Neither progress events nor the log may exist while the user types the query.
+ assertEquals("", console.toString(StandardCharsets.UTF_8));
+ assertFalse(hasLog(target));
+ // End of input abandons the export before anything starts.
+ throw new EndOfFileException();
+ });
+ } finally {
+ System.setOut(originalOut);
+ }
+ assertTrue(prompted.get());
+ assertEquals(Constants.CODE_ERROR, code);
+ assertEquals(
+ CliMessages.EXCEPTION_EXPORT_CANCELLED_85D44F54 + System.lineSeparator(),
+ console.toString(StandardCharsets.UTF_8));
+ assertFalse(hasLog(target));
+ }
+
+ private static boolean hasLog(Path directory) throws IOException {
+ try (Stream<Path> files = Files.list(directory)) {
+ return files.anyMatch(file -> file.getFileName().toString().endsWith(".log"));
+ }
+ }
+
+ @Test
+ public void cancelledFinishReportsSignalStatusWithoutUnopenedOutput() throws Exception {
+ StringWriter log = new StringWriter();
+ ExportConfig config = config();
+ try (ExportProgressReporter reporter = reporter(log)) {
+ reporter.taskStarted(
+ new ExportTask(
+ "select s from root.sg.d", config.targetDirectory.resolve("dump0"), null, null));
+ reporter.cancel(130);
+ assertThrows(IOException.class, reporter::checkActive);
+ reporter.finish(ExportProgressReporter.Status.CANCELLED, null);
+ }
+ String content = log.toString();
+ // The task has not opened a file, so no output is named.
+ assertFalse(content, content.contains("\"output\":"));
+ assertTrue(content, content.contains(member("status", "CANCELLED")));
+ // The terminating signal decides the process status, which FINISH reports.
+ assertTrue(content, content.contains(member("exit_code", 130)));
+ }
+
+ @Test
+ public void queryModeReportsItsTimeRangeAsIgnored() throws Exception {
+ StringWriter queryLog = new StringWriter();
+ ByteArrayOutputStream console = new ByteArrayOutputStream();
+ try (ExportProgressReporter reporter =
+ new ExportProgressReporter(
+ queryLog,
+ "tree",
+ System.nanoTime(),
+ new PrintStream(console, true, StandardCharsets.UTF_8))) {
+ reporter.start(
+ config("-q", "select * from root.**", "-start_time", "100"),
+ true,
+ () -> fail("unexpected log error"));
+ reporter.finish(ExportProgressReporter.Status.SUCCESS, null);
+ }
+ // The query runs as written, so START must not suggest a time filter.
+ String start = event(queryLog, "START");
+ assertFalse(start, start.contains("\"start_time\":"));
+ String warning = event(queryLog, "WARNING");
+ assertTrue(warning, warning.contains(member("level", "WARN")));
+ assertTrue(warning, warning.contains(member("start_time", "100")));
+ String shown = console.toString(StandardCharsets.UTF_8);
+ assertTrue(
+ shown,
+ shown
+ .lines()
+ .anyMatch(
+ String.format(
+ CliMessages.LOG_WARNING_ARG_05BE32E9,
+ CliMessages
+ .LOG_START_TIME_AND_END_TIME_DO_NOT_APPLY_TO_QUERIES_AND_ARE_IGNORED_02DC37CC)
+ ::equals));
+
+ StringWriter measurementLog = new StringWriter();
+ try (ExportProgressReporter reporter = reporter(measurementLog)) {
+ reporter.start(
+ config("-mf", input("root.sg.d.s\n").toString(), "-start_time", "100"),
+ false,
+ () -> fail("unexpected log error"));
+ reporter.finish(ExportProgressReporter.Status.SUCCESS, null);
+ }
+ assertTrue(event(measurementLog, "START").contains(member("start_time", "100")));
+ assertFalse(measurementLog.toString().contains(member("event", "WARNING")));
+ }
+
+ @Test
+ public void tableQueryReportsItsTableAsIgnored() throws Exception {
+ StringWriter queryLog = new StringWriter();
+ try (ExportProgressReporter reporter = reporter(queryLog)) {
+ reporter.start(
+ config("-sql_dialect", "table", "-db", "db", "-table", "t", "-q", "select * from t"),
+ true,
+ () -> fail("unexpected log error"));
+ reporter.finish(ExportProgressReporter.Status.SUCCESS, null);
+ }
+ // SQL and TsFile output take the table from the query, so -table has no effect.
+ String warning = event(queryLog, "WARNING");
+ assertTrue(
+ warning,
+ warning.contains(
+ member(
+ "message",
+ CliMessages.LOG_TABLE_DOES_NOT_APPLY_TO_QUERIES_AND_IS_IGNORED_DBA31209)));
+ assertTrue(warning, warning.contains(member("table", "t")));
+
+ StringWriter tableLog = new StringWriter();
+ try (ExportProgressReporter reporter = reporter(tableLog)) {
+ reporter.start(
+ config("-sql_dialect", "table", "-db", "db", "-table", "t"),
+ false,
+ () -> fail("unexpected log error"));
+ reporter.finish(ExportProgressReporter.Status.SUCCESS, null);
+ }
+ assertFalse(tableLog.toString().contains(member("event", "WARNING")));
+ }
+
+ private static String event(StringWriter log, String event) {
+ return log.toString()
+ .lines()
+ .filter(line -> line.contains(member("event", event)))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError(log.toString()));
+ }
+
+ /** A member as the JSON Lines log writes it, such as "status":"FAILED" or "exit_code":130. */
+ static String member(String key, Object value) {
+ return '"' + key + "\":" + (value instanceof Number ? value : "\"" + value + '"');
+ }
+
+ /** Whether some console line came from {@code template}, whatever values filled it in. */
+ static boolean printed(String console, String template) {
+ String pattern =
+ Arrays.stream(template.split("%[sd]", -1))
+ .map(Pattern::quote)
+ .collect(Collectors.joining(".*?"));
+ return Pattern.compile(pattern).matcher(console).find();
+ }
+
+ @Test
+ public void timerLogFailureIsVisibleToWorkerAndFailureHandler() throws Exception {
+ CountDownLatch failed = new CountDownLatch(1);
+ AtomicBoolean failWrites = new AtomicBoolean();
+ Writer log =
+ new Writer() {
+ @Override
+ public void write(char[] data, int offset, int length) throws IOException {
+ if (failWrites.get()) {
+ throw new IOException("disk failure");
+ }
+ }
+
+ @Override
+ public void flush() {}
+
+ @Override
+ public void close() {}
+ };
+ ByteArrayOutputStream console = new ByteArrayOutputStream();
+ try (ExportProgressReporter reporter =
+ new ExportProgressReporter(log, "tree", System.nanoTime(), new PrintStream(console))) {
+ reporter.start(config(), false, failed::countDown);
+ failWrites.set(true);
+ assertTrue(failed.await(5, TimeUnit.SECONDS));
+ assertThrows(IOException.class, reporter::checkActive);
+ assertThrows(
+ IOException.class, () -> reporter.finish(ExportProgressReporter.Status.SUCCESS, null));
+ assertEquals(ExportProgressReporter.Status.FAILED, reporter.status());
+ }
+ // The log cannot hold its own failure, so the console summary names it.
+ String shown = console.toString(StandardCharsets.UTF_8);
+ assertTrue(shown, printed(shown, CliMessages.LOG_EXPORT_FAILED_AFTER_ARG_ARG_C123F0EF));
+ assertTrue(shown, shown.contains("disk failure"));
+ assertFalse(
+ shown,
+ printed(shown, CliMessages.LOG_EXPORT_COMPLETELY_ROWS_ARG_FILES_ARG_ELAPSED_ARG_A94BBB8B));
+ }
+
+ @Test
+ public void resultCloseFailureNeverCompletesTheTask() throws Exception {
+ Session session = mock(Session.class);
+ SessionDataSet data = mock(SessionDataSet.class);
+ SessionDataSet.DataIterator iterator = mock(SessionDataSet.DataIterator.class);
+ when(data.getColumnNames()).thenReturn(Arrays.asList("Time", "root.sg.d.s"));
+ when(data.iterator()).thenReturn(iterator);
+ when(iterator.getColumnTypeList()).thenReturn(Arrays.asList("INT64", "INT32"));
+ doThrow(new IOExceptionForTest()).when(data).close();
+ when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(data);
+ StringWriter log = new StringWriter();
+ ExportConfig config = config();
+ try (ExportProgressReporter reporter = reporter(log);
+ ExportDataTree exporter = new ExportDataTree(config, reporter, session)) {
+ assertThrows(
+ org.apache.iotdb.rpc.IoTDBConnectionException.class,
+ () ->
+ exporter.export(
+ new ExportTask(
+ "select s from root.sg.d",
+ config.targetDirectory.resolve("dump0"),
+ null,
+ null)));
+ reporter.finish(ExportProgressReporter.Status.FAILED, new IOException("close failure"));
+ }
+ assertFalse(log.toString().contains(member("event", "TASK_COMPLETE")));
+ verify(data, times(1)).close();
+ verify(session).close();
+ }
+
+ private static class IOExceptionForTest extends org.apache.iotdb.rpc.IoTDBConnectionException {
+ IOExceptionForTest() {
+ super("close failure");
+ }
+ }
+
+ private ExportConfig config(String... extras) throws Exception {
+ List<String> args =
+ new ArrayList<>(Arrays.asList("-ft", "csv", "-t", folder.getRoot().toString()));
+ args.addAll(Arrays.asList(extras));
+ return ExportConfig.parse(args.toArray(new String[0]), START);
+ }
+
+ private Path input(String contents) throws IOException {
+ Path path = folder.newFile().toPath();
+ return Files.writeString(path, contents, StandardCharsets.UTF_8);
+ }
+
+ private ExportProgressReporter reporter(StringWriter log) {
+ return new ExportProgressReporter(
+ log, "tree", System.nanoTime(), new PrintStream(new ByteArrayOutputStream()));
+ }
+
+ private SessionDataSet metadata(String... paths) throws Exception {
+ SessionDataSet data = mock(SessionDataSet.class);
+ when(data.getColumnNames())
+ .thenReturn(Arrays.asList("Timeseries", "DataType", "Encoding", "Compression"));
+ List<RowRecord> records = new ArrayList<>();
+ for (String path : paths) {
+ List<Field> fields = new ArrayList<>();
+ for (String value : Arrays.asList(path, "INT32", "RLE", "LZ4")) {
+ Field field = new Field(TSDataType.TEXT);
+ field.setBinaryV(new Binary(value, StandardCharsets.UTF_8));
+ fields.add(field);
+ }
+ records.add(new RowRecord(0, fields));
+ }
+ java.util.concurrent.atomic.AtomicInteger cursor =
+ new java.util.concurrent.atomic.AtomicInteger();
+ when(data.hasNext()).thenAnswer(call -> cursor.get() < records.size());
+ when(data.next()).thenAnswer(call -> records.get(cursor.getAndIncrement()));
+ return data;
+ }
+}
diff --git a/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ExportDataRegressionTest.java b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ExportDataRegressionTest.java
new file mode 100644
index 00000000000..925c88cc99c
--- /dev/null
+++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ExportDataRegressionTest.java
@@ -0,0 +1,300 @@
+/*
+ * 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.tool.data;
+
+import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.session.Session;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.io.StringWriter;
+import java.nio.file.Files;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Guards existing query output while the export lifecycle evolves. */
+public class ExportDataRegressionTest {
+
+ @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void treeCsvKeepsHeaderNullValuesAndSplits() throws Exception {
+ SessionDataSet dataSet = treeDataSet();
+ Session session = mock(Session.class);
+ when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(dataSet);
+ ExportConfig config = config("csv", false);
+ try (ExportProgressReporter progress = progress();
+ ExportDataTree exporter = new ExportDataTree(config, progress, session)) {
+ exporter.export(task(config, 3));
+ }
+
+ assertEquals("Time,root.sg.d.s\r\n100,11\r\n200,\r\n", read("dump3_0.csv"));
+ assertEquals("Time,root.sg.d.s\r\n300,33\r\n", read("dump3_1.csv"));
+ verify(dataSet).close();
+ }
+
+ @Test
+ public void treeSqlKeepsTimestampAndSkipsRowsWithoutValues() throws Exception {
+ SessionDataSet dataSet = treeDataSet();
+ Session session = mock(Session.class);
+ when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(dataSet);
+ ExportConfig config = config("sql", false);
+ try (ExportProgressReporter progress = progress();
+ ExportDataTree exporter = new ExportDataTree(config, progress, session)) {
+ exporter.export(task(config, 0));
+ // The row at 200 has no value, so it writes no statement and is not counted.
+ assertEquals(2, progress.rows());
+ }
+
+ assertEquals(
+ "INSERT INTO root.sg.d(TIMESTAMP,s) VALUES (100,11);\n"
+ + "INSERT INTO root.sg.d(TIMESTAMP,s) VALUES (300,33);\n",
+ read("dump0_0.sql"));
+ assertFalse(Files.exists(temporaryFolder.getRoot().toPath().resolve("dump0_1.sql")));
+ }
+
+ @Test
+ public void treeSqlCountsAMultiDeviceRowOnceAndKeepsItInOneFile() throws Exception {
+ SessionDataSet dataSet = mock(SessionDataSet.class);
+ SessionDataSet.DataIterator iterator = mock(SessionDataSet.DataIterator.class);
+ when(dataSet.getColumnNames())
+ .thenReturn(new ArrayList<>(Arrays.asList("Time", "root.sg.d1.s", "root.sg.d2.s")));
+ when(dataSet.iterator()).thenReturn(iterator);
+ when(iterator.getColumnTypeList()).thenReturn(Arrays.asList("INT64", "INT32", "INT32"));
+ AtomicInteger cursor = new AtomicInteger(-1);
+ when(iterator.next()).thenAnswer(call -> cursor.incrementAndGet() < 3);
+ when(iterator.getLong(1)).thenAnswer(call -> (cursor.get() + 1L) * 100);
+ when(iterator.getString(2)).thenAnswer(call -> "1" + cursor.get());
+ when(iterator.getString(3)).thenAnswer(call -> "2" + cursor.get());
+ Session session = mock(Session.class);
+ when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(dataSet);
+ ExportConfig config = config("sql", false);
+ StringWriter log = new StringWriter();
+ try (ExportProgressReporter progress = progress(log);
+ ExportDataTree exporter = new ExportDataTree(config, progress, session)) {
+ exporter.export(task(config, 0));
+ assertEquals(3, progress.rows());
+ }
+
+ // -lpf 2 counts data rows, so the second row's statements share the first file.
+ assertEquals(
+ "INSERT INTO root.sg.d1(TIMESTAMP,s) VALUES (100,10);\n"
+ + "INSERT INTO root.sg.d2(TIMESTAMP,s) VALUES (100,20);\n"
+ + "INSERT INTO root.sg.d1(TIMESTAMP,s) VALUES (200,11);\n"
+ + "INSERT INTO root.sg.d2(TIMESTAMP,s) VALUES (200,21);\n",
+ read("dump0_0.sql"));
+ assertEquals(
+ "INSERT INTO root.sg.d1(TIMESTAMP,s) VALUES (300,12);\n"
+ + "INSERT INTO root.sg.d2(TIMESTAMP,s) VALUES (300,22);\n",
+ read("dump0_1.sql"));
+ String content = log.toString();
+ assertTrue(content, content.contains(ExportDataFeatureTest.member("file_rows", 2)));
+ assertTrue(content, content.contains(ExportDataFeatureTest.member("file_rows", 1)));
+ assertTrue(content, content.contains(ExportDataFeatureTest.member("task_rows", 3)));
+ }
+
+ @Test
+ public void treeSqlGroupsInterleavedColumnsByDevice() throws Exception {
+ SessionDataSet dataSet = mock(SessionDataSet.class);
+ SessionDataSet.DataIterator iterator = mock(SessionDataSet.DataIterator.class);
+ when(dataSet.getColumnNames())
+ .thenReturn(
+ new ArrayList<>(Arrays.asList("Time", "root.sg.d1.a", "root.sg.d2.x", "root.sg.d1.b")));
+ when(dataSet.iterator()).thenReturn(iterator);
+ when(iterator.getColumnTypeList())
+ .thenReturn(Arrays.asList("INT64", "INT32", "INT32", "INT32"));
+ when(iterator.next()).thenReturn(true, false);
+ when(iterator.getLong(1)).thenReturn(100L);
+ when(iterator.getString(2)).thenReturn("1");
+ when(iterator.getString(3)).thenReturn("2");
+ when(iterator.getString(4)).thenReturn("3");
+ Session session = mock(Session.class);
+ when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(dataSet);
+ ExportConfig config = config("sql", false);
+ try (ExportProgressReporter progress = progress();
+ ExportDataTree exporter = new ExportDataTree(config, progress, session)) {
+ exporter.export(task(config, 0));
+ }
+
+ assertEquals(
+ "INSERT INTO root.sg.d1(TIMESTAMP,a,b) VALUES (100,1,3);\n"
+ + "INSERT INTO root.sg.d2(TIMESTAMP,x) VALUES (100,2);\n",
+ read("dump0_0.sql"));
+ }
+
+ @Test
+ public void treeSqlLeavesOutNullColumnsOfADevice() throws Exception {
+ SessionDataSet dataSet = mock(SessionDataSet.class);
+ SessionDataSet.DataIterator iterator = mock(SessionDataSet.DataIterator.class);
+ when(dataSet.getColumnNames())
+ .thenReturn(
+ new ArrayList<>(Arrays.asList("Time", "root.sg.d.a", "root.sg.d.b", "root.sg.d.c")));
+ when(dataSet.iterator()).thenReturn(iterator);
+ when(iterator.getColumnTypeList())
+ .thenReturn(Arrays.asList("INT64", "INT32", "INT32", "INT32"));
+ String[][] rows = {{"1", null, "3"}, {null, null, "6"}, {"7", "8", "9"}};
+ AtomicInteger cursor = new AtomicInteger(-1);
+ when(iterator.next()).thenAnswer(call -> cursor.incrementAndGet() < rows.length);
+ when(iterator.getLong(1)).thenAnswer(call -> (cursor.get() + 1L) * 100);
+ when(iterator.isNull(anyInt()))
+ .thenAnswer(call -> rows[cursor.get()][(int) call.getArgument(0) - 2] == null);
+ when(iterator.getString(anyInt()))
+ .thenAnswer(call -> rows[cursor.get()][(int) call.getArgument(0) - 2]);
+ Session session = mock(Session.class);
+ when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(dataSet);
+ ExportConfig config = config("sql", false);
+ try (ExportProgressReporter progress = progress();
+ ExportDataTree exporter = new ExportDataTree(config, progress, session)) {
+ exporter.export(task(config, 0));
+ }
+
+ assertEquals(
+ "INSERT INTO root.sg.d(TIMESTAMP,a,c) VALUES (100,1,3);\n"
+ + "INSERT INTO root.sg.d(TIMESTAMP,c) VALUES (200,6);\n",
+ read("dump0_0.sql"));
+ assertEquals(
+ "INSERT INTO root.sg.d(TIMESTAMP,a,b,c) VALUES (300,7,8,9);\n", read("dump0_1.sql"));
+ }
+
+ @Test
+ public void treeSqlAlignedByDeviceNamesEachRowsDevice() throws Exception {
+ SessionDataSet dataSet = mock(SessionDataSet.class);
+ SessionDataSet.DataIterator iterator = mock(SessionDataSet.DataIterator.class);
+ when(dataSet.getColumnNames())
+ .thenReturn(new ArrayList<>(Arrays.asList("Time", "Device", "s1", "s2")));
+ when(dataSet.iterator()).thenReturn(iterator);
+ when(iterator.getColumnTypeList()).thenReturn(Arrays.asList("INT64", "TEXT", "INT32", "TEXT"));
+ AtomicInteger cursor = new AtomicInteger(-1);
+ when(iterator.next()).thenAnswer(call -> cursor.incrementAndGet() < 2);
+ when(iterator.getLong(1)).thenAnswer(call -> (cursor.get() + 1L) * 100);
+ when(iterator.getString(2)).thenAnswer(call -> "root.sg.d" + (cursor.get() + 1));
+ when(iterator.getString(3)).thenAnswer(call -> String.valueOf(cursor.get()));
+ when(iterator.isNull(4)).thenAnswer(call -> cursor.get() == 1);
+ when(iterator.getString(4)).thenReturn("a\"b");
+ Session session = mock(Session.class);
+ when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(dataSet);
+ ExportConfig config = config("sql", false);
+ try (ExportProgressReporter progress = progress();
+ ExportDataTree exporter = new ExportDataTree(config, progress, session)) {
+ exporter.export(task(config, 0));
+ }
+
+ assertEquals(
+ "INSERT INTO root.sg.d1(TIMESTAMP,s1,s2) ALIGNED VALUES (100,0,'a\"b');\n"
+ + "INSERT INTO root.sg.d2(TIMESTAMP,s1,s2) ALIGNED VALUES (200,1,NULL);\n",
+ read("dump0_0.sql"));
+ }
+
+ @Test
+ public void tableCsvKeepsColumnOrderAndNullValues() throws Exception {
+ SessionDataSet dataSet = mock(SessionDataSet.class);
+ SessionDataSet.DataIterator iterator = mock(SessionDataSet.DataIterator.class);
+ when(dataSet.getColumnNames()).thenReturn(Arrays.asList("time", "value"));
+ when(dataSet.iterator()).thenReturn(iterator);
+ when(iterator.getColumnTypeList()).thenReturn(Arrays.asList("TIMESTAMP", "INT32"));
+ when(iterator.next()).thenReturn(true, true, true, false);
+ when(iterator.getLong(1)).thenReturn(100L, 200L, 300L);
+ when(iterator.isNull(2)).thenReturn(false, true, false);
+ when(iterator.getString(2)).thenReturn("11", "33");
+ ITableSession session = mock(ITableSession.class);
+ when(session.executeQueryStatement(anyString(), anyLong())).thenReturn(dataSet);
+ ExportConfig config = config("csv", true);
+ try (ExportProgressReporter progress = progress();
+ ExportDataTable exporter = new ExportDataTable(config, progress, session)) {
+ exporter.export(
+ new ExportTask(
+ "select * from metrics", config.targetDirectory.resolve("dump0"), null, "metrics"));
+ }
+
+ assertEquals("time,value\r\n100,11\r\n200,\r\n", read("dump0_0.csv"));
+ assertEquals("time,value\r\n300,33\r\n", read("dump0_1.csv"));
+ verify(dataSet).close();
+ }
+
+ private SessionDataSet treeDataSet() throws Exception {
+ SessionDataSet dataSet = mock(SessionDataSet.class);
+ SessionDataSet.DataIterator iterator = mock(SessionDataSet.DataIterator.class);
+ when(dataSet.getColumnNames())
+ .thenReturn(new ArrayList<>(Arrays.asList("Time", "root.sg.d.s")));
+ when(dataSet.iterator()).thenReturn(iterator);
+ when(iterator.getColumnTypeList()).thenReturn(Arrays.asList("INT64", "INT32"));
+ AtomicInteger cursor = new AtomicInteger(-1);
+ when(iterator.next()).thenAnswer(call -> cursor.incrementAndGet() < 3);
+ when(iterator.getLong(1)).thenAnswer(call -> (cursor.get() + 1L) * 100);
+ when(iterator.isNull(2)).thenAnswer(call -> cursor.get() == 1);
+ when(iterator.getString(2))
+ .thenAnswer(call -> cursor.get() == 0 ? "11" : cursor.get() == 1 ? null : "33");
+ return dataSet;
+ }
+
+ private String read(String name) throws Exception {
+ return Files.readString(temporaryFolder.getRoot().toPath().resolve(name));
+ }
+
+ private ExportConfig config(String format, boolean table) throws Exception {
+ List<String> args =
+ new ArrayList<>(
+ Arrays.asList(
+ "-ft",
+ format,
+ "-t",
+ temporaryFolder.getRoot().toString(),
+ "-lpf",
+ "2",
+ "-tf",
+ "timestamp"));
+ if (table) {
+ args.addAll(Arrays.asList("-sql_dialect", "table", "-db", "db"));
+ }
+ return ExportConfig.parse(args.toArray(new String[0]), ZonedDateTime.now());
+ }
+
+ private ExportProgressReporter progress() {
+ return progress(new StringWriter());
+ }
+
+ private ExportProgressReporter progress(StringWriter log) {
+ return new ExportProgressReporter(
+ log, "tree", System.nanoTime(), new PrintStream(OutputStream.nullOutputStream()));
+ }
+
+ private ExportTask task(ExportConfig config, int index) {
+ return new ExportTask(
+ "select s from root.sg.d", config.targetDirectory.resolve("dump" + index), null, null);
+ }
+}
diff --git a/scripts/tools/README.md b/scripts/tools/README.md
index cd6cea184a1..3ed99f76594 100644
--- a/scripts/tools/README.md
+++ b/scripts/tools/README.md
@@ -154,3 +154,133 @@ tools/import-wal.sh \
The process exits with `0` after a successful import and `1` when argument parsing, replay, or
source-file handling fails.
+
+# ExportData
+
+`tools/export-data.sh` (Windows: `tools/windows/export-data.bat`) exports Tree or Table results
+as CSV, SQL or TsFile. Run `-help csv`, `-help sql` or `-help tsfile` for the format's options.
+
+## Tree measurement files
+
+```bash
+tools/export-data.sh -h 127.0.0.1 -p 6667 -u root -pw secret \
+ -ft csv -t ./export -mf ./measurements.txt \
+ -start_time 2026-10-14T00:00:00+08:00 -end_time 2026-10-14T23:59:59+08:00
+```
+
+`-mf/--measurement_file` accepts a UTF-8 file containing one IoTDB path pattern per line:
+
+```text
+# Exact paths and IoTDB patterns can be combined.
+root.factory.device01.temperature
+root.factory.*.pre*
+root.factory.**
+```
+
+A first-line BOM, blank lines, surrounding whitespace and lines beginning with `#` are accepted.
+The entire file is validated before any server query. Each distinct pattern is resolved with
+`SHOW TIMESERIES`; `*`, `**` and `pre*` use IoTDB path syntax. Actual measurements are merged and
+deduplicated in first-discovery order, then exported sequentially. Each unmatched pattern produces
+a warning that names the pattern and its line, such as
+`WARNING: No measurements matched root.factory.missing.* (measurement file line 5).`; the log also
+records the source file. The command fails if no actual measurements remain.
+`-mf` is available only in Tree mode and cannot be combined with `-q`.
+
+Each measurement uses its actual full path as the output filename. CSV and SQL are split by
+`-lpf/--lines_per_file` (default 10,000 rows); TsFile produces one file per measurement:
+
+```text
+export/root.factory.device01.temperature_0.csv
+export/root.factory.device01.temperature_1.csv
+export/root.factory.device01.temperature.tsfile
+```
+
+The existing `-pfn/--prefix_file_name` supplies an optional extra prefix in measurement-file mode:
+`-pfn backup` produces `backup_root.factory.device01.temperature_0.csv`. No default `dump` prefix
+is added in this mode, and `-f` is not an alias. Query and Table exports retain names such as
+`dump0_0.csv`, `dump0_0.sql` and `dump0.tsfile`. Characters that are unsafe in filenames are encoded
+as `%HH`. Oversized filenames and case/normalization collisions fail before exporting data.
+
+`-start_time` and `-end_time` form an inclusive time interval. Values may be integer timestamps in
+the server's precision or date-time strings interpreted in the session's timezone (`-tz`). A
+single bound is supported; a reversed or invalid interval fails before data queries. These bounds
+also apply to automatic Table exports. Queries, whether given with `-q` or at the Tree prompt, run
+as written: the bounds are ignored and reported as a `WARNING`.
+A measurement with no rows in the interval still counts as completed; CSV/SQL create no data file
+for it. SQL exports contain INSERT statements for reimport into compatible existing schema;
+TsFile retains measurement types, encoding, compression and device alignment.
+
+## Queries and Table exports
+
+```bash
+tools/export-data.sh -ft csv -t ./export -q 'SELECT * FROM root.factory.**'
+tools/export-data.sh -sql_dialect table -db factory -table readings \
+ -ft tsfile -t ./export -start_time 100 -end_time 200
+tools/export-data.sh -sql_dialect table -db factory \
+ -ft sql -t ./export -q 'SELECT * FROM readings WHERE time >= 100'
+```
+
+Tree mode prompts for a query when neither `-mf` nor `-q` is supplied. Table mode automatically
+exports the selected table or all tables in `-db`. Each query in `-q` writes its own files
+(`dump0_0.csv`, `dump1_0.csv`, ...). A Table query exported as SQL or TsFile is written to the table
+it reads: it must read a single table of `-db` without joins, subqueries or CTEs, and return the
+time column and original column names; TsFile queries must include all tags to preserve device
+identity. All queries are checked before the first one runs. Like the time bounds, `-table` does
+not apply to queries and is reported as a `WARNING`.
+
+SQL output quotes strings and dates in single quotes, doubling a quote inside them. A Tree query
+row that spans several devices becomes one SQL INSERT per device that has values in the row, and
+columns without a value are left out. These statements stay in the same file and count as one data
+row for `-lpf`, `exported_data_rows` and `file_rows`. `ALIGN BY DEVICE` and Table rows write `NULL`
+for a missing value. With `-dt true`, CSV headers in both models add each value column's type, such
+as `s1(DOUBLE)`.
+
+## Progress, logs and failures
+
+Every invocation writes a UTF-8 append log and reports progress every two seconds, including while
+waiting for queries. When Tree mode prompts for a query, logging and progress start after the query
+is entered. The default log path uses the command's startup time:
+`<target_dir>/log_export_data_yyyyMMddHHmmss.log`. Override it with `-l/--export_data_log <path>`.
+The output directory is created automatically; an overridden log's parent directory must exist.
+
+The console is for people. In a terminal one progress line is redrawn in place, cutting the current
+measurement from the left to fit the terminal width. When output goes to a file, `nohup` or a
+container log, or when the progress line would not fit the terminal, each update is a line of its
+own:
+
+```text
+Exporting 59% 121/204 3,535,481 rows 00:00:02 ETA 00:00:01 ...line19.s0
+[59%] 121/204 measurements, 3,535,481 rows, elapsed 00:00:02, ETA 00:00:01, current root.factory.line19.s0
+```
+
+The shell launcher passes the terminal width to the tool as `COLUMNS`. The Windows launcher does
+not, so 80 columns are assumed there unless `COLUMNS` is set.
+
+The console also shows the start, warnings and the log path, then ends with one summary line such
+as `Export completely! Rows: 5, files: 1, elapsed: 00:00:00.` Query and Table modes have no total,
+so they show rows, elapsed time and rows per second instead of a percentage and ETA. The launcher
+applies `conf/logback-tool.xml` when present, so library logging shows errors only.
+
+The log is for programs: JSON Lines, one event object per line, with counts as JSON numbers. Each
+event starts with `progress_id` (the operating system process ID), `timestamp` (ISO 8601 with
+offset), `model` (`tree` or `table`), `status` (`RUNNING`, `SUCCESS`, `FAILED` or `CANCELLED`),
+`exported_data_rows`, `elapsed_time` (`HH:mm:ss`) and `output`, the file being written (omitted
+until a task opens one), followed by `event`, `level`, `phase` and the event's details. Events
+include `START`, `DISCOVERY`, `WARNING`, `TASK_START`, `TASK_COMPLETE`, `FILE_START`,
+`FILE_COMPLETE`, `PROGRESS` and one `FINISH`. The log records source patterns, queries, output
+files, rows, completed files, elapsed time and errors; command credentials are excluded. For
+example, `jq -c 'select(.event == "FINISH")' export/log_export_data_*.log` prints each result.
+
+Measurement-file mode adds `exported_measurement_num`, `total_measurement_num`, `exported_percent`
+and `estimated_remaining` between `exported_data_rows` and `elapsed_time`. A measurement completes
+only after all output files and its query handle close successfully. `estimated_remaining` uses the
+average measurement duration (`HH:mm:ss`). Measurements differ in size, so it stays `UNKNOWN`, and
+the console shows `ETA --`, until 5% of the measurements, and at least three, have completed. No
+extra `COUNT(*)` scan is made, and no `-progress` option is needed.
+
+The command returns `0` on success and `1` for invalid arguments, input, queries, output or log
+errors; failures record `FAILED`. Ctrl+C or `SIGTERM` records `CANCELLED`. The export stops at the
+next row and closes its query and session; a server call that is still blocked after three seconds
+is closed. The process then exits with the signal's status (`130` for Ctrl+C, `143` for `SIGTERM`),
+which `FINISH` reports as `exit_code`. Already generated files are retained for inspection, and
+incomplete files are not marked `FILE_COMPLETE`.
diff --git a/scripts/tools/export-data.sh b/scripts/tools/export-data.sh
index d98d1363aa4..72405e62ee2 100644
--- a/scripts/tools/export-data.sh
+++ b/scripts/tools/export-data.sh
@@ -57,5 +57,21 @@ done
MAIN_CLASS=org.apache.iotdb.tool.data.ExportData
-"$JAVA" $JVM_OPTS -DIOTDB_HOME=${IOTDB_HOME} -cp "$CLASSPATH" "$MAIN_CLASS" "$@"
+# Library logs would interleave with the progress line; the tool config keeps errors only.
+LOG_CONFIG="${IOTDB_HOME}/conf/logback-tool.xml"
+if [ ! -f "$LOG_CONFIG" ]; then
+ LOG_CONFIG=
+fi
+
+# Shells do not export COLUMNS. The tool redraws the progress line in place only while it fits.
+if [ -t 1 ]; then
+ if [ -z "${COLUMNS:-}" ]; then
+ COLUMNS=$(stty size </dev/tty 2>/dev/null | awk '{print $2}')
+ fi
+ if [ -n "${COLUMNS:-}" ]; then
+ export COLUMNS
+ fi
+fi
+
+"$JAVA" $JVM_OPTS -DIOTDB_HOME=${IOTDB_HOME} ${LOG_CONFIG:+"-Dlogback.configurationFile=$LOG_CONFIG"} -cp "$CLASSPATH" "$MAIN_CLASS" "$@"
exit $?
\ No newline at end of file
diff --git a/scripts/tools/windows/export-data.bat b/scripts/tools/windows/export-data.bat
index 06d489a2d30..43b61b8f82b 100644
--- a/scripts/tools/windows/export-data.bat
+++ b/scripts/tools/windows/export-data.bat
@@ -42,9 +42,13 @@ set JAVA_OPTS=-ea^
@REM ***** CLASSPATH library setting *****
set CLASSPATH=%CLASSPATH%;"%IOTDB_HOME%\lib\*"
+@REM Library logs would interleave with the progress line; the tool config keeps errors only.
+set LOG_CONFIG_OPTS=
+if exist "%IOTDB_HOME%\conf\logback-tool.xml" set LOG_CONFIG_OPTS=-Dlogback.configurationFile="%IOTDB_HOME%\conf\logback-tool.xml"
+
REM -----------------------------------------------------------------------------
-"%JAVA_HOME%\bin\java" -DIOTDB_HOME="%IOTDB_HOME%" %JAVA_OPTS% -cp %CLASSPATH% %MAIN_CLASS% %*
+"%JAVA_HOME%\bin\java" -DIOTDB_HOME="%IOTDB_HOME%" %JAVA_OPTS% %LOG_CONFIG_OPTS% -cp %CLASSPATH% %MAIN_CLASS% %*
set ret_code=%ERRORLEVEL%
goto finally