diff --git a/iotdb-client/cli/src/assembly/resources/tools/export-data.bat b/iotdb-client/cli/src/assembly/resources/tools/export-data.bat index 2178482f36910..574caa77e45b5 100644 --- a/iotdb-client/cli/src/assembly/resources/tools/export-data.bat +++ b/iotdb-client/cli/src/assembly/resources/tools/export-data.bat @@ -57,6 +57,4 @@ pause @REM ----------------------------------------------------------------------------- :finally -ENDLOCAL - -EXIT /B %ret_code% +ENDLOCAL & EXIT /B %ret_code% diff --git a/iotdb-client/cli/src/assembly/resources/tools/export-schema.bat b/iotdb-client/cli/src/assembly/resources/tools/export-schema.bat index dab5dfaf66799..fa51997660f26 100644 --- a/iotdb-client/cli/src/assembly/resources/tools/export-schema.bat +++ b/iotdb-client/cli/src/assembly/resources/tools/export-schema.bat @@ -57,6 +57,4 @@ pause @REM ----------------------------------------------------------------------------- :finally -ENDLOCAL - -EXIT /B %ret_code% +ENDLOCAL & EXIT /B %ret_code% diff --git a/iotdb-client/cli/src/assembly/resources/tools/export-tsfile.bat b/iotdb-client/cli/src/assembly/resources/tools/export-tsfile.bat index 2c85f42bd6278..9ea66f90c187b 100644 --- a/iotdb-client/cli/src/assembly/resources/tools/export-tsfile.bat +++ b/iotdb-client/cli/src/assembly/resources/tools/export-tsfile.bat @@ -57,6 +57,4 @@ pause @REM ----------------------------------------------------------------------------- :finally -ENDLOCAL - -EXIT /B %ret_code% +ENDLOCAL & EXIT /B %ret_code% diff --git a/iotdb-client/cli/src/assembly/resources/tools/import-data.bat b/iotdb-client/cli/src/assembly/resources/tools/import-data.bat index cbb71dcf68d27..0a5cfb2e9825b 100644 --- a/iotdb-client/cli/src/assembly/resources/tools/import-data.bat +++ b/iotdb-client/cli/src/assembly/resources/tools/import-data.bat @@ -58,6 +58,4 @@ pause @REM ----------------------------------------------------------------------------- :finally -ENDLOCAL - -EXIT /B %ret_code% +ENDLOCAL & EXIT /B %ret_code% diff --git a/iotdb-client/cli/src/assembly/resources/tools/import-schema.bat b/iotdb-client/cli/src/assembly/resources/tools/import-schema.bat index fbf5236128be6..1c94c1e5e7d71 100644 --- a/iotdb-client/cli/src/assembly/resources/tools/import-schema.bat +++ b/iotdb-client/cli/src/assembly/resources/tools/import-schema.bat @@ -58,6 +58,4 @@ pause @REM ----------------------------------------------------------------------------- :finally -ENDLOCAL - -EXIT /B %ret_code% +ENDLOCAL & EXIT /B %ret_code% 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 11f4ed56effcb..bc6c8e8a79aa3 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.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 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 @@ protected static void writeDataAlignedByDevice( AtomicInteger pointSize = new AtomicInteger(0); - ArrayList> failedRecords = new ArrayList<>(); + BatchContext batchContext = new BatchContext(); records.forEach( recordObj -> { @@ -421,13 +437,27 @@ protected static void writeDataAlignedByDevice( } 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,12 @@ protected static void writeDataAlignedByDevice( } } 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); + times.add(timestamp); typesList.add(types); valuesList.add(values); measurementsList.add(measurements); @@ -497,15 +529,147 @@ protected static void writeDataAlignedByDevice( }); 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 enum BatchResult { + SUCCESS, + STATEMENT_FAILURE, + CONNECTION_FAILURE + } + + private static final class BatchContext { + private final Map> failedRecords = new LinkedHashMap<>(); + private final List batchRecords = new ArrayList<>(); + + private void recordFailure(CSVRecord record) { + failedRecords.computeIfAbsent( + record.getRecordNumber(), ignored -> record.stream().collect(Collectors.toList())); + } + + private void recordBatch(CSVRecord record) { + batchRecords.add(record); + } + + private void finishBatch( + BatchResult result, GroupInsert groupInsert, Session reconnectableSession, int retryTime) { + if (result == BatchResult.CONNECTION_FAILURE) { + recordRemainingFailures(0); + } else if (result == BatchResult.STATEMENT_FAILURE) { + retryGroups(groupInsert, reconnectableSession, retryTime); + } + batchRecords.clear(); + } + + private void retryGroups(GroupInsert groupInsert, Session reconnectableSession, int retryTime) { + for (int index = 0; index < batchRecords.size(); index++) { + final int groupIndex = index; + BatchResult result = + executeBatchWithRetry( + () -> groupInsert.insert(groupIndex), reconnectableSession, retryTime, false); + if (result == BatchResult.CONNECTION_FAILURE) { + recordRemainingFailures(index); + return; + } + if (result == BatchResult.STATEMENT_FAILURE) { + recordFailure(batchRecords.get(index)); + markCurrentImportFailed(); + } + } + } + + private void recordRemainingFailures(int from) { + for (int index = from; index < batchRecords.size(); index++) { + recordFailure(batchRecords.get(index)); + } + markCurrentImportFailed(); + } + } + + @FunctionalInterface + private interface GroupInsert { + void insert(int index) throws IoTDBConnectionException, StatementExecutionException; + } + + private static BatchResult executeBatchWithRetry( + BatchInsert batchInsert, Session reconnectableSession, int retryTime) { + return executeBatchWithRetry(batchInsert, reconnectableSession, retryTime, true); + } + + private static BatchResult executeBatchWithRetry( + BatchInsert batchInsert, Session reconnectableSession, int retryTime, boolean reportFailure) { + for (int attempt = 0; ; attempt++) { + try { + batchInsert.insert(); + return BatchResult.SUCCESS; + } catch (IoTDBConnectionException e) { + if (attempt >= retryTime) { + reportInsertFailure(e, reportFailure); + return BatchResult.CONNECTION_FAILURE; + } + if (reconnectableSession != null) { + try { + reconnectableSession.open(); + } catch (IoTDBConnectionException reconnectException) { + if (attempt + 1 >= retryTime) { + reportInsertFailure(reconnectException, reportFailure); + return BatchResult.CONNECTION_FAILURE; + } + } + } + } catch (StatementExecutionException e) { + reportInsertFailure(e, reportFailure); + return BatchResult.STATEMENT_FAILURE; + } + } + } + + private static void reportInsertFailure(Exception e, boolean reportFailure) { + if (reportFailure) { + ioTPrinter.println(INSERT_CSV_MEET_ERROR_MSG + e.getMessage()); + } + } + + private static BatchResult insertBatch( + SessionPool sessionPool, + List deviceIds, + List times, + List> typesList, + List> valuesList, + List> 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 +677,46 @@ private static void writeAndEmptyDataSet( List> typesList, List> valuesList, List> 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(); - } + BatchResult result = + 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( + result, + index -> { + if (Boolean.FALSE.equals(aligned)) { + sessionPool.insertRecordsOfOneDevice( + device, + Arrays.asList(times.get(index)), + Arrays.asList(measurementsList.get(index)), + Arrays.asList(typesList.get(index)), + Arrays.asList(valuesList.get(index))); + } else { + sessionPool.insertAlignedRecordsOfOneDevice( + device, + Arrays.asList(times.get(index)), + Arrays.asList(measurementsList.get(index)), + Arrays.asList(typesList.get(index)), + Arrays.asList(valuesList.get(index))); + } + }, + null, + retryTime); + times.clear(); + typesList.clear(); + valuesList.clear(); + measurementsList.clear(); } private static void writeAndEmptyDataSet( @@ -545,38 +726,46 @@ private static void writeAndEmptyDataSet( List> typesList, List> valuesList, List> 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(); - } + BatchResult result = + 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( + result, + index -> { + if (Boolean.FALSE.equals(aligned)) { + session.insertRecords( + Arrays.asList(deviceIds.get(index)), + Arrays.asList(times.get(index)), + Arrays.asList(measurementsList.get(index)), + Arrays.asList(typesList.get(index)), + Arrays.asList(valuesList.get(index))); + } else { + session.insertAlignedRecords( + Arrays.asList(deviceIds.get(index)), + Arrays.asList(times.get(index)), + Arrays.asList(measurementsList.get(index)), + Arrays.asList(typesList.get(index)), + Arrays.asList(valuesList.get(index))); + } + }, + session, + retryTime); + deviceIds.clear(); + times.clear(); + typesList.clear(); + valuesList.clear(); + measurementsList.clear(); } private static void writeAndEmptyDataSet( @@ -586,38 +775,46 @@ private static void writeAndEmptyDataSet( List> typesList, List> valuesList, List> 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(); - } + BatchResult result = + 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( + result, + index -> { + if (Boolean.FALSE.equals(aligned)) { + session.insertRecordsOfOneDevice( + device, + Arrays.asList(times.get(index)), + Arrays.asList(measurementsList.get(index)), + Arrays.asList(typesList.get(index)), + Arrays.asList(valuesList.get(index))); + } else { + session.insertAlignedRecordsOfOneDevice( + device, + Arrays.asList(times.get(index)), + Arrays.asList(measurementsList.get(index)), + Arrays.asList(typesList.get(index)), + Arrays.asList(valuesList.get(index))); + } + }, + session, + retryTime); + times.clear(); + typesList.clear(); + valuesList.clear(); + measurementsList.clear(); } private static void writeAndEmptyDataSet( @@ -627,28 +824,37 @@ private static void writeAndEmptyDataSet( List> typesList, List> valuesList, List> 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(); - } + BatchResult result = + insertBatch( + sessionPool, deviceIds, times, typesList, valuesList, measurementsList, retryTime); + batchContext.finishBatch( + result, + index -> { + if (Boolean.FALSE.equals(aligned)) { + sessionPool.insertRecords( + Arrays.asList(deviceIds.get(index)), + Arrays.asList(times.get(index)), + Arrays.asList(measurementsList.get(index)), + Arrays.asList(typesList.get(index)), + Arrays.asList(valuesList.get(index))); + } else { + sessionPool.insertAlignedRecords( + Arrays.asList(deviceIds.get(index)), + Arrays.asList(times.get(index)), + Arrays.asList(measurementsList.get(index)), + Arrays.asList(typesList.get(index)), + Arrays.asList(valuesList.get(index))); + } + }, + null, + retryTime); + deviceIds.clear(); + times.clear(); + typesList.clear(); + valuesList.clear(); + measurementsList.clear(); } private static void writeFailedLinesFile( @@ -985,7 +1191,7 @@ protected static void writeDataAlignedByTime( AtomicReference hasStarted = new AtomicReference<>(false); AtomicInteger pointSize = new AtomicInteger(0); - ArrayList> failedRecords = new ArrayList<>(); + BatchContext batchContext = new BatchContext(); records.forEach( recordObj -> { @@ -993,7 +1199,14 @@ protected static void writeDataAlignedByTime( 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 +1255,9 @@ protected static void writeDataAlignedByTime( } } if (!measurements.isEmpty()) { - times.add(parseTimestamp(recordObj.get(timeColumn))); + long timestamp = parseTimestamp(recordObj.get(timeColumn)); + batchContext.recordBatch(recordObj); + times.add(timestamp); deviceIds.add(deviceId); typesList.add(types); valuesList.add(values); @@ -1050,17 +1265,18 @@ protected static void writeDataAlignedByTime( } } 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 +1333,7 @@ protected static void writeDataAlignedByDevice( AtomicInteger pointSize = new AtomicInteger(0); - ArrayList> failedRecords = new ArrayList<>(); + BatchContext batchContext = new BatchContext(); records.forEach( recordObj -> { @@ -1127,13 +1343,27 @@ protected static void writeDataAlignedByDevice( } 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 +1423,12 @@ protected static void writeDataAlignedByDevice( } } 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); + times.add(timestamp); typesList.add(types); valuesList.add(values); measurementsList.add(measurements); @@ -1204,13 +1436,23 @@ protected static void writeDataAlignedByDevice( }); 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())); + } + if (!isCurrentImportFailed()) { + ioTPrinter.println("Import completely!"); } - ioTPrinter.println("Import completely!"); } /** @@ -1243,7 +1485,7 @@ protected static void writeDataAlignedByTime( AtomicReference hasStarted = new AtomicReference<>(false); AtomicInteger pointSize = new AtomicInteger(0); - ArrayList> failedRecords = new ArrayList<>(); + BatchContext batchContext = new BatchContext(); records.forEach( recordObj -> { @@ -1251,7 +1493,14 @@ protected static void writeDataAlignedByTime( 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 +1546,9 @@ protected static void writeDataAlignedByTime( } } if (!measurements.isEmpty()) { - times.add(parseTimestamp(recordObj.get(timeColumn))); + long timestamp = parseTimestamp(recordObj.get(timeColumn)); + batchContext.recordBatch(recordObj); + times.add(timestamp); deviceIds.add(deviceId); typesList.add(types); valuesList.add(values); @@ -1305,20 +1556,22 @@ protected static void writeDataAlignedByTime( } } 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())) { + if (Boolean.TRUE.equals(hasStarted.get()) && !isCurrentImportFailed()) { ioTPrinter.println("Import completely!"); - } else { + } else if (Boolean.FALSE.equals(hasStarted.get())) { ioTPrinter.println("No records!"); } } 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 6500ef670834f..71f384060cc5b 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.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 @@ protected static void setTimeZone() throws IoTDBConnectionException, StatementEx @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 @@ protected static void processSuccessFile() { 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> failedRecords = new ArrayList<>(); + boolean fileFailed = false; + boolean failedCounted = false; String failedFilePath; if (failedFileDirectory == null) { failedFilePath = file.getAbsolutePath() + ".failed"; @@ -87,13 +108,20 @@ private static void importFromSqlFile(File file) { 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 @@ private static void importFromSqlFile(File file) { 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 @@ private static void importFromSqlFile(File file) { 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 headerNames = csvRecords.getHeaderNames(); Stream 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 @@ private static void importFromSingleFile(File file) { } 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 e2a08af86d825..5a451ccc4bc2a 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.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 static void main(String[] args) { 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 @@ protected static void setTimeZone() throws IoTDBConnectionException, StatementEx * @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 @@ private static void dumpFromSqlFile(String filePath) throws IOException { * @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 headers = new ArrayList<>(); List names = sessionDataSet.getColumnNames(); List 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 @@ private static void exportToSqlFileWithAlignDevice( SessionDataSet.DataIterator iterator = sessionDataSet.iterator(); List 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 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 @@ private static void exportToSqlFileWithAlignDevice( private static void exportToSqlFileWithoutAlign( SessionDataSet sessionDataSet, String filePath, List headers) throws IoTDBConnectionException, StatementExecutionException, IOException { - - List measurementNames = new ArrayList<>(); - String deviceName = null; if (CollectionUtils.isEmpty(headers) || headers.size() <= 1) { return; - } else { - List 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> deviceColumnIndices = new LinkedHashMap<>(); + Map 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 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 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> entry : deviceColumnIndices.entrySet()) { + List names = new ArrayList<>(); + List 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 @@ private static void exportToSqlFileWithoutAlign( 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 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 753713b9d1f68..31cde58b6bf8b 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 static void main(String[] args) throws IoTDBConnectionException { 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() { @@ -591,10 +586,20 @@ private static void asyncImportDataFiles() { ioTPrinter.println("ImportData thread join interrupted: " + e.getMessage()); } }); - ioTPrinter.println("Import completely!"); + if (!AsyncImportData.hasImportFailed() + && ImportDataScanTool.getTsFileQueueSize() == 0 + && !Thread.currentThread().isInterrupted()) { + ioTPrinter.println("Import completely!"); + } } private static int importFromTargetPathAsync() { + AsyncImportData.resetImportStatus(); + final File file = new File(targetPath); + if (!file.isFile() && !file.isDirectory()) { + ioTPrinter.println(String.format("Source file or directory %s does not exist", targetPath)); + return CODE_ERROR; + } try { sessionPool = new SessionPool.Builder() @@ -612,14 +617,13 @@ private static int importFromTargetPathAsync() { AsyncImportData.setSessionPool(sessionPool); AsyncImportData.setTimeZone(); ImportDataScanTool.setSourceFullPath(targetPath); - final File file = new File(targetPath); - if (!file.isFile() && !file.isDirectory()) { - ioTPrinter.println(String.format("Source file or directory %s does not exist", targetPath)); - System.exit(CODE_ERROR); - } 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(); @@ -627,6 +631,10 @@ private static int importFromTargetPathAsync() { } catch (Exception e) { ioTPrinter.println(String.format("Import tsfile fail: %s", e.getMessage())); return CODE_ERROR; + } finally { + if (sessionPool != null) { + sessionPool.close(); + } } } @@ -652,33 +660,7 @@ public static int importFromTargetPath( session.open(false); timeZoneID = timeZone; setTimeZone(); - - File file = new File(targetPath); - if (file.isFile()) { - if (file.getName().endsWith(SQL_SUFFIXS)) { - importFromSqlFile(session, file); - } else { - importFromSingleFile(session, file); - } - } else if (file.isDirectory()) { - File[] files = file.listFiles(); - if (files == null) { - return CODE_OK; - } - - for (File subFile : files) { - if (subFile.isFile()) { - if (subFile.getName().endsWith(SQL_SUFFIXS)) { - importFromSqlFile(session, subFile); - } else { - importFromSingleFile(session, subFile); - } - } - } - } else { - ioTPrinter.println("File not found!"); - return CODE_ERROR; - } + return importFromTargetPath(session, new File(targetPath)); } catch (IoTDBConnectionException | StatementExecutionException e) { ioTPrinter.println("Encounter an error when connecting to server, because " + e.getMessage()); return CODE_ERROR; @@ -687,7 +669,35 @@ public static int importFromTargetPath( session.close(); } } - return CODE_OK; + } + + static int importFromTargetPath(Session session, File file) { + resetCurrentImportFailed(); + if (file.isFile()) { + importFile(session, file); + } else if (file.isDirectory()) { + File[] files = file.listFiles(); + if (files == null) { + return CODE_ERROR; + } + for (File subFile : files) { + if (subFile.isFile()) { + importFile(session, subFile); + } + } + } else { + ioTPrinter.println("File not found!"); + return CODE_ERROR; + } + return isCurrentImportFailed() ? CODE_ERROR : CODE_OK; + } + + private static void importFile(Session session, File file) { + if (file.getName().endsWith(SQL_SUFFIXS)) { + importFromSqlFile(session, file); + } else { + importFromSingleFile(session, file); + } } /** @@ -697,15 +707,16 @@ public static int importFromTargetPath( */ private static void importFromSingleFile(Session session, 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 headerNames = csvRecords.getHeaderNames(); Stream records = csvRecords.stream(); if (headerNames.isEmpty()) { + markCurrentImportFailed(); ioTPrinter.println("Empty file!"); return; } if (!timeColumn.equalsIgnoreCase(filterBomHeader(headerNames.get(0)))) { + markCurrentImportFailed(); ioTPrinter.println("The first field of header must be `Time`!"); return; } @@ -721,9 +732,11 @@ private static void importFromSingleFile(Session session, File file) { writeDataAlignedByDevice(session, headerNames, records, failedFilePath); } } catch (IOException | IllegalPathException e) { + markCurrentImportFailed(); ioTPrinter.println("CSV file read exception because: " + e.getMessage()); } } else { + markCurrentImportFailed(); ioTPrinter.println("The file name must end with \"csv\" or \"txt\"!"); } } @@ -731,6 +744,7 @@ private static void importFromSingleFile(Session session, File file) { @SuppressWarnings("java:S2259") private static void importFromSqlFile(Session session, File file) { ArrayList> failedRecords = new ArrayList<>(); + boolean fileFailed = false; String failedFilePath = null; if (failedFileDirectory == null) { failedFilePath = file.getAbsolutePath() + ".failed"; @@ -743,13 +757,19 @@ private static void importFromSqlFile(Session session, File file) { try { session.executeNonQueryStatement(sql); } catch (IoTDBConnectionException | StatementExecutionException e) { + fileFailed = true; failedRecords.add(Arrays.asList(sql)); } } - ioTPrinter.println(file.getName() + " Import completely!"); } catch (IOException e) { + fileFailed = true; ioTPrinter.println("SQL file read exception because: " + e.getMessage()); } + if (fileFailed) { + markCurrentImportFailed(); + } else { + ioTPrinter.println(file.getName() + " Import completely!"); + } if (!failedRecords.isEmpty()) { FileWriter writer = null; try { @@ -758,6 +778,7 @@ private static void importFromSqlFile(Session session, File file) { writer.write(failedRecord.get(0).toString() + "\n"); } } catch (IOException e) { + markCurrentImportFailed(); ioTPrinter.println("Cannot dump fail result because: " + e.getMessage()); } finally { if (ObjectUtils.isNotEmpty(writer)) { 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 cdfbd5b4437ad..b1ec3c86b3da5 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 @@ -33,7 +33,10 @@ import java.io.File; import java.io.IOException; import java.net.UnknownHostException; +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.Collections; @@ -368,8 +371,7 @@ public static int importFromTargetPath() { 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 +411,8 @@ private static void processSetParams() { successDir, successOperation, failDir, failOperation); } - public static void asyncImportTsFiles() { + public static int asyncImportTsFiles() { + ImportTsFileBase.resetCounters(); final List list = new ArrayList<>(threadNum); for (int i = 0; i < threadNum; i++) { final Thread thread = @@ -429,6 +432,9 @@ public static void asyncImportTsFiles() { IOT_PRINTER.println("ImportTsFile thread join interrupted: " + e.getMessage()); } }); + return Thread.currentThread().isInterrupted() || ImportTsFileBase.hasFailures() + ? CODE_ERROR + : CODE_OK; } public enum Operation { @@ -439,6 +445,82 @@ public enum Operation { DELETE, ; + /** + * Process a TsFile and its optional sidecar files without replacing a destination member. + * + * @param source the source TsFile + * @param target the destination TsFile + * @throws FileAlreadyExistsException when a destination member already exists + * @throws IOException when a file operation fails + */ + public void processFileGroup(Path source, Path target) throws IOException { + List suffixes = existingSuffixes(source); + if (requiresDestinationCheck()) { + checkDestinations(target, suffixes); + 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); + Operation operation = this == HARDLINK && !suffix.isEmpty() ? CP : this; + operation.processFile(sourceFile, targetFile); + } + } + + private List existingSuffixes(Path source) { + List suffixes = new ArrayList<>(); + suffixes.add(""); + for (String suffix : new String[] {".resource", ".mods"}) { + if (Files.exists(Paths.get(source + suffix))) { + suffixes.add(suffix); + } + } + return suffixes; + } + + private boolean requiresDestinationCheck() { + return this == CP || this == HARDLINK || this == MV; + } + + private static void checkDestinations(Path target, List suffixes) + throws FileAlreadyExistsException { + for (String suffix : suffixes) { + Path destination = Paths.get(target + suffix); + if (Files.exists(destination, LinkOption.NOFOLLOW_LINKS)) { + throw new FileAlreadyExistsException(destination.toString()); + } + } + } + + 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; + } + } + public static boolean isValidOperation(String operation) { return "none".equalsIgnoreCase(operation) || "mv".equalsIgnoreCase(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 6d6f95dcb827a..ccdc81faa6e66 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 @@ -21,12 +21,9 @@ import org.apache.iotdb.cli.utils.IoTPrinter; -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; @@ -100,107 +97,26 @@ protected static void processSuccessFile(final String filePath) { } } - 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 ImportTsFile.Operation 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) { 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 fe520502ec3b2..5332319607a41 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 static int getSourceFullPathLength() { : 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 0000000000000..4506e4848837c --- /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 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 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/ImportDataBatchTest.java b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ImportDataBatchTest.java new file mode 100644 index 0000000000000..7e9576772d628 --- /dev/null +++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ImportDataBatchTest.java @@ -0,0 +1,247 @@ +/* + * 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.IoTDBConnectionException; +import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.session.Session; +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 org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.mockito.stubbing.Answer; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.RETURNS_DEFAULTS; +import static org.mockito.Mockito.mock; + +@RunWith(Parameterized.class) +public class ImportDataBatchTest { + + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private final boolean usePool; + private final boolean byDevice; + private final boolean aligned; + private final List> requests = new ArrayList<>(); + private Boolean originalAligned; + private int originalBatchSize; + private Path failed; + private String csv; + + @Parameterized.Parameters(name = "pool={0}, byDevice={1}, aligned={2}") + public static Collection parameters() { + return Arrays.asList( + new Object[][] { + {false, false, false}, + {false, false, true}, + {false, true, false}, + {false, true, true}, + {true, false, false}, + {true, false, true}, + {true, true, false}, + {true, true, true} + }); + } + + public ImportDataBatchTest(boolean usePool, boolean byDevice, boolean aligned) { + this.usePool = usePool; + this.byDevice = byDevice; + this.aligned = aligned; + } + + @Before + public void setUp() { + originalAligned = AbstractDataTool.aligned; + originalBatchSize = AbstractDataTool.batchPointSize; + AbstractDataTool.aligned = aligned; + AbstractDataTool.batchPointSize = 100_000; + AbstractDataTool.resetCurrentImportFailed(); + failed = temporaryFolder.getRoot().toPath().resolve("failed"); + csv = header() + row(1) + row(2) + row(3); + } + + @After + public void tearDown() { + AbstractDataTool.aligned = originalAligned; + AbstractDataTool.batchPointSize = originalBatchSize; + AbstractDataTool.resetCurrentImportFailed(); + } + + @Test + public void connectionFailureRetriesTheBatchOnly() throws Exception { + writeCsv( + times -> { + throw new IoTDBConnectionException("connection failed"); + }); + + assertEquals(4, requests.size()); + for (List request : requests) { + assertEquals(Arrays.asList(1L, 2L, 3L), request); + } + assertTrue(AbstractDataTool.isCurrentImportFailed()); + assertFailedCsv(csv); + } + + @Test + public void statementFailureRecoversGoodRows() throws Exception { + List imported = new ArrayList<>(); + writeCsv( + times -> { + if (times.size() > 1 || times.get(0) == 2L) { + throw new StatementExecutionException("bad row"); + } + imported.addAll(times); + }); + + assertEquals(4, requests.size()); + assertEquals(Arrays.asList(1L, 3L), imported); + assertTrue(AbstractDataTool.isCurrentImportFailed()); + assertFailedCsv(header() + row(2)); + } + + @Test + public void connectionFailureDuringRowRecoveryStopsRemainingRetries() throws Exception { + writeCsv( + times -> { + if (times.size() > 1) { + throw new StatementExecutionException("batch failed"); + } + throw new IoTDBConnectionException("connection lost during row recovery"); + }); + + assertEquals(5, requests.size()); + assertEquals(Arrays.asList(1L, 2L, 3L), requests.get(0)); + for (int index = 1; index < requests.size(); index++) { + assertEquals(Arrays.asList(1L), requests.get(index)); + } + assertTrue(AbstractDataTool.isCurrentImportFailed()); + assertFailedCsv(csv); + } + + @Test + public void transientConnectionFailureKeepsTheWholeBatch() throws Exception { + writeCsv( + times -> { + if (requests.size() == 1) { + throw new IoTDBConnectionException("temporary outage"); + } + }); + + assertEquals(2, requests.size()); + assertEquals(requests.get(0), requests.get(1)); + assertFalse(AbstractDataTool.isCurrentImportFailed()); + assertFalse(Files.exists(Path.of(failed + "_0"))); + } + + @Test + public void recoveryUsesTheCurrentBatchAfterFlushing() throws Exception { + AbstractDataTool.batchPointSize = 2; + writeCsv( + times -> { + if (times.size() > 1 || times.get(0) == 3L) { + throw new StatementExecutionException("bad batch or last row"); + } + }); + + assertEquals( + Arrays.asList( + Arrays.asList(1L, 2L), + Arrays.asList(1L), + Arrays.asList(2L), + Arrays.asList(3L), + Arrays.asList(3L)), + requests); + assertTrue(AbstractDataTool.isCurrentImportFailed()); + assertFailedCsv(header() + row(3)); + } + + private void writeCsv(InsertBehavior behavior) throws Exception { + String method = + (aligned ? "insertAlignedRecords" : "insertRecords") + (byDevice ? "OfOneDevice" : ""); + Answer answer = + invocation -> { + if (invocation.getMethod().getName().startsWith("insert")) { + assertEquals(method, invocation.getMethod().getName()); + List times = invocation.getArgument(1); + List> values = invocation.getArgument(4); + for (int index = 0; index < times.size(); index++) { + assertEquals(times.get(index).intValue(), values.get(index).get(0)); + } + requests.add(new ArrayList<>(times)); + behavior.insert(times); + return null; + } + return RETURNS_DEFAULTS.answer(invocation); + }; + Object client = usePool ? mock(SessionPool.class, answer) : mock(Session.class, answer); + Path input = temporaryFolder.newFile("input.csv").toPath(); + Files.writeString(input, csv); + try (CSVParser parser = AbstractDataTool.readCsvFile(input.toString())) { + if (usePool) { + if (byDevice) { + AbstractDataTool.writeDataAlignedByDevice( + (SessionPool) client, parser.getHeaderNames(), parser.stream(), failed.toString()); + } else { + AbstractDataTool.writeDataAlignedByTime( + (SessionPool) client, parser.getHeaderNames(), parser.stream(), failed.toString()); + } + } else if (byDevice) { + AbstractDataTool.writeDataAlignedByDevice( + (Session) client, parser.getHeaderNames(), parser.stream(), failed.toString()); + } else { + AbstractDataTool.writeDataAlignedByTime( + (Session) client, parser.getHeaderNames(), parser.stream(), failed.toString()); + } + } + } + + private String header() { + return byDevice ? "Time,Device,s1(INT32)\n" : "Time,root.d.s1(INT32)\n"; + } + + private String row(int timestamp) { + return timestamp + (byDevice ? ",root.d," : ",") + timestamp + "\n"; + } + + private void assertFailedCsv(String expected) throws Exception { + assertEquals(expected, Files.readString(Path.of(failed + "_0")).replace("\r\n", "\n")); + } + + @FunctionalInterface + private interface InsertBehavior { + void insert(List times) throws IoTDBConnectionException, StatementExecutionException; + } +} 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 0000000000000..ae2d790d4148c --- /dev/null +++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ImportDataFailureTest.java @@ -0,0 +1,198 @@ +/* + * 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.IoTDBConnectionException; +import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.session.Session; +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.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +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 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")); + } + + @Test + public void connectionFailureDoesNotRetryEachRow() throws Exception { + SessionPool sessionPool = mock(SessionPool.class); + doThrow(new IoTDBConnectionException("connection failed")) + .when(sessionPool) + .insertRecords(anyList(), anyList(), anyList(), anyList(), anyList()); + + Path input = temporaryFolder.newFile("input.csv").toPath(); + Files.writeString(input, "Time,root.d.s1(INT32)\n1,1\n2,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()); + } + + verify(sessionPool, times(4)) + .insertRecords(anyList(), anyList(), anyList(), anyList(), anyList()); + assertEquals( + "Time,root.d.s1(INT32)\n1,1\n2,2\n", + Files.readString(Path.of(failed + "_0")).replace("\r\n", "\n")); + } + + @Test + public void conversionFailureOnlyWritesFailedRow() throws Exception { + SessionPool sessionPool = mock(SessionPool.class); + Path input = temporaryFolder.newFile("input.csv").toPath(); + Files.writeString(input, "Time,root.d.s1(INT32)\n1,not-an-int\n"); + Path failed = temporaryFolder.getRoot().toPath().resolve("failed"); + AbstractDataTool.resetCurrentImportFailed(); + + try (CSVParser parser = AbstractDataTool.readCsvFile(input.toString())) { + AbstractDataTool.writeDataAlignedByTime( + sessionPool, parser.getHeaderNames(), parser.stream(), failed.toString()); + } + + assertFalse(AbstractDataTool.isCurrentImportFailed()); + assertEquals( + "Time,root.d.s1(INT32)\n1,not-an-int\n", + Files.readString(Path.of(failed + "_0")).replace("\r\n", "\n")); + } + + @Test + public void sessionBatchFailureSetsImportFailureStatus() throws Exception { + Session session = mock(Session.class); + doThrow(new StatementExecutionException("session insert failed")) + .when(session) + .insertRecords(anyList(), anyList(), anyList(), anyList(), anyList()); + + Path input = temporaryFolder.newFile("input.csv").toPath(); + Files.writeString(input, "Time,root.d.s1(INT32)\n1,1\n"); + Path failed = temporaryFolder.getRoot().toPath().resolve("failed"); + AbstractDataTool.resetCurrentImportFailed(); + + try (CSVParser parser = AbstractDataTool.readCsvFile(input.toString())) { + AbstractDataTool.writeDataAlignedByTime( + session, parser.getHeaderNames(), parser.stream(), failed.toString()); + } + + assertTrue(AbstractDataTool.isCurrentImportFailed()); + } + + @Test + public void sessionDeviceBatchFailureSetsImportFailureStatus() throws Exception { + Session session = mock(Session.class); + doThrow(new StatementExecutionException("session device insert failed")) + .when(session) + .insertRecordsOfOneDevice(anyString(), anyList(), anyList(), anyList(), anyList()); + + Path input = temporaryFolder.newFile("input.csv").toPath(); + Files.writeString(input, "Time,Device,s1(INT32)\n1,root.d,1\n"); + Path failed = temporaryFolder.getRoot().toPath().resolve("failed"); + AbstractDataTool.resetCurrentImportFailed(); + + try (CSVParser parser = AbstractDataTool.readCsvFile(input.toString())) { + AbstractDataTool.writeDataAlignedByDevice( + session, parser.getHeaderNames(), parser.stream(), failed.toString()); + } + + assertTrue(AbstractDataTool.isCurrentImportFailed()); + } +} diff --git a/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ImportDataStatusTest.java b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ImportDataStatusTest.java new file mode 100644 index 0000000000000..44d51ea7aa6e1 --- /dev/null +++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/data/ImportDataStatusTest.java @@ -0,0 +1,268 @@ +/* + * 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.utils.IoTPrinter; +import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.session.Session; +import org.apache.iotdb.session.pool.SessionPool; + +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.ByteArrayOutputStream; +import java.io.PrintStream; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; +import java.nio.charset.StandardCharsets; +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.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +public class ImportDataStatusTest { + + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private final Map originalFields = new HashMap<>(); + private final Map originalPrinters = new HashMap<>(); + private final ByteArrayOutputStream output = new ByteArrayOutputStream(); + private Field printerStream; + private SessionPool sessionPool; + + @Before + public void setUp() throws Exception { + printerStream = IoTPrinter.class.getDeclaredField("screenPrinter"); + printerStream.setAccessible(true); + for (Class owner : + List.of( + AbstractDataTool.class, + ImportData.class, + AsyncImportData.class, + ImportDataScanTool.class)) { + for (Field field : owner.getDeclaredFields()) { + if (Modifier.isStatic(field.getModifiers()) && !Modifier.isFinal(field.getModifiers())) { + field.setAccessible(true); + originalFields.put(field, field.get(null)); + } + } + if (owner != ImportDataScanTool.class) { + Field printer = owner.getDeclaredField("ioTPrinter"); + printer.setAccessible(true); + Object instance = printer.get(null); + originalPrinters.put(instance, printerStream.get(instance)); + printerStream.set(instance, new PrintStream(output, true, StandardCharsets.UTF_8)); + } + } + drainQueue(); + AbstractDataTool.resetCurrentImportFailed(); + AsyncImportData.resetImportStatus(); + AbstractDataTool.loadFileFailedNum.reset(); + AbstractDataTool.loadFileSuccessfulNum.reset(); + AbstractDataTool.failedFileDirectory = null; + AbstractDataTool.aligned = false; + AbstractDataTool.host = "127.0.0.1"; + AbstractDataTool.port = "6667"; + AbstractDataTool.username = "root"; + AbstractDataTool.password = "root"; + AbstractDataTool.timeZoneID = null; + setImportOption("threadNum", 1); + setImportOption("aligned", false); + sessionPool = mock(SessionPool.class); + AsyncImportData.setSessionPool(sessionPool); + } + + @After + public void tearDown() throws Exception { + drainQueue(); + AbstractDataTool.resetCurrentImportFailed(); + AsyncImportData.resetImportStatus(); + AbstractDataTool.loadFileFailedNum.reset(); + AbstractDataTool.loadFileSuccessfulNum.reset(); + for (Map.Entry entry : originalFields.entrySet()) { + entry.getKey().set(null, entry.getValue()); + } + for (Map.Entry entry : originalPrinters.entrySet()) { + printerStream.set(entry.getKey(), entry.getValue()); + } + } + + @Test + public void workerContinuesAfterFileFailureWithoutPrintingSuccess() throws Exception { + Path invalid = file("invalid.csv", "Wrong,root.d.s1(INT32)\n1,1\n"); + Path valid = file("valid.csv", "Time,root.d.s1(INT32)\n2,2\n"); + ImportDataScanTool.putToQueue(invalid.toString()); + ImportDataScanTool.putToQueue(valid.toString()); + + invokeImport("asyncImportDataFiles"); + + assertTrue(AsyncImportData.hasImportFailed()); + assertEquals(1, AbstractDataTool.loadFileFailedNum.sum()); + assertEquals(1, AbstractDataTool.loadFileSuccessfulNum.sum()); + assertEquals(0, ImportDataScanTool.getTsFileQueueSize()); + assertFalse(output.toString(StandardCharsets.UTF_8).contains("Import completely!")); + verify(sessionPool).insertRecords(anyList(), anyList(), anyList(), anyList(), anyList()); + } + + @Test + public void asyncMissingSourceReturnsErrorWithoutCreatingSessionPool() throws Exception { + Path missing = temporaryFolder.getRoot().toPath().resolve("missing"); + setImportOption("targetPath", missing.toString()); + setImportOption("sessionPool", null); + + assertEquals(AbstractDataTool.CODE_ERROR, invokeImport("importFromTargetPathAsync")); + assertEquals( + "Source file or directory " + missing + " does not exist" + System.lineSeparator(), + output.toString(StandardCharsets.UTF_8)); + Field pool = ImportData.class.getDeclaredField("sessionPool"); + pool.setAccessible(true); + assertNull(pool.get(null)); + assertEquals(0, ImportDataScanTool.getTsFileQueueSize()); + } + + @Test + public void asyncExitStatusResetsAndConversionFailureRemainsNonfatal() throws Exception { + Path invalid = file("invalid.csv", "Wrong,root.d.s1(INT32)\n1,1\n"); + setImportOption("targetPath", invalid.toString()); + assertEquals(AbstractDataTool.CODE_ERROR, invokeImport("importFromTargetPathAsync")); + assertTrue(AsyncImportData.hasImportFailed()); + assertFalse(output.toString(StandardCharsets.UTF_8).contains("Import completely!")); + + Path conversion = file("conversion.csv", "Time,root.d.s1(INT32)\n1,abc\n"); + setImportOption("targetPath", conversion.toString()); + output.reset(); + assertEquals(AbstractDataTool.CODE_OK, invokeImport("importFromTargetPathAsync")); + assertFalse(AsyncImportData.hasImportFailed()); + assertTrue(output.toString(StandardCharsets.UTF_8).contains("Import completely!")); + assertEquals( + "Time,root.d.s1(INT32)\n1,abc\n", + Files.readString(Path.of(conversion + ".failed_0")).replace("\r\n", "\n")); + } + + @Test + public void asyncUnprocessedFilesReturnErrorWithoutPrintingSuccess() throws Exception { + Path input = file("input.csv", "Time,root.d.s1(INT32)\n1,1\n"); + setImportOption("targetPath", input.toString()); + setImportOption("threadNum", 0); + assertEquals(AbstractDataTool.CODE_ERROR, invokeImport("importFromTargetPathAsync")); + assertEquals(1, ImportDataScanTool.getTsFileQueueSize()); + assertFalse(output.toString(StandardCharsets.UTF_8).contains("Import completely!")); + } + + @Test + public void syncCsvReturnsErrorAndResetsBeforeNextCommand() throws Exception { + Session session = mock(Session.class); + doAnswer( + invocation -> { + if (((List) invocation.getArgument(1)).contains(1L)) { + throw new StatementExecutionException("bad row"); + } + return null; + }) + .when(session) + .insertRecords(anyList(), anyList(), anyList(), anyList(), anyList()); + Path failed = file("failed.csv", "Time,root.d.s1(INT32)\n1,1\n"); + assertEquals( + AbstractDataTool.CODE_ERROR, ImportData.importFromTargetPath(session, failed.toFile())); + assertFalse(output.toString(StandardCharsets.UTF_8).contains("Import completely!")); + + output.reset(); + Path valid = file("valid.csv", "Time,root.d.s1(INT32)\n2,2\n"); + assertEquals( + AbstractDataTool.CODE_OK, ImportData.importFromTargetPath(session, valid.toFile())); + assertTrue(output.toString(StandardCharsets.UTF_8).contains("Import completely!")); + } + + @Test + public void syncSqlReturnsErrorAndKeepsProcessingStatements() throws Exception { + Session session = mock(Session.class); + doThrow(new StatementExecutionException("bad statement")) + .when(session) + .executeNonQueryStatement("bad"); + Path failed = file("failed.sql", "bad\naccepted\n"); + assertEquals( + AbstractDataTool.CODE_ERROR, ImportData.importFromTargetPath(session, failed.toFile())); + verify(session).executeNonQueryStatement("accepted"); + assertEquals("bad\n", Files.readString(Path.of(failed + ".failed")).replace("\r\n", "\n")); + assertFalse(output.toString(StandardCharsets.UTF_8).contains("Import completely!")); + + output.reset(); + Path valid = file("valid.sql", "valid\n"); + assertEquals( + AbstractDataTool.CODE_OK, ImportData.importFromTargetPath(session, valid.toFile())); + assertTrue(output.toString(StandardCharsets.UTF_8).contains("valid.sql Import completely!")); + } + + @Test + public void asyncSqlFailureSetsRunStatus() throws Exception { + doThrow(new StatementExecutionException("bad statement")) + .when(sessionPool) + .executeNonQueryStatement("bad"); + Path failed = file("failed.sql", "bad\naccepted\n"); + ImportDataScanTool.putToQueue(failed.toString()); + + invokeImport("asyncImportDataFiles"); + + assertTrue(AsyncImportData.hasImportFailed()); + assertEquals(1, AbstractDataTool.loadFileFailedNum.sum()); + assertEquals(0, AbstractDataTool.loadFileSuccessfulNum.sum()); + assertFalse(output.toString(StandardCharsets.UTF_8).contains("Import completely!")); + verify(sessionPool).executeNonQueryStatement("accepted"); + } + + private Path file(String name, String content) throws Exception { + Path path = temporaryFolder.newFile(name).toPath(); + Files.writeString(path, content); + return path; + } + + private static Object invokeImport(String name) throws Exception { + Method method = ImportData.class.getDeclaredMethod(name); + method.setAccessible(true); + return method.invoke(null); + } + + private static void setImportOption(String name, Object value) throws Exception { + Field field = ImportData.class.getDeclaredField(name); + field.setAccessible(true); + field.set(null, value); + } + + private static void drainQueue() { + while (ImportDataScanTool.pollFromQueue() != null) { + // Drain any files left by a failing assertion before restoring shared state. + } + } +} 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 0000000000000..feca35d513c79 --- /dev/null +++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/tsfile/ImportTsFileTest.java @@ -0,0 +1,208 @@ +/* + * 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.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 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(), + ImportTsFile.Operation.CP, + archive.toString(), + ImportTsFile.Operation.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 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 (ImportTsFile.Operation operation : + List.of( + ImportTsFile.Operation.CP, + ImportTsFile.Operation.HARDLINK, + ImportTsFile.Operation.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); + } +}