This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch codex/backport-import-export-1.3 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d980d94b078f3dd0ff0ef4d6cba90380fc362316 Author: Caideyipi <[email protected]> AuthorDate: Thu Oct 22 15:59:17 2026 +0800 [To dev/1.3] Fix import-data and export-data correctness and error reporting Backport the tree-model CLI fixes from 0acff4971750046b6973c654fe16d7f5c3e9a333. Adapt CSV batch failure recovery, SQL export, TsFile archival and exit status handling to the 1.3 implementation. Exclude the 2.x table model, relational grammar and relocated Windows wrappers. --- .../iotdb/tool/common/ImportTsFileOperation.java | 96 ++++ .../apache/iotdb/tool/data/AbstractDataTool.java | 530 +++++++++++++++------ .../apache/iotdb/tool/data/AsyncImportData.java | 58 ++- .../org/apache/iotdb/tool/data/ExportData.java | 189 ++++---- .../org/apache/iotdb/tool/data/ImportData.java | 14 +- .../org/apache/iotdb/tool/tsfile/ImportTsFile.java | 9 +- .../apache/iotdb/tool/tsfile/ImportTsFileBase.java | 147 ++---- .../iotdb/tool/tsfile/ImportTsFileScanTool.java | 11 + .../org/apache/iotdb/tool/data/ExportDataTest.java | 92 ++++ .../iotdb/tool/data/ImportDataFailureTest.java | 108 +++++ .../apache/iotdb/tool/tsfile/ImportTsFileTest.java | 204 ++++++++ 11 files changed, 1097 insertions(+), 361 deletions(-) diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/ImportTsFileOperation.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/ImportTsFileOperation.java new file mode 100644 index 00000000000..c5652ac48d5 --- /dev/null +++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/ImportTsFileOperation.java @@ -0,0 +1,96 @@ +/* + * 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.common; + +import java.io.IOException; +import java.nio.file.FileAlreadyExistsException; +import java.nio.file.Files; +import java.nio.file.LinkOption; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; + +/** Operations for a TsFile and the sidecar files that belong to it. */ +public enum ImportTsFileOperation { + NONE, + MV, + HARDLINK, + CP, + DELETE, + ; + + /** Process a TsFile group without replacing an existing destination member. */ + public void processFileGroup(Path source, Path target) throws IOException { + List<String> suffixes = new ArrayList<>(); + suffixes.add(""); + for (String suffix : new String[] {".resource", ".mods"}) { + if (Files.exists(Paths.get(source + suffix))) { + suffixes.add(suffix); + } + } + + if (this == CP || this == HARDLINK || this == MV) { + // Check every destination before changing any source in the file group. + for (String suffix : suffixes) { + Path destination = Paths.get(target + suffix); + if (Files.exists(destination, LinkOption.NOFOLLOW_LINKS)) { + throw new FileAlreadyExistsException(destination.toString()); + } + } + Path parent = target.getParent(); + if (parent != null) { + Files.createDirectories(parent); + } + } + + for (String suffix : suffixes) { + Path sourceFile = Paths.get(source + suffix); + Path targetFile = Paths.get(target + suffix); + ImportTsFileOperation operation = this == HARDLINK && !suffix.isEmpty() ? CP : this; + operation.processFile(sourceFile, targetFile); + } + } + + private void processFile(Path source, Path target) throws IOException { + switch (this) { + case DELETE: + Files.deleteIfExists(source); + break; + case CP: + Files.copy(source, target); + break; + case HARDLINK: + try { + Files.createLink(target, source); + } catch (FileAlreadyExistsException e) { + throw e; + } catch (IOException | UnsupportedOperationException e) { + Files.copy(source, target); + } + break; + case MV: + Files.move(source, target); + break; + default: + break; + } + } +} diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractDataTool.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractDataTool.java index 11f4ed56eff..b562b6404d0 100644 --- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractDataTool.java +++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractDataTool.java @@ -63,6 +63,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -178,6 +179,21 @@ public abstract class AbstractDataTool { protected static Boolean aligned; protected static Session session; protected static final LongAdder loadFileSuccessfulNum = new LongAdder(); + protected static final LongAdder loadFileFailedNum = new LongAdder(); + private static final ThreadLocal<Boolean> CURRENT_IMPORT_FAILED = + ThreadLocal.withInitial(() -> false); + + protected static void markCurrentImportFailed() { + CURRENT_IMPORT_FAILED.set(true); + } + + protected static boolean isCurrentImportFailed() { + return CURRENT_IMPORT_FAILED.get(); + } + + protected static void resetCurrentImportFailed() { + CURRENT_IMPORT_FAILED.set(false); + } protected static final String DATATYPE_BOOLEAN = "boolean"; protected static final String DATATYPE_INT = "int"; @@ -411,7 +427,7 @@ public abstract class AbstractDataTool { AtomicInteger pointSize = new AtomicInteger(0); - ArrayList<List<Object>> failedRecords = new ArrayList<>(); + BatchContext batchContext = new BatchContext(); records.forEach( recordObj -> { @@ -421,13 +437,27 @@ public abstract class AbstractDataTool { } else if (!Objects.equals(deviceName.get(), recordObj.get(1))) { // if device changed writeAndEmptyDataSet( - sessionPool, deviceName.get(), times, typesList, valuesList, measurementsList, 3); + sessionPool, + deviceName.get(), + times, + typesList, + valuesList, + measurementsList, + batchContext, + 3); deviceName.set(recordObj.get(1)); pointSize.set(0); } else if (pointSize.get() >= batchPointSize) { // insert a batch writeAndEmptyDataSet( - sessionPool, deviceName.get(), times, typesList, valuesList, measurementsList, 3); + sessionPool, + deviceName.get(), + times, + typesList, + valuesList, + measurementsList, + batchContext, + 3); pointSize.set(0); } @@ -486,10 +516,13 @@ public abstract class AbstractDataTool { } } if (Boolean.TRUE.equals(isFail.get())) { - failedRecords.add(recordObj.stream().collect(Collectors.toList())); + batchContext.recordFailure(recordObj); } if (!measurements.isEmpty()) { - times.add(parseTimestamp(recordObj.get(timeColumn))); + long timestamp = parseTimestamp(recordObj.get(timeColumn)); + batchContext.recordBatch( + recordObj, deviceName.get(), timestamp, types, values, measurements); + times.add(timestamp); typesList.add(types); valuesList.add(values); measurementsList.add(measurements); @@ -497,15 +530,154 @@ public abstract class AbstractDataTool { }); if (!times.isEmpty()) { writeAndEmptyDataSet( - sessionPool, deviceName.get(), times, typesList, valuesList, measurementsList, 3); + sessionPool, + deviceName.get(), + times, + typesList, + valuesList, + measurementsList, + batchContext, + 3); pointSize.set(0); } - if (!failedRecords.isEmpty()) { - writeFailedLinesFile(headerNames, failedFilePath, failedRecords); + if (!batchContext.failedRecords.isEmpty()) { + writeFailedLinesFile( + headerNames, failedFilePath, new ArrayList<>(batchContext.failedRecords.values())); } // ioTPrinter.println("Import completely!"); } + @FunctionalInterface + private interface BatchInsert { + void insert() throws IoTDBConnectionException, StatementExecutionException; + } + + private static final class BatchContext { + private final Map<Long, List<Object>> failedRecords = new LinkedHashMap<>(); + private final List<PendingGroup> batchGroups = new ArrayList<>(); + + private void recordFailure(CSVRecord record) { + failedRecords.putIfAbsent( + record.getRecordNumber(), record.stream().collect(Collectors.toList())); + markCurrentImportFailed(); + } + + private void recordBatch( + CSVRecord record, + String device, + long timestamp, + List<TSDataType> types, + List<Object> values, + List<String> measurements) { + batchGroups.add( + new PendingGroup( + record.getRecordNumber(), + record.stream().collect(Collectors.toList()), + device, + timestamp, + new ArrayList<>(types), + new ArrayList<>(values), + new ArrayList<>(measurements))); + } + + private void finishBatch( + boolean succeeded, GroupInsert groupInsert, Session reconnectableSession, int retryTime) { + if (!succeeded) { + for (PendingGroup group : batchGroups) { + boolean groupSucceeded = + executeBatchWithRetry( + () -> groupInsert.insert(group), reconnectableSession, retryTime); + if (!groupSucceeded) { + failedRecords.putIfAbsent(group.recordNumber, group.record); + markCurrentImportFailed(); + } + } + } + batchGroups.clear(); + } + } + + @FunctionalInterface + private interface GroupInsert { + void insert(PendingGroup group) throws IoTDBConnectionException, StatementExecutionException; + } + + private static final class PendingGroup { + private final long recordNumber; + private final List<Object> record; + private final String device; + private final long timestamp; + private final List<TSDataType> types; + private final List<Object> values; + private final List<String> measurements; + + private PendingGroup( + long recordNumber, + List<Object> record, + String device, + long timestamp, + List<TSDataType> types, + List<Object> values, + List<String> measurements) { + this.recordNumber = recordNumber; + this.record = record; + this.device = device; + this.timestamp = timestamp; + this.types = types; + this.values = values; + this.measurements = measurements; + } + } + + private static boolean executeBatchWithRetry( + BatchInsert batchInsert, Session reconnectableSession, int retryTime) { + for (int attempt = 0; ; attempt++) { + try { + batchInsert.insert(); + return true; + } catch (IoTDBConnectionException e) { + if (attempt >= retryTime) { + ioTPrinter.println(INSERT_CSV_MEET_ERROR_MSG + e.getMessage()); + return false; + } + if (reconnectableSession != null) { + try { + reconnectableSession.open(); + } catch (IoTDBConnectionException reconnectException) { + if (attempt + 1 >= retryTime) { + ioTPrinter.println(INSERT_CSV_MEET_ERROR_MSG + reconnectException.getMessage()); + return false; + } + } + } + } catch (StatementExecutionException e) { + ioTPrinter.println(INSERT_CSV_MEET_ERROR_MSG + e.getMessage()); + return false; + } + } + } + + private static boolean insertBatch( + SessionPool sessionPool, + List<String> deviceIds, + List<Long> times, + List<List<TSDataType>> typesList, + List<List<Object>> valuesList, + List<List<String>> measurementsList, + int retryTime) { + return executeBatchWithRetry( + () -> { + if (Boolean.FALSE.equals(aligned)) { + sessionPool.insertRecords(deviceIds, times, measurementsList, typesList, valuesList); + } else { + sessionPool.insertAlignedRecords( + deviceIds, times, measurementsList, typesList, valuesList); + } + }, + null, + retryTime); + } + private static void writeAndEmptyDataSet( SessionPool sessionPool, String device, @@ -513,29 +685,46 @@ public abstract class AbstractDataTool { List<List<TSDataType>> typesList, List<List<Object>> valuesList, List<List<String>> measurementsList, + BatchContext batchContext, int retryTime) { - try { - if (Boolean.FALSE.equals(aligned)) { - sessionPool.insertRecordsOfOneDevice( - device, times, measurementsList, typesList, valuesList); - } else { - sessionPool.insertAlignedRecordsOfOneDevice( - device, times, measurementsList, typesList, valuesList); - } - } catch (IoTDBConnectionException e) { - if (retryTime > 0) { - writeAndEmptyDataSet( - sessionPool, device, times, typesList, valuesList, measurementsList, --retryTime); - } - } catch (StatementExecutionException e) { - ioTPrinter.println(INSERT_CSV_MEET_ERROR_MSG + e.getMessage()); - System.exit(1); - } finally { - times.clear(); - typesList.clear(); - valuesList.clear(); - measurementsList.clear(); - } + boolean succeeded = + executeBatchWithRetry( + () -> { + if (Boolean.FALSE.equals(aligned)) { + sessionPool.insertRecordsOfOneDevice( + device, times, measurementsList, typesList, valuesList); + } else { + sessionPool.insertAlignedRecordsOfOneDevice( + device, times, measurementsList, typesList, valuesList); + } + }, + null, + retryTime); + batchContext.finishBatch( + succeeded, + group -> { + if (Boolean.FALSE.equals(aligned)) { + sessionPool.insertRecordsOfOneDevice( + group.device, + Arrays.asList(group.timestamp), + Arrays.asList(group.measurements), + Arrays.asList(group.types), + Arrays.asList(group.values)); + } else { + sessionPool.insertAlignedRecordsOfOneDevice( + group.device, + Arrays.asList(group.timestamp), + Arrays.asList(group.measurements), + Arrays.asList(group.types), + Arrays.asList(group.values)); + } + }, + null, + retryTime); + times.clear(); + typesList.clear(); + valuesList.clear(); + measurementsList.clear(); } private static void writeAndEmptyDataSet( @@ -545,38 +734,46 @@ public abstract class AbstractDataTool { List<List<TSDataType>> typesList, List<List<Object>> valuesList, List<List<String>> measurementsList, + BatchContext batchContext, int retryTime) { - try { - if (Boolean.FALSE.equals(aligned)) { - session.insertRecords(deviceIds, times, measurementsList, typesList, valuesList); - } else { - session.insertAlignedRecords(deviceIds, times, measurementsList, typesList, valuesList); - } - } catch (IoTDBConnectionException e) { - if (retryTime > 0) { - try { - session.open(); - } catch (IoTDBConnectionException ex) { - ioTPrinter.println(INSERT_CSV_MEET_ERROR_MSG + e.getMessage()); - } - writeAndEmptyDataSet( - session, deviceIds, times, typesList, valuesList, measurementsList, --retryTime); - } - } catch (StatementExecutionException e) { - ioTPrinter.println(INSERT_CSV_MEET_ERROR_MSG + e.getMessage()); - try { - session.close(); - } catch (IoTDBConnectionException ex) { - // do nothing - } - System.exit(1); - } finally { - deviceIds.clear(); - times.clear(); - typesList.clear(); - valuesList.clear(); - measurementsList.clear(); - } + boolean succeeded = + executeBatchWithRetry( + () -> { + if (Boolean.FALSE.equals(aligned)) { + session.insertRecords(deviceIds, times, measurementsList, typesList, valuesList); + } else { + session.insertAlignedRecords( + deviceIds, times, measurementsList, typesList, valuesList); + } + }, + session, + retryTime); + batchContext.finishBatch( + succeeded, + group -> { + if (Boolean.FALSE.equals(aligned)) { + session.insertRecords( + Arrays.asList(group.device), + Arrays.asList(group.timestamp), + Arrays.asList(group.measurements), + Arrays.asList(group.types), + Arrays.asList(group.values)); + } else { + session.insertAlignedRecords( + Arrays.asList(group.device), + Arrays.asList(group.timestamp), + Arrays.asList(group.measurements), + Arrays.asList(group.types), + Arrays.asList(group.values)); + } + }, + session, + retryTime); + deviceIds.clear(); + times.clear(); + typesList.clear(); + valuesList.clear(); + measurementsList.clear(); } private static void writeAndEmptyDataSet( @@ -586,38 +783,46 @@ public abstract class AbstractDataTool { List<List<TSDataType>> typesList, List<List<Object>> valuesList, List<List<String>> measurementsList, + BatchContext batchContext, int retryTime) { - try { - if (Boolean.FALSE.equals(aligned)) { - session.insertRecordsOfOneDevice(device, times, measurementsList, typesList, valuesList); - } else { - session.insertAlignedRecordsOfOneDevice( - device, times, measurementsList, typesList, valuesList); - } - } catch (IoTDBConnectionException e) { - if (retryTime > 0) { - try { - session.open(); - } catch (IoTDBConnectionException ex) { - ioTPrinter.println(INSERT_CSV_MEET_ERROR_MSG + e.getMessage()); - } - writeAndEmptyDataSet( - session, device, times, typesList, valuesList, measurementsList, --retryTime); - } - } catch (StatementExecutionException e) { - ioTPrinter.println(INSERT_CSV_MEET_ERROR_MSG + e.getMessage()); - try { - session.close(); - } catch (IoTDBConnectionException ex) { - // do nothing - } - System.exit(1); - } finally { - times.clear(); - typesList.clear(); - valuesList.clear(); - measurementsList.clear(); - } + boolean succeeded = + executeBatchWithRetry( + () -> { + if (Boolean.FALSE.equals(aligned)) { + session.insertRecordsOfOneDevice( + device, times, measurementsList, typesList, valuesList); + } else { + session.insertAlignedRecordsOfOneDevice( + device, times, measurementsList, typesList, valuesList); + } + }, + session, + retryTime); + batchContext.finishBatch( + succeeded, + group -> { + if (Boolean.FALSE.equals(aligned)) { + session.insertRecordsOfOneDevice( + group.device, + Arrays.asList(group.timestamp), + Arrays.asList(group.measurements), + Arrays.asList(group.types), + Arrays.asList(group.values)); + } else { + session.insertAlignedRecordsOfOneDevice( + group.device, + Arrays.asList(group.timestamp), + Arrays.asList(group.measurements), + Arrays.asList(group.types), + Arrays.asList(group.values)); + } + }, + session, + retryTime); + times.clear(); + typesList.clear(); + valuesList.clear(); + measurementsList.clear(); } private static void writeAndEmptyDataSet( @@ -627,28 +832,37 @@ public abstract class AbstractDataTool { List<List<TSDataType>> typesList, List<List<Object>> valuesList, List<List<String>> measurementsList, + BatchContext batchContext, int retryTime) { - try { - if (Boolean.FALSE.equals(aligned)) { - sessionPool.insertRecords(deviceIds, times, measurementsList, typesList, valuesList); - } else { - sessionPool.insertAlignedRecords(deviceIds, times, measurementsList, typesList, valuesList); - } - } catch (IoTDBConnectionException e) { - if (retryTime > 0) { - writeAndEmptyDataSet( - sessionPool, deviceIds, times, typesList, valuesList, measurementsList, --retryTime); - } - } catch (StatementExecutionException e) { - ioTPrinter.println(INSERT_CSV_MEET_ERROR_MSG + e.getMessage()); - System.exit(1); - } finally { - deviceIds.clear(); - times.clear(); - typesList.clear(); - valuesList.clear(); - measurementsList.clear(); - } + boolean succeeded = + insertBatch( + sessionPool, deviceIds, times, typesList, valuesList, measurementsList, retryTime); + batchContext.finishBatch( + succeeded, + group -> { + if (Boolean.FALSE.equals(aligned)) { + sessionPool.insertRecords( + Arrays.asList(group.device), + Arrays.asList(group.timestamp), + Arrays.asList(group.measurements), + Arrays.asList(group.types), + Arrays.asList(group.values)); + } else { + sessionPool.insertAlignedRecords( + Arrays.asList(group.device), + Arrays.asList(group.timestamp), + Arrays.asList(group.measurements), + Arrays.asList(group.types), + Arrays.asList(group.values)); + } + }, + null, + retryTime); + deviceIds.clear(); + times.clear(); + typesList.clear(); + valuesList.clear(); + measurementsList.clear(); } private static void writeFailedLinesFile( @@ -985,7 +1199,7 @@ public abstract class AbstractDataTool { AtomicReference<Boolean> hasStarted = new AtomicReference<>(false); AtomicInteger pointSize = new AtomicInteger(0); - ArrayList<List<Object>> failedRecords = new ArrayList<>(); + BatchContext batchContext = new BatchContext(); records.forEach( recordObj -> { @@ -993,7 +1207,14 @@ public abstract class AbstractDataTool { hasStarted.set(true); } else if (pointSize.get() >= batchPointSize) { writeAndEmptyDataSet( - sessionPool, deviceIds, times, typesList, valuesList, measurementsList, 3); + sessionPool, + deviceIds, + times, + typesList, + valuesList, + measurementsList, + batchContext, + 3); pointSize.set(0); } @@ -1042,7 +1263,9 @@ public abstract class AbstractDataTool { } } if (!measurements.isEmpty()) { - times.add(parseTimestamp(recordObj.get(timeColumn))); + long timestamp = parseTimestamp(recordObj.get(timeColumn)); + batchContext.recordBatch(recordObj, deviceId, timestamp, types, values, measurements); + times.add(timestamp); deviceIds.add(deviceId); typesList.add(types); valuesList.add(values); @@ -1050,17 +1273,18 @@ public abstract class AbstractDataTool { } } if (isFail) { - failedRecords.add(recordObj.stream().collect(Collectors.toList())); + batchContext.recordFailure(recordObj); } }); if (!deviceIds.isEmpty()) { writeAndEmptyDataSet( - sessionPool, deviceIds, times, typesList, valuesList, measurementsList, 3); + sessionPool, deviceIds, times, typesList, valuesList, measurementsList, batchContext, 3); pointSize.set(0); } - if (!failedRecords.isEmpty()) { - writeFailedLinesFile(headerNames, failedFilePath, failedRecords); + if (!batchContext.failedRecords.isEmpty()) { + writeFailedLinesFile( + headerNames, failedFilePath, new ArrayList<>(batchContext.failedRecords.values())); } // if (Boolean.TRUE.equals(hasStarted.get())) { // ioTPrinter.println("Import completely!"); @@ -1117,7 +1341,7 @@ public abstract class AbstractDataTool { AtomicInteger pointSize = new AtomicInteger(0); - ArrayList<List<Object>> failedRecords = new ArrayList<>(); + BatchContext batchContext = new BatchContext(); records.forEach( recordObj -> { @@ -1127,13 +1351,27 @@ public abstract class AbstractDataTool { } else if (!Objects.equals(deviceName.get(), recordObj.get(1))) { // if device changed writeAndEmptyDataSet( - session, deviceName.get(), times, typesList, valuesList, measurementsList, 3); + session, + deviceName.get(), + times, + typesList, + valuesList, + measurementsList, + batchContext, + 3); deviceName.set(recordObj.get(1)); pointSize.set(0); } else if (pointSize.get() >= batchPointSize) { // insert a batch writeAndEmptyDataSet( - session, deviceName.get(), times, typesList, valuesList, measurementsList, 3); + session, + deviceName.get(), + times, + typesList, + valuesList, + measurementsList, + batchContext, + 3); pointSize.set(0); } @@ -1193,10 +1431,13 @@ public abstract class AbstractDataTool { } } if (Boolean.TRUE.equals(isFail.get())) { - failedRecords.add(recordObj.stream().collect(Collectors.toList())); + batchContext.recordFailure(recordObj); } if (!measurements.isEmpty()) { - times.add(parseTimestamp(recordObj.get(timeColumn))); + long timestamp = parseTimestamp(recordObj.get(timeColumn)); + batchContext.recordBatch( + recordObj, deviceName.get(), timestamp, types, values, measurements); + times.add(timestamp); typesList.add(types); valuesList.add(values); measurementsList.add(measurements); @@ -1204,11 +1445,19 @@ public abstract class AbstractDataTool { }); if (!times.isEmpty()) { writeAndEmptyDataSet( - session, deviceName.get(), times, typesList, valuesList, measurementsList, 3); + session, + deviceName.get(), + times, + typesList, + valuesList, + measurementsList, + batchContext, + 3); pointSize.set(0); } - if (!failedRecords.isEmpty()) { - writeFailedLinesFile(headerNames, failedFilePath, failedRecords); + if (!batchContext.failedRecords.isEmpty()) { + writeFailedLinesFile( + headerNames, failedFilePath, new ArrayList<>(batchContext.failedRecords.values())); } ioTPrinter.println("Import completely!"); } @@ -1243,7 +1492,7 @@ public abstract class AbstractDataTool { AtomicReference<Boolean> hasStarted = new AtomicReference<>(false); AtomicInteger pointSize = new AtomicInteger(0); - ArrayList<List<Object>> failedRecords = new ArrayList<>(); + BatchContext batchContext = new BatchContext(); records.forEach( recordObj -> { @@ -1251,7 +1500,14 @@ public abstract class AbstractDataTool { hasStarted.set(true); } else if (pointSize.get() >= batchPointSize) { writeAndEmptyDataSet( - session, deviceIds, times, typesList, valuesList, measurementsList, 3); + session, + deviceIds, + times, + typesList, + valuesList, + measurementsList, + batchContext, + 3); pointSize.set(0); } @@ -1297,7 +1553,9 @@ public abstract class AbstractDataTool { } } if (!measurements.isEmpty()) { - times.add(parseTimestamp(recordObj.get(timeColumn))); + long timestamp = parseTimestamp(recordObj.get(timeColumn)); + batchContext.recordBatch(recordObj, deviceId, timestamp, types, values, measurements); + times.add(timestamp); deviceIds.add(deviceId); typesList.add(types); valuesList.add(values); @@ -1305,16 +1563,18 @@ public abstract class AbstractDataTool { } } if (isFail) { - failedRecords.add(recordObj.stream().collect(Collectors.toList())); + batchContext.recordFailure(recordObj); } }); if (!deviceIds.isEmpty()) { - writeAndEmptyDataSet(session, deviceIds, times, typesList, valuesList, measurementsList, 3); + writeAndEmptyDataSet( + session, deviceIds, times, typesList, valuesList, measurementsList, batchContext, 3); pointSize.set(0); } - if (!failedRecords.isEmpty()) { - writeFailedLinesFile(headerNames, failedFilePath, failedRecords); + if (!batchContext.failedRecords.isEmpty()) { + writeFailedLinesFile( + headerNames, failedFilePath, new ArrayList<>(batchContext.failedRecords.values())); } if (Boolean.TRUE.equals(hasStarted.get())) { ioTPrinter.println("Import completely!"); diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AsyncImportData.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AsyncImportData.java index 6500ef67083..71f384060cc 100644 --- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AsyncImportData.java +++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AsyncImportData.java @@ -37,12 +37,14 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Stream; public class AsyncImportData extends AbstractDataTool implements Runnable { private static final IoTPrinter ioTPrinter = new IoTPrinter(System.out); private static SessionPool sessionPool; + private static final AtomicBoolean importFailed = new AtomicBoolean(); protected static void setTimeZone() throws IoTDBConnectionException, StatementExecutionException { if (timeZoneID != null) { @@ -54,17 +56,20 @@ public class AsyncImportData extends AbstractDataTool implements Runnable { @Override public void run() { String filePath; - try { - while ((filePath = ImportDataScanTool.pollFromQueue()) != null) { - File file = new File(filePath); + while ((filePath = ImportDataScanTool.pollFromQueue()) != null) { + resetCurrentImportFailed(); + File file = new File(filePath); + try { if (file.getName().endsWith(SQL_SUFFIXS)) { importFromSqlFile(file); } else { importFromSingleFile(file); } + } catch (Exception e) { + importFailed.set(true); + loadFileFailedNum.increment(); + ioTPrinter.println(file.getName() + ": " + e.getMessage()); } - } catch (Exception e) { - ioTPrinter.println("Unexpected error occurred: " + e.getMessage()); } } @@ -72,9 +77,25 @@ public class AsyncImportData extends AbstractDataTool implements Runnable { loadFileSuccessfulNum.increment(); } + protected static void resetImportStatus() { + importFailed.set(false); + } + + protected static boolean hasImportFailed() { + return importFailed.get(); + } + + private static void processFailedFile() { + markCurrentImportFailed(); + importFailed.set(true); + loadFileFailedNum.increment(); + } + @SuppressWarnings("java:S2259") private static void importFromSqlFile(File file) { ArrayList<List<Object>> failedRecords = new ArrayList<>(); + boolean fileFailed = false; + boolean failedCounted = false; String failedFilePath; if (failedFileDirectory == null) { failedFilePath = file.getAbsolutePath() + ".failed"; @@ -87,13 +108,20 @@ public class AsyncImportData extends AbstractDataTool implements Runnable { try { sessionPool.executeNonQueryStatement(sql); } catch (IoTDBConnectionException | StatementExecutionException e) { + fileFailed = true; failedRecords.add(Arrays.asList(sql)); } } - processSuccessFile(); } catch (IOException e) { + fileFailed = true; ioTPrinter.println("SQL file read exception because: " + e.getMessage()); } + if (fileFailed) { + processFailedFile(); + failedCounted = true; + } else { + processSuccessFile(); + } if (!failedRecords.isEmpty()) { FileWriter writer = null; try { @@ -102,6 +130,11 @@ public class AsyncImportData extends AbstractDataTool implements Runnable { writer.write(failedRecord.get(0).toString() + "\n"); } } catch (IOException e) { + markCurrentImportFailed(); + importFailed.set(true); + if (!failedCounted) { + processFailedFile(); + } ioTPrinter.println("Cannot dump fail result because: " + e.getMessage()); } finally { if (ObjectUtils.isNotEmpty(writer)) { @@ -118,15 +151,16 @@ public class AsyncImportData extends AbstractDataTool implements Runnable { private static void importFromSingleFile(File file) { if (file.getName().endsWith(CSV_SUFFIXS) || file.getName().endsWith(TXT_SUFFIXS)) { - try { - CSVParser csvRecords = readCsvFile(file.getAbsolutePath()); + try (CSVParser csvRecords = readCsvFile(file.getAbsolutePath())) { List<String> headerNames = csvRecords.getHeaderNames(); Stream<CSVRecord> records = csvRecords.stream(); if (headerNames.isEmpty()) { + processFailedFile(); ioTPrinter.println("Empty file!"); return; } if (!timeColumn.equalsIgnoreCase(filterBomHeader(headerNames.get(0)))) { + processFailedFile(); ioTPrinter.println("The first field of header must be `Time`!"); return; } @@ -141,11 +175,17 @@ public class AsyncImportData extends AbstractDataTool implements Runnable { } else { writeDataAlignedByDevice(sessionPool, headerNames, records, failedFilePath); } - processSuccessFile(); + if (isCurrentImportFailed()) { + processFailedFile(); + } else { + processSuccessFile(); + } } catch (IOException | IllegalPathException e) { + processFailedFile(); ioTPrinter.println("CSV file read exception because: " + e.getMessage()); } } else { + processFailedFile(); ioTPrinter.println("The file name must end with \"csv\" or \"txt\"!"); } } 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 e2a08af86d8..5a451ccc4bc 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 @@ -59,8 +59,11 @@ import java.time.ZonedDateTime; import java.time.format.DateTimeFormatter; import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Locale; +import java.util.Map; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -275,13 +278,19 @@ public class ExportData extends AbstractDataTool { ioTPrinter.println(sql); String[] values = sql.trim().split(";"); for (int i = 0; i < values.length; i++) { - dumpResult(values[i], i); + if (!dumpResult(values[i], i)) { + exitCode = CODE_ERROR; + } } } else { - dumpFromSqlFile(sqlFile); + if (!dumpFromSqlFile(sqlFile)) { + exitCode = CODE_ERROR; + } } } else { - dumpResult(queryCommand, 0); + if (!dumpResult(queryCommand, 0)) { + exitCode = CODE_ERROR; + } } } catch (IOException e) { @@ -571,15 +580,17 @@ public class ExportData extends AbstractDataTool { * @param filePath sql file path * @throws IOException exception */ - private static void dumpFromSqlFile(String filePath) throws IOException { + private static boolean dumpFromSqlFile(String filePath) throws IOException { + boolean succeeded = true; try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) { String sql; int index = 0; while ((sql = reader.readLine()) != null) { - dumpResult(sql, index); + succeeded &= dumpResult(sql, index); index++; } } + return succeeded; } /** @@ -588,36 +599,36 @@ public class ExportData extends AbstractDataTool { * @param sql export the result of executing the sql * @param index used to create dump file name */ - private static void dumpResult(String sql, int index) { + private static boolean dumpResult(String sql, int index) { if (EXPORT_SQL_TYPE_NAME.equalsIgnoreCase(exportType)) { legalCheck(sql); } final String path = targetDirectory + targetFile + index; - try { - SessionDataSet sessionDataSet = session.executeQueryStatement(sql, timeout); + try (SessionDataSet sessionDataSet = session.executeQueryStatement(sql, timeout)) { List<Object> headers = new ArrayList<>(); List<String> names = sessionDataSet.getColumnNames(); List<String> types = sessionDataSet.getColumnTypes(); if (EXPORT_SQL_TYPE_NAME.equalsIgnoreCase(exportType)) { writeSqlFile(sessionDataSet, path, names); } else { - if (Boolean.TRUE.equals(needDataTypePrinted)) { - for (int i = 0; i < names.size(); i++) { - if (!"Time".equals(names.get(i)) && !"Device".equals(names.get(i))) { - headers.add(String.format("%s(%s)", names.get(i), types.get(i))); - } else { - headers.add(names.get(i)); - } + for (int i = 0; i < names.size(); i++) { + String name = names.get(i); + if (Boolean.TRUE.equals(needDataTypePrinted) + && !"Time".equalsIgnoreCase(name) + && !"Device".equals(name)) { + headers.add(String.format("%s(%s)", name, types.get(i))); + } else { + headers.add(name); } - } else { - headers.addAll(names); } writeCsvFile(sessionDataSet, path, headers, linesPerFile); } sessionDataSet.closeOperationHandle(); ioTPrinter.println("Export completely!"); + return true; } catch (StatementExecutionException | IoTDBConnectionException | IOException e) { ioTPrinter.println("Cannot dump result because: " + e.getMessage()); + return false; } } @@ -728,41 +739,24 @@ public class ExportData extends AbstractDataTool { 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; try { while (iterator.next()) { - if (writer == null) { - writer = new FileWriter(String.format(filePathTemplate, fileIndex)); - } - deviceName = iterator.getString(2); + String deviceName = iterator.getString(2); if (deviceName.startsWith(SYSTEM_DATABASE + ".")) { continue; } + if (writer == null) { + writer = new FileWriter(String.format(filePathTemplate, fileIndex)); + } List<String> values = new ArrayList<>(); for (int index = 2; index < totalColumns; index++) { String dataType = columnTypeList.get(index); - String value = iterator.getString(index + 1); - if (value == null) { - values.add("null"); - continue; - } - if ("TEXT".equalsIgnoreCase(dataType) - || "STRING".equalsIgnoreCase(dataType) - || "DATE".equalsIgnoreCase(dataType)) { - values.add(String.format("\"%s\"", value)); - } else if ("BLOB".equalsIgnoreCase(dataType)) { - if (value.length() >= 2 && (value.startsWith("0x") || value.startsWith("0X"))) { - values.add(String.format("X'%s'", value.substring(2))); - } else { - values.add(String.format("X'%s'", value)); - } - } else { - values.add(value); - } + String value = iterator.isNull(index + 1) ? null : iterator.getString(index + 1); + values.add(formatSqlValue(value, dataType)); } long timestamp = iterator.getLong(1); writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values))); @@ -788,79 +782,61 @@ public class ExportData extends AbstractDataTool { private static void exportToSqlFileWithoutAlign( SessionDataSet sessionDataSet, String filePath, List<String> headers) throws IoTDBConnectionException, StatementExecutionException, IOException { - - List<String> measurementNames = new ArrayList<>(); - String deviceName = null; if (CollectionUtils.isEmpty(headers) || headers.size() <= 1) { return; - } else { - List<String> localHeaders = new ArrayList<>(headers); - localHeaders.remove("Time"); - if (CollectionUtils.isEmpty(localHeaders)) { - return; - } - Path path = new Path(localHeaders.get(0), true); - deviceName = path.getDevice(); - for (String header : localHeaders) { - path = new Path(header, true); - String meas = path.getMeasurement(); - if (path.getDevice().equals(deviceName)) { - measurementNames.add(meas); - } - } } - if (deviceName.startsWith(SYSTEM_DATABASE + ".")) { - return; + + 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.getDevice(); + if (device.startsWith(SYSTEM_DATABASE + ".")) { + continue; + } + deviceColumnIndices.computeIfAbsent(device, key -> new ArrayList<>()).add(i); + measurements.put(i, path.getMeasurement()); } - String sqlPrefix = - String.format( - "INSERT INTO %s(TIMESTAMP,%s) VALUES (%s);\n", - "%s", String.join(",", measurementNames), "%d,%s"); SessionDataSet.DataIterator iterator = sessionDataSet.iterator(); List<String> columnTypeList = iterator.getColumnTypeList(); - int totalColumns = measurementNames.size(); int fileIndex = 0; int currentLines = 0; - String filePathTemplate = filePath + "_%d" + ".sql"; FileWriter writer = null; try { while (iterator.next()) { - if (writer == null) { - writer = new FileWriter(String.format(filePathTemplate, fileIndex)); - } - List<String> values = new ArrayList<>(); - for (int index = 0; index < totalColumns; index++) { - String dataType = columnTypeList.get(index + 1); - String value = iterator.getString(index + 2); - if (value == null) { - values.add("null"); + for (Map.Entry<String, List<Integer>> entry : deviceColumnIndices.entrySet()) { + List<String> names = new ArrayList<>(); + List<String> values = new ArrayList<>(); + for (int columnIndex : entry.getValue()) { + if (!iterator.isNull(columnIndex + 1)) { + names.add(measurements.get(columnIndex)); + values.add( + formatSqlValue( + iterator.getString(columnIndex + 1), columnTypeList.get(columnIndex))); + } + } + // A timestamp present in another device must not create an empty record here. + if (values.isEmpty()) { continue; } - if ("TEXT".equalsIgnoreCase(dataType) - || "STRING".equalsIgnoreCase(dataType) - || "DATE".equalsIgnoreCase(dataType)) { - values.add(String.format("\"%s\"", value)); - } else if ("BLOB".equalsIgnoreCase(dataType)) { - if (value.length() >= 2 && (value.startsWith("0x") || value.startsWith("0X"))) { - values.add(String.format("X'%s'", value.substring(2))); - } else { - values.add(String.format("X'%s'", value)); - } - } else { - values.add(value); + if (writer == null) { + writer = new FileWriter(filePath + "_" + fileIndex + ".sql"); + } + 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.flush(); + writer.close(); + writer = null; + fileIndex++; + currentLines = 0; } - } - long timestamp = iterator.getLong(1); - writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values))); - currentLines += 1; - - if (currentLines >= linesPerFile) { - writer.flush(); - writer.close(); - fileIndex += 1; - writer = null; - currentLines = 0; } } } finally { @@ -872,6 +848,23 @@ public class ExportData extends AbstractDataTool { ioTPrinter.print("\n"); } + private static String formatSqlValue(String value, String dataType) { + if (value == null) { + return "NULL"; + } + if ("TEXT".equalsIgnoreCase(dataType) + || "STRING".equalsIgnoreCase(dataType) + || "DATE".equalsIgnoreCase(dataType)) { + return "'" + value.replace("'", "''") + "'"; + } + if ("BLOB".equalsIgnoreCase(dataType)) { + String hexValue = + value.startsWith("0x") || value.startsWith("0X") ? value.substring(2) : value; + return "X'" + hexValue + "'"; + } + return value; + } + public static void writeSqlFile( SessionDataSet sessionDataSet, String filePath, List<String> headers) throws IOException, IoTDBConnectionException, StatementExecutionException { diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ImportData.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ImportData.java index 753713b9d1f..86a369b8009 100644 --- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ImportData.java +++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ImportData.java @@ -567,12 +567,7 @@ public class ImportData extends AbstractDataTool { ioTPrinter.println("Encounter an error, because: " + e.getMessage()); System.exit(CODE_ERROR); } - final int resultCode = importFromTargetPathAsync(); - if (ImportDataScanTool.getTsFileQueueSize() <= 0) { - System.exit(CODE_OK); - } - asyncImportDataFiles(); - System.exit(resultCode); + System.exit(importFromTargetPathAsync()); } private static void asyncImportDataFiles() { @@ -611,6 +606,7 @@ public class ImportData extends AbstractDataTool { AsyncImportData.setAligned(aligned); AsyncImportData.setSessionPool(sessionPool); AsyncImportData.setTimeZone(); + AsyncImportData.resetImportStatus(); ImportDataScanTool.setSourceFullPath(targetPath); final File file = new File(targetPath); if (!file.isFile() && !file.isDirectory()) { @@ -619,7 +615,11 @@ public class ImportData extends AbstractDataTool { } ImportDataScanTool.traverseAndCollectFiles(); asyncImportDataFiles(); - return CODE_OK; + return AsyncImportData.hasImportFailed() + || ImportDataScanTool.getTsFileQueueSize() > 0 + || Thread.currentThread().isInterrupted() + ? CODE_ERROR + : CODE_OK; } catch (InterruptedException e) { ioTPrinter.println(String.format("Import tsfile fail: %s", e.getMessage())); Thread.currentThread().interrupt(); diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFile.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFile.java index cdfbd5b4437..9ae03c8f55d 100644 --- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFile.java +++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFile.java @@ -368,8 +368,7 @@ public class ImportTsFile extends AbstractTsFileTool { ImportTsFileScanTool.traverseAndCollectFiles(); ImportTsFileScanTool.addNoResourceOrModsToQueue(); - asyncImportTsFiles(); - return CODE_OK; + return asyncImportTsFiles(); } catch (InterruptedException e) { IOT_PRINTER.println(String.format("Import tsfile fail: %s", e.getMessage())); Thread.currentThread().interrupt(); @@ -409,7 +408,8 @@ public class ImportTsFile extends AbstractTsFileTool { successDir, successOperation, failDir, failOperation); } - public static void asyncImportTsFiles() { + public static int asyncImportTsFiles() { + ImportTsFileBase.resetCounters(); final List<Thread> list = new ArrayList<>(threadNum); for (int i = 0; i < threadNum; i++) { final Thread thread = @@ -429,6 +429,9 @@ public class ImportTsFile extends AbstractTsFileTool { IOT_PRINTER.println("ImportTsFile thread join interrupted: " + e.getMessage()); } }); + return Thread.currentThread().isInterrupted() || ImportTsFileBase.hasFailures() + ? CODE_ERROR + : CODE_OK; } public enum Operation { diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileBase.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileBase.java index 6d6f95dcb82..b6ab21bde7b 100644 --- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileBase.java +++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileBase.java @@ -20,13 +20,11 @@ package org.apache.iotdb.tool.tsfile; import org.apache.iotdb.cli.utils.IoTPrinter; +import org.apache.iotdb.tool.common.ImportTsFileOperation; -import java.io.File; -import java.nio.file.FileAlreadyExistsException; -import java.nio.file.Files; +import java.io.IOException; import java.nio.file.Path; import java.nio.file.Paths; -import java.nio.file.StandardCopyOption; import java.util.Objects; import java.util.concurrent.atomic.LongAdder; @@ -40,9 +38,9 @@ public abstract class ImportTsFileBase implements Runnable { private static final LongAdder processingLoadFailedFileSuccessfulNum = new LongAdder(); private static String timePrecision = "ms"; private static String successDir; - private static ImportTsFile.Operation successOperation; + private static ImportTsFileOperation successOperation; private static String failDir; - private static ImportTsFile.Operation failOperation; + private static ImportTsFileOperation failOperation; @Override public void run() { @@ -100,107 +98,26 @@ public abstract class ImportTsFileBase implements Runnable { } } - public static void processingFile(final String filePath, final boolean isSuccess) { - final String relativePath = - filePath.substring(ImportTsFileScanTool.getSourceFullPathLength() + 1); - final Path sourcePath = Paths.get(filePath); - - final String target = - isSuccess - ? successDir - : failDir + File.separator + relativePath.replace(File.separator, "_"); - final Path targetPath = Paths.get(target); - - final String RESOURCE = ".resource"; - Path sourceResourcePath = Paths.get(sourcePath + RESOURCE); - sourceResourcePath = Files.exists(sourceResourcePath) ? sourceResourcePath : null; - final Path targetResourcePath = Paths.get(target + RESOURCE); - - final String MODS = ".mods"; - Path sourceModsPath = Paths.get(sourcePath + MODS); - sourceModsPath = Files.exists(sourceModsPath) ? sourceModsPath : null; - final Path targetModsPath = Paths.get(target + MODS); - - switch (isSuccess ? successOperation : failOperation) { - case DELETE: - { - try { - Files.deleteIfExists(sourcePath); - if (null != sourceResourcePath) { - Files.deleteIfExists(sourceResourcePath); - } - if (null != sourceModsPath) { - Files.deleteIfExists(sourceModsPath); - } - } catch (final Exception e) { - IOT_PRINTER.println(String.format("Failed to delete file: %s", e.getMessage())); - } - break; - } - case CP: - { - try { - Files.copy(sourcePath, targetPath, StandardCopyOption.REPLACE_EXISTING); - if (null != sourceResourcePath) { - Files.copy( - sourceResourcePath, targetResourcePath, StandardCopyOption.REPLACE_EXISTING); - } - if (null != sourceModsPath) { - Files.copy(sourceModsPath, targetModsPath, StandardCopyOption.REPLACE_EXISTING); - } - } catch (final Exception e) { - IOT_PRINTER.println(String.format("Failed to copy file: %s", e.getMessage())); - } - break; - } - case HARDLINK: - { - try { - Files.createLink(targetPath, sourcePath); - } catch (FileAlreadyExistsException e) { - IOT_PRINTER.println("Hardlink already exists: " + e.getMessage()); - } catch (final Exception e) { - try { - Files.copy(sourcePath, targetPath, StandardCopyOption.REPLACE_EXISTING); - } catch (final Exception copyException) { - IOT_PRINTER.println( - String.format("Failed to copy file: %s", copyException.getMessage())); - } - } - - try { - if (null != sourceResourcePath) { - Files.copy( - sourceResourcePath, targetResourcePath, StandardCopyOption.REPLACE_EXISTING); - } - if (null != sourceModsPath) { - Files.copy(sourceModsPath, targetModsPath, StandardCopyOption.REPLACE_EXISTING); - } - } catch (final Exception e) { - IOT_PRINTER.println( - String.format("Failed to copy resource or mods file: %s", e.getMessage())); - } - break; - } - case MV: - { - try { - Files.move(sourcePath, targetPath, StandardCopyOption.REPLACE_EXISTING); - if (null != sourceResourcePath) { - Files.move( - sourceResourcePath, targetResourcePath, StandardCopyOption.REPLACE_EXISTING); - } - if (null != sourceModsPath) { - Files.move(sourceModsPath, targetModsPath, StandardCopyOption.REPLACE_EXISTING); - } - } catch (final Exception e) { - IOT_PRINTER.println(String.format("Failed to move file: %s", e.getMessage())); - } - break; - } - default: - break; - } + public static void processingFile(final String filePath, final boolean isSuccess) + throws IOException { + final String relativePath = ImportTsFileScanTool.getRelativePath(filePath); + final Path target = Paths.get(isSuccess ? successDir : failDir).resolve(relativePath); + final ImportTsFileOperation operation = isSuccess ? successOperation : failOperation; + operation.processFileGroup(Paths.get(filePath), target); + } + + protected static void resetCounters() { + loadFileSuccessfulNum.reset(); + loadFileFailedNum.reset(); + processingLoadSuccessfulFileSuccessfulNum.reset(); + processingLoadFailedFileSuccessfulNum.reset(); + } + + protected static boolean hasFailures() { + return loadFileFailedNum.sum() > 0 + || loadFileSuccessfulNum.sum() != processingLoadSuccessfulFileSuccessfulNum.sum() + || loadFileFailedNum.sum() != processingLoadFailedFileSuccessfulNum.sum() + || ImportTsFileScanTool.getTsFileQueueSize() > 0; } protected static void printResult(final long startTime) { @@ -232,15 +149,27 @@ public abstract class ImportTsFileBase implements Runnable { public static void setSuccessAndFailDirAndOperation( final String successDir, - final ImportTsFile.Operation successOperation, + final ImportTsFileOperation successOperation, final String failDir, - final ImportTsFile.Operation failOperation) { + final ImportTsFileOperation failOperation) { ImportTsFileBase.successDir = successDir; ImportTsFileBase.successOperation = successOperation; ImportTsFileBase.failDir = failDir; ImportTsFileBase.failOperation = failOperation; } + public static void setSuccessAndFailDirAndOperation( + final String successDir, + final ImportTsFile.Operation successOperation, + final String failDir, + final ImportTsFile.Operation failOperation) { + setSuccessAndFailDirAndOperation( + successDir, + successOperation == null ? null : ImportTsFileOperation.valueOf(successOperation.name()), + failDir, + failOperation == null ? null : ImportTsFileOperation.valueOf(failOperation.name())); + } + protected void setTimePrecision(String timePrecision) { this.timePrecision = timePrecision; } diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileScanTool.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileScanTool.java index fe520502ec3..5332319607a 100644 --- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileScanTool.java +++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileScanTool.java @@ -20,6 +20,8 @@ package org.apache.iotdb.tool.tsfile; import java.io.File; +import java.nio.file.Path; +import java.nio.file.Paths; import java.util.HashSet; import java.util.Set; import java.util.concurrent.LinkedBlockingQueue; @@ -36,6 +38,9 @@ public class ImportTsFileScanTool { private static String sourceFullPath; public static void traverseAndCollectFiles() throws InterruptedException { + tsfileQueue.clear(); + tsfileSet.clear(); + resourceOrModsSet.clear(); traverseAndCollectFilesBySourceFullPath(new File(sourceFullPath)); } @@ -92,6 +97,12 @@ public class ImportTsFileScanTool { : new File(ImportTsFileScanTool.sourceFullPath).getParent().length(); } + public static String getRelativePath(String file) { + Path source = Paths.get(sourceFullPath).toAbsolutePath().normalize(); + Path sourceDirectory = source.toFile().isDirectory() ? source : source.getParent(); + return sourceDirectory.relativize(Paths.get(file).toAbsolutePath().normalize()).toString(); + } + public static int getTsFileQueueSize() { return ImportTsFileScanTool.tsfileQueue.size(); } diff --git a/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ExportDataTest.java b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ExportDataTest.java new file mode 100644 index 00000000000..4506e484883 --- /dev/null +++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ExportDataTest.java @@ -0,0 +1,92 @@ +/* + * 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.SessionDataSet; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class ExportDataTest { + + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void exportsInterleavedDevicesWithoutEmptyRecords() throws Exception { + SessionDataSet dataSet = mock(SessionDataSet.class); + SessionDataSet.DataIterator iterator = mock(SessionDataSet.DataIterator.class); + when(dataSet.iterator()).thenReturn(iterator); + when(iterator.getColumnTypeList()) + .thenReturn(List.of("INT64", "TEXT", "TEXT", "INT32", "INT32")); + when(iterator.next()).thenReturn(true, true, false); + when(iterator.getLong(1)).thenReturn(1L, 2L); + when(iterator.isNull(2)).thenReturn(false, true); + when(iterator.isNull(3)).thenReturn(true, false); + when(iterator.isNull(4)).thenReturn(false, true); + when(iterator.isNull(5)).thenReturn(true); + when(iterator.getString(2)).thenReturn("O'Reilly"); + when(iterator.getString(3)).thenReturn("a \"quoted\" value"); + when(iterator.getString(4)).thenReturn("7"); + List<String> headers = List.of("Time", "root.d.s1", "root.e.s1", "root.d.s2", "root.e.s2"); + Path output = temporaryFolder.getRoot().toPath().resolve("export"); + + ExportData.writeSqlFile(dataSet, output.toString(), headers); + + assertEquals( + "INSERT INTO root.d(TIMESTAMP,s1,s2) VALUES (1,'O''Reilly',7);\n" + + "INSERT INTO root.e(TIMESTAMP,s1) VALUES (2,'a \"quoted\" value');\n", + Files.readString(Path.of(output + "_0.sql")).replace("\r\n", "\n")); + assertEquals("Time", headers.get(0)); + } + + @Test + public void alignedExportPreservesNullsAndTypedValues() throws Exception { + SessionDataSet dataSet = mock(SessionDataSet.class); + SessionDataSet.DataIterator iterator = mock(SessionDataSet.DataIterator.class); + when(dataSet.iterator()).thenReturn(iterator); + when(iterator.getColumnTypeList()) + .thenReturn(List.of("INT64", "TEXT", "TEXT", "STRING", "BLOB", "DATE")); + when(iterator.next()).thenReturn(true, false); + when(iterator.getLong(1)).thenReturn(3L); + when(iterator.getString(2)).thenReturn("root.d"); + when(iterator.isNull(3)).thenReturn(true); + when(iterator.getString(4)).thenReturn("O'Reilly"); + when(iterator.getString(5)).thenReturn("0xABCD"); + when(iterator.getString(6)).thenReturn("2026-10-22"); + List<String> headers = List.of("Time", "Device", "s1", "s2", "s3", "s4"); + Path output = temporaryFolder.getRoot().toPath().resolve("aligned"); + + ExportData.writeSqlFile(dataSet, output.toString(), headers); + + assertEquals( + "INSERT INTO root.d(TIMESTAMP,s1,s2,s3,s4) ALIGNED VALUES (3,NULL,'O''Reilly',X'ABCD','2026-10-22');\n", + Files.readString(Path.of(output + "_0.sql")).replace("\r\n", "\n")); + assertEquals(List.of("Time", "Device", "s1", "s2", "s3", "s4"), headers); + } +} diff --git a/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ImportDataFailureTest.java b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ImportDataFailureTest.java new file mode 100644 index 00000000000..2204ab4018e --- /dev/null +++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ImportDataFailureTest.java @@ -0,0 +1,108 @@ +/* + * 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.rpc.StatementExecutionException; +import org.apache.iotdb.session.pool.SessionPool; + +import org.apache.commons.csv.CSVParser; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.junit.Assert.assertEquals; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; + +public class ImportDataFailureTest { + + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Before + public void setUp() { + AbstractDataTool.aligned = false; + } + + @After + public void tearDown() { + AbstractDataTool.aligned = null; + } + + @Test + public void failedBatchWritesEachOriginalCsvRecordOnce() throws Exception { + SessionPool sessionPool = mock(SessionPool.class); + doAnswer( + invocation -> { + throw new StatementExecutionException("insert failed"); + }) + .when(sessionPool) + .insertRecords(anyList(), anyList(), anyList(), anyList(), anyList()); + + Path input = temporaryFolder.newFile("input.csv").toPath(); + Files.writeString(input, "Time,root.d.s1(INT32),root.e.s1(INT32)\n1,1,2\n"); + Path failed = temporaryFolder.getRoot().toPath().resolve("failed"); + + try (CSVParser parser = AbstractDataTool.readCsvFile(input.toString())) { + AbstractDataTool.writeDataAlignedByTime( + sessionPool, parser.getHeaderNames(), parser.stream(), failed.toString()); + } + + assertEquals( + "Time,root.d.s1(INT32),root.e.s1(INT32)\n1,1,2\n", + Files.readString(Path.of(failed + "_0")).replace("\r\n", "\n")); + } + + @Test + public void failedDeviceGroupWritesTheWholeOriginalRecordOnce() throws Exception { + SessionPool sessionPool = mock(SessionPool.class); + doAnswer( + invocation -> { + if (((java.util.List<?>) invocation.getArgument(0)).size() == 1 + && "root.e".equals(((java.util.List<?>) invocation.getArgument(0)).get(0))) { + throw new StatementExecutionException("device insert failed"); + } + if (((java.util.List<?>) invocation.getArgument(0)).size() > 1) { + throw new StatementExecutionException("batch insert failed"); + } + return null; + }) + .when(sessionPool) + .insertRecords(anyList(), anyList(), anyList(), anyList(), anyList()); + + Path input = temporaryFolder.newFile("input.csv").toPath(); + Files.writeString(input, "Time,root.d.s1(INT32),root.e.s1(INT32)\n1,1,2\n"); + Path failed = temporaryFolder.getRoot().toPath().resolve("failed"); + + try (CSVParser parser = AbstractDataTool.readCsvFile(input.toString())) { + AbstractDataTool.writeDataAlignedByTime( + sessionPool, parser.getHeaderNames(), parser.stream(), failed.toString()); + } + + assertEquals( + "Time,root.d.s1(INT32),root.e.s1(INT32)\n1,1,2\n", + Files.readString(Path.of(failed + "_0")).replace("\r\n", "\n")); + } +} diff --git a/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/tsfile/ImportTsFileTest.java b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/tsfile/ImportTsFileTest.java new file mode 100644 index 00000000000..229ae608487 --- /dev/null +++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/tsfile/ImportTsFileTest.java @@ -0,0 +1,204 @@ +/* + * 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.tsfile; + +import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.session.pool.SessionPool; +import org.apache.iotdb.tool.common.ImportTsFileOperation; + +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.lang.reflect.Field; +import java.lang.reflect.Modifier; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +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; + +public class ImportTsFileTest extends AbstractTsFileTool { + + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private final Map<Field, Object> originalFields = new HashMap<>(); + private SessionPool sessionPool; + private Path input; + private Path archive; + + @Before + public void setUp() throws Exception { + for (Class<?> owner : + List.of( + ImportTsFile.class, + ImportTsFileBase.class, + ImportTsFileLocally.class, + ImportTsFileRemotely.class, + ImportTsFileScanTool.class)) { + for (Field field : owner.getDeclaredFields()) { + if (Modifier.isStatic(field.getModifiers()) && !Modifier.isFinal(field.getModifiers())) { + field.setAccessible(true); + originalFields.put(field, field.get(null)); + } + } + } + sessionPool = mock(SessionPool.class); + ImportTsFileLocally.setSessionPool(sessionPool); + setImportOption("isRemoteLoad", false); + setImportOption("threadNum", 2); + input = Files.createDirectories(temporaryFolder.getRoot().toPath().resolve("input")); + archive = Files.createDirectories(temporaryFolder.getRoot().toPath().resolve("archive")); + ImportTsFileScanTool.setSourceFullPath(input.toString()); + ImportTsFileBase.setSuccessAndFailDirAndOperation( + archive.toString(), ImportTsFileOperation.CP, archive.toString(), ImportTsFileOperation.CP); + } + + @After + public void tearDown() throws Exception { + while (ImportTsFileScanTool.pollFromQueue() != null) { + // Drain any files left by a failing assertion before restoring shared state. + } + ImportTsFileBase.resetCounters(); + for (Map.Entry<Field, Object> entry : originalFields.entrySet()) { + entry.getKey().set(null, entry.getValue()); + } + } + + @Test + public void testSuccessfulImportReturnsZero() throws Exception { + Path source = createTsFileGroup("one.tsfile"); + ImportTsFileScanTool.putToQueue(source.toString()); + assertEquals(CODE_OK, ImportTsFile.asyncImportTsFiles()); + assertArchived("one.tsfile"); + verify(sessionPool).executeNonQueryStatement(anyString()); + assertNull(ImportTsFileScanTool.pollFromQueue()); + } + + @Test + public void testArchiveConflictsReturnErrorAndContinueImport() throws Exception { + for (ImportTsFileOperation operation : + List.of( + ImportTsFileOperation.CP, ImportTsFileOperation.HARDLINK, ImportTsFileOperation.MV)) { + ImportTsFileBase.setSuccessAndFailDirAndOperation( + archive.toString(), operation, archive.toString(), operation); + for (String conflict : List.of("", ".resource", ".mods")) { + String name = operation + conflict + ".tsfile"; + Path source = createTsFileGroup(name); + Path occupied = archive.resolve(name + conflict); + Files.writeString(occupied, "existing"); + String goodName = "good-" + name; + Path good = createTsFileGroup(goodName); + ImportTsFileScanTool.putToQueue(source.toString()); + ImportTsFileScanTool.putToQueue(good.toString()); + + assertEquals(CODE_ERROR, ImportTsFile.asyncImportTsFiles()); + assertEquals("existing", Files.readString(occupied)); + for (String suffix : List.of("", ".resource", ".mods")) { + assertEquals("source" + suffix, Files.readString(Path.of(source + suffix))); + if (!suffix.equals(conflict)) { + assertFalse(Files.exists(archive.resolve(name + suffix))); + } + } + assertArchived(goodName); + assertNull(ImportTsFileScanTool.pollFromQueue()); + } + } + verify(sessionPool, times(18)).executeNonQueryStatement(anyString()); + } + + @Test + public void testArchiveIoFailureReturnsError() throws Exception { + Files.createDirectories(input.resolve("nested")); + Path source = createTsFileGroup("nested/one.tsfile"); + Files.writeString(archive.resolve("nested"), "blocking file"); + ImportTsFileScanTool.putToQueue(source.toString()); + assertEquals(CODE_ERROR, ImportTsFile.asyncImportTsFiles()); + assertEquals("source", Files.readString(source)); + assertEquals("blocking file", Files.readString(archive.resolve("nested"))); + } + + @Test + public void testLoadFailureReturnsErrorWhenArchiveSucceeds() throws Exception { + Path source = createTsFileGroup("one.tsfile"); + doThrow(new StatementExecutionException("load failed")) + .when(sessionPool) + .executeNonQueryStatement(anyString()); + ImportTsFileScanTool.putToQueue(source.toString()); + assertEquals(CODE_ERROR, ImportTsFile.asyncImportTsFiles()); + assertArchived("one.tsfile"); + verify(sessionPool).executeNonQueryStatement(anyString()); + } + + @Test + public void testExitCodeResetsForNextImport() throws Exception { + Path source = createTsFileGroup("one.tsfile"); + Path occupied = archive.resolve("one.tsfile"); + Files.writeString(occupied, "existing"); + ImportTsFileScanTool.putToQueue(source.toString()); + assertEquals(CODE_ERROR, ImportTsFile.asyncImportTsFiles()); + Files.delete(occupied); + ImportTsFileScanTool.putToQueue(source.toString()); + assertEquals(CODE_OK, ImportTsFile.asyncImportTsFiles()); + assertArchived("one.tsfile"); + verify(sessionPool, times(2)).executeNonQueryStatement(anyString()); + } + + @Test + public void testImportFromTargetPathReturnsErrorForUnprocessedFiles() throws Exception { + setImportOption("threadNum", 0); + setImportOption("isRemoteLoad", true); + Path source = createTsFileGroup("one.tsfile"); + setImportOption("source", source.toString()); + assertEquals(CODE_ERROR, ImportTsFile.importFromTargetPath()); + assertEquals(source.toString(), ImportTsFileScanTool.pollFromQueue()); + } + + private Path createTsFileGroup(String name) throws Exception { + Path source = input.resolve(name); + for (String suffix : List.of("", ".resource", ".mods")) { + Files.writeString(Path.of(source + suffix), "source" + suffix); + } + return source; + } + + private void assertArchived(String name) throws Exception { + for (String suffix : List.of("", ".resource", ".mods")) { + assertEquals("source" + suffix, Files.readString(archive.resolve(name + suffix))); + } + } + + private static void setImportOption(String name, Object value) throws Exception { + Field field = ImportTsFile.class.getDeclaredField(name); + field.setAccessible(true); + field.set(null, value); + } +}
