This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch support_wal_in_import_data in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit a25258ac70aff80e3af284f99208c29db1194390 Author: Tian Jiang <[email protected]> AuthorDate: Thu Aug 13 17:05:40 2026 +0800 add on_success --- .../apache/iotdb/db/i18n/ImportWALMessages.java | 10 ++++ .../apache/iotdb/db/i18n/ImportWALMessages.java | 10 ++++ .../java/org/apache/iotdb/db/tools/ImportWAL.java | 61 ++++++++++++++++++++- .../org/apache/iotdb/db/tools/ImportWALTest.java | 63 ++++++++++++++++++++++ 4 files changed, 143 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ImportWALMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ImportWALMessages.java index 2ed53656ec0..0dd487bfe8b 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ImportWALMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ImportWALMessages.java @@ -38,6 +38,9 @@ public final class ImportWALMessages { public static final String MESSAGE_PASSWORD_PROMPT_F2D0E794 = "Password: "; public static final String MESSAGE_TARGET_DATABASE_FOR_TABLE_MODEL_WAL_ENTRIES_27BACD1C = "Target database for table-model WAL entries."; + public static final String + MESSAGE_WHEN_ALL_WAL_FILES_ARE_REPLAYED_SUCCESSFULLY_DO_OPERATION_ON_SOURCE_WAL_FILES_OPTIONAL_PARAMETERS_ARE_NONE_DEFAULT_AND_DELETE_41963A66 = + "When all WAL files are replayed successfully, do operation on source WAL files. Optional parameters are none (default) and delete."; public static final String MESSAGE_PRINT_THIS_HELP_MESSAGE_E800AF7A = "Print this help message."; public static final String MESSAGE_ARGUMENT_ERROR_ARG_A9767F62 = "Argument error: %s"; @@ -58,8 +61,15 @@ public final class ImportWALMessages { public static final String MESSAGE_IMPORT_DURATION_ARG_SECONDS_TOTAL_SIZE_ARG_BYTES_AVERAGE_RATE_ARG_MB_PER_SECOND_4B4EA58D = "Import duration: %.1f s; total size: %d bytes; average rate: %.1f MB/s."; + public static final String MESSAGE_DELETED_ARG_SOURCE_WAL_FILES_C7A5AA1B = + "Deleted %d source WAL files."; public static final String EXCEPTION_FAILED_TO_REPLAY_WAL_FILE_ARG_AT_OFFSET_ARG_ARG_FCFAF7F9 = "Failed to replay WAL file %s at offset %d: %s"; + public static final String EXCEPTION_FAILED_TO_DELETE_SOURCE_WAL_FILE_ARG_ARG_236AF580 = + "Failed to delete source WAL file %s: %s"; + public static final String + EXCEPTION_UNSUPPORTED_ON_SUCCESS_VALUE_ARG_EXPECTED_NONE_OR_DELETE_F1C8EACE = + "Unsupported on_success value: %s. Expected none or delete."; public static final String EXCEPTION_TABLE_MODEL_WAL_ENTRIES_REQUIRE_DB_DATABASE_F7597726 = "Table-model WAL entries require -db/--database."; public static final String EXCEPTION_UNSUPPORTED_WAL_OPERATION_ARG_ABD227A0 = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ImportWALMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ImportWALMessages.java index c8dc2266dc8..57052048e63 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ImportWALMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ImportWALMessages.java @@ -38,6 +38,9 @@ public final class ImportWALMessages { public static final String MESSAGE_PASSWORD_PROMPT_F2D0E794 = "密码："; public static final String MESSAGE_TARGET_DATABASE_FOR_TABLE_MODEL_WAL_ENTRIES_27BACD1C = "表模型 WAL 条目的目标数据库。"; + public static final String + MESSAGE_WHEN_ALL_WAL_FILES_ARE_REPLAYED_SUCCESSFULLY_DO_OPERATION_ON_SOURCE_WAL_FILES_OPTIONAL_PARAMETERS_ARE_NONE_DEFAULT_AND_DELETE_41963A66 = + "所有 WAL 文件成功重放后，对源 WAL 文件执行操作。可选参数为 none（默认）和 delete。"; public static final String MESSAGE_PRINT_THIS_HELP_MESSAGE_E800AF7A = "打印帮助信息。"; public static final String MESSAGE_ARGUMENT_ERROR_ARG_A9767F62 = "参数错误：%s"; public static final String MESSAGE_WAL_IMPORT_FAILED_ARG_55C014BA = "WAL 导入失败：%s"; @@ -57,8 +60,15 @@ public final class ImportWALMessages { public static final String MESSAGE_IMPORT_DURATION_ARG_SECONDS_TOTAL_SIZE_ARG_BYTES_AVERAGE_RATE_ARG_MB_PER_SECOND_4B4EA58D = "导入耗时：%.1f 秒；文件总大小：%d 字节；平均速率：%.1f MB/s。"; + public static final String MESSAGE_DELETED_ARG_SOURCE_WAL_FILES_C7A5AA1B = + "已删除 %d 个源 WAL 文件。"; public static final String EXCEPTION_FAILED_TO_REPLAY_WAL_FILE_ARG_AT_OFFSET_ARG_ARG_FCFAF7F9 = "重放 WAL 文件 %s 时失败，偏移量 %d：%s"; + public static final String EXCEPTION_FAILED_TO_DELETE_SOURCE_WAL_FILE_ARG_ARG_236AF580 = + "删除源 WAL 文件 %s 失败：%s"; + public static final String + EXCEPTION_UNSUPPORTED_ON_SUCCESS_VALUE_ARG_EXPECTED_NONE_OR_DELETE_F1C8EACE = + "不支持的 on_success 值：%s。应为 none 或 delete。"; public static final String EXCEPTION_TABLE_MODEL_WAL_ENTRIES_REQUIRE_DB_DATABASE_F7597726 = "表模型 WAL 条目要求指定 -db/--database。"; public static final String EXCEPTION_UNSUPPORTED_WAL_OPERATION_ARG_ABD227A0 = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java index f8316499bdb..c1281c00ee3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java @@ -118,6 +118,8 @@ public class ImportWAL { final Path source = Paths.get(commandLine.getOptionValue("file")); final List<Path> walFiles = collectWALFiles(source); final String database = commandLine.getOptionValue("database"); + final boolean deleteSource = + shouldDeleteSource(commandLine.getOptionValue("on_success", "none")); final String password = getPassword(commandLine); final Session treeSession = createSession( @@ -141,7 +143,8 @@ public class ImportWAL { tableSession.open(false); } final ReplayStatistics statistics = - replayWALFiles(walFiles, new WALReplayer(treeSession, tableSession, database), out); + replayWALFiles( + walFiles, new WALReplayer(treeSession, tableSession, database), out, deleteSource); out.printf( ImportWALMessages .MESSAGE_REPLAYED_ARG_OPERATIONS_FROM_ARG_WAL_FILES_SKIPPED_ARG_ENTRIES_F0D37E3A, @@ -156,6 +159,11 @@ public class ImportWAL { statistics.getTotalBytes(), statistics.getAverageRateMbPerSecond()); out.println(); + if (deleteSource) { + out.printf( + ImportWALMessages.MESSAGE_DELETED_ARG_SOURCE_WAL_FILES_C7A5AA1B, walFiles.size()); + out.println(); + } } finally { closeSession(tableSession); closeSession(treeSession); @@ -211,6 +219,15 @@ public class ImportWAL { .hasArg() .desc(ImportWALMessages.MESSAGE_TARGET_DATABASE_FOR_TABLE_MODEL_WAL_ENTRIES_27BACD1C) .build()); + options.addOption( + Option.builder("os") + .longOpt("on_success") + .argName("on_success") + .hasArg() + .desc( + ImportWALMessages + .MESSAGE_WHEN_ALL_WAL_FILES_ARE_REPLAYED_SUCCESSFULLY_DO_OPERATION_ON_SOURCE_WAL_FILES_OPTIONAL_PARAMETERS_ARE_NONE_DEFAULT_AND_DELETE_41963A66) + .build()); options.addOption( Option.builder() .longOpt("help") @@ -242,6 +259,21 @@ public class ImportWAL { return new String(password); } + static boolean shouldDeleteSource(final String onSuccess) { + final String normalizedOnSuccess = onSuccess.trim(); + if ("none".equalsIgnoreCase(normalizedOnSuccess)) { + return false; + } + if ("delete".equalsIgnoreCase(normalizedOnSuccess)) { + return true; + } + throw new IllegalArgumentException( + String.format( + ImportWALMessages + .EXCEPTION_UNSUPPORTED_ON_SUCCESS_VALUE_ARG_EXPECTED_NONE_OR_DELETE_F1C8EACE, + onSuccess)); + } + private static boolean containsHelpOption(final String[] args) { if (args == null) { return false; @@ -368,6 +400,15 @@ public class ImportWAL { static ReplayStatistics replayWALFiles( final List<Path> walFiles, final WALReplayer replayer, final PrintStream progressStream) throws IOException { + return replayWALFiles(walFiles, replayer, progressStream, false); + } + + static ReplayStatistics replayWALFiles( + final List<Path> walFiles, + final WALReplayer replayer, + final PrintStream progressStream, + final boolean deleteSource) + throws IOException { final ReplayStatistics statistics = new ReplayStatistics(); final long startNanos = System.nanoTime(); for (final Path walFile : walFiles) { @@ -436,9 +477,27 @@ public class ImportWAL { e); } } + if (deleteSource) { + deleteSourceWALFiles(walFiles); + } return statistics; } + private static void deleteSourceWALFiles(final List<Path> walFiles) throws IOException { + for (final Path walFile : walFiles) { + try { + Files.delete(walFile); + } catch (final IOException e) { + throw new IOException( + String.format( + ImportWALMessages.EXCEPTION_FAILED_TO_DELETE_SOURCE_WAL_FILE_ARG_ARG_236AF580, + walFile, + e.getMessage()), + e); + } + } + } + static class WALReplayer { private final Session treeSession; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java index 787e06e1975..c8efb24ef9f 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java @@ -104,6 +104,26 @@ public class ImportWALTest { files); } + /** Covers CLI discovery of the source-file operation without opening a Session. */ + @Test + public void testHelpDescribesDeleteSourceOption() { + final ByteArrayOutputStream output = new ByteArrayOutputStream(); + + final int exitCode = + ImportWAL.run(new String[] {"--help"}, new PrintStream(output), new PrintStream(output)); + + assertEquals(0, exitCode); + assertTrue(output.toString().contains("--on_success")); + } + + /** Covers the default retention value, deletion value, normalization, and invalid input. */ + @Test + public void testParseOnSuccessOption() { + assertFalse(ImportWAL.shouldDeleteSource("none")); + assertTrue(ImportWAL.shouldDeleteSource(" DELETE ")); + assertThrows(IllegalArgumentException.class, () -> ImportWAL.shouldDeleteSource("unsupported")); + } + /** * Covers a real WAL file containing one tree insert and one internal signal. The insert must be * sent once as a Tablet, while the signal is counted as skipped and no corruption is reported. @@ -126,6 +146,49 @@ public class ImportWALTest { verify(treeSession).insertTablet(tabletCaptor.capture()); assertEquals("root.sg.d1", tabletCaptor.getValue().getDeviceId()); assertEquals(100, tabletCaptor.getValue().getTimestamp(0)); + assertTrue(walFile.exists()); + } + + /** Covers opt-in deletion after every source WAL file completes successfully. */ + @Test + public void testReplayDeletesSourceFilesAfterAllFilesSucceed() throws Exception { + final File firstWALFile = createWALFile(0); + final File secondWALFile = createWALFile(1); + writeWAL( + firstWALFile, new WALInfoEntry(1, WALTestUtils.getInsertRowNode("root.sg.delete.d1", 1))); + writeWAL( + secondWALFile, new WALInfoEntry(2, WALTestUtils.getInsertRowNode("root.sg.delete.d2", 2))); + + ImportWAL.replayWALFiles( + Arrays.asList(firstWALFile.toPath(), secondWALFile.toPath()), + new ImportWAL.WALReplayer(mock(Session.class), null, null), + null, + true); + + assertFalse(firstWALFile.exists()); + assertFalse(secondWALFile.exists()); + } + + /** Covers all-or-nothing replay gating: a later failure must retain every source WAL file. */ + @Test + public void testReplayRetainsAllSourceFilesWhenAnyFileFails() throws Exception { + final File validWALFile = createWALFile(0); + final File corruptedWALFile = createWALFile(1); + writeWAL( + validWALFile, new WALInfoEntry(1, WALTestUtils.getInsertRowNode("root.sg.retain.d1", 1))); + Files.write(corruptedWALFile.toPath(), new byte[] {WALEntryType.INSERT_ROW_NODE.getCode()}); + + assertThrows( + IOException.class, + () -> + ImportWAL.replayWALFiles( + Arrays.asList(validWALFile.toPath(), corruptedWALFile.toPath()), + new ImportWAL.WALReplayer(mock(Session.class), null, null), + null, + true)); + + assertTrue(validWALFile.exists()); + assertTrue(corruptedWALFile.exists()); } @Test
