add detailed prompt
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 3061a18..1be122a 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
@@ -82,8 +82,14 @@ "Table-model WAL entries require -db/--database."; public static final String EXCEPTION_UNSUPPORTED_WAL_OPERATION_ARG_ABD227A0 = "Unsupported WAL operation: %s"; - public static final String MESSAGE_UNSUPPORTED_WAL_OPERATION_ARG_SKIP_THIS_ENTRY_Y_N_DAFBE650 = - "Unsupported WAL operation: %s. Skip this entry? [y/N]: "; + public static final String + MESSAGE_TREE_MODEL_DELETE_OPERATION_DETECTED_ARG_CHOOSE_E_EXECUTE_S_SKIP_A_EXECUTE_ALL_L_SKIP_ALL_Q_QUIT_11E39FD7 = + "Tree-model delete operation detected: %s. Choose e=execute, s=skip, a=execute all, l=skip all, q=quit: "; + public static final String + MESSAGE_UNSUPPORTED_WAL_OPERATION_ARG_CHOOSE_S_SKIP_L_SKIP_ALL_Q_QUIT_0A734E52 = + "Unsupported WAL operation: %s. Choose s=skip, l=skip all, q=quit: "; + public static final String EXCEPTION_WAL_REPLAY_WAS_TERMINATED_BY_THE_USER_E0BD6197 = + "WAL replay was terminated by the user."; public static final String EXCEPTION_INSERT_NODE_ARG_CONTAINS_NO_REPLAYABLE_DATA_5DA13453 = "Insert node %s contains no replayable data."; public static final String EXCEPTION_UNSUPPORTED_SNAPSHOT_DATA_TYPE_ARG_7A32D312 =
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 120c326..912ab66 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
@@ -81,8 +81,14 @@ "表模型 WAL 条目要求指定 -db/--database。"; public static final String EXCEPTION_UNSUPPORTED_WAL_OPERATION_ARG_ABD227A0 = "不支持的 WAL 操作：%s"; - public static final String MESSAGE_UNSUPPORTED_WAL_OPERATION_ARG_SKIP_THIS_ENTRY_Y_N_DAFBE650 = - "不支持的 WAL 操作：%s。是否跳过此条目？[y/N]："; + public static final String + MESSAGE_TREE_MODEL_DELETE_OPERATION_DETECTED_ARG_CHOOSE_E_EXECUTE_S_SKIP_A_EXECUTE_ALL_L_SKIP_ALL_Q_QUIT_11E39FD7 = + "检测到树模型删除操作：%s。请选择 e=执行、s=跳过、a=全部执行、l=全部跳过、q=终止重放："; + public static final String + MESSAGE_UNSUPPORTED_WAL_OPERATION_ARG_CHOOSE_S_SKIP_L_SKIP_ALL_Q_QUIT_0A734E52 = + "不支持的 WAL 操作：%s。请选择 s=跳过、l=全部跳过、q=终止重放："; + public static final String EXCEPTION_WAL_REPLAY_WAS_TERMINATED_BY_THE_USER_E0BD6197 = + "用户终止了 WAL 重放。"; public static final String EXCEPTION_INSERT_NODE_ARG_CONTAINS_NO_REPLAYABLE_DATA_5DA13453 = "Insert node %s 不包含可重放数据。"; public static final String EXCEPTION_UNSUPPORTED_SNAPSHOT_DATA_TYPE_ARG_7A32D312 =
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 2214b69..dd1aca2 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
@@ -98,7 +98,6 @@ private static final String DEFAULT_USER = "root"; private static final int DEFAULT_THREAD_NUM = 1; private static final int SNAPSHOT_TABLET_ROW_LIMIT = 1024; - private static final Object CONSOLE_PROMPT_LOCK = new Object(); private ImportWAL() {} @@ -136,11 +135,15 @@ final String host = commandLine.getOptionValue("host", DEFAULT_HOST); final int port = parsePort(commandLine.getOptionValue("port", String.valueOf(DEFAULT_PORT))); final String username = commandLine.getOptionValue("username", DEFAULT_USER); + final WALReplayer.ReplayDecisionController replayDecisionController = + new WALReplayer.ReplayDecisionController(System.console()); final ReplayStatistics statistics = replayWALDirectories( walFiles, threadNum, - () -> createWALReplayWorker(host, port, username, password, database), + () -> + createWALReplayWorker( + host, port, username, password, database, replayDecisionController), out, deleteSource); out.printf( @@ -353,7 +356,8 @@ final int port, final String username, final String password, - final String database) + final String database, + final WALReplayer.ReplayDecisionController replayDecisionController) throws IOException { final Session treeSession = createSession(host, port, username, password, null); final Session tableSession = @@ -363,7 +367,7 @@ if (tableSession != null) { tableSession.open(false); } - return new SessionWALReplayer(treeSession, tableSession, database); + return new SessionWALReplayer(treeSession, tableSession, database, replayDecisionController); } catch (final IoTDBConnectionException e) { closeSession(tableSession); closeSession(treeSession); @@ -702,8 +706,11 @@ private final Session tableSession; private SessionWALReplayer( - final Session treeSession, final Session tableSession, final String tableDatabaseName) { - super(treeSession, tableSession, tableDatabaseName); + final Session treeSession, + final Session tableSession, + final String tableDatabaseName, + final WALReplayer.ReplayDecisionController replayDecisionController) { + super(treeSession, tableSession, tableDatabaseName, replayDecisionController); this.treeSession = treeSession; this.tableSession = tableSession; } @@ -720,7 +727,7 @@ private final Session treeSession; private final Session tableSession; private final ConsensusLogToTabletConverter converter; - private final UnsupportedEntryPrompt unsupportedEntryPrompt; + private final ReplayDecisionPrompt replayDecisionPrompt; private final Map<String, List<IMeasurementSchema>> tableTagSchemas = new HashMap<>(); WALReplayer( @@ -729,17 +736,17 @@ treeSession, tableSession, tableDatabaseName, - createUnsupportedEntryPrompt(System.console())); + new ReplayDecisionController(System.console())); } WALReplayer( final Session treeSession, final Session tableSession, final String tableDatabaseName, - final UnsupportedEntryPrompt unsupportedEntryPrompt) { + final ReplayDecisionPrompt replayDecisionPrompt) { this.treeSession = treeSession; this.tableSession = tableSession; - this.unsupportedEntryPrompt = unsupportedEntryPrompt; + this.replayDecisionPrompt = replayDecisionPrompt; converter = new ConsensusLogToTabletConverter( null, null, ColumnFilterMatcher.matchAll(), tableDatabaseName); @@ -757,45 +764,111 @@ return true; } if (entry.getValue() instanceof DeleteDataNode deleteDataNode) { + final ReplayDecision decision = replayDecisionPrompt.decide(entry, true); + if (decision == ReplayDecision.SKIP || decision == ReplayDecision.SKIP_ALL) { + return false; + } + if (decision == ReplayDecision.TERMINATE) { + throw replayTerminatedByUser(); + } replayTreeDelete(deleteDataNode); return true; } if (entry.getValue() instanceof RelationalDeleteDataNode || entry.getValue() instanceof ObjectNode) { - // A null prompt means no interactive console is available, so preserve fail-fast behavior. - if (unsupportedEntryPrompt != null && unsupportedEntryPrompt.shouldSkip(entry)) { + final ReplayDecision decision = replayDecisionPrompt.decide(entry, false); + if (decision == ReplayDecision.SKIP || decision == ReplayDecision.SKIP_ALL) { return false; } + if (decision == ReplayDecision.TERMINATE) { + throw replayTerminatedByUser(); + } throw unsupportedOperation(entry); } return false; } - private static UnsupportedEntryPrompt createUnsupportedEntryPrompt(final Console console) { - if (console == null) { - return null; - } - return entry -> { - synchronized (CONSOLE_PROMPT_LOCK) { - final String answer = - console.readLine( - ImportWALMessages - .MESSAGE_UNSUPPORTED_WAL_OPERATION_ARG_SKIP_THIS_ENTRY_Y_N_DAFBE650, - entry.getType()); - return isSkipConfirmation(answer); - } - }; - } - - static boolean isSkipConfirmation(final String answer) { - return answer != null - && ("y".equalsIgnoreCase(answer.trim()) || "yes".equalsIgnoreCase(answer.trim())); + enum ReplayDecision { + EXECUTE, + SKIP, + EXECUTE_ALL, + SKIP_ALL, + TERMINATE } @FunctionalInterface - interface UnsupportedEntryPrompt { + interface ReplayDecisionPrompt { - boolean shouldSkip(WALEntry entry); + ReplayDecision decide(WALEntry entry, boolean treeDelete); + } + + static class ReplayDecisionController implements ReplayDecisionPrompt { + + private final Console console; + private ReplayDecision treeDeleteDecision; + private boolean skipAllUnsupportedEntries; + + ReplayDecisionController(final Console console) { + this.console = console; + } + + // The controller is shared by parallel workers so an "all" choice applies to the whole + // import rather than only to the WAL files assigned to one worker. + @Override + public synchronized ReplayDecision decide(final WALEntry entry, final boolean treeDelete) { + if (treeDelete && treeDeleteDecision != null) { + return treeDeleteDecision == ReplayDecision.EXECUTE_ALL + ? ReplayDecision.EXECUTE + : ReplayDecision.SKIP; + } + if (!treeDelete && skipAllUnsupportedEntries) { + return ReplayDecision.SKIP; + } + if (console == null) { + return ReplayDecision.TERMINATE; + } + final String answer = + console.readLine( + treeDelete + ? ImportWALMessages + .MESSAGE_TREE_MODEL_DELETE_OPERATION_DETECTED_ARG_CHOOSE_E_EXECUTE_S_SKIP_A_EXECUTE_ALL_L_SKIP_ALL_Q_QUIT_11E39FD7 + : ImportWALMessages + .MESSAGE_UNSUPPORTED_WAL_OPERATION_ARG_CHOOSE_S_SKIP_L_SKIP_ALL_Q_QUIT_0A734E52, + entry.getType()); + final ReplayDecision decision = parseDecision(answer, treeDelete); + rememberAllDecision(decision, treeDelete); + return decision; + } + + private void rememberAllDecision(final ReplayDecision decision, final boolean treeDelete) { + if (treeDelete + && (decision == ReplayDecision.EXECUTE_ALL || decision == ReplayDecision.SKIP_ALL)) { + treeDeleteDecision = decision; + } else if (!treeDelete && decision == ReplayDecision.SKIP_ALL) { + skipAllUnsupportedEntries = true; + } + } + + static ReplayDecision parseDecision(final String answer, final boolean treeDelete) { + if (answer == null) { + return ReplayDecision.TERMINATE; + } + return switch (answer.trim().toLowerCase(Locale.ROOT)) { + case "e", "execute", "yes", "y" -> + treeDelete ? ReplayDecision.EXECUTE : ReplayDecision.TERMINATE; + case "s", "skip", "no", "n" -> ReplayDecision.SKIP; + case "a", "all", "execute_all" -> + treeDelete ? ReplayDecision.EXECUTE_ALL : ReplayDecision.TERMINATE; + case "l", "skip_all" -> ReplayDecision.SKIP_ALL; + case "q", "quit", "terminate", "t" -> ReplayDecision.TERMINATE; + default -> ReplayDecision.TERMINATE; + }; + } + } + + private static StatementExecutionException replayTerminatedByUser() { + return new StatementExecutionException( + ImportWALMessages.EXCEPTION_WAL_REPLAY_WAS_TERMINATED_BY_THE_USER_E0BD6197); } private static StatementExecutionException unsupportedOperation(final WALEntry entry) {
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 a40f020..6cef003 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
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; @@ -75,6 +76,7 @@ import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; @@ -412,23 +414,132 @@ 20); final Session treeSession = mock(Session.class); - new ImportWAL.WALReplayer(treeSession, null, null).replay(new WALInfoEntry(1, deleteNode)); + new ImportWAL.WALReplayer( + treeSession, + null, + null, + (entry, treeDelete) -> ImportWAL.WALReplayer.ReplayDecision.EXECUTE) + .replay(new WALInfoEntry(1, deleteNode)); verify(treeSession) .deleteData(eq(Arrays.asList("root.sg.d1.s1", "root.sg.d2.*")), eq(10L), eq(20L)); } + @Test + public void testReplayTreeDeleteSkipsAfterConfirmation() throws Exception { + final DeleteDataNode deleteNode = + new DeleteDataNode( + new PlanNodeId(""), List.of(new MeasurementPath("root.sg.d1.s1")), 10, 20); + final Session treeSession = mock(Session.class); + + final boolean replayed = + new ImportWAL.WALReplayer( + treeSession, + null, + null, + (entry, treeDelete) -> ImportWAL.WALReplayer.ReplayDecision.SKIP) + .replay(new WALInfoEntry(1, deleteNode)); + + assertFalse(replayed); + verify(treeSession, never()).deleteData(any(), anyLong(), anyLong()); + } + + @Test + public void testReplayTreeDeleteExecuteAllAndSkipAllDecisions() throws Exception { + final DeleteDataNode deleteNode = + new DeleteDataNode( + new PlanNodeId(""), List.of(new MeasurementPath("root.sg.d1.s1")), 10, 20); + final Session treeSession = mock(Session.class); + final AtomicInteger executeAllPromptCount = new AtomicInteger(); + final ImportWAL.WALReplayer.ReplayDecisionPrompt executeAllPrompt = + (entry, treeDelete) -> { + executeAllPromptCount.incrementAndGet(); + return ImportWAL.WALReplayer.ReplayDecision.EXECUTE_ALL; + }; + final ImportWAL.WALReplayer firstExecuteAllReplayer = + new ImportWAL.WALReplayer(treeSession, null, null, executeAllPrompt); + final ImportWAL.WALReplayer secondExecuteAllReplayer = + new ImportWAL.WALReplayer(treeSession, null, null, executeAllPrompt); + + assertTrue(firstExecuteAllReplayer.replay(new WALInfoEntry(1, deleteNode))); + assertTrue(secondExecuteAllReplayer.replay(new WALInfoEntry(2, deleteNode))); + assertEquals(2, executeAllPromptCount.get()); + verify(treeSession, times(2)).deleteData(any(), eq(10L), eq(20L)); + + final Session skippedTreeSession = mock(Session.class); + final AtomicInteger skipAllPromptCount = new AtomicInteger(); + final ImportWAL.WALReplayer.ReplayDecisionPrompt skipAllPrompt = + (entry, treeDelete) -> { + skipAllPromptCount.incrementAndGet(); + return ImportWAL.WALReplayer.ReplayDecision.SKIP_ALL; + }; + final ImportWAL.WALReplayer firstSkipAllReplayer = + new ImportWAL.WALReplayer(skippedTreeSession, null, null, skipAllPrompt); + final ImportWAL.WALReplayer secondSkipAllReplayer = + new ImportWAL.WALReplayer(skippedTreeSession, null, null, skipAllPrompt); + + assertFalse(firstSkipAllReplayer.replay(new WALInfoEntry(1, deleteNode))); + assertFalse(secondSkipAllReplayer.replay(new WALInfoEntry(2, deleteNode))); + assertEquals(2, skipAllPromptCount.get()); + verify(skippedTreeSession, never()).deleteData(any(), anyLong(), anyLong()); + } + + @Test + public void testReplayTreeDeleteTerminatesAfterConfirmation() throws Exception { + final DeleteDataNode deleteNode = + new DeleteDataNode( + new PlanNodeId(""), List.of(new MeasurementPath("root.sg.d1.s1")), 10, 20); + + assertThrows( + StatementExecutionException.class, + () -> + new ImportWAL.WALReplayer( + mock(Session.class), + null, + null, + (entry, treeDelete) -> ImportWAL.WALReplayer.ReplayDecision.TERMINATE) + .replay(new WALInfoEntry(1, deleteNode))); + } + /** Covers an unsupported entry when the interactive user explicitly chooses to skip it. */ @Test public void testReplayUnsupportedEntrySkipsAfterConfirmation() throws Exception { final WALEntry entry = mockUnsupportedEntry(); final boolean replayed = - new ImportWAL.WALReplayer(mock(Session.class), null, null, ignored -> true).replay(entry); + new ImportWAL.WALReplayer( + mock(Session.class), + null, + null, + (ignored, treeDelete) -> ImportWAL.WALReplayer.ReplayDecision.SKIP) + .replay(entry); assertFalse(replayed); } + @Test + public void testReplayUnsupportedEntriesSkipAllAfterConfirmation() throws Exception { + final AtomicInteger promptCount = new AtomicInteger(); + final ImportWAL.WALReplayer.ReplayDecisionPrompt skipAllPrompt = + (entry, treeDelete) -> { + promptCount.incrementAndGet(); + return ImportWAL.WALReplayer.ReplayDecision.SKIP_ALL; + }; + final ImportWAL.WALReplayer relationalDeleteReplayer = + new ImportWAL.WALReplayer(mock(Session.class), null, null, skipAllPrompt); + + assertFalse(relationalDeleteReplayer.replay(mockUnsupportedEntry())); + assertFalse(relationalDeleteReplayer.replay(mockUnsupportedEntry())); + + final WALEntry objectEntry = mock(WALEntry.class); + when(objectEntry.getType()).thenReturn(WALEntryType.OBJECT_FILE_NODE); + when(objectEntry.getValue()).thenReturn(mock(ObjectNode.class)); + final ImportWAL.WALReplayer objectNodeReplayer = + new ImportWAL.WALReplayer(mock(Session.class), null, null, skipAllPrompt); + assertFalse(objectNodeReplayer.replay(objectEntry)); + assertEquals(3, promptCount.get()); + } + /** Covers an unsupported entry when the interactive user declines the skip prompt. */ @Test public void testReplayUnsupportedEntryFailsAfterDecliningSkip() { @@ -437,7 +548,11 @@ assertThrows( StatementExecutionException.class, () -> - new ImportWAL.WALReplayer(mock(Session.class), null, null, ignored -> false) + new ImportWAL.WALReplayer( + mock(Session.class), + null, + null, + (ignored, treeDelete) -> ImportWAL.WALReplayer.ReplayDecision.TERMINATE) .replay(entry)); } @@ -448,17 +563,38 @@ assertThrows( StatementExecutionException.class, - () -> new ImportWAL.WALReplayer(mock(Session.class), null, null, null).replay(entry)); + () -> + new ImportWAL.WALReplayer( + mock(Session.class), + null, + null, + new ImportWAL.WALReplayer.ReplayDecisionController((java.io.Console) null)) + .replay(entry)); } - /** Covers accepted confirmations and the safe default for all other prompt answers. */ @Test - public void testUnsupportedEntrySkipConfirmationParsing() { - assertTrue(ImportWAL.WALReplayer.isSkipConfirmation("y")); - assertTrue(ImportWAL.WALReplayer.isSkipConfirmation(" YES ")); - assertFalse(ImportWAL.WALReplayer.isSkipConfirmation("n")); - assertFalse(ImportWAL.WALReplayer.isSkipConfirmation("")); - assertFalse(ImportWAL.WALReplayer.isSkipConfirmation(null)); + public void testReplayDecisionParsing() { + assertEquals( + ImportWAL.WALReplayer.ReplayDecision.EXECUTE, + ImportWAL.WALReplayer.ReplayDecisionController.parseDecision("e", true)); + assertEquals( + ImportWAL.WALReplayer.ReplayDecision.SKIP, + ImportWAL.WALReplayer.ReplayDecisionController.parseDecision("s", true)); + assertEquals( + ImportWAL.WALReplayer.ReplayDecision.EXECUTE_ALL, + ImportWAL.WALReplayer.ReplayDecisionController.parseDecision("a", true)); + assertEquals( + ImportWAL.WALReplayer.ReplayDecision.SKIP_ALL, + ImportWAL.WALReplayer.ReplayDecisionController.parseDecision("l", true)); + assertEquals( + ImportWAL.WALReplayer.ReplayDecision.TERMINATE, + ImportWAL.WALReplayer.ReplayDecisionController.parseDecision("a", false)); + assertEquals( + ImportWAL.WALReplayer.ReplayDecision.SKIP_ALL, + ImportWAL.WALReplayer.ReplayDecisionController.parseDecision("l", false)); + assertEquals( + ImportWAL.WALReplayer.ReplayDecision.TERMINATE, + ImportWAL.WALReplayer.ReplayDecisionController.parseDecision("q", true)); } /** Covers a non-aligned snapshot whose measurements have independent time axes. */